Workers P2: Feed-Stillstand bei dichten Ereignissen behoben (Sprung hinter die Spitze)

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 <noreply@anthropic.com>

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

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, [])
+})