commit each intake batch, its deliveries, and its cursor in one transaction
19 unmodified lines20212223242425262728176 unmodified linesExpand all20520620720820920921021121221362 unmodified lines19 unmodified lines readonly webhookEventsChannel: string readonly webhookCacheMax: number readonly webhookCacheTtlMs: number readonly intakeQueueMax: number readonly intakeConcurrency: number readonly intakeBatchMax: number readonly intakeEventMaxBytes: number readonly intakeRecordKeyMax: number readonly registryActiveSubscriptionsMax: number readonly registryOwnerActiveRecordsMax: number176 unmodified linesExpand all webhookEventsChannel: channel(env.WEBHOOK_EVENTS_CHANNEL), webhookCacheMax: integer('WEBHOOK_CACHE_MAX', env.WEBHOOK_CACHE_MAX, 1_000, 1, 10_000), webhookCacheTtlMs: integer('WEBHOOK_CACHE_TTL_MS', env.WEBHOOK_CACHE_TTL_MS, 60_000, 1_000, 3_600_000), intakeQueueMax: integer('WEBHOOK_INTAKE_QUEUE_MAX', env.WEBHOOK_INTAKE_QUEUE_MAX, 512, 1, 10_000), intakeConcurrency: integer('WEBHOOK_INTAKE_CONCURRENCY', env.WEBHOOK_INTAKE_CONCURRENCY, 1, 1, 32), intakeBatchMax: integer('WEBHOOK_INTAKE_BATCH_MAX', env.WEBHOOK_INTAKE_BATCH_MAX, 128, 1, 1_000), intakeEventMaxBytes: integer( 'WEBHOOK_INTAKE_EVENT_MAX_BYTES', env.WEBHOOK_INTAKE_EVENT_MAX_BYTES, 1_048_576,62 unmodified lines26 unmodified lines272829303132333435363738394041424344454647481392 unmodified linesExpand all144114421443144414311445144614471448144914501451145211 unmodified lines146414651466146714511468146914701471147298 unmodified lines15711572157315741558157515761577157815791580158115601582158315841585158628 unmodified lines16151616161716181597159816191620162116221623162416251626162716011628162916301631163248 unmodified lines168116821683168416581685168616871688168916601690169116921693169442 unmodified lines1737173817391740174117421743174417453 unmodified lines17491750175117521722175317541755175617575 unmodified lines176317641765176617361767176817691770177156 unmodified lines18281829183018311801183218331834183518361837183818391840180518411842184318441845112 unmodified linesExpand all1958195919601961196219631964196519661967196819691932197019711972197319741975197673 unmodified lines205020512052205320142054205520562057205820592060386 unmodified linesExpand all26 unmodified lines
export const db = new SQL(validatedDatabaseUrl)
/** * A statement runner: the shared pool, or one open transaction on it. * * Intake commits a whole batch through one executor - deliveries, backlink * state, registry mutations, and the stream cursor - so a crash replays that * batch instead of leaving the cursor ahead of the effects it claims. */export type SqlExecutor = typeof db
/** Run one intake batch so its cursor and its effects commit together. */export async function runIntakeBatch<T>(work: (sql: SqlExecutor) => Promise<T>): Promise<T> { return db.begin(async (tx) => work(tx as unknown as SqlExecutor)) as Promise<T>}
/** * Serialize all schema bootstrap work across replicas. The transaction-scoped * advisory lock is released automatically on commit/rollback. lock_timeout * makes a wedged migration fail startup rather than wait forever. */1392 unmodified linesExpand all/** * Record only a hashed event identity and a finite reason. Rejected record * bytes, URLs, credentials, and payload text are deliberately never stored. */export async function recordWebhookIntakeQuarantine(input: WebhookIntakeQuarantineInput): Promise<void> {export async function recordWebhookIntakeQuarantine( input: WebhookIntakeQuarantineInput, sql: SqlExecutor = db,): Promise<void> { if ( !Number.isSafeInteger(input.timeUs) || input.timeUs < 0 || !isNonEmptyBoundedString(input.did, 2048) ||11 unmodified lines const eventAtUriHash = createHash('sha256').update(`wisp-webhook-event/v1\0${eventAtUri}`).digest('hex') const quarantineId = createHash('sha256') .update(`wisp-webhook-quarantine/v1\0${relay}\0${input.timeUs}\0${input.rev}\0${eventAtUri}`) .digest('hex') await db` await sql` INSERT INTO webhook_intake_quarantines (quarantine_id, relay_id, source_time_us, source_revision, event_at_uri_hash, reason, created_at, last_seen_at) VALUES (${quarantineId}, ${relay}, ${input.timeUs}, ${input.rev}, ${eventAtUriHash}, ${input.reason}, NOW(), NOW()) ON CONFLICT (quarantine_id) DO UPDATE SET98 unmodified lines return Math.max(0, sequence - CURSOR_REWIND_US)}
/** Persist a nondecreasing cursor for one consumer and one normalized relay. */export async function saveJetstreamCursor(consumer: JetstreamConsumer, relay: string, sequence: number): Promise<void> {export async function saveJetstreamCursor( consumer: JetstreamConsumer, relay: string, sequence: number, sql: SqlExecutor = db,): Promise<void> { const id = assertCursorInput(consumer, relay, sequence) await db` await sql` INSERT INTO jetstream_cursors (consumer_id, relay_id, seq, saved_at) VALUES (${consumer}, ${id}, ${sequence}, NOW()) ON CONFLICT (consumer_id, relay_id) DO UPDATE SET seq = GREATEST(jetstream_cursors.seq, EXCLUDED.seq),28 unmodified lines throw new Error('Invalid Jetstream consumer') return loadJetstreamCursor(streamId, relay)}
/** Alias used by newer intake code. */export async function saveCursorForStream(streamId: string, cursor: number, relay: string): Promise<void> {/** Alias used by newer intake code. Pass the batch executor to commit it with the batch. */export async function saveCursorForStream( streamId: string, cursor: number, relay: string, sql: SqlExecutor = db,): Promise<void> { if (streamId !== 'direct' && streamId !== 'backlink' && streamId !== 'registry') throw new Error('Invalid Jetstream consumer') await saveJetstreamCursor(streamId, relay, cursor) await saveJetstreamCursor(streamId, relay, cursor, sql)}
export const MAX_BACKLINK_REFERENCE_ROWS = 100_000export const MAX_BACKLINK_REFERENCES_PER_EVENT = 10048 unmodified lines return decoded}
/** Load bounded prior relevant backlink references after restart. */export async function loadPriorReferenceIndex(eventAtUri: string): Promise<PriorReferenceIndexEntry | undefined> {export async function loadPriorReferenceIndex( eventAtUri: string, sql: SqlExecutor = db,): Promise<PriorReferenceIndexEntry | undefined> { const key = normalizeAtUri(eventAtUri) const rows = await db<Array<{ refs: unknown; last_seq: number | string | bigint; last_rev: string }>>` const rows = await sql<Array<{ refs: unknown; last_seq: number | string | bigint; last_rev: string }>>` SELECT refs, last_seq, last_rev FROM webhook_backlink_references WHERE event_at_uri = ${key} AND expires_at > NOW() LIMIT 142 unmodified lines eventAtUri: string, references: readonly string[], timeUs: number, rev: string, sql: SqlExecutor = db,): Promise<PriorReferenceWrite> { const key = normalizeAtUri(eventAtUri) if (!Number.isSafeInteger(timeUs) || timeUs < 0) throw new Error('Invalid backlink reference') assertAtprotoRevision(rev)3 unmodified lines if (new TextEncoder().encode(JSON.stringify(normalized)).byteLength > MAX_BACKLINK_REFERENCE_BYTES) throw new Error('Backlink references are too large')
if (normalized.length === 0) { const cleared = await db<Array<{ event_at_uri: string }>>` const cleared = await sql<Array<{ event_at_uri: string }>>` UPDATE webhook_backlink_references SET refs = '[]'::jsonb, last_seq = ${timeUs}::bigint, last_rev = ${rev}::text,5 unmodified lines ` return cleared.length > 0 ? 'cleared' : 'stale' }
const rows = await db<Array<{ admitted: boolean; written: boolean }>>` const rows = await sql<Array<{ admitted: boolean; written: boolean }>>` WITH admission AS ( SELECT EXISTS (SELECT 1 FROM webhook_backlink_references WHERE event_at_uri = ${key}) OR (SELECT count(*) FROM webhook_backlink_references) < ${MAX_BACKLINK_REFERENCE_ROWS} AS admitted56 unmodified lines return { keys: rows.slice(0, limit).map((row) => row.event_at_uri), complete: rows.length <= limit }}
/** Delete only if a later event has not already repopulated the same record. */export async function deletePriorReferenceIndex(eventAtUri: string, timeUs: number, rev: string): Promise<boolean> {export async function deletePriorReferenceIndex( eventAtUri: string, timeUs: number, rev: string, sql: SqlExecutor = db,): Promise<boolean> { const key = normalizeAtUri(eventAtUri) if (!Number.isSafeInteger(timeUs) || timeUs < 0) throw new Error('Invalid backlink reference') assertAtprotoRevision(rev) const rows = await db<Array<{ event_at_uri: string }>>` const rows = await sql<Array<{ event_at_uri: string }>>` DELETE FROM webhook_backlink_references WHERE event_at_uri = ${key} AND (last_seq < ${timeUs} OR (last_seq = ${timeUs} AND last_rev <= ${rev})) RETURNING event_at_uri112 unmodified linesExpand allexport async function enqueueWebhookDeliveryOutbox( event: NewWebhookDeliveryEventRow, rows: readonly NewWebhookDeliveryOutboxRow[], ensureEvent = true, /** When intake supplies its batch executor, these rows commit with the cursor. */ sql?: SqlExecutor,): Promise<{ enqueued: number; deduplicated: number }> { if (rows.length > 1_000 || typeof ensureEvent !== 'boolean') throw new Error('Webhook delivery batch is too large') assertOutboxEventRow(event) for (const row of rows) assertOutboxRow(row, event.eventId) if (rows.length === 0) return { enqueued: 0, deduplicated: 0 }
return db.begin(async (tx) => { // An intake batch already owns a transaction; joining it keeps these rows and // the stream cursor in one commit. Standalone callers still get their own. const write = async (tx: SqlExecutor): Promise<{ enqueued: number; deduplicated: number }> => { // A collision or mismatched replay must not silently attach rows to a // different immutable body. The deterministic event ID normally makes this // impossible, but the check also detects corruption. if (ensureEvent) {73 unmodified lines ON CONFLICT (delivery_id) DO NOTHING RETURNING delivery_id ` return { enqueued: inserted.length, deduplicated: rows.length - inserted.length } }) }
return sql ? write(sql) : db.begin(async (tx) => write(tx as unknown as SqlExecutor))}
export interface ClaimedWebhookDelivery { deliveryId: string386 unmodified linesExpand all14 unmodified lines151617181920212223140 unmodified linesExpand all16416516616716716816917017117222 unmodified lines19519619719819819920020120220319920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125210 unmodified lines26326426526622626726826927027114 unmodified lines28628728828924929029129229329441 unmodified lines3363373383392993403413423433441 unmodified line3463473483493093503513523533543553153563573583593603203613623633643651 unmodified line36736836937033037137237337437537637737833837938038138238314 unmodified lines loadPriorReferenceIndex: async () => undefined, savePriorReferenceIndex: async () => 'stored', deletePriorReferenceIndex: async () => false, loadPriorReferenceKeys: async () => ({ keys: [], complete: true }), runIntakeBatch: async (work: (sql: unknown) => Promise<unknown>) => work(undefined), enqueueWebhookDeliveries: async () => ({ enqueued: 0, deduplicated: 0 }), recordWebhookIntakeQuarantine: async () => undefined, getWebhookRecord: async () => undefined, upsertWebhookRecord: async () => false,140 unmodified linesExpand all recordWebhookIntakeQuarantine: async () => undefined, }) const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE)) expect(direct?.options.wantedDids).toEqual(expect.arrayContaining([OWNER, SCOPE])) await direct?.options.onEvent(commit(10)) await direct?.options.onEvents([commit(10)]) expect(enqueued).toEqual([1]) })
test('quarantines deterministic poison, commits its cursor path, and continues with a valid event', async () => {22 unmodified lines }, }) const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE)) expect(direct).toBeDefined() const poisonEvent = commit(20) await direct?.options.onEvent(poisonEvent) await direct?.options.onAcknowledged?.(poisonEvent) const validEvent = commit(21) await direct?.options.onEvent(validEvent) await direct?.options.onAcknowledged?.(validEvent) await direct?.options.onEvents([commit(20)]) await direct?.options.onEvents([commit(21)]) expect(quarantines).toEqual(['payload_too_large']) // Each batch commits its own cursor, quarantined poison included. expect(saved).toEqual([20, 21]) })
test('commits deliveries and the stream cursor in one batch, and neither when it fails', async () => { initScopeDids([{ did: OWNER, rkey: 'hook', record: record() }]) const created: FakeJetstream[] = [] const effects: string[] = [] let failing = false await startFirehose({ ...testDependencies(created), runIntakeBatch: async (work) => { try { const result = await work(undefined) effects.push('commit') return result } catch (error) { effects.push('rollback') throw error } }, cursorRepository: { load: async () => undefined, save: async (_stream, cursor) => { effects.push(`cursor:${cursor}`) }, }, enqueueWebhookDeliveries: async (entries) => { effects.push('enqueue') if (failing) throw new Error('outbox unavailable') return { enqueued: entries.length, deduplicated: 0 } }, recordWebhookIntakeQuarantine: async () => undefined, }) const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE)) expect(direct).toBeDefined() await direct?.options.onEvents([commit(10)]) // The cursor is the last statement of the same transaction as the delivery. expect(effects).toEqual(['enqueue', 'cursor:10', 'commit'])
failing = true effects.length = 0 await expect(direct?.options.onEvents([commit(11)])).rejects.toThrow('Intake event processing failed') // A rolled back batch never advances the cursor past work it lost. expect(effects).toEqual(['enqueue', 'rollback']) })
test('bounds a direct-PDS owner flood deterministically at the per-owner admission cap', async () => { const rows = Array.from({ length: 101 }, (_, index) => ({ did: OWNER, rkey: `hook-${String(index).padStart(3, '0')}`,10 unmodified lines }, recordWebhookIntakeQuarantine: async () => undefined, }) const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE)) await direct?.options.onEvent(commit(30)) await direct?.options.onEvents([commit(30)]) expect(candidateCount).toBe(100) expect(getEventStats().rejectedSubscriptionAdmissions).toBeGreaterThanOrEqual(1) })
14 unmodified lines }, recordWebhookIntakeQuarantine: async () => undefined, }) const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE)) await direct?.options.onEvent(commit(40)) await direct?.options.onEvents([commit(40)]) expect(candidateCount).toBe(1) })})
41 unmodified lines test('a relay record that references nothing in scope costs no durable read or write', async () => { const created: FakeJetstream[] = [] const probe = referenceProbe() const backlink = await startBacklinkIntake(created, probe) await backlink.options.onEvent(likeCommit(10, [])) await backlink.options.onEvents([likeCommit(10, [])]) expect(probe.loads).toEqual([]) expect(probe.writes).toEqual([]) })
1 unmodified line const created: FakeJetstream[] = [] const probe = referenceProbe() const backlink = await startBacklinkIntake(created, probe)
await backlink.options.onEvent(likeCommit(10, [SUBJECT])) await backlink.options.onEvents([likeCommit(10, [SUBJECT])]) expect(probe.loads).toEqual([]) expect(probe.writes).toEqual([VISITOR_KEY]) expect(probe.rows.get(VISITOR_KEY)).toEqual([SUBJECT])
// The same key may now own durable state, so its next version is read. await backlink.options.onEvent(likeCommit(11, [], 'update')) await backlink.options.onEvents([likeCommit(11, [], 'update')]) expect(probe.loads).toEqual([VISITOR_KEY]) expect(probe.rows.get(VISITOR_KEY)).toEqual([])
// Cleared state is forgotten again: no further round trip for this key. await backlink.options.onEvent(likeCommit(12, [], 'update')) await backlink.options.onEvents([likeCommit(12, [], 'update')]) expect(probe.loads).toEqual([VISITOR_KEY]) expect(probe.writes).toEqual([VISITOR_KEY, VISITOR_KEY]) })
1 unmodified line const created: FakeJetstream[] = [] const probe = referenceProbe() probe.complete = false const backlink = await startBacklinkIntake(created, probe) await backlink.options.onEvent(likeCommit(10, [])) await backlink.options.onEvents([likeCommit(10, [])]) expect(probe.loads).toEqual([VISITOR_KEY]) })
test('hydrates keys that already own references so their next event is matched', async () => { const created: FakeJetstream[] = [] const probe = referenceProbe([[VISITOR_KEY, [SUBJECT]]]) const backlink = await startBacklinkIntake(created, probe) await backlink.options.onEvent(likeCommit(10, [], 'update')) await backlink.options.onEvents([likeCommit(10, [], 'update')]) expect(probe.loads).toEqual([VISITOR_KEY]) expect(probe.rows.get(VISITOR_KEY)).toEqual([]) })})42 unmodified lines4344454647484950515253544955565758592 unmodified lines62636465606166676869707172737463757677787911 unmodified lines9192939495969798992 unmodified lines1021031041051061071081091101114 unmodified lines11611711811912012112212312434 unmodified lines15916016116214714816316416516683 unmodified lines2502512522532402542552562572583 unmodified lines2622632642652522532662672682692702717 unmodified lines2792802812822692832842852862872882892902762912922932942802952962822972982842993002863013022883033042903053063073083091 unmodified line311312313314300301315316317318319320321322323324325326327328329330331332333334303335336337338339457 unmodified linesExpand all79779879980080180280380480524 unmodified lines83083183283380183483583683783883984084180880981081181281381481581684284384484584684784884985085185285385485585673 unmodified lines9309319329339349359369379019389399409419429439449459469479489492 unmodified lines9529539549559189569579589599609619626 unmodified lines969970971972973974975976977978979939980981982983984985945986987988989949990991992993953994995996997998999100010011002100310049631005100610071008100924 unmodified lines1034103510361037996103810391040104110421001104310441045104610471048100710081049105010511052105310544 unmodified lines105910601061106210211063106410651066102510261067106810691070107110301072107310741075107610771078107910801081108210831084103210331085108610871088108910901091103510921093109410951096109710981099110011011042104310441045104610471048104910501051105210531054105511021103110411054 unmodified lines11101111111211131068111411151116111711181073107411191120112111221123191 unmodified linesExpand all131513161317131813191320132113221323163 unmodified linesExpand all14871488148914901445149142 unmodified linesconst BACKLINK_STREAM = 'backlink'const REGISTRY_STREAM = 'registry'type StreamName = typeof DIRECT_STREAM | typeof BACKLINK_STREAM | typeof REGISTRY_STREAM
/** The executor of the current intake batch; every durable effect joins it. */type BatchSql = database.SqlExecutor | undefined
/** Runs one batch so its effects and its cursor commit together. */type IntakeBatchRunner = <T>(work: (sql: BatchSql) => Promise<T>) => Promise<T>
export interface CursorRepository { load(stream: StreamName, relay: string): Promise<number | undefined> save(stream: StreamName, cursor: number, relay: string): Promise<void> save(stream: StreamName, cursor: number, relay: string, sql?: BatchSql): Promise<void>}
interface PriorReferenceSnapshot { references: readonly string[]2 unmodified lines}
/** Durable, bounded backlink state. It deliberately stores references, never whole records. */export interface PriorReferenceRepository { load(eventAtUri: string): Promise<PriorReferenceSnapshot | undefined> save(eventAtUri: string, references: readonly string[], timeUs: number, rev: string): Promise<PriorReferenceWrite> load(eventAtUri: string, sql?: BatchSql): Promise<PriorReferenceSnapshot | undefined> save( eventAtUri: string, references: readonly string[], timeUs: number, rev: string, sql?: BatchSql, ): Promise<PriorReferenceWrite> /** True only when this event removed the durable row. */ delete(eventAtUri: string, timeUs: number, rev: string): Promise<boolean> delete(eventAtUri: string, timeUs: number, rev: string, sql?: BatchSql): Promise<boolean> /** Bounded hydration of the keys that own durable references, when available. */ keys?(): Promise<{ keys: readonly string[]; complete: boolean }>}
11 unmodified lines
type EnqueueWebhookDeliveries = ( entries: readonly WebhookEntry[], event: DurableDeliveryEvent, sql?: BatchSql,) => Promise<{ enqueued: number; deduplicated: number }>
interface FirehoseStartOptions { /** Test seam. Production uses the durable DB cursor repository. */2 unmodified lines referenceRepository?: PriorReferenceRepository enqueueWebhookDeliveries?: EnqueueWebhookDeliveries /** Test seam. Production writes an idempotent redacted intake quarantine row. */ recordWebhookIntakeQuarantine?: typeof database.recordWebhookIntakeQuarantine /** Test seam. Production wraps each batch in one database transaction. */ runIntakeBatch?: IntakeBatchRunner createJetstreamClient?: (options: JetstreamOptions) => JetstreamClient /** Compatibility seam for tests; each stream remains independent in production. */ cursors?: Partial<Record<StreamName, number>> /** Capture relay events without invoking handlers until startup reconciliation finishes. */4 unmodified lines cursorRepository: CursorRepository referenceRepository: PriorReferenceRepository enqueueWebhookDeliveries: EnqueueWebhookDeliveries recordWebhookIntakeQuarantine: typeof database.recordWebhookIntakeQuarantine runIntakeBatch: IntakeBatchRunner createJetstreamClient: (options: JetstreamOptions) => JetstreamClient}
interface WhMutationResult {34 unmodified lines// for webhook registrations, including an owner's first-ever record.let registryJetstream: JetstreamClient | null = nullconst allClients = new Set<JetstreamClient>()
const cursorWriteTails = new Map<StreamName, Promise<void>>()
function safeError(stream: StreamName): void { logger.warn(`[${stream}] intake event failed; cursor was not advanced`)}
83 unmodified lines
function defaultCursorRepository(): CursorRepository { const api = database as typeof database & { loadCursorForStream?: (stream: string, relay: string) => Promise<number | undefined> saveCursorForStream?: (stream: string, cursor: number, relay: string) => Promise<void> saveCursorForStream?: (stream: string, cursor: number, relay: string, sql?: BatchSql) => Promise<void> saveCursor?: | ((cursor: number, relay: string) => Promise<void>) | ((stream: string, cursor: number, relay: string) => Promise<void>) }3 unmodified lines // Legacy installs have only one cursor. Never reuse it for unrelated streams. if (stream === DIRECT_STREAM) return database.loadCursor(relay) throw new Error('Independent durable cursor storage is unavailable') }, async save(stream, cursor, relay) { if (api.saveCursorForStream) return api.saveCursorForStream(stream, cursor, relay) async save(stream, cursor, relay, sql) { if (api.saveCursorForStream) return api.saveCursorForStream(stream, cursor, relay, sql) if (api.saveCursor && api.saveCursor.length >= 3) { return (api.saveCursor as (stream: string, cursor: number, relay: string) => Promise<void>)( stream, cursor,7 unmodified lines}
function defaultReferenceRepository(): PriorReferenceRepository { const api = database as typeof database & { loadPriorReferenceIndex?: (eventAtUri: string) => Promise<PriorReferenceSnapshot | undefined> loadPriorReferenceIndex?: (eventAtUri: string, sql?: BatchSql) => Promise<PriorReferenceSnapshot | undefined> savePriorReferenceIndex?: ( eventAtUri: string, references: readonly string[], timeUs: number, rev: string, sql?: BatchSql, ) => Promise<PriorReferenceWrite> deletePriorReferenceIndex?: (eventAtUri: string, timeUs: number, rev: string) => Promise<boolean> deletePriorReferenceIndex?: (eventAtUri: string, timeUs: number, rev: string, sql?: BatchSql) => Promise<boolean> loadPriorReferenceKeys?: () => Promise<{ keys: readonly string[]; complete: boolean }> } return { async load(key) { async load(key, sql) { if (!api.loadPriorReferenceIndex) throw new Error('Durable prior-reference storage is unavailable') return api.loadPriorReferenceIndex(key) return api.loadPriorReferenceIndex(key, sql) }, async save(key, references, timeUs, rev) { async save(key, references, timeUs, rev, sql) { if (!api.savePriorReferenceIndex) throw new Error('Durable prior-reference storage is unavailable') return api.savePriorReferenceIndex(key, references, timeUs, rev) return api.savePriorReferenceIndex(key, references, timeUs, rev, sql) }, async delete(key, timeUs, rev) { async delete(key, timeUs, rev, sql) { if (!api.deletePriorReferenceIndex) throw new Error('Durable prior-reference storage is unavailable') return api.deletePriorReferenceIndex(key, timeUs, rev) return api.deletePriorReferenceIndex(key, timeUs, rev, sql) }, async keys() { if (!api.loadPriorReferenceKeys) throw new Error('Durable prior-reference storage is unavailable') return api.loadPriorReferenceKeys()1 unmodified line }}
function defaultEnqueueWebhookDeliveries(): EnqueueWebhookDeliveries { const api = delivery as typeof delivery & { enqueueWebhookDeliveries?: EnqueueWebhookDeliveries } return async (entries, event) => { const api = delivery as typeof delivery & { enqueueWebhookDeliveries?: ( entries: readonly WebhookEntry[], event: DurableDeliveryEvent, options?: delivery.EnqueueWebhookDeliveriesOptions, ) => Promise<{ enqueued: number; deduplicated: number }> } return async (entries, event, sql) => { if (!api.enqueueWebhookDeliveries) throw new Error('Durable webhook delivery queue is unavailable') // Outbox rows join the intake batch, so the cursor can never advance past // a delivery that was rolled back. const options = sql ? { enqueueOutbox: ( outboxEvent: Parameters<typeof database.enqueueWebhookDeliveryOutbox>[0], rows: Parameters<typeof database.enqueueWebhookDeliveryOutbox>[1], ensureEvent?: boolean, ) => database.enqueueWebhookDeliveryOutbox(outboxEvent, rows, ensureEvent, sql), } : undefined return api.enqueueWebhookDeliveries(entries, event) return api.enqueueWebhookDeliveries(entries, event, options) }}
function currentDependencies(): RuntimeDependencies {457 unmodified linesExpand all event: JetstreamEvent, backlinkRecord: unknown, deliveryRecord: unknown, excludedWebhookKey?: string, sql?: BatchSql,): Promise<void> { if (event.kind !== 'commit' || !event.commit) return const { did, time_us: timeUs, commit } = event const candidates = await getWebhooksForEvent(did, backlinkRecord)24 unmodified lines ...(commit.cid ? { cid: commit.cid } : {}), ...(deliveryRecord === undefined ? {} : { record: deliveryRecord }), } try { const result = await currentDependencies().enqueueWebhookDeliveries(matched, durableEvent) const result = await currentDependencies().enqueueWebhookDeliveries(matched, durableEvent, sql) totalMatched += result.enqueued + result.deduplicated } catch (error) { if (error instanceof delivery.WebhookDeliveryInputError) { // Deterministic untrusted payload/subscription input is ackable. Store a // bounded redacted identity first; a failure to persist that row remains // retryable and deliberately keeps the cursor behind. await currentDependencies().recordWebhookIntakeQuarantine( await currentDependencies().recordWebhookIntakeQuarantine({ relay: normalizeRelayIdentity(config.jetstreamUrl), timeUs, rev: commit.rev, did, collection: commit.collection, rkey: commit.rkey, reason: error.kind, }) { relay: normalizeRelayIdentity(config.jetstreamUrl), timeUs, rev: commit.rev, did, collection: commit.collection, rkey: commit.rkey, reason: error.kind, }, sql, ) invalidDeliveryInputEvents++ logger.warn('[delivery] invalid durable input; event quarantined') return }73 unmodified lines key: string, references: readonly string[], timeUs: number, rev: string, sql?: BatchSql,): Promise<void> { if (references.length === 0 && !mayOwnPriorReferences(key)) return const owned = PRIOR_STATE_AFTER_WRITE[await currentDependencies().referenceRepository.save(key, references, timeUs, rev)] PRIOR_STATE_AFTER_WRITE[await currentDependencies().referenceRepository.save(key, references, timeUs, rev, sql)] if (owned !== undefined) trackPriorReferences(key, owned)}
async function loadReferenceLifecycle( event: JetstreamEvent, deliveryRecord: unknown, sql?: BatchSql,): Promise<ReferenceLifecycle | undefined> { if (event.kind !== 'commit' || !event.commit || backlinkScopeDids.size === 0) return undefined const { did, commit } = event const key = eventAtUri(did, commit.collection, commit.rkey)2 unmodified lines } const current = commit.operation === 'delete' ? { references: [], tooComplex: false } : currentReferences(deliveryRecord) // Only a key that may own durable state is worth reading it. const snapshot = mayOwnPriorReferences(key) ? await currentDependencies().referenceRepository.load(key) : undefined const snapshot = mayOwnPriorReferences(key) ? await currentDependencies().referenceRepository.load(key, sql) : undefined const oldReferences = normalizeReferences(snapshot?.references ?? []) const combined = [...new Set([...oldReferences, ...current.references])] return { key,6 unmodified lines
async function commitReferenceLifecycle( event: JetstreamEvent, lifecycle: ReferenceLifecycle | undefined, sql?: BatchSql,): Promise<void> { if (!lifecycle || event.kind !== 'commit' || !event.commit) return const { time_us: timeUs, commit } = event if (lifecycle.tooComplex) { // Preserve a monotonic empty tombstone at this event version. A later delete // therefore cannot resurrect stale references from an earlier known state. await writePriorReferences(lifecycle.key, [], timeUs, commit.rev) await writePriorReferences(lifecycle.key, [], timeUs, commit.rev, sql) return } if (commit.operation === 'delete') { // Delete only after all deliveries using the old durable snapshot are enqueued. if (!mayOwnPriorReferences(lifecycle.key)) return if (await currentDependencies().referenceRepository.delete(lifecycle.key, timeUs, commit.rev)) { if (await currentDependencies().referenceRepository.delete(lifecycle.key, timeUs, commit.rev, sql)) { trackPriorReferences(lifecycle.key, false) } } else { await writePriorReferences(lifecycle.key, lifecycle.newReferences, timeUs, commit.rev) await writePriorReferences(lifecycle.key, lifecycle.newReferences, timeUs, commit.rev, sql) }}
async function clearPriorReferenceState(event: JetstreamEvent): Promise<void> {async function clearPriorReferenceState(event: JetstreamEvent, sql?: BatchSql): Promise<void> { if (event.kind !== 'commit' || !event.commit) return await writePriorReferences( eventAtUri(event.did, event.commit.collection, event.commit.rkey), [], event.time_us, event.commit.rev, sql, )}
async function processCommit(stream: StreamName, event: JetstreamEvent): Promise<void> {async function processCommit(stream: StreamName, event: JetstreamEvent, sql?: BatchSql): Promise<void> { if (event.kind !== 'commit' || !event.commit) return const { did, commit } = event if (commit.operation !== 'create' && commit.operation !== 'update' && commit.operation !== 'delete') return
24 unmodified lines selfKey = mutation.selfKey ackableInvalid = mutation.ackableInvalid === true } if (ackableInvalid) { await clearPriorReferenceState(event) await clearPriorReferenceState(event, sql) totalEvents++ return }
const lifecycle = await loadReferenceLifecycle(event, deliveryRecord) const lifecycle = await loadReferenceLifecycle(event, deliveryRecord, sql) if (lifecycle?.tooComplex) { tooComplexBacklinkEvents++ logger.warn('[backlink] record references too complex; backlink matching skipped') // Passing an empty backlink snapshot preserves direct matching while ensuring // no truncated backlink result is ever partially delivered or indexed. await enqueueMatched(event, [], deliveryRecord, selfKey) await commitReferenceLifecycle(event, lifecycle) await enqueueMatched(event, [], deliveryRecord, selfKey, sql) await commitReferenceLifecycle(event, lifecycle, sql) totalEvents++ return } // A full-backlink relay event with neither prior nor current relevant reference4 unmodified lines lifecycle && lifecycle.oldReferences.length === 0 && lifecycle.newReferences.length === 0 ) { await commitReferenceLifecycle(event, lifecycle) await commitReferenceLifecycle(event, lifecycle, sql) return }
await enqueueMatched(event, lifecycle?.matchRecord ?? deliveryRecord, deliveryRecord, selfKey) await commitReferenceLifecycle(event, lifecycle) await enqueueMatched(event, lifecycle?.matchRecord ?? deliveryRecord, deliveryRecord, selfKey, sql) await commitReferenceLifecycle(event, lifecycle, sql) totalEvents++}
async function handleStreamEvent(stream: StreamName, event: JetstreamEvent): Promise<void> {/** * One relay batch is one durable transaction: deliveries, backlink state, and * the stream cursor commit together. A crash therefore replays the batch rather * than resuming past effects the outbox never received. * * Registry record mutations keep their own short transactions. They are rare * and monotonic by source revision, and they take table-wide advisory locks * that do not belong inside a relay-paced batch. */async function handleStreamEvents(stream: StreamName, events: readonly JetstreamEvent[]): Promise<void> { const last = events[events.length - 1] if (!last) return lastEventTime = Date.now() if (event.kind !== 'commit' || !event.commit) return const key = `${event.did}/${event.commit.collection}/${event.commit.rkey}` const relay = normalizeRelayIdentity(config.jetstreamUrl) const dependencies = currentDependencies() try { await dependencies.runIntakeBatch(async (sql) => { for (const event of events) { if (event.kind !== 'commit' || !event.commit) continue const key = `${event.did}/${event.commit.collection}/${event.commit.rkey}` await recordExecutor.run(key, () => processCommit(stream, event)) await recordExecutor.run(key, () => processCommit(stream, event, sql)) } await dependencies.cursorRepository.save(stream, last.time_us, relay, sql) }) } catch { safeError(stream) throw new Error('Intake event processing failed') }}
async function persistCursor(stream: StreamName, event: JetstreamEvent): Promise<void> { const relay = normalizeRelayIdentity(config.jetstreamUrl) const previous = cursorWriteTails.get(stream) ?? Promise.resolve() const current = previous .catch(() => undefined) .then(() => currentDependencies().cursorRepository.save(stream, event.time_us, relay)) cursorWriteTails.set(stream, current) try { await current } finally { if (cursorWriteTails.get(stream) === current) cursorWriteTails.delete(stream) }}
function createClient( stream: StreamName, cursor: number | undefined, wantedDids?: readonly string[],4 unmodified lines ...(wantedDids ? { wantedDids } : {}), ...(wantedCollections ? { wantedCollections } : {}), cursor, maxQueue: config.intakeQueueMax, concurrency: config.intakeConcurrency, batchMax: config.intakeBatchMax, maxEventBytes: config.intakeEventMaxBytes, reconnectMinMs: config.jetstreamReconnectMinMs, reconnectMaxMs: config.jetstreamReconnectMaxMs, reconnectMaxExponent: config.jetstreamReconnectMaxExponent, onEvent: (event) => handleStreamEvent(stream, event), onAcknowledged: (event) => persistCursor(stream, event), onEvents: (events) => handleStreamEvents(stream, events), onConnect: () => logger.info(`[${stream}] Jetstream connected`), onDisconnect: () => logger.warn(`[${stream}] Jetstream disconnected`), onError: () => safeError(stream), })191 unmodified linesExpand all cursorRepository: options.cursorRepository ?? defaultCursorRepository(), referenceRepository: options.referenceRepository ?? defaultReferenceRepository(), enqueueWebhookDeliveries: options.enqueueWebhookDeliveries ?? defaultEnqueueWebhookDeliveries(), recordWebhookIntakeQuarantine: options.recordWebhookIntakeQuarantine ?? database.recordWebhookIntakeQuarantine, runIntakeBatch: options.runIntakeBatch ?? database.runIntakeBatch, createJetstreamClient: options.createJetstreamClient ?? ((clientOptions) => new JetstreamClient(clientOptions)), } firehoseStarted = true try {163 unmodified linesExpand all backlinkScopeDids = new Set<string>() directWantedDids = new Set<string>() directSubscriptionUrlBytes = Buffer.byteLength(new URL(config.jetstreamUrl).toString()) clearRegistryCache() cursorWriteTails.clear()}59 unmodified lines60616263646564656667686869707171727374757677757677787980811 unmodified line8384858687878889909191929395949597969799989910010210110210310610710810911011110410510610710810911011111211311412011511611711812411912012112712812212312412512612712813412913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016114414516216316416516616712 unmodified lines1801811821831661841851861871881891901911921751931941951781961971981821831841992002012022032042051891901912062072082093 unmodified lines21321421521620321721821920622022122222322422521322622722821721821922022923023123223327 unmodified lines2612622632642562652662672602612622682642692702712722732742752762722732742752762772782792802782792802812822832848 unmodified lines2932942952962972982973002982993003014 unmodified lines30630730830931331031131531631731831231331432231531631732632731831932032132232933033132332432532632732832933034033134234334433234533334734834935035135233433533633733833934034136036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540634234334434534634734841434935035135235316 unmodified lines59 unmodified lines globalThis.WebSocket = realWebSocket FakeWebSocket.instances = []})
describe('JetstreamClient durable acknowledgement and protocol quarantine', () => { test('does not advance cursor until handler and durable acknowledgement resolve', async () => {describe('JetstreamClient durable batch acknowledgement and protocol quarantine', () => { test('does not advance the cursor until the batch commits', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket let release: (() => void) | undefined const processed = new Promise<void>((resolve) => { const committed = new Promise<void>((resolve) => { release = resolve }) const acknowledged: number[] = [] const batches: number[][] = [] const client = new JetstreamClient({ url: 'wss://relay.example/subscribe', cursor: 10, onEvent: () => processed, onAcknowledged: async (value) => { acknowledged.push(value.time_us) onEvents: (events) => { batches.push(events.map((value) => value.time_us)) return committed }, }) client.start() const socket = FakeWebSocket.instances[0]1 unmodified line socket?.open() socket?.message(event(11)) await flush() expect(client.cursor).toBe(10) expect(acknowledged).toEqual([]) release?.() await flush() expect(batches).toEqual([[11]]) expect(client.cursor).toBe(11) expect(acknowledged).toEqual([11]) client.destroy() })
test('acknowledges the longest completed prefix with one durable cursor write', async () => { test('hands the queued events to one batch and advances to its last event', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket const acknowledged: number[] = [] const batches: number[][] = [] let release: (() => void) | undefined const firstWrite = new Promise<void>((resolve) => { const firstBatch = 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 onEvents: (events) => { batches.push(events.map((value) => value.time_us)) return batches.length === 1 ? firstBatch : Promise.resolve() }, }) 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. // These arrive while the first batch is still committing. socket?.message(event(52)) socket?.message(event(53)) await flush() expect(acknowledged).toEqual([51]) expect(batches).toEqual([[51]]) release?.() await flush() // 52 and 53 are one contiguous completed prefix: one write, latest cursor. expect(acknowledged).toEqual([51, 53]) // One transaction, one cursor write, both remaining events. expect(batches).toEqual([[51], [52, 53]]) expect(client.cursor).toBe(53) expect(client.queued).toBe(0) client.destroy() })
test('stays healthy while a bounded queue drains, and stops when acknowledgement stalls', async () => { test('never hands over more than the batch bound at once', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket const sizes: number[] = [] const client = new JetstreamClient({ url: 'wss://relay.example/subscribe', cursor: 80, batchMax: 2, maxQueue: 8, onEvents: async (events) => { sizes.push(events.length) }, }) client.start() const socket = FakeWebSocket.instances[0] expect(socket).toBeDefined() socket?.open() for (const time of [81, 82, 83, 84, 85]) socket?.message(event(time)) await flush() expect(Math.max(...sizes)).toBeLessThanOrEqual(2) expect(client.cursor).toBe(85) client.destroy() })
test('stays healthy while a bounded queue drains, and stops when progress 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, batchMax: 1, onEvents: () => held, progressStaleMs: 60_000, }) client.start() const socket = FakeWebSocket.instances[0]12 unmodified lines expect(client.isProgressing).toBe(true)
const stalled = new JetstreamClient({ url: 'wss://relay.example/subscribe', onEvent: async () => undefined, onEvents: async () => undefined, progressStaleMs: 1, }) // Never acknowledged anything and not connected: not serving. expect(stalled.isProgressing).toBe(false) client.destroy() stalled.destroy() })
test('retries a failed event in place instead of discarding accepted work', async () => { test('retries a failed batch in place instead of replaying the stream', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket let failures = 0 const acknowledged: number[] = [] const client = new JetstreamClient({ url: 'wss://relay.example/subscribe', cursor: 70, eventRetryDelayMs: 1, onEvent: async (value) => { if (value.time_us === 71 && failures === 0) { batchRetryDelayMs: 1, onEvents: async () => { if (failures === 0) { failures++ throw new Error('transient') } }, onAcknowledged: async (value) => { acknowledged.push(value.time_us) }, }) client.start() const socket = FakeWebSocket.instances[0] expect(socket).toBeDefined()3 unmodified lines await flush() await Bun.sleep(10) await flush() expect(failures).toBe(1) // The failure never reached the stream: no replay connection, no lost sibling. // The failure never reached the stream: no replay connection, nothing dropped. expect(FakeWebSocket.instances.length).toBe(1) expect(client.cursor).toBe(72) expect(acknowledged).toEqual([72]) 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({ url: 'wss://relay.example/subscribe', cursor: 40, onEvent: async () => undefined, onAcknowledged: async (value) => { acknowledgements.push(value.time_us) }, onEvents: async () => undefined, reconnectMinMs: 1, reconnectMaxMs: 1, reconnectMaxExponent: 0, })27 unmodified lines
test('retains a full bounded queue, drains it in order, then reconnects from the latest ack', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket let releaseFirst: (() => void) | undefined let releaseSecond: (() => void) | undefined const first = new Promise<void>((resolve) => { releaseFirst = resolve }) const second = new Promise<void>((resolve) => { releaseSecond = resolve }) const processed: number[] = [] const acknowledged: number[] = [] const client = new JetstreamClient({ url: 'wss://relay.example/subscribe', cursor: 10, maxQueue: 2, batchMax: 1, reconnectMinMs: 1, reconnectMaxMs: 1, reconnectMaxExponent: 0, onEvent: async (value) => { const time = value.time_us processed.push(time) if (time === 11) await first if (time === 12) await second onEvents: async (events) => { for (const value of events) processed.push(value.time_us) if (events[0]?.time_us === 11) await first }, onAcknowledged: async (value) => { acknowledged.push(value.time_us) }, }) client.start() const socket = FakeWebSocket.instances[0] socket?.open()8 unmodified lines // from the durable cursor on the replacement connection. socket?.message(event(13)) releaseFirst?.() await flush() releaseSecond?.() await flush() expect(processed).toEqual([11, 12]) expect(acknowledged).toEqual([11, 12]) expect(client.cursor).toBe(12) await Bun.sleep(5) expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2) const replay = FakeWebSocket.instances[FakeWebSocket.instances.length - 1]4 unmodified lines expect(processed).toEqual([11, 12, 13]) client.destroy() })
test('ignores late frames after retained work fails, then reconnects after asynchronous close', async () => { test('replays from the last durable batch when a batch keeps failing', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket let release: (() => void) | undefined const pending = new Promise<void>((resolve) => { release = resolve }) const client = new JetstreamClient({ url: 'wss://relay.example/subscribe', cursor: 20, maxQueue: 2, reconnectMinMs: 1, reconnectMaxMs: 1, reconnectMaxExponent: 0, onEvent: async (value) => { if (value.time_us === 21) await pending // This exercises stream-level failure, so give the batch no retries. batchMaxAttempts: 1, onEvents: async () => { throw new Error('durable batch failed') }, onAcknowledged: async () => { throw new Error('cursor store unavailable') }, }) client.start() const socket = FakeWebSocket.instances[0] expect(socket).toBeDefined() if (!socket) throw new Error('missing test socket') socket.closeImmediately = false socket.open() socket.message(event(21)) socket.message(event(22)) await flush() release?.() await flush() expect(client.failureKind).toBe('cursor') expect(client.failureKind).toBe('handler') expect(client.isConnected).toBe(false) expect(client.queued).toBe(0) expect(socket.closeRequested).toBe(true) // The peer can still deliver buffered frames before its close callback. // A failed client must ignore them or it wedges on an undrainable queue. socket.message(event(23)) await flush() expect(client.queued).toBe(0) expect(client.cursor).toBe(20) socket.finishClose() await Bun.sleep(5) expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2) expect(FakeWebSocket.instances[FakeWebSocket.instances.length - 1]?.url).toContain('cursor=20') client.destroy() })
test('waits for sibling handlers before replaying a failure after disconnect', async () => { globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket let rejectFirst: ((error: Error) => void) | undefined let releaseSecond: (() => void) | undefined const first = new Promise<void>((_resolve, reject) => { rejectFirst = reject }) const second = new Promise<void>((resolve) => { releaseSecond = resolve }) const client = new JetstreamClient({ url: 'wss://relay.example/subscribe', cursor: 20, reconnectMinMs: 1, reconnectMaxMs: 1, reconnectMaxExponent: 0, // This exercises stream-level failure, so give the event no retries. eventMaxAttempts: 1, onEvent: async (value) => { if (value.time_us === 21) await first if (value.time_us === 22) await second }, onAcknowledged: async () => undefined, }) client.start() const socket = FakeWebSocket.instances[0] expect(socket).toBeDefined() if (!socket) throw new Error('missing test socket') socket.open() socket.message(event(21)) socket.message(event(22)) await flush() // The relay disconnects while both handlers are still active. socket.finishClose() rejectFirst?.(new Error('handler failed')) await flush() expect(client.failureKind).toBe('handler') expect(client.queued).toBe(0) expect(FakeWebSocket.instances).toHaveLength(1) releaseSecond?.() await flush() await Bun.sleep(5) expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2) expect(FakeWebSocket.instances[FakeWebSocket.instances.length - 1]?.url).toContain('cursor=20') client.destroy() })
test('rejects malformed relay fields and caller query injection', () => { expect(parseJetstreamEvent({ did: DID, time_us: 1, kind: 'commit', commit: { rev: 'not-a-tid' } })).toBeNull() expect(parseJetstreamEvent({ did: DID, time_us: 1.5, kind: 'identity' })).toBeNull() expect( () => new JetstreamClient({ url: 'wss://relay.example/subscribe?wantedDids=did:plc:evil', onEvent: async () => undefined, onEvents: async () => undefined, }), ).toThrow('Unsafe Jetstream relay URL') })
16 unmodified lines