Workers P2: Feed-Steuerung und Hooks (Zustand 15 s, Ereignisse 5 s)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
ClaudeandClaude Sonnet 5.5 committed 2026-10-18 01:34:20 +00:00
1 parent 97fdfbfcf0
commit 0bfad0ae0b
3 files changed
+630

No files matched your search

+111
View File
@@ -0,0 +1,111 @@
import { useCallback, useEffect, useRef, useState } from 'react'
import { createFeedController } from '../lib/workersFeed'
import { createRequestGuard } from '../lib/requestGuard'
// Spur eines Hustlers wird bei sichtbarem Tab alle 15 s neu geholt
const TRACE_MS = 15000
const UNKNOWN_ERROR = 'Unbekannter Fehler'
const isVisible = () => document.visibilityState !== 'hidden'
// { state, error, loaded, refresh }: Zustand der Hustler (alle 15 s) und Ereignisse (alle 5 s, über onEvents) von der
// Dienst-API. state ist null bis zum ersten Erfolg, loaded wird dann wahr und bleibt es. error ist der deutsche Fehlertext
// (oder null); die Abfrage läuft bei einem Fehler weiter. Im Hintergrund-Tab ruht alles, beim Zurückkehren wird sofort
// abgefragt. refresh() holt den Zustand sofort (z. B. nach dem Pausenschalter). onEvents darf bei jedem Render neu sein.
export function useWorkersFeed(api, { onEvents } = {}) {
const [state, setState] = useState(null)
const [error, setError] = useState(null)
const [loaded, setLoaded] = useState(false)
const onEventsRef = useRef(onEvents)
const controllerRef = useRef(null)
useEffect(() => { onEventsRef.current = onEvents }, [onEvents])
useEffect(() => {
setError(null)
const controller = createFeedController({
api,
setTimer: (fn, ms) => window.setTimeout(fn, ms),
clearTimer: (id) => window.clearTimeout(id),
isVisible,
onSnapshot: (snapshot) => {
setState(snapshot)
setLoaded(true)
},
onEvents: (list) => { if (onEventsRef.current) onEventsRef.current(list) },
onError: (err) => setError(err ? err.message || UNKNOWN_ERROR : null),
})
controllerRef.current = controller
const onVisibility = () => controller.visibilityChanged()
document.addEventListener('visibilitychange', onVisibility)
controller.start()
return () => {
document.removeEventListener('visibilitychange', onVisibility)
controller.stop()
controllerRef.current = null
}
}, [api])
const refresh = useCallback(() => {
if (controllerRef.current) controllerRef.current.refreshState()
}, [])
return { state, error, loaded, refresh }
}
// { trace, loading, error }: Spur eines Hustlers (api.trace). workerId null ergibt { trace: null } ohne Abruf. Beim
// Wechsel der workerId wird neu geladen (loading, bis die erste Antwort da ist), danach alle 15 s bei sichtbarem Tab und
// sofort beim Zurückkehren. Eine späte Antwort für einen früheren Hustler oder eine ältere Anfrage wird verworfen.
export function useWorkerTrace(api, workerId) {
const [result, setResult] = useState({ key: null, trace: null, error: null })
useEffect(() => {
if (!workerId) return undefined
const guard = createRequestGuard()
let timer = null
const stopTimer = () => {
if (timer === null) return
window.clearTimeout(timer)
timer = null
}
const load = async () => {
stopTimer()
const id = guard.next()
try {
const trace = await api.trace(workerId)
if (!guard.isCurrent(id)) return
setResult({ key: workerId, trace, error: null })
} catch (err) {
if (!guard.isCurrent(id)) return
// die letzte Spur desselben Hustlers bleibt stehen, der Fehler kommt dazu
setResult((prev) => ({ key: workerId, trace: prev.key === workerId ? prev.trace : null, error: err?.message || UNKNOWN_ERROR }))
}
timer = window.setTimeout(tick, TRACE_MS)
}
// Im Hintergrund kein Abruf, nur neu planen
const tick = () => {
timer = null
if (isVisible()) load()
else timer = window.setTimeout(tick, TRACE_MS)
}
const onVisibility = () => { if (isVisible()) load() }
document.addEventListener('visibilitychange', onVisibility)
load()
return () => {
guard.next()
stopTimer()
document.removeEventListener('visibilitychange', onVisibility)
}
}, [api, workerId])
const current = Boolean(workerId) && result.key === workerId
return {
trace: current ? result.trace : null,
loading: Boolean(workerId) && !current,
error: current ? result.error : null,
}
}
+198
View File
@@ -0,0 +1,198 @@
// Polling-Steuerung der Hustler-Seite: Zustand alle 15 s, Ereignisse alle 5 s (Übersicht §6.2).
// Kein React, kein Appwrite, kein window/document: Timer und Sichtbarkeit werden injiziert (unter Node testbar).
// Je Schleife läuft höchstens ein Abruf und höchstens ein wartender Timer; der nächste Timer wird erst nach
// Abschluss des Abrufs gesetzt (kein Intervall, kein Überlappen).
import { takeNewEvents, latestTs, needsStateRefresh } from './workersEvents.js'
const EPOCH = '1970-01-01T00:00:00.000Z'
const noop = () => {}
const isRecord = (v) => v !== null && typeof v === 'object' && !Array.isArray(v)
/**
* Steuerung für den Feed. Rückrufe: onSnapshot(state) bei jedem geholten Zustand, onEvents(liste) einmal je Runde
* mit allen neuen Ereignissen (aufsteigend), onError(err) einmal je Fehlerfolge und onError(null) beim nächsten Erfolg.
* Der Dienst liefert bei events(after) auch Einträge der letzten 5 s davor; die Deduplizierung übernimmt takeNewEvents.
*/
export function createFeedController({
api, setTimer, clearTimer, isVisible,
onSnapshot = noop, onEvents = noop, onError = noop,
stateMs = 15000, eventsMs = 5000, pageLimit = 200, maxPages = 5,
}) {
let running = false
let gen = 0 // wird bei start/stop erhöht: Antworten einer früheren Generation ändern nichts
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 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 }
const inFlight = { state: false, events: false }
const failure = { state: null, events: null }
function cancel(loop) {
if (timer[loop] === null) return
clearTimer(timer[loop])
timer[loop] = null
}
function schedule(loop) {
if (!running) return
cancel(loop)
timer[loop] = setTimer(() => tick(loop), loop === 'state' ? stateMs : eventsMs)
}
// Im Hintergrund kein Abruf, nur neu planen; beim Zurückkehren holt visibilityChanged() sofort nach
function tick(loop) {
timer[loop] = null
if (!running) return
if (!isVisible()) {
schedule(loop)
return
}
if (loop === 'state') void loadState()
else void loadEvents()
}
// Ein Fehler gilt, bis beide Schleifen wieder laufen; gemeldet wird nur, wenn sich der Text ändert
function publishError() {
const err = failure.state || failure.events || null
const message = err ? String(err.message ?? err) : null
if (message === reported) return
reported = message
onError(err)
}
async function loadState() {
if (!running || inFlight.state) return
cancel('state')
inFlight.state = true
const myGen = gen
let state = null
let failed = null
try {
state = await api.state()
} catch (err) {
failed = err
}
if (myGen !== gen) return
inFlight.state = false
failure.state = failed
let first = false
try {
if (failed === null) {
const snapshot = isRecord(state) ? state : {}
if (cursor === null) {
first = true
// fehlt lastEventTs oder ist es kein Datum, gibt es keine Vorgeschichte und der Cursor beginnt bei 1970
const lastMs = typeof snapshot.lastEventTs === 'string' ? Date.parse(snapshot.lastEventTs) : NaN
cursor = Number.isNaN(lastMs) ? EPOCH : snapshot.lastEventTs
historyMs = lastMs
}
onSnapshot(snapshot)
}
publishError()
} finally {
if (myGen === gen) {
if (stateAgain) {
stateAgain = false
void loadState()
} else {
schedule('state')
}
if (first) {
if (isVisible()) void loadEvents()
else schedule('events')
}
}
}
}
// Vorgeschichte: in der ersten Runde nach dem Zustand zählen Ereignisse bis lastEventTs als gesehen, aber nicht neu
function isHistory(e) {
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
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)
if (myGen !== gen) break
const list = isRecord(res) && Array.isArray(res.events) ? res.events : []
for (const e of takeNewEvents(seen, list)) {
if (!isHistory(e)) fresh.push(e)
}
cursor = latestTs(list, cursor)
if (list.length < pageLimit) break
}
} catch (err) {
failed = err
}
return { fresh, failed }
}
async function loadEvents() {
if (!running || cursor === null || inFlight.events) return
cancel('events')
inFlight.events = true
const myGen = gen
const { fresh, failed } = await fetchRound(myGen)
if (myGen !== gen) return
inFlight.events = false
failure.events = failed
if (failed === null) historyMs = NaN
try {
// Bei einem Fehler nach einer vollen Seite wird das bereits Geholte trotzdem gemeldet (es gilt schon als gesehen)
if (fresh.length > 0) onEvents(fresh)
publishError()
if (needsStateRefresh(fresh)) refreshState()
} finally {
if (myGen === gen) schedule('events')
}
}
function refreshState() {
if (!running) return
if (inFlight.state) {
stateAgain = true
return
}
void loadState()
}
function visibilityChanged() {
if (!running || !isVisible()) return
void loadState()
void loadEvents()
}
function start() {
if (running) return
running = true
gen += 1
cursor = null
historyMs = NaN
seen = new Set()
stateAgain = false
reported = null
failure.state = null
failure.events = null
if (isVisible()) void loadState()
else schedule('state')
}
function stop() {
running = false
gen += 1
cancel('state')
cancel('events')
inFlight.state = false
inFlight.events = false
stateAgain = false
}
return { start, stop, visibilityChanged, refreshState }
}
+321
View File
@@ -0,0 +1,321 @@
import test from 'node:test'
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import { loadModule } from './load.mjs'
// Feed-Steuerung der Hustler-Seite: Zustand 15 s, Ereignisse 5 s, Cursor, Seiten, Sichtbarkeit, Fehler.
// Alles mit Fake-Timern und einer Fake-API (kein Netz, keine echte Zeit).
const { createFeedController } = await loadModule('src/lib/workersFeed.js')
const nextTick = () => new Promise((resolve) => setImmediate(resolve))
function deferred() {
let resolve
const promise = new Promise((r) => { resolve = r })
return { promise, resolve }
}
// Fake-Timer + Fake-API. Ereignisse kommen entweder aus der Antwortliste `eventReplies` (ein Error darin wird zur
// Ablehnung) oder, mit `store`, aus einer veränderlichen Liste und verhalten sich wie der Dienst P1:
// events(after) liefert aufsteigend alle Einträge mit ts >= after - 5000 (5 s Überlappung), höchstens `limit`.
// `stateReplies` wie eventReplies für den Zustand; `h.gate` hält state() an, bis das Versprechen erfüllt wird.
function makeFeedHarness({ state, eventReplies = [], eventsHang = false, stateReplies = [], store = null, opts = {} } = {}) {
let now = 0
let nextId = 1
const timers = new Map()
const log = { state: 0, events: [], snapshots: [], batches: [], errors: [] }
const visible = { v: true }
const h = { gate: null, timers, log, visible }
const base = state ?? { lastEventTs: '2026-10-17T20:00:00.000Z', workers: [] }
const setTimer = (fn, ms) => {
const id = nextId
nextId += 1
timers.set(id, { at: now + ms, fn })
return id
}
const clearTimer = (id) => { timers.delete(id) }
const api = {
async state() {
log.state += 1
if (h.gate) await h.gate.promise
const r = stateReplies.shift()
if (r instanceof Error) throw r
return r ?? base
},
async events(after, limit) {
log.events.push([after, limit])
if (eventsHang) return new Promise(() => {})
if (store) {
const from = Date.parse(after) - 5000
const list = store.filter((e) => Date.parse(e.ts) >= from).sort((a, b) => Date.parse(a.ts) - Date.parse(b.ts))
return { events: list.slice(0, limit) }
}
const r = eventReplies.shift() ?? []
if (r instanceof Error) throw r
return { events: r }
},
}
h.c = createFeedController({
api,
setTimer,
clearTimer,
isVisible: () => visible.v,
onSnapshot: (s) => { log.snapshots.push(s) },
onEvents: (list) => { log.batches.push(list) },
onError: (e) => { log.errors.push(e ? e.message : null) },
...opts,
})
// feuert fällige Timer in Zeitreihenfolge; je Timer lässt ein setImmediate alle Antworten durchlaufen
h.advance = async (ms) => {
const target = now + ms
while (true) {
let pick = null
for (const [id, t] of timers) {
if (t.at <= target && (pick === null || t.at < pick.t.at)) pick = { id, t }
}
if (pick === null) break
timers.delete(pick.id)
now = pick.t.at
pick.t.fn()
await nextTick()
}
now = target
}
h.settle = async () => {
await nextTick()
await nextTick()
}
return h
}
const mk = (i) => ({ id: `e${i}`, ts: new Date(Date.UTC(2026, 9, 17, 20, 0, 1 + i)).toISOString(), workerId: 'kim', kind: 'fund', text: 't' })
const ids = (batches) => batches.map((b) => b.map((e) => e.id))
test('Start: Zustand, Cursor aus lastEventTs, dann Ereignisse (Review Focus 3)', async () => {
const h = makeFeedHarness({})
h.c.start()
await h.settle()
assert.equal(h.log.snapshots.length, 1)
assert.deepEqual(h.log.events[0], ['2026-10-17T20:00:00.000Z', 200])
const h2 = makeFeedHarness({ state: { workers: [] } })
h2.c.start()
await h2.settle()
assert.equal(h2.log.events[0][0], '1970-01-01T00:00:00.000Z')
// ein unlesbares lastEventTs wird nie als Cursor an den Dienst geschickt
const h3 = makeFeedHarness({ state: { lastEventTs: 'kaputt', workers: [] } })
h3.c.start()
await h3.settle()
assert.equal(h3.log.events[0][0], '1970-01-01T00:00:00.000Z')
})
test('Takt: Ereignisse alle 5 s mit weitergeschobenem Cursor, Zustand alle 15 s', async () => {
const h = makeFeedHarness({ eventReplies: [[mk(1)], [], []] })
h.c.start()
await h.settle()
await h.advance(5000)
assert.equal(h.log.events[1][0], mk(1).ts)
await h.advance(10000)
assert.equal(h.log.state, 2)
assert.equal(h.log.events.length, 4)
})
test('Hintergrund: keine Abrufe, beim Zurückkehren sofort beide', async () => {
const h = makeFeedHarness({})
h.c.start()
await h.settle()
h.visible.v = false
await h.advance(60000)
assert.equal(h.log.state, 1)
assert.equal(h.log.events.length, 1)
h.visible.v = true
h.c.visibilityChanged()
await h.settle()
assert.equal(h.log.state, 2)
assert.equal(h.log.events.length, 2)
await h.advance(5000)
assert.equal(h.log.events.length, 3)
assert.equal(h.timers.size, 2) // eine Kette je Schleife, kein doppelter Timer nach der Rückkehr
})
test('Doppelte Ereignisse werden nie zweimal gemeldet (Review Focus 3)', async () => {
const h = makeFeedHarness({ eventReplies: [[mk(1), mk(2)], [mk(1), mk(2), mk(3)]] })
h.c.start()
await h.settle()
await h.advance(5000)
assert.deepEqual(h.log.batches.map((b) => b.map((e) => e.id)), [['e1', 'e2'], ['e3']])
})
test('Rückstau: volle Seiten werden sofort nachgeladen, höchstens 5 je Runde (Review Focus 2)', async () => {
const page = (n) => Array.from({ length: 200 }, (_, i) => mk(n * 200 + i))
const h = makeFeedHarness({ eventReplies: [page(0), page(1), page(2), page(3), page(4), page(5)] })
h.c.start()
await h.settle()
assert.equal(h.log.events.length, 5)
assert.equal(h.log.events[1][0], mk(199).ts)
// alle Seiten einer Runde kommen in einem einzigen onEvents-Aufruf
assert.equal(h.log.batches.length, 1)
assert.equal(h.log.batches[0].length, 1000)
})
test('keine überlappenden Abrufe, Fehler laufen weiter, Zustand-Auffrischung bei wichtigen Ereignissen', async () => {
const hang = makeFeedHarness({ eventsHang: true })
hang.c.start()
await hang.settle()
await hang.advance(60000)
assert.equal(hang.log.events.length, 1)
const err = makeFeedHarness({ eventReplies: [new Error('Der Hustler-Dienst ist gerade nicht erreichbar.'), []] })
err.c.start()
await err.settle()
await err.advance(5000)
assert.deepEqual(err.log.errors, ['Der Hustler-Dienst ist gerade nicht erreichbar.', null])
const deal = makeFeedHarness({ eventReplies: [[{ ...mk(1), kind: 'deal' }]] })
deal.c.start()
await deal.settle()
assert.equal(deal.log.state, 2)
})
test('stop räumt auf, start ist idempotent, späte Antworten werden verworfen', async () => {
const h = makeFeedHarness({})
h.c.start()
h.c.start()
await h.settle()
assert.equal(h.log.state, 1)
h.c.stop()
assert.equal(h.timers.size, 0)
await h.advance(60000)
assert.equal(h.log.state, 1)
})
test('Hooks: Sichtbarkeit und Aufräumen im Quelltext', () => {
const src = readFileSync(new URL('../../src/hooks/useWorkersFeed.js', import.meta.url), 'utf8')
assert.match(src, /visibilitychange/)
assert.match(src, /removeEventListener\('visibilitychange'/)
assert.match(src, /\.stop\(\)/)
assert.match(src, /createRequestGuard/)
})
// ---- Ergänzungen: Vorgeschichte, Überlappung des Dienstes, Teilrunden, Spätantworten ----
test('Vorgeschichte: Ereignisse bis lastEventTs gelten als gesehen und werden nie gemeldet', async () => {
const hist1 = { ...mk(0), id: 'h1', ts: '2026-10-17T19:59:58.000Z' }
const hist2 = { ...mk(0), id: 'h2', ts: '2026-10-17T20:00:00.000Z' } // genau am Cursor
const h = makeFeedHarness({ eventReplies: [[hist1, hist2, mk(1)], [hist1, hist2, mk(1), mk(2)]] })
h.c.start()
await h.settle()
await h.advance(5000)
assert.deepEqual(ids(h.log.batches), [['e1'], ['e2']])
// ohne lastEventTs im Zustand (nichts davor) wird alles gemeldet
const none = makeFeedHarness({ state: { workers: [] }, eventReplies: [[hist1, mk(1)]] })
none.c.start()
await none.settle()
assert.deepEqual(ids(none.log.batches), [['h1', 'e1']])
})
test('wie der Dienst: 5 s Überlappung liefert jedes Ereignis genau einmal', async () => {
const store = [{ ...mk(0), id: 'alt', ts: '2026-10-17T19:59:57.000Z' }, { ...mk(0), id: 'am-cursor', ts: '2026-10-17T20:00:00.000Z' }]
const h = makeFeedHarness({ store })
h.c.start()
await h.settle()
assert.equal(h.log.batches.length, 0) // die Vorgeschichte aus der Überlappung wird nicht nachgespielt
store.push(mk(1), mk(2))
await h.advance(5000)
store.push(mk(3))
await h.advance(5000)
await h.advance(5000)
await h.advance(5000)
assert.deepEqual(ids(h.log.batches), [['e1', 'e2'], ['e3']])
})
test('wie der Dienst: 450 neue Ereignisse über drei Seiten, einmal gemeldet, aufsteigend', async () => {
const store = Array.from({ length: 450 }, (_, i) => mk(i))
const h = makeFeedHarness({ store })
h.c.start()
await h.settle()
assert.equal(h.log.events.length, 3)
assert.equal(h.log.batches.length, 1)
assert.equal(h.log.batches[0].length, 450)
assert.deepEqual(h.log.batches[0].map((e) => e.id), store.map((e) => e.id))
await h.advance(5000)
assert.equal(h.log.batches.length, 1)
})
test('Fehler nach einer Seite: bereits Geholtes wird gemeldet, danach der Fehler', async () => {
const page = Array.from({ length: 200 }, (_, i) => mk(i))
const h = makeFeedHarness({ eventReplies: [page, new Error('Der Hustler-Dienst ist gerade nicht erreichbar.'), []] })
h.c.start()
await h.settle()
assert.equal(h.log.batches.length, 1)
assert.equal(h.log.batches[0].length, 200)
assert.deepEqual(h.log.errors, ['Der Hustler-Dienst ist gerade nicht erreichbar.'])
await h.advance(5000)
assert.deepEqual(h.log.errors, ['Der Hustler-Dienst ist gerade nicht erreichbar.', null])
assert.equal(h.log.batches.length, 1) // die Seite wird nicht noch einmal gemeldet
})
test('erster Zustand schlägt fehl: kein Abruf der Ereignisse, später Cursor und Fehler zurückgesetzt', async () => {
const h = makeFeedHarness({ stateReplies: [new Error('Der Hustler-Dienst antwortet nicht richtig.')] })
h.c.start()
await h.settle()
assert.equal(h.log.events.length, 0)
assert.equal(h.log.snapshots.length, 0)
assert.deepEqual(h.log.errors, ['Der Hustler-Dienst antwortet nicht richtig.'])
await h.advance(15000)
assert.equal(h.log.snapshots.length, 1)
assert.equal(h.log.events.length, 1)
assert.equal(h.log.events[0][0], '2026-10-17T20:00:00.000Z')
assert.deepEqual(h.log.errors, ['Der Hustler-Dienst antwortet nicht richtig.', null])
})
test('Zustand-Auffrischung während eines laufenden Abrufs wird danach nachgeholt', async () => {
const h = makeFeedHarness({})
h.c.start()
await h.settle()
h.gate = deferred()
h.c.refreshState()
await h.settle()
assert.equal(h.log.state, 2)
h.c.refreshState()
h.c.refreshState()
await h.settle()
assert.equal(h.log.state, 2) // kein zweiter Abruf daneben
const gate = h.gate
h.gate = null
gate.resolve()
await h.settle()
assert.equal(h.log.state, 3)
assert.equal(h.log.snapshots.length, 3)
assert.equal(h.timers.size, 2)
})
test('Rückkehr in den Vordergrund startet keinen zweiten Abruf neben einem hängenden', async () => {
const h = makeFeedHarness({ eventsHang: true })
h.c.start()
await h.settle()
h.c.visibilityChanged()
await h.settle()
assert.equal(h.log.events.length, 1)
assert.equal(h.log.state, 2) // der Zustand ist frei und wird geholt
})
test('späte Antwort nach stop wird verworfen, ein neuer Start beginnt frisch', async () => {
const h = makeFeedHarness({ eventReplies: [[mk(1)]] })
h.gate = deferred()
h.c.start()
await h.settle()
h.c.stop()
const gate = h.gate
h.gate = null
gate.resolve()
await h.settle()
assert.equal(h.log.snapshots.length, 0)
assert.equal(h.log.events.length, 0)
assert.equal(h.timers.size, 0)
h.c.start()
await h.settle()
assert.equal(h.log.snapshots.length, 1)
assert.deepEqual(ids(h.log.batches), [['e1']])
assert.equal(h.timers.size, 2)
})