gate intake readiness on durable progress instead of socket state
118 unmodified linesExpand all119120121122123124125126127128390 unmodified linesExpand all118 unmodified linesExpand all dropped: boundedNumber(redis.dropped), }, intake: { connected: intake.connected, // Socket state and durable progress differ under bounded backpressure. progressing: intake.directProgressing && intake.backlinkProgressing && intake.registryProgressing, started: intake.started, queued: Math.min(intake.queued, config.intakeQueueMax), directConnected: intake.directConnected, backlinkConnected: intake.backlinkConnected,390 unmodified linesExpand all1337 unmodified linesExpand all133813391340134113421343134413451346134713481349135013511352135313541355135613571358135928 unmodified lines1388138913901391138313921393139413951396139744 unmodified lines1337 unmodified linesExpand all const directConnected = !directRequired || directJetstream?.isConnected === true const backlinkConnected = !backlinkRequired || backlinkJetstream?.isConnected === true const registryConnected = registryJetstream?.isConnected === true const connected = registryConnected && directConnected && backlinkConnected // Readiness follows durable progress, not socket state. A stream draining its // bounded queue after backpressure is serving; one that acknowledges nothing // is not. Both are reported so the socket view stays visible. const directProgressing = !directRequired || directJetstream?.isProgressing === true const backlinkProgressing = !backlinkRequired || backlinkJetstream?.isProgressing === true const registryProgressing = registryJetstream?.isProgressing === true return { connected, directConnected, backlinkConnected, registryConnected, directProgressing, backlinkProgressing, registryProgressing, started: firehoseStarted, paused: intakePaused, queued: (directJetstream?.queued ?? 0) + (backlinkJetstream?.queued ?? 0) + (registryJetstream?.queued ?? 0), activeRecordKeys: recordExecutor.size,28 unmodified lines failureKind: registryJetstream?.failureKind, lastProgressAt: registryJetstream?.lastProgressTime, }, }, healthy: firehoseStarted && registryConnected && directConnected && backlinkConnected && !registrySnapshotOverflow, healthy: firehoseStarted && registryProgressing && directProgressing && backlinkProgressing && !registrySnapshotOverflow, }}
export function getEventStats() {44 unmodified lines129 unmodified linesExpand all130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178218 unmodified linesExpand all129 unmodified linesExpand all expect(client.queued).toBe(0) client.destroy() })
test('stays healthy while a bounded queue drains, and stops when acknowledgement stalls', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket let release: (() => void) | undefined const held = new Promise<void>((resolve) => { release = resolve }) const client = new JetstreamClient({ url: 'wss://relay.example/subscribe', cursor: 60, maxQueue: 2, onEvent: async () => held, onAcknowledged: async () => undefined, progressStaleMs: 60_000, }) client.start() const socket = FakeWebSocket.instances[0] if (!socket) throw new Error('missing test socket') socket.closeImmediately = false socket.open() socket.message(event(61)) socket.message(event(62)) await flush() // Backpressure closed the socket; the retained queue is still real work. expect(client.isConnected).toBe(false) expect(client.failureKind).toBe('queue') release?.() await flush() expect(client.cursor).toBe(62) expect(client.isProgressing).toBe(true)
const stalled = new JetstreamClient({ url: 'wss://relay.example/subscribe', onEvent: async () => undefined, progressStaleMs: 1, }) // Never acknowledged anything and not connected: not serving. expect(stalled.isProgressing).toBe(false) client.destroy() stalled.destroy() })
test('quarantines malformed bytes, ignores late frames, then resets only after an acked replay', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket const acknowledgements: number[] = [] const client = new JetstreamClient({218 unmodified linesExpand all42 unmodified lines434445464748495051523 unmodified lines5657585960616263646566148 unmodified linesExpand all21521621721821922022122222318 unmodified lines242243244245246247248249250291 unmodified linesExpand all54254354454554654754854955055155255355455555655755855956056180 unmodified lines42 unmodified lines maxEventBytes?: number reconnectMinMs?: number reconnectMaxMs?: number reconnectMaxExponent?: number /** How long a recovering stream may go without durable progress before it is stalled. */ progressStaleMs?: number}
interface QueuedEvent { event: JetstreamEvent3 unmodified linesfunction isStableDid(value: string): boolean { return isCanonicalWebhookDid(value)}
/** A recovering stream stays healthy only while it keeps acknowledging work. */const DEFAULT_PROGRESS_STALE_MS = 60_000
const COLLECTION_RE = /^[A-Za-z0-9.-]{1,253}$/const RKEY_RE = /^[A-Za-z0-9._~:%@+-]{1,512}$/
function callback(callback: (() => void) | undefined): void {148 unmodified linesExpand all private readonly maxEventBytes: number private readonly reconnectMinMs: number private readonly reconnectMaxMs: number private readonly reconnectMaxExponent: number private readonly progressStaleMs: number
constructor(private readonly opts: JetstreamOptions) { let relay: URL try {18 unmodified lines this.maxEventBytes = opts.maxEventBytes ?? config.intakeEventMaxBytes this.reconnectMinMs = opts.reconnectMinMs ?? config.jetstreamReconnectMinMs this.reconnectMaxMs = opts.reconnectMaxMs ?? config.jetstreamReconnectMaxMs this.reconnectMaxExponent = opts.reconnectMaxExponent ?? config.jetstreamReconnectMaxExponent this.progressStaleMs = opts.progressStaleMs ?? DEFAULT_PROGRESS_STALE_MS if ( !Number.isSafeInteger(this.maxQueue) || this.maxQueue < 1 || !Number.isSafeInteger(this.concurrency) ||291 unmodified linesExpand all get isConnected(): boolean { return this.connected && !this.failed && !this.quarantined && !this.backpressured }
/** * Intake health is durable progress, not socket state. A bounded queue that * closes the socket, drains, and replays from its last acknowledgement is * working exactly as designed; only a quarantined stream, or one that has * acknowledged nothing inside the window, has actually stopped serving. */ get isProgressing(): boolean { if (this.destroyed || this.quarantined) return false if (this.isConnected) return true return this.lastProgressAt !== 0 && Date.now() - this.lastProgressAt < this.progressStaleMs }
get isQuarantined(): boolean { return this.quarantined }
80 unmodified lines