diff --git a/README.md b/README.md index e163b27..68127d5 100644 --- a/README.md +++ b/README.md @@ -203,11 +203,16 @@ index.php api/items.php FeedService::refreshStale() ← runs on both entry points, respects TTL │ ├─ FeedRepository::syncFeeds() ensure every configured feed has a row - ├─ FeedRepository::staleFeeds() feeds older than TTL (or error cooldown) - └─ for each stale feed: + ├─ flock data/refresh.lock single-flight: concurrent callers skip + ├─ FeedRepository::staleFeeds() due feeds, minus retry backoff, jittered TTL + ├─ host pacing per-refresh cap + cross-process host interval + ├─ all stale Reddit feeds → ONE combined request (/r/a+b+c/.rss), + │ items attributed back to each subreddit by link + └─ for each other stale feed: FeedFetcher::fetch() Guzzle GET → laminas parse → normalize FeedRepository::upsertItems() INSERT OR IGNORE (dedup by sha256(link)) - FeedRepository::recordAttempt() success/error + timestamps + FeedRepository::recordAttempt() success → clears error/fail_count; + failure → fail_count++, next_retry_at FeedService::page(source, page, pageSize, since) ← read path └─ FeedRepository::getItems()/countItems() SQL over the SQLite cache @@ -230,14 +235,27 @@ button when `has_more` is false. - `items` and `feeds` live in `data/feeds.sqlite` (WAL mode, outside the web root). - Only **enabled** feeds are refreshed and served — disabling one from the manage page stops both without deleting its cached items. -- A feed is **stale** when `now - last_attempt_at >= cache_ttl`, or, if its last - attempt errored, when `now - last_attempt_at >= error_retry_ttl + jitter` - (jitter derived from `crc32(url)` so rate-limited feeds don't all retry at once). -- Fetch failures are recorded (`last_error`) and **never delete cached items** — - the dashboard keeps serving the last good data. -- Reddit rate-limits hard. `throttled_hosts` caps how many feeds of a given host - are fetched per refresh (`throttled_host_max_per_refresh`), and requests to the - same host are spaced by `host_interval_ms`. Only 5xx responses retry in-request. +- A feed is **due** when its retry backoff has elapsed (`next_retry_at`) and its + jittered TTL has passed: `now - last_attempt_at >= cache_ttl + crc32(url) % ttl_jitter` + (jitter spreads expiry so feeds don't all refresh at once). +- **One refresh at a time.** `refreshStale()` takes a non-blocking lock on + `data/refresh.lock`; concurrent page/API/cron requests serve the cache and + return immediately instead of stampeding providers. +- **Rate limits are respected.** Throttled hosts (`throttled_hosts`) are limited + to `throttled_host_max_per_refresh` fetches per refresh *and* to a per-host + minimum interval enforced across processes. The gap between requests to the + same host is also paced in-request (`host_interval_ms`). +- **Reddit is fetched as one request.** All due subreddits are combined into a + single `reddit.com/r/a+b+c/.rss` call and the items are attributed back to each + subreddit by their links. If that fails with 404/403 (a private or banned + subreddit poisons the combined feed), the app falls back to individual fetches + so the culprit is isolated. +- **429 handling.** A `429` carries the server's own window (`Retry-After`, or + Reddit's `x-ratelimit-reset`); the feed is parked until then. Other failures + back off exponentially (`error_retry_ttl` doubling per consecutive failure, up + to `error_retry_cap`, plus jitter). Success resets the counters. +- Fetch failures are recorded (`last_error`, `fail_count`, `next_retry_at`) and + **never delete cached items** — the dashboard keeps serving the last good data. --- @@ -255,6 +273,9 @@ button when `has_more` is false. │ ├── bootstrap.php Wires config + classes together, returns the container │ ├── Database.php PDO SQLite connection + schema migration │ ├── FeedFetcher.php HTTP + parse + normalize a single feed +│ ├── FeedFetchException.php Fetch failure carrying its HTTP status +│ ├── RateLimitException.php HTTP 429, carries the server's retry window +│ ├── Backoff.php Retry policy: server hint, else capped exponential + jitter │ ├── FeedRepository.php All SQL: upsert/query/count, staleness, feed management │ ├── FeedService.php Orchestration: refreshStale(), page(), sources() │ ├── FeedManager.php Manual selection: addReddit()/addYouTube()/remove()/setEnabled() @@ -279,8 +300,9 @@ button when `has_more` is false. |---|---|---| | `App\Database` | Creates the SQLite file/dir, applies schema, sets WAL + busy timeout | `connect()` | | `App\FeedFetcher` | Fetch one feed (with host throttle + 5xx retry), parse with laminas, extract media fields, normalize to a plain array | `fetch()` | -| `App\FeedRepository` | Every SQL statement; returns arrays/DTO-less rows | `syncFeeds()`, `enabledFeeds()`, `staleFeeds()`, `upsertItems()`, `recordAttempt()`, `getItems()`, `countItems()`, `addFeed()`, `setFeedEnabled()`, `deleteFeed()`, `feedsStatus()` | -| `App\FeedService` | Business flow: refresh policy + paginated reads | `refreshStale()`, `page()`, `sources()`, `feedsStatus()` | +| `App\Backoff` | Turns a failure (fail count + optional server hint) into the next retry time | `nextRetryAt()` | +| `App\FeedRepository` | Every SQL statement; returns arrays/DTO-less rows | `syncFeeds()`, `enabledFeeds()`, `staleFeeds()`, `hostLastAttempt()`, `upsertItems()`, `recordAttempt()`, `getItems()`, `countItems()`, `addFeed()`, `setFeedEnabled()`, `deleteFeed()`, `feedsStatus()` | +| `App\FeedService` | Business flow: single-flight refresh (throttled hosts, combined Reddit fetch) + paginated reads | `refreshStale()`, `page()`, `sources()`, `feedsStatus()` | | `App\FeedManager` | Manual selection changes; validates new feeds with one immediate fetch (404 rejects the entry) | `addReddit()`, `addYouTube()`, `remove()`, `setEnabled()` | | `App\ChannelResolver` | YouTube input (`@handle`, `UC…` ID or channel URL) → Atom feed URL; resolves handles once via the channel page (youtube.com only) | `resolve()` | | `App\ManageAuth` | Session gate for `manage.php`: password login, CSRF tokens, flash messages | `login()`, `authed()`, `csrfToken()`, `checkCsrf()` | @@ -324,6 +346,8 @@ Indexes: `idx_items_published(published_at DESC)`, | `last_error` | last failure message, `NULL` on success | | `origin` | `config` (from `config/feeds.php`) or `manual` (added via manage.php) | | `enabled` | `1` = selected; disabled feeds are neither refreshed nor shown | +| `fail_count` | consecutive failures; resets on success (drives exponential backoff) | +| `next_retry_at` | ISO-8601 UTC; the feed is skipped until this time (rate-limit / backoff) | Only items whose `feed_url` belongs to an enabled feed are served — disabling or removing a feed hides its cached stories without deleting them. @@ -338,10 +362,12 @@ removing a feed hides its cached stories without deleting them. |---|---|---| | `db_path` | `data/feeds.sqlite` | SQLite file location | | `cache_ttl` | `1800` | seconds before a healthy feed is re-fetched | -| `error_retry_ttl` | `120` | base retry delay for failed feeds (plus jitter) | -| `host_interval_ms` | `1000` | min gap between requests to the same host | -| `throttled_hosts` | `['www.reddit.com']` | hosts fetched sparingly | -| `throttled_host_max_per_refresh` | `1` | max feeds of a throttled host per refresh | +| `error_retry_ttl` | `300` | base backoff for failed feeds (doubles per consecutive failure, plus jitter) | +| `error_retry_cap` | `7200` | maximum retry backoff for a failing feed | +| `ttl_jitter` | `300` | random seconds added to `cache_ttl` so feeds don't all expire at once | +| `host_interval_ms` | `1000` | min gap between requests to the same host within one refresh | +| `throttled_hosts` | `['www.reddit.com' => 60, 'www.youtube.com' => 15]` | host => min seconds between requests (enforced across processes) | +| `throttled_host_max_per_refresh` | `1` | max feeds of a throttled host fetched per refresh (combined Reddit = 1) | | `page_size` | `20` | items per page | | `ranges` | `1h,2h,6h,24h,7d` | time-filter windows (key → seconds) | | `request_timeout` / `connect_timeout` | `15` / `10` | Guzzle timeouts | @@ -472,7 +498,7 @@ The source tab and filter pill are generated automatically from `sources()`. | Symptom | Cause / fix | |---|---| | `cURL error 7 ... Could not connect` | SELinux: enable `httpd_can_network_connect` (see above). | -| Reddit feeds empty / `HTTP 429` | Reddit's unauthenticated limit (~1 req/window). It self-heals via `error_retry_ttl` + jitter and the per-host cap; cached items still show. | +| Reddit feeds empty / `HTTP 429` | Reddit's unauthenticated limit is tight (~1 request/min/IP). The app now combines all subreddits into one request, honors `x-ratelimit-reset`, backs off, and serves cached items meanwhile. If it persists, set a real `user_agent` contact and/or lower the Reddit refresh rate (`throttled_hosts`). | | `403` on `/src/...`, `/config/...` | Expected — `.htaccess` denies those paths. | | `unable to open database file` | `data/` not writable by the web user, or wrong SELinux type on `data/`. | | Blank page / 500 | Check the PHP error log; run `php -l` on changed files. | diff --git a/config/app.php b/config/app.php index c04b201..a255cf0 100644 --- a/config/app.php +++ b/config/app.php @@ -8,13 +8,27 @@ return [ 'cache_ttl' => 1800, // Shorter retry interval for feeds whose last fetch failed (e.g. rate limited). - 'error_retry_ttl' => 120, + // Backoff grows from here (doubling per consecutive failure) and is capped + // by error_retry_cap; a server Retry-After hint always wins. + 'error_retry_ttl' => 300, - // Minimum milliseconds between requests to the same host. + // Maximum retry backoff for a failing feed (2 hours). + 'error_retry_cap' => 7200, + + // Random extra seconds added to the healthy TTL so feeds do not all expire + // (and get fetched) at the same moment. + 'ttl_jitter' => 300, + + // Minimum milliseconds between requests to the same host within one refresh. 'host_interval_ms' => 1000, - // Hosts with aggressive rate limits: only fetch this many of their feeds per refresh. - 'throttled_hosts' => ['www.reddit.com'], + // Hosts with aggressive rate limits: minimum seconds between requests to + // that host (enforced across processes via the cache DB), and at most + // throttled_host_max_per_refresh feeds of it per refresh. + 'throttled_hosts' => [ + 'www.reddit.com' => 60, + 'www.youtube.com' => 15, + ], 'throttled_host_max_per_refresh' => 1, // Number of items per page. @@ -33,8 +47,9 @@ return [ 'request_timeout' => 15, 'connect_timeout' => 10, - // Identify the client to feed providers (important for Reddit). - 'user_agent' => 'RSSFeed/1.0 (+https://localhost; contact: admin@localhost)', + // Identify the client to feed providers (important for Reddit). Reddit asks + // for a descriptive, unique agent — set a real contact before deploying. + 'user_agent' => 'RSSFeed/1.0 (self-hosted dashboard; contact: admin@example.com)', // Password for manage.php (manual subreddit/channel configuration). // Accepts a plain password or a password_hash() value; empty disables the diff --git a/src/Backoff.php b/src/Backoff.php new file mode 100644 index 0000000..e056177 --- /dev/null +++ b/src/Backoff.php @@ -0,0 +1,35 @@ +baseSeconds = max(1, $baseSeconds); + $this->capSeconds = max($this->baseSeconds, $capSeconds); + } + + public function nextRetryAt(int $failCount, ?int $retryAfterSeconds, string $url): string + { + if ($retryAfterSeconds !== null && $retryAfterSeconds > 0) { + // Honour the server's own window; no jitter so we retry when it says. + $delay = min($retryAfterSeconds, $this->capSeconds); + } else { + $delay = min($this->capSeconds, $this->baseSeconds * (2 ** min($failCount, 10))); + $delay += crc32($url) % $this->baseSeconds; + } + + return gmdate('c', time() + $delay); + } +} diff --git a/src/Database.php b/src/Database.php index 0c57e25..2c5f21a 100644 --- a/src/Database.php +++ b/src/Database.php @@ -67,6 +67,8 @@ final class Database // SQLite has no "ADD COLUMN IF NOT EXISTS": migrate pre-existing tables. self::addColumn($pdo, 'feeds', 'origin', "ALTER TABLE feeds ADD COLUMN origin TEXT NOT NULL DEFAULT 'config'"); self::addColumn($pdo, 'feeds', 'enabled', 'ALTER TABLE feeds ADD COLUMN enabled INTEGER NOT NULL DEFAULT 1'); + self::addColumn($pdo, 'feeds', 'fail_count', 'ALTER TABLE feeds ADD COLUMN fail_count INTEGER NOT NULL DEFAULT 0'); + self::addColumn($pdo, 'feeds', 'next_retry_at', 'ALTER TABLE feeds ADD COLUMN next_retry_at TEXT'); } private static function addColumn(PDO $pdo, string $table, string $column, string $sql): void diff --git a/src/FeedFetchException.php b/src/FeedFetchException.php new file mode 100644 index 0000000..73863ae --- /dev/null +++ b/src/FeedFetchException.php @@ -0,0 +1,22 @@ +status; + } +} diff --git a/src/FeedFetcher.php b/src/FeedFetcher.php index 2581603..f3fddcf 100644 --- a/src/FeedFetcher.php +++ b/src/FeedFetcher.php @@ -9,6 +9,7 @@ use DOMElement; use DOMXPath; use GuzzleHttp\Client; use Laminas\Feed\Reader\Reader; +use Psr\Http\Message\ResponseInterface; final class FeedFetcher { @@ -124,7 +125,17 @@ final class FeedFetcher return (string) $response->getBody(); } - // Only retry transient server errors; client errors (incl. 429) are + // Rate limiting is handled by the caller: it backs off for the + // server-suggested window instead of retrying in-request. + if ($lastStatus === 429) { + throw new RateLimitException( + sprintf('HTTP 429 while fetching %s', $url), + 429, + self::retryAfterSeconds($response) + ); + } + + // Only retry transient server errors; other client errors are // handled by the caller with a staggered cooldown. if ($lastStatus < 500) { break; @@ -137,7 +148,33 @@ final class FeedFetcher } } - throw new \RuntimeException(sprintf('HTTP %d while fetching %s', $lastStatus, $url)); + throw new FeedFetchException(sprintf('HTTP %d while fetching %s', $lastStatus, $url), $lastStatus); + } + + /** + * Seconds to wait before retrying, from Retry-After (seconds or HTTP date) + * or Reddit's x-ratelimit-reset (seconds). Null when the server is silent. + */ + private static function retryAfterSeconds(ResponseInterface $response): ?int + { + $header = trim($response->getHeaderLine('Retry-After')); + if ($header !== '') { + if (ctype_digit($header)) { + return max(1, (int) $header); + } + + $timestamp = strtotime($header); + if ($timestamp !== false) { + return max(1, $timestamp - time()); + } + } + + $reset = trim($response->getHeaderLine('X-Ratelimit-Reset')); + if ($reset !== '' && is_numeric($reset)) { + return max(1, (int) ceil((float) $reset)); + } + + return null; } /** diff --git a/src/FeedManager.php b/src/FeedManager.php index ec672e4..602efd0 100644 --- a/src/FeedManager.php +++ b/src/FeedManager.php @@ -14,6 +14,7 @@ final class FeedManager private FeedRepository $repository, private FeedFetcher $fetcher, private ChannelResolver $resolver, + private Backoff $backoff, ) { } @@ -121,16 +122,17 @@ final class FeedManager return ['ok' => true, 'message' => sprintf('Added "%s" — %d stories cached.', $label, $newItems)]; } catch (\Throwable $e) { - $error = $e->getMessage(); - $this->repository->recordAttempt($url, false, $error); + $retryAfter = $e instanceof RateLimitException ? $e->getRetryAfter() : null; + $failCount = (int) ($this->repository->findFeed($url)['fail_count'] ?? 0); + $this->repository->recordAttempt($url, false, $e->getMessage(), $this->backoff->nextRetryAt($failCount, $retryAfter, $url)); - if (str_contains($error, 'HTTP 404')) { + if ($e instanceof FeedFetchException && $e->getStatus() === 404) { $this->repository->deleteFeed($url); return ['ok' => false, 'message' => sprintf('"%s" was not found (HTTP 404); nothing was added.', $displayInput)]; } - return ['ok' => true, 'message' => sprintf('Added "%s" — not fetched yet (%s); it will fill in on a later refresh.', $label, $error)]; + return ['ok' => true, 'message' => sprintf('Added "%s" — not fetched yet (%s); it will fill in on a later refresh.', $label, $e->getMessage())]; } } diff --git a/src/FeedRepository.php b/src/FeedRepository.php index f421138..8066709 100644 --- a/src/FeedRepository.php +++ b/src/FeedRepository.php @@ -48,7 +48,7 @@ final class FeedRepository public function enabledFeeds(): array { return $this->pdo->query( - 'SELECT url, COALESCE(label, url) AS label, source FROM feeds WHERE enabled = 1 ORDER BY rowid' + 'SELECT url, COALESCE(label, url) AS label, source, fail_count FROM feeds WHERE enabled = 1 ORDER BY rowid' )->fetchAll(); } @@ -87,7 +87,7 @@ final class FeedRepository */ public function findFeed(string $url): ?array { - $stmt = $this->pdo->prepare('SELECT url, source, label, origin, enabled FROM feeds WHERE url = :url'); + $stmt = $this->pdo->prepare('SELECT url, source, label, origin, enabled, fail_count FROM feeds WHERE url = :url'); $stmt->execute([':url' => $url]); $row = $stmt->fetch(); @@ -101,23 +101,26 @@ final class FeedRepository } /** - * Return the subset of feeds whose last attempt is missing or older than TTL. - * Feeds that previously failed use a shorter retry interval. + * Return the subset of feeds that are due for a fetch. A feed is skipped + * while it is inside its retry backoff window (next_retry_at), and healthy + * feeds are only due once their jittered TTL has elapsed so a batch of + * feeds does not all expire (and get fetched) at the same moment. * * @param array> $feeds * @return array> */ - public function staleFeeds(array $feeds, int $ttlSeconds, int $errorTtlSeconds): array + public function staleFeeds(array $feeds, int $ttlSeconds, int $ttlJitter): array { - $attempts = []; - foreach ($this->pdo->query('SELECT url, last_attempt_at, last_error FROM feeds') as $row) { - $attempts[(string) $row['url']] = $row; + $state = []; + foreach ($this->pdo->query('SELECT url, last_attempt_at, last_error, next_retry_at FROM feeds') as $row) { + $state[(string) $row['url']] = $row; } $now = time(); $stale = []; foreach ($feeds as $feed) { - $row = $attempts[(string) $feed['url']] ?? null; + $url = (string) $feed['url']; + $row = $state[$url] ?? null; $last = $row['last_attempt_at'] ?? null; if ($last === null) { @@ -125,15 +128,22 @@ final class FeedRepository continue; } - $age = $now - (int) strtotime((string) $last); - $isError = ($row['last_error'] ?? null) !== null; - $threshold = $ttlSeconds; - - if ($isError) { - // Stagger retries so rate-limited feeds do not all retry at once. - $threshold = $errorTtlSeconds + (crc32((string) $feed['url']) % max(1, $errorTtlSeconds)); + $nextRetry = $row['next_retry_at'] ?? null; + if (is_string($nextRetry) && $nextRetry !== '' && (int) strtotime($nextRetry) > $now) { + // Still inside the rate-limit / error backoff window. + continue; } + // Rows that failed before backoff tracking existed retry once now, + // then receive a proper next_retry_at from the failure path. + if (($row['last_error'] ?? null) !== null && ($nextRetry === null || $nextRetry === '')) { + $stale[] = $feed; + continue; + } + + $age = $now - (int) strtotime((string) $last); + $threshold = $ttlSeconds + (int) (crc32($url) % max(1, $ttlJitter)); + if ($age >= $threshold) { $stale[] = $feed; } @@ -142,6 +152,30 @@ final class FeedRepository return $stale; } + /** + * Most recent attempt timestamp per host, used to space requests to + * rate-limited hosts across separate PHP processes (not just within one). + * + * @return array host => unix timestamp + */ + public function hostLastAttempt(): array + { + $hosts = []; + foreach ($this->pdo->query('SELECT url, last_attempt_at FROM feeds WHERE last_attempt_at IS NOT NULL') as $row) { + $host = strtolower((string) parse_url((string) $row['url'], PHP_URL_HOST)); + if ($host === '') { + continue; + } + + $ts = (int) strtotime((string) $row['last_attempt_at']); + if (!isset($hosts[$host]) || $ts > $hosts[$host]) { + $hosts[$host] = $ts; + } + } + + return $hosts; + } + /** * Insert new items, ignoring ones already stored (deduplicated by link hash). * @@ -188,13 +222,14 @@ final class FeedRepository return $inserted; } - public function recordAttempt(string $url, bool $success, ?string $error = null): void + public function recordAttempt(string $url, bool $success, ?string $error = null, ?string $nextRetryAt = null): void { $now = gmdate('c'); if ($success) { $stmt = $this->pdo->prepare( - 'UPDATE feeds SET last_attempt_at = :now, last_success_at = :now2, last_error = NULL WHERE url = :url' + 'UPDATE feeds SET last_attempt_at = :now, last_success_at = :now2, last_error = NULL, + fail_count = 0, next_retry_at = NULL WHERE url = :url' ); $stmt->execute([':now' => $now, ':now2' => $now, ':url' => $url]); @@ -202,9 +237,10 @@ final class FeedRepository } $stmt = $this->pdo->prepare( - 'UPDATE feeds SET last_attempt_at = :now, last_error = :error WHERE url = :url' + 'UPDATE feeds SET last_attempt_at = :now, last_error = :error, + fail_count = fail_count + 1, next_retry_at = :retry WHERE url = :url' ); - $stmt->execute([':now' => $now, ':error' => $error, ':url' => $url]); + $stmt->execute([':now' => $now, ':error' => $error, ':retry' => $nextRetryAt, ':url' => $url]); } /** @@ -274,7 +310,8 @@ final class FeedRepository public function feedsStatus(): array { return $this->pdo->query( - 'SELECT url, label, source, origin, enabled, last_success_at, last_error FROM feeds ORDER BY source, rowid' + 'SELECT url, label, source, origin, enabled, last_success_at, last_error, fail_count, next_retry_at + FROM feeds ORDER BY source, rowid' )->fetchAll(); } } diff --git a/src/FeedService.php b/src/FeedService.php index ccf1070..384f396 100644 --- a/src/FeedService.php +++ b/src/FeedService.php @@ -6,25 +6,36 @@ namespace App; final class FeedService { + private const REDDIT_FEED = '~^https?://(?:www\.)?reddit\.com/r/([A-Za-z0-9_]+)/\.rss$~i'; + private const REDDIT_LINK = '~^https?://(?:www\.)?reddit\.com/r/([A-Za-z0-9_]+)/~i'; + + /** @var array host => minimum seconds between requests */ + private array $throttledHostIntervals; + /** * @param array> $feeds + * @param array|array $throttledHosts host => seconds, or a list of hosts */ public function __construct( private FeedRepository $repository, private FeedFetcher $fetcher, private array $feeds, private int $ttlSeconds, - private int $errorRetryTtlSeconds = 300, - private array $throttledHosts = [], + private Backoff $backoff, + array $throttledHosts = [], private int $throttledHostMax = 1, + private int $ttlJitter = 300, + private ?string $lockFile = null, ) { + $this->throttledHostIntervals = self::normalizeHostIntervals($throttledHosts); } /** - * Refresh every feed whose cached copy is stale. Failures are recorded and - * skipped so previously cached items remain available. + * Refresh every feed that is due, collapsing all Reddit feeds into a single + * request and spacing throttled hosts. Only one process refreshes at a time + * (file lock); concurrent callers serve the cache and return immediately. * - * @return array{attempted: int, succeeded: int, failed: int, new_items: int, errors: array} + * @return array{attempted: int, succeeded: int, failed: int, new_items: int, locked: bool, errors: array} */ public function refreshStale(): array { @@ -35,37 +46,49 @@ final class FeedService 'succeeded' => 0, 'failed' => 0, 'new_items' => 0, + 'locked' => false, 'errors' => [], ]; - $hostCounts = []; - $activeFeeds = $this->repository->enabledFeeds(); + $lock = $this->acquireLock(); + if ($lock === false) { + $report['locked'] = true; - foreach ($this->repository->staleFeeds($activeFeeds, $this->ttlSeconds, $this->errorRetryTtlSeconds) as $feed) { - $url = (string) $feed['url']; - $host = (string) parse_url($url, PHP_URL_HOST); + return $report; + } - if (in_array($host, $this->throttledHosts, true)) { - $seen = $hostCounts[$host] ?? 0; - if ($seen >= $this->throttledHostMax) { - // Leave it stale so a later request picks it up. + try { + $activeFeeds = $this->repository->enabledFeeds(); + $stale = $this->repository->staleFeeds($activeFeeds, $this->ttlSeconds, $this->ttlJitter); + + $reddit = []; + $others = []; + foreach ($stale as $feed) { + if (self::isRedditFeed((string) $feed['url'])) { + $reddit[] = $feed; + } else { + $others[] = $feed; + } + } + + $now = time(); + $hostLast = $this->repository->hostLastAttempt(); + $hostCounts = []; + + foreach ($others as $feed) { + if (!$this->hostAllows((string) $feed['url'], $hostLast, $hostCounts, $now)) { continue; } - $hostCounts[$host] = $seen + 1; + $this->fetchOne($feed, $report); } - $report['attempted']++; - - try { - $result = $this->fetcher->fetch($feed); - $report['new_items'] += $this->repository->upsertItems($result['items']); - $this->repository->recordAttempt($url, true); - $report['succeeded']++; - } catch (\Throwable $e) { - $this->repository->recordAttempt($url, false, $e->getMessage()); - $report['failed']++; - $report['errors'][] = ($feed['label'] ?? $url) . ': ' . $e->getMessage(); + // All Reddit feeds are fetched as one combined request, so they count + // as a single request against the host cap and interval. + if ($reddit !== [] && $this->hostAllows((string) $reddit[0]['url'], $hostLast, $hostCounts, $now)) { + $this->fetchRedditGroup($reddit, $report); } + } finally { + $this->releaseLock($lock); } return $report; @@ -114,4 +137,210 @@ final class FeedService return array_keys($sources); } + + /** + * @param array $feed + * @param array $report + */ + private function fetchOne(array $feed, array &$report): void + { + try { + $result = $this->fetcher->fetch($feed); + $this->recordSuccess($feed, $report, $this->repository->upsertItems($result['items'])); + } catch (\Throwable $e) { + $this->recordFailure($feed, $e, $report); + } + } + + /** + * Fetch every due Reddit feed with one /r/a+b+c/.rss request and attribute + * the returned items back to their individual subreddit feeds by link. + * + * @param array> $feeds + * @param array $report + */ + private function fetchRedditGroup(array $feeds, array &$report): void + { + $bySub = []; + foreach ($feeds as $feed) { + if (preg_match(self::REDDIT_FEED, (string) $feed['url'], $m)) { + $bySub[strtolower($m[1])] = $feed; + } + } + + $combined = ['source' => 'reddit', 'label' => 'Reddit', 'url' => self::combinedRedditUrl(array_keys($bySub))]; + + try { + $result = $this->fetcher->fetch($combined); + } catch (FeedFetchException $e) { + // A banned/private subreddit poisons the whole combined feed; fall + // back to individual fetches so the culprit is isolated per feed. + if ($e->getStatus() === 404 || $e->getStatus() === 403) { + foreach ($feeds as $feed) { + $this->fetchOne($feed, $report); + } + + return; + } + + foreach ($feeds as $feed) { + $this->recordFailure($feed, $e, $report); + } + + return; + } catch (\Throwable $e) { + foreach ($feeds as $feed) { + $this->recordFailure($feed, $e, $report); + } + + return; + } + + $grouped = []; + foreach ($result['items'] as $item) { + if (!preg_match(self::REDDIT_LINK, (string) $item['link'], $m)) { + continue; + } + + $member = $bySub[strtolower($m[1])] ?? null; + if ($member === null) { + continue; + } + + $url = (string) $member['url']; + $item['feed_url'] = $url; + $item['feed_label'] = (string) ($member['label'] ?? $url); + $grouped[$url][] = $item; + } + + foreach ($feeds as $feed) { + $url = (string) $feed['url']; + $this->recordSuccess($feed, $report, $this->repository->upsertItems($grouped[$url] ?? [])); + } + } + + /** + * @param array $feed + * @param array $report + */ + private function recordSuccess(array $feed, array &$report, int $newItems): void + { + $this->repository->recordAttempt((string) $feed['url'], true); + $report['attempted']++; + $report['succeeded']++; + $report['new_items'] += $newItems; + } + + /** + * @param array $feed + * @param array $report + */ + private function recordFailure(array $feed, \Throwable $e, array &$report): void + { + $url = (string) $feed['url']; + $retryAfter = $e instanceof RateLimitException ? $e->getRetryAfter() : null; + $nextRetry = $this->backoff->nextRetryAt((int) ($feed['fail_count'] ?? 0), $retryAfter, $url); + + $this->repository->recordAttempt($url, false, $e->getMessage(), $nextRetry); + $report['attempted']++; + $report['failed']++; + $report['errors'][] = ($feed['label'] ?? $url) . ': ' . $e->getMessage(); + } + + /** + * Decide whether a host may be contacted now, honouring both the per-refresh + * cap and the cross-process minimum interval for throttled hosts. + * + * @param array $hostLast + * @param array $hostCounts + */ + private function hostAllows(string $url, array $hostLast, array &$hostCounts, int $now): bool + { + $host = strtolower((string) parse_url($url, PHP_URL_HOST)); + $interval = $this->throttledHostIntervals[$host] ?? null; + + if ($interval === null) { + return true; + } + + $last = $hostLast[$host] ?? null; + if ($last !== null && ($now - $last) < $interval) { + return false; + } + + $seen = $hostCounts[$host] ?? 0; + if ($seen >= $this->throttledHostMax) { + return false; + } + + $hostCounts[$host] = $seen + 1; + + return true; + } + + /** + * @param array $subs lower-cased subreddit names + */ + private static function combinedRedditUrl(array $subs): string + { + return 'https://www.reddit.com/r/' . implode('+', $subs) . '/.rss'; + } + + private static function isRedditFeed(string $url): bool + { + return (bool) preg_match(self::REDDIT_FEED, $url); + } + + /** + * @param array|array $hosts + * @return array + */ + private static function normalizeHostIntervals(array $hosts): array + { + $normalized = []; + foreach ($hosts as $key => $value) { + if (is_int($key) && is_string($value)) { + // Legacy list form: ['www.reddit.com']. + $normalized[strtolower($value)] = 60; + } elseif (is_string($key)) { + $normalized[strtolower($key)] = max(0, (int) $value); + } + } + + return $normalized; + } + + /** + * @return resource|bool|null handle when acquired, false when held, null when disabled + */ + private function acquireLock() + { + if ($this->lockFile === null || $this->lockFile === '') { + return null; + } + + $handle = @fopen($this->lockFile, 'c'); + if ($handle === false) { + return null; + } + + if (!flock($handle, LOCK_EX | LOCK_NB)) { + fclose($handle); + + return false; + } + + return $handle; + } + + /** + * @param resource|bool|null $lock + */ + private function releaseLock($lock): void + { + if (is_resource($lock)) { + flock($lock, LOCK_UN); + fclose($lock); + } + } } diff --git a/src/RateLimitException.php b/src/RateLimitException.php new file mode 100644 index 0000000..788d987 --- /dev/null +++ b/src/RateLimitException.php @@ -0,0 +1,22 @@ +retryAfter; + } +} diff --git a/src/bootstrap.php b/src/bootstrap.php index c3668c5..e69e744 100644 --- a/src/bootstrap.php +++ b/src/bootstrap.php @@ -11,15 +11,21 @@ $pdo = App\Database::connect($config['db_path']); $repository = new App\FeedRepository($pdo); $fetcher = new App\FeedFetcher($config); $resolver = new App\ChannelResolver($config); -$manager = new App\FeedManager($repository, $fetcher, $resolver); +$backoff = new App\Backoff( + (int) ($config['error_retry_ttl'] ?? 300), + (int) ($config['error_retry_cap'] ?? 7200) +); +$manager = new App\FeedManager($repository, $fetcher, $resolver, $backoff); $service = new App\FeedService( $repository, $fetcher, $feeds, (int) $config['cache_ttl'], - (int) ($config['error_retry_ttl'] ?? 300), + $backoff, (array) ($config['throttled_hosts'] ?? []), - (int) ($config['throttled_host_max_per_refresh'] ?? 1) + (int) ($config['throttled_host_max_per_refresh'] ?? 1), + (int) ($config['ttl_jitter'] ?? 300), + dirname($config['db_path']) . '/refresh.lock' ); return [