From 67df2fd90295623f569ae17102f970fce3185dbd Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Oct 2026 01:41:30 +0000 Subject: [PATCH] Workers P2: Feed-Stillstand bei dichten Ereignissen behoben (Sprung hinter die Spitze) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Der Dienst liefert bei events(after) die ältesten pageLimit Einträge ab after - 5 s. Mit mehr als 200 Einträgen im Fenster kam mit after = jüngster ts immer dieselbe, schon bekannte Seite und das Neuere nie an. Eine volle Seite ohne neues Ereignis setzt jetzt eine Abrufmarke hinter die Spitze (cursor + 5 s, bei erneutem Stillstand + 1 ms), die über Runden hinweg gilt. Co-Authored-By: Claude Sonnet 5.5 --- src/lib/workersFeed.js | 28 ++++++++++-- tests/unit/workersFeed.test.mjs | 77 +++++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 4 deletions(-) diff --git a/src/lib/workersFeed.js b/src/lib/workersFeed.js index 1a75164..f553e07 100644 --- a/src/lib/workersFeed.js +++ b/src/lib/workersFeed.js @@ -5,6 +5,8 @@ import { takeNewEvents, latestTs, needsStateRefresh } from './workersEvents.js' const EPOCH = '1970-01-01T00:00:00.000Z' +// Der Dienst liefert bei events(after) alle Einträge ab after - 5 s (P1 buildEvents), die ältesten pageLimit Stück +const OVERLAP_MS = 5000 const noop = () => {} const isRecord = (v) => v !== null && typeof v === 'object' && !Array.isArray(v) @@ -24,6 +26,7 @@ export function createFeedController({ let cursor = null // null bis der erste Zustand da ist; danach der jüngste gesehene Zeitstempel let historyMs = NaN // lastEventTs des ersten Zustands: Ereignisse bis dahin sind Vorgeschichte let seen = new Set() + let resume = null // Abrufmarke nach einer vollen Seite ohne Neues: der nächste Abruf fragt ab hier statt ab cursor let stateAgain = false // refreshState() kam während eines laufenden Zustand-Abrufs let reported = null // zuletzt gemeldete Fehlermeldung (null = kein Fehler) const timer = { state: null, events: null } @@ -113,20 +116,36 @@ export function createFeedController({ return !Number.isNaN(historyMs) && Date.parse(e.ts) <= historyMs } - // Holt Seiten, solange sie voll sind (höchstens maxPages); wirft nie, ein Fehler steht im Ergebnis + // Zeitstempel hinter der bisher gesehenen Spitze: die Marke after, bei der der Dienst (ab after - 5 s) wieder ab dem + // jüngsten gesehenen Eintrag liefert; strict: ab 1 ms danach + function skipTo(strict) { + return new Date(Date.parse(cursor) + OVERLAP_MS + (strict ? 1 : 0)).toISOString() + } + + // Holt Seiten, solange sie voll sind (höchstens maxPages); wirft nie, ein Fehler steht im Ergebnis. + // Stillstand-Schutz: Der Dienst liefert die ältesten pageLimit Einträge ab after - 5 s. Liegen mehr als pageLimit + // Einträge im Fenster vor cursor, bestünde eine volle Seite nur aus schon bekannten, und mit after = cursor käme + // immer dieselbe Seite (das Neuere käme nie an). Eine volle Seite ohne ein einziges neues Ereignis + // setzt darum `resume` hinter die Spitze: after = cursor + 5 s fragt ab cursor (Doppelte fängt takeNewEvents ab). + // Bleibt auch diese Seite voll und ohne Neues (pageLimit Einträge mit demselben ts), geht es ab cursor + 1 ms weiter. + // `resume` gilt über die Runde hinaus, bis eine Seite Neues bringt oder nicht voll ist. async function fetchRound(myGen) { const fresh = [] let failed = null try { for (let page = 0; page < maxPages; page += 1) { - const res = await api.events(cursor, pageLimit) + const jumped = resume !== null + const res = await api.events(jumped ? resume : cursor, pageLimit) if (myGen !== gen) break const list = isRecord(res) && Array.isArray(res.events) ? res.events : [] - for (const e of takeNewEvents(seen, list)) { + const unseen = takeNewEvents(seen, list) + for (const e of unseen) { if (!isHistory(e)) fresh.push(e) } cursor = latestTs(list, cursor) - if (list.length < pageLimit) break + const full = list.length >= pageLimit + resume = full && unseen.length === 0 ? skipTo(jumped) : null + if (!full) break } } catch (err) { failed = err @@ -176,6 +195,7 @@ export function createFeedController({ cursor = null historyMs = NaN seen = new Set() + resume = null stateAgain = false reported = null failure.state = null diff --git a/tests/unit/workersFeed.test.mjs b/tests/unit/workersFeed.test.mjs index 1276310..4351988 100644 --- a/tests/unit/workersFeed.test.mjs +++ b/tests/unit/workersFeed.test.mjs @@ -319,3 +319,80 @@ test('späte Antwort nach stop wird verworfen, ein neuer Start beginnt frisch', assert.deepEqual(ids(h.log.batches), [['e1']]) assert.equal(h.timers.size, 2) }) + +// ---- Dichte Ereignisse: mehr als pageLimit Einträge im 5-s-Fenster des Dienstes (Fix-Runde 1) ---- +// Der Dienst liefert bei events(after) die ÄLTESTEN pageLimit Einträge ab after - 5 s. Mit after = jüngster ts käme bei +// mehr als 200 Einträgen im Fenster immer dieselbe, schon bekannte Seite: der Feed stünde still, ohne Fehler. + +// n Ereignisse im Abstand von step ms, Standard ab 20:00:01 (nach dem Standard-Zustand lastEventTs 20:00:00) +const dense = (n, { from = Date.UTC(2026, 9, 17, 20, 0, 1), step = 10, prefix = 'd' } = {}) => Array.from({ length: n }, (_, i) => ({ + id: `${prefix}${i}`, + ts: new Date(from + i * step).toISOString(), + workerId: 'kim', + kind: 'fund', + text: 't', +})) +const flat = (batches) => batches.flat().map((e) => e.id) + +test('dichter Speicher wie der Dienst: 450 Ereignisse im Abstand von 10 ms, kein Stillstand', async () => { + const store = dense(450) + const h = makeFeedHarness({ store }) + h.c.start() + await h.settle() + // alles in einer Runde, ein onEvents, aufsteigend, höchstens maxPages Abrufe + assert.equal(h.log.batches.length, 1) + assert.deepEqual(h.log.batches[0].map((e) => e.id), store.map((e) => e.id)) + assert.ok(h.log.events.length <= 5) + // volle Seite ohne Neues: der nächste Abruf fragt hinter die Spitze (Dienst liefert dann ab dem jüngsten ts) + assert.equal(h.log.events[1][0], store[199].ts) + assert.equal(h.log.events[2][0], new Date(Date.parse(store[199].ts) + 5000).toISOString()) + for (let i = 0; i < 4; i += 1) await h.advance(5000) + assert.equal(h.log.batches.length, 1) // nichts doppelt + // ein Ereignis nach dem Stau kommt an + store.push({ ...mk(0), id: 'spaet', ts: '2026-10-17T20:00:30.000Z' }) + await h.advance(5000) + assert.deepEqual(ids(h.log.batches).slice(1), [['spaet']]) + assert.deepEqual(h.log.errors, []) +}) + +for (const maxPages of [1, 2]) { + test(`dichter Speicher mit maxPages ${maxPages}: der Sprung gilt über Runden hinweg, alles kommt genau einmal an`, async () => { + const store = dense(450) + const h = makeFeedHarness({ store, opts: { maxPages } }) + h.c.start() + await h.settle() + for (let i = 0; i < 8; i += 1) await h.advance(5000) + assert.deepEqual(flat(h.log.batches), store.map((e) => e.id)) + assert.deepEqual(h.log.errors, []) + }) +} + +test('dichte Vorgeschichte: über 200 Ereignisse kurz vor lastEventTs werden nie gemeldet, das Neue danach schon', async () => { + const history = dense(300, { from: Date.UTC(2026, 9, 17, 19, 59, 57), prefix: 'alt' }) // 19:59:57.000 bis 19:59:59.990 + const store = [...history] + const h = makeFeedHarness({ store, state: { lastEventTs: history[299].ts, workers: [] } }) + h.c.start() + await h.settle() + assert.equal(h.log.batches.length, 0) + store.push({ ...mk(0), id: 'neu1', ts: '2026-10-17T20:00:01.000Z' }) + await h.advance(5000) + store.push({ ...mk(0), id: 'neu2', ts: '2026-10-17T20:00:07.000Z' }) + await h.advance(5000) + assert.deepEqual(ids(h.log.batches), [['neu1'], ['neu2']]) +}) + +test('200 Ereignisse mit demselben ts halten den Feed nicht an (Fremd-API ohne eindeutige ts)', async () => { + // P1 vergibt strikt aufsteigende ts; hier geht es nur darum, dass auch dann nichts dauerhaft hängen bleibt + const store = Array.from({ length: 250 }, (_, i) => ({ ...mk(0), id: `g${i}`, ts: '2026-10-17T20:00:01.000Z' })) + const h = makeFeedHarness({ store }) + h.c.start() + await h.settle() + assert.ok(h.log.events.length <= 5) + assert.equal(flat(h.log.batches).length, 200) // mehr als pageLimit Einträge mit einem ts sind über die API nicht erreichbar + store.push({ ...mk(0), id: 'spaet', ts: '2026-10-17T20:00:30.000Z' }) + await h.advance(5000) + const all = flat(h.log.batches) + assert.equal(all[all.length - 1], 'spaet') + assert.equal(new Set(all).size, all.length) + assert.deepEqual(h.log.errors, []) +})