harry/RSS-Feed
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Eliminate Reddit/YouTube 429s with efficient request handling

Browse Source 
Reddit allows roughly one request per minute per IP on unauthenticated
.rss, but the app fired six separate subreddit requests, retried 429s
every couple of minutes, and only paced hosts within a single process.
Concurrent page/API/cron requests stampeded providers.

- Combine all due subreddits into one /r/a+b+c/.rss request and attribute
  items back to each subreddit by link; fall back to individual fetches
  when a banned/private subreddit poisons the combined feed.
- Single-flight refresh via data/refresh.lock so concurrent callers serve
  the cache instead of refetching.
- Enforce throttled-host spacing across processes (feeds.hostLastAttempt)
  in addition to the per-refresh cap; add www.youtube.com to throttled hosts.
- Honor 429 Retry-After / x-ratelimit-reset and back off exponentially
  (fail_count doubling, capped at error_retry_cap, plus jitter), reset on
  success. New feeds columns fail_count and next_retry_at.
- Jitter the healthy TTL so feeds don't all expire at once.
- Detect add-feed 404 by status instead of matching the error string.
- Update README (semantics, data model, config) and default user agent.
This commit is contained in:
Harwinder Sekhon
2026-10-07 20:34:16 -04:00
parent a6c27404e5
commit 397ee1a413
11 changed files with 513 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+44 -18
View File
@@ -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. |
config/app.php
+21 -6
View File
@@ -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
src/Backoff.php
+35
View File
@@ -0,0 +1,35 @@
<?php
declare(strict_types=1);
namespace App;
/**
* Retry policy for failed feeds. A server-provided delay (Retry-After /
* x-ratelimit-reset) wins; otherwise the delay grows exponentially with the
* number of consecutive failures, capped and jittered per URL so a batch of
* rate-limited feeds does not all retry at the same moment.
*/
final class Backoff
{
public function __construct(
private int $baseSeconds = 300,
private int $capSeconds = 7200,
) {
$this->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);
}
}
src/Database.php
+2
View File
@@ -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
src/FeedFetchException.php
+22
View File
@@ -0,0 +1,22 @@
<?php
declare(strict_types=1);
namespace App;
/**
* A feed fetch failed with a non-success HTTP status. Carries the status code
* so callers can distinguish removable errors (404) from transient ones.
*/
class FeedFetchException extends \RuntimeException
{
public function __construct(string $message, private int $status = 0)
{
parent::__construct($message);
}
public function getStatus(): int
{
return $this->status;
}
}
src/FeedFetcher.php
+39 -2
View File
@@ -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;
}
/**
src/FeedManager.php
+6 -4
View File
@@ -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())];
}
}
src/FeedRepository.php
+58 -21
View File
@@ -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<int, array<string, mixed>> $feeds
* @return array<int, array<string, mixed>>
*/
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<string, int> 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();
}
}
src/FeedService.php
+257 -28
View File
@@ -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<string, int> host => minimum seconds between requests */
private array $throttledHostIntervals;
/**
* @param array<int, array<string, mixed>> $feeds
* @param array<string, int>|array<int, string> $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<int, string>}
* @return array{attempted: int, succeeded: int, failed: int, new_items: int, locked: bool, errors: array<int, string>}
*/
public function refreshStale(): array
{
@@ -35,39 +46,51 @@ 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);
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.
continue;
return $report;
}
$hostCounts[$host] = $seen + 1;
}
$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();
$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;
}
$this->fetchOne($feed, $report);
}
// 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<string, mixed> $feed
* @param array<string, mixed> $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<int, array<string, mixed>> $feeds
* @param array<string, mixed> $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<string, mixed> $feed
* @param array<string, mixed> $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<string, mixed> $feed
* @param array<string, mixed> $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<string, int> $hostLast
* @param array<string, int> $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<int, string> $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<string, int>|array<int, string> $hosts
* @return array<string, int>
*/
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);
}
}
}
src/RateLimitException.php
+22
View File
@@ -0,0 +1,22 @@
<?php
declare(strict_types=1);
namespace App;
/**
* The provider asked us to slow down (HTTP 429). retryAfter is the delay in
* seconds suggested by the server (Retry-After / x-ratelimit-reset), if any.
*/
final class RateLimitException extends FeedFetchException
{
public function __construct(string $message, int $status = 429, private ?int $retryAfter = null)
{
parent::__construct($message, $status);
}
public function getRetryAfter(): ?int
{
return $this->retryAfter;
}
}
src/bootstrap.php
+9 -3
View File
@@ -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 [