acknowledge the completed jetstream prefix with one cursor write
90 unmodified lines919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137218 unmodified linesExpand all90 unmodified lines expect(acknowledged).toEqual([11]) client.destroy() })
test('acknowledges the longest completed prefix with one durable cursor write', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket const acknowledged: number[] = [] let release: (() => void) | undefined const firstWrite = new Promise<void>((resolve) => { release = resolve }) let writes = 0 const client = new JetstreamClient({ url: 'wss://relay.example/subscribe', cursor: 50, concurrency: 4, onEvent: async () => undefined, onAcknowledged: async (value) => { writes++ acknowledged.push(value.time_us) if (writes === 1) await firstWrite }, }) client.start() const socket = FakeWebSocket.instances[0] expect(socket).toBeDefined() socket?.open() socket?.message(event(51)) await flush() // The first write is in flight; the next events complete behind it. socket?.message(event(52)) socket?.message(event(53)) await flush() expect(acknowledged).toEqual([51]) release?.() await flush() // 52 and 53 are one contiguous completed prefix: one write, latest cursor. expect(acknowledged).toEqual([51, 53]) expect(client.cursor).toBe(53) expect(client.queued).toBe(0) client.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 all378 unmodified linesExpand all379380381382383384385386387388389383390391392393394395396397398399400401402403404405406397407408409410411211 unmodified linesExpand all378 unmodified linesExpand all if (this.flushing || this.destroyed || this.failed) return this.flushing = true try { while (!this.destroyed && !this.failed && this.queue[0]?.state === 'complete') { // One durable write per completed prefix, never per event. The cursor // is a resume point, so intermediate values only add a round trip // each, and a crash mid-prefix replays into deduplicated durable // state. Under a backlog this is the difference between the relay // round-trip rate and the handler rate. let prefix = 0 while (this.queue[prefix]?.state === 'complete') prefix++ const item = this.queue[0] const item = this.queue[prefix - 1] if (!item) break try { await this.opts.onAcknowledged?.(item.event) } catch { this.fail(new Error('Jetstream cursor persistence failed'), 'cursor') return } // Failure or teardown during that write owns the queue instead. if (this.destroyed || this.failed) return this.lastAckedCursor = item.event.time_us // A durable acknowledgement proves the connection carried useful work; // only then reset exponential reconnect backoff. this.reconnectAttempts = 0 this.protocolFailures = 0 this.quarantined = false this.lastProgressAt = Date.now() this.queue.shift() this.queue.splice(0, prefix) this.notifyDrained() } } finally { this.flushing = false211 unmodified linesExpand all