Agregacja newsów w webie: pipeline, który przetrwa skok ruchu przy breaking news

Każdy agregator newsów ma w sobie coś, co wygląda jak zwykły cron: pobierz kanał RSS, wyciągnij pozycje, zapisz do bazy, pokaż na stronie. Działa tygodniami, aż pewnego dnia jedna depesza zbiera kilkadziesiąt tysięcy odsłon w kwadrans i okazuje się, że cała ścieżka odczytu idzie prosto do Postgresa, worker pobierający kanały dostał właśnie 429 od trzech źródeł, a retry bez backoffu dokłada kolejne setki nieudanych żądań.

W naszym wewnętrznym rankingu wpisy z kanału google_news dostają score od 0 do 100. Pozycja z najwyższym score i stabilnym kierunkiem to zwykle materiał, który w kilkanaście minut ściąga ruch rzędu kilkudziesięciu razy większego niż normalna doba. Przykład z tego tygodnia to depesza AP News — ale ten tekst nie jest o niej. Jest o rurociągu, który ją do nas dostarczył, i o tym, dlaczego przy takim wpisie nie pęka.

Poniżej rozkładam pipeline na cztery niezależne warstwy: model danych z idempotentnym kluczem, pobieranie z warunkowymi żądaniami i circuit breakerem, kolejkę z realnym backpressure oraz ścieżkę odczytu odseparowaną od ścieżki zapisu. Na końcu jest checklista, którą można przejść przed najbliższym breaking news.

Breaking news to dwa różne problemy naraz

Warto nazwać to wprost: skok ruchu przy gorącym newsie nie jest problemem jednego zasobu. To dwa niezależne zjawiska o różnych skalach czasu.

Pierwsze to odczyt. Ruch rośnie skokowo, w sekundach, i dotyczy małego zbioru kluczy — strony głównej, kategorii, jednego konkretnego wpisu. Klasyczny problem cache miss storm: tysiące żądań w tej samej sekundzie pytają o ten sam nieistniejący jeszcze w cache klucz.

Drugie to zapis. Źródła, które normalnie aktualizują się co kilka minut, przy breaking news aktualizują się co kilkanaście sekund. Polling, który był tani, nagle generuje kilkanaście razy więcej żądań na źródło, a każde z nich może wracać błędem, bo dostawca też ma skok.

Praktyczna konsekwencja: ścieżka odczytu nigdy nie powinna wykonywać zapytań do tej samej bazy, do której pisze ingestion. Czytelnik i worker nie mogą dzielić poola połączeń ani limitu CPU w sposób, w którym jeden zagłodzi drugi.

Idempotentny klucz wpisu

Zanim cokolwiek zoptymalizujesz, ustal, czym jest „ten sam wpis”. GUID z RSS bywa niestabilny: zmienia się przy korekcie redakcyjnej, przy zmianie linku kanonicznego, przy przekierowaniu na inną domenę. Bez własnego klucza po trzech pobraniach masz trzy rekordy tej samej depeszy i trzy wejścia w feedzie.

Klucz budujemy jako sha256(source_id + '|' + canonical_id), gdzie canonical_id to pierwszy dostępny z listy: guid, atom:id, znormalizowany link, a w ostateczności tytuł|data publikacji. Normalizacja linku musi obciąć fragment, ujednolicić wielkość liter w hoście, usunąć parametry śledzące i posortować query.

import { createHash } from 'node:crypto';

const TRACKING = /^(utm_|fbclid$|gclid$|mc_cid$|mc_eid$)/i;

function canonicalizeUrl(raw, base) {
  const u = new URL(raw, base);
  u.hash = '';
  u.hostname = u.hostname.toLowerCase();
  if (u.pathname !== '/') u.pathname = u.pathname.replace(/\/+$/, '');
  const params = [...u.searchParams.entries()]
    .filter(([k]) => !TRACKING.test(k))
    .sort(([a], [b]) => a.localeCompare(b));
  u.search = new URLSearchParams(params).toString();
  return u.toString();
}

function entryKey(sourceId, item) {
  const canonicalId =
    item.guid?.trim() ||
    item.id?.trim() ||
    canonicalizeUrl(item.link, item.baseUrl) ||
    `${item.title}|${item.pubDate}`;
  return createHash('sha256')
    .update(`${sourceId}|${canonicalId}`, 'utf8')
    .digest('hex');
}

Po stronie bazy wystarczy entry_key jako klucz główny i upsert, który nie tworzy nowej wersji, jeśli treść się nie zmieniła. Osobny content_hash pozwala odróżnić nowy wpis od edycji istniejącego — i to jego, a nie updated_at, warto używać do decyzji o publikacji.

create table entries (
  entry_key     text primary key,
  source_id     text not null,
  canonical_id  text not null,
  content_hash  text not null,
  title         text not null,
  url           text not null,
  published_at  timestamptz,
  first_seen_at timestamptz not null default now(),
  updated_at    timestamptz not null default now()
);

insert into entries (entry_key, source_id, canonical_id, content_hash, title, url, published_at)
values ($1, $2, $3, $4, $5, $6, $7)
on conflict (entry_key) do update
  set content_hash = excluded.content_hash,
      title        = excluded.title,
      url          = excluded.url,
      updated_at   = now()
  where entries.content_hash <> excluded.content_hash
returning entry_key, (xmax = 0) as inserted;

Flaga xmax = 0 to trik specyficzny dla Postgresa, ale bardzo wygodny: w jednym zapytaniu wiesz, czy wpis jest nowy, czy zaktualizowany, więc możesz od razu wypchnąć tylko nowe pozycje do kolejki powiadomień.

Pobieranie: warunkowe GET, twarde timeouty, breaker na źródło

Reguła pierwsza: źródło nie może blokować pipeline’u. Każde żądanie ma własny timeout, a każde źródło własny circuit breaker. Gdy jedno z dziesięciu źródeł zaczyna odpowiadać 503, pozostałe dziewięć ma działać dalej.

Reguła druga: nie pobieraj tego, co już masz. ETag i Last-Modified zapisane per źródło zamieniają pełny transfer w odpowiedź 304, która przy breaking news jest większością ruchu.

async function fetchSource(source) {
  const res = await fetch(source.url, {
    signal: AbortSignal.timeout(8_000),
    redirect: 'follow',
    headers: {
      'user-agent': 'opole-aggregator/1.0 (+https://webdev.opole.pl)',
      accept: 'application/rss+xml, application/atom+xml, application/xml;q=0.9',
      ...(source.etag ? { 'if-none-match': source.etag } : {}),
      ...(source.lastModified ? { 'if-modified-since': source.lastModified } : {}),
    },
  });

  if (res.status === 304) return { notModified: true };
  if (!res.ok) throw Object.assign(new Error(`HTTP ${res.status}`), {
    status: res.status,
    retryable: res.status === 429 || res.status >= 500,
  });

  return {
    body: await res.text(),
    etag: res.headers.get('etag'),
    lastModified: res.headers.get('last-modified'),
  };
}

Breaker trzymaj per source_id: licz kolejne błędy, po przekroczeniu progu otwieraj obwód na kilkadziesiąt sekund, a w stanie half-open przepuszczaj dokładnie jedno żądanie. Przy newsach interwał pollingu warto uzależnić od tego, czy źródło jest „gorące” — ale zmiana harmonogramu nie może omijać breakera, bo to dokładnie wtedy źródła padają najczęściej.

Kolejka, backpressure i DLQ

Najczęstszy błąd w agregatorach to nieograniczona kolejka. Worker pobiera 200 pozycji, wrzuca je do kolejki w pamięci, pobiera kolejne 200, a gdy downstream (baza, indeks, wysyłka push) zwalnia, proces rośnie w RAM aż do OOM. Kolejka musi mieć limit i musi umieć powiedzieć „nie”.

Backpressure realizuj przez ograniczenie współbieżności i oczekiwanie producenta, a nie przez ciche odrzucanie. Odrzucenie jest OK tylko wtedy, gdy jest mierzone.

import pLimit from 'p-limit';

const perSource = pLimit(4);
const perItem = pLimit(32);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

async function withRetry(fn, { attempts = 4, baseMs = 250 } = {}) {
  let lastErr;
  for (let i = 0; i < attempts; i++) {
    try {
      return await fn();
    } catch (err) {
      lastErr = err;
      if (!err.retryable || i === attempts - 1) break;
      await sleep(Math.random() * baseMs * 2 ** i); // full jitter
    }
  }
  throw lastErr;
}

async function runSource(source) {
  const payload = await withRetry(() => fetchSource(source));
  if (payload.notModified) return;

  for (const item of parseFeed(payload.body)) {
    await perItem(async () => {
      try {
        await ingestItem(source, item);
      } catch (err) {
        await deadLetter.push({ sourceId: source.id, item, err: String(err) });
      }
    });
  }
}

await Promise.all(sources.map((s) => perSource(() => runSource(s))));

Full jitter (Math.random() * base * 2^i) jest tu istotny bardziej niż wykładniczy backoff sam w sobie: przy breaking news większość instancji agregatora uderza w to samo źródło w tej samej sekundzie, a bez losowości retry tworzą idealnie zsynchronizowaną falę.

DLQ nie jest miejscem, gdzie rzeczy giną. To kolejka z TTL, metryką głębokości i alertem po przekroczeniu progu — wpis, który wypadł z trzech prób, prawie zawsze oznacza zmianę formatu kanału albo cichy rate limit, a nie pojedynczą awarię.

Ścieżka odczytu z dala od bazy

Przy breaking news serwis musi odpowiadać z cache’u, nie z bazy. Trzy elementy, które to załatwiają:

• Read model w KV. Writer po każdym batchu liczy gotowy payload (lista wpisów dla strony głównej, kategorii, top 20) i zapisuje go pod kluczem w Redisie lub KV na brzegu. Odczyt to jedno GET, nie JOIN po trzech tabelach.
• Cache-control z SWR. Cache-Control: public, s-maxage=15, stale-while-revalidate=60, stale-if-error=600. Piętnaście sekund świeżości to dla newsa wystarczająco dużo, a stale-if-error ratuje stronę, gdy origin właśnie się dławi.
• Single-flight. Przy miss nie puszczaj stu równoległych zapytań o ten sam klucz — jedna instancja liczy, reszta czeka na wynik.

async function readModel(key, ttlMs, build) {
  const hit = await kv.get(key);
  if (hit) return JSON.parse(hit);

  return inflight(`model:${key}`, async () => {
    const fresh = await build();
    await kv.set(key, JSON.stringify(fresh), { PX: ttlMs });
    return fresh;
  });
}

Osobna decyzja: czy feed ma być paginowany po published_at, czy po score. Przy skoku ruchu paginacja po score powoduje, że każda strona przelicza ranking. Lepiej zamrozić kolejność w read modelu na okres TTL i nie liczyć jej pod żądaniem.

Co mierzyć i jak to przetestować

Minimum metryk, bez których nie ma o czym rozmawiać:

• lag kolejki (p50, p99) — czas od pobrania wpisu do jego dostępności w read modelu,
• error rate per źródło i osobno rate 304 — spadek 304 przy rosnącym ruchu oznacza, że źródło przestało wysyłać walidatory,
• głębokość DLQ z alertem po progu,
• cache hit ratio na brzegu oraz p99 TTFB dla ścieżki odczytu,
• liczba otwartych breakerów w danym momencie.

Test obciążeniowy rób scenariuszem skoku, nie stałym ruchem. k6 z ramping-arrival-rate modeluje to lepiej niż constant-vus, bo interesuje cię zachowanie przy zmianie nachylenia, a nie przy plateau.

import http from 'k6/http';
import { check, sleep } from 'k6';

export const options = {
  scenarios: {
    breaking: {
      executor: 'ramping-arrival-rate',
      startRate: 5,
      timeUnit: '1s',
      preAllocatedVUs: 200,
      maxVUs: 2000,
      stages: [
        { target: 5,   duration: '30s' },
        { target: 400, duration: '20s' },
        { target: 400, duration: '2m' },
        { target: 5,   duration: '30s' },
      ],
    },
  },
  thresholds: {
    http_req_failed: ['rate<0.01'],
    http_req_duration: ['p(95)<800', 'p(99)<2000'],
  },
};

export default function () {
  const res = http.get(`${__ENV.BASE_URL}/api/news?limit=20`);
  check(res, { 'status 200': (r) => r.status === 200 });
  sleep(Math.random() * 2);
}

Osobno przetestuj sam ingestion: zablokuj jedno źródło (np. przez /etc/hosts) i sprawdź, czy pozostałe kończą cykl w rozsądnym czasie, a breaker otwiera się po zadanej liczbie błędów, zamiast walić w martwy endpoint do końca okna.

Checklista przed najbliższym breaking news

• Każdy wpis ma entry_key z source_id i znormalizowanego canonical_id; ponowne pobranie nie tworzy duplikatu.
• Upsert porównuje content_hash — edycja aktualizuje rekord, identyczna treść nie generuje zdarzenia.
• Każde źródło ma zapisane ETag i Last-Modified, a 304 nie jest traktowane jak błąd.
• Timeouty są ustawione jawnie (connect i total), a nie domyślne.
• Breaker jest per źródło, half-open przepuszcza jedno żądanie, stan jest widoczny w metrykach.
• Kolejka ma jawny limit; przy zapełnieniu producent czeka, a odrzucenia są liczone.
• Retry ma full jitter i limit prób; 4xx poza 429 nie jest retryowane.
• DLQ ma TTL, metrykę głębokości i alert.
• Ścieżka odczytu nie dotyka bazy zapisowej — czyta z KV albo z brzegu.
• Nagłówki cache zawierają s-maxage, stale-while-revalidate i stale-if-error.
• Ranking i paginacja są liczone w writerze, nie pod żądaniem użytkownika.
• Scenariusz k6 ze skokiem przechodzi progi przy zimnym cache.

Wnioski

Breaking news nie psuje agregatora dlatego, że ruch jest duży — psuje go dlatego, że duży ruch trafia w tę samą ścieżkę, która właśnie wykonuje najwięcej zapisów. Rozdzielenie odczytu od zapisu, idempotentny klucz wpisu i kolejka, która potrafi powiedzieć „stop”, rozwiązują większość problemów, zanim zdążą się pojawić. Reszta to obserwowalność: jeśli nie widzisz lagu kolejki i error rate per źródło, to przy następnym gorącym wpisie dowiesz się o awarii z Twittera, a nie z dashboardu.

Warto też pamiętać, że źródła newsowe nie są ci nic winne. Ich rate limity, zmiany formatu i ciche przekierowania to norma, nie wyjątek — dlatego cała warstwa pobierania powinna być zaprojektowana tak, jakby każde źródło miało się zepsuć w najgorszym możliwym momencie. Zwykle ma.