/**
* Ordered relay cursor accounting primitives.
* Kept independent from the runtime singleton so they are deterministic and testable.
*/
/** A cursor reservation that is completed only after its site work is safe. */
export interface CursorReservation {
readonly seq: number
complete(): void
fail(): void
}
interface PendingCursorGroup {
seq: number
remaining: number
failed: boolean
sealed: boolean
generation: number
}
interface CursorWaiter {
seq: number
resolve: (reservation: CursorReservation | undefined) => void
}
export function isValidFirehoseSeq(seq: unknown): seq is number {
return typeof seq === 'number' && Number.isSafeInteger(seq) && seq >= 0
}
/**
* Tracks received firehose sequences in arrival order.
*
* A relay sequence can contain several commit operations. They share one
* bounded queue slot and count down together, so a small queue cannot deadlock
* while a single commit contains several operations. A group is sealed by the
* next sequence or source shutdown before it becomes checkpointable.
*/
export class OrderedCursorTracker {
private readonly pending: PendingCursorGroup[] = []
private readonly waiters: CursorWaiter[] = []
private accepting = true
private generation = 0
private confirmedCursor: number | undefined
private openGroup: PendingCursorGroup | null = null
constructor(
private readonly maxPendingEvents: number,
initialCursor?: number,
) {
if (!Number.isSafeInteger(maxPendingEvents) || maxPendingEvents < 1) {
throw new Error('maxPendingEvents must be a positive safe integer')
}
this.confirmedCursor = isValidFirehoseSeq(initialCursor) ? initialCursor : undefined
}
/** The newest fully completed and sealed relay sequence. */
get cursor(): number | undefined {
return this.confirmedCursor
}
/**
* The cursor safe to persist or use for reconnecting. If the first received
* event has not completed yet, use the sequence immediately before it so it
* is replayed rather than lost.
*/
get resumableCursor(): number | undefined {
if (this.confirmedCursor !== undefined) return this.confirmedCursor
const first = this.pending[0]?.seq
return first !== undefined && first > 0 ? first - 1 : undefined
}
/** Number of bounded relay-sequence groups awaiting a checkpoint. */
get pendingCount(): number {
return this.pending.length
}
get isAccepting(): boolean {
return this.accepting
}
/**
* Reserve capacity before accepting an event. Callers await this when the
* bounded queue is full, which lets Firehose apply natural backpressure.
* Invalid, duplicate, and rewound sequences are ignored and never regress a
* confirmed cursor.
*/
reserve(seq: number): Promise<CursorReservation | undefined> {
if (!this.accepting || !isValidFirehoseSeq(seq) || this.isStaleOrRewound(seq)) {
return Promise.resolve(undefined)
}
const reservation = this.admit(seq)
if (reservation !== null) return Promise.resolve(reservation)
return new Promise((resolve) => {
this.waiters.push({ seq, resolve })
})
}
/** Stop admitting waiting events and make them return without scheduling work. */
stopAccepting(): void {
this.accepting = false
for (const waiter of this.waiters.splice(0)) waiter.resolve(undefined)
}
/** Seal the trailing relay sequence after the source has been destroyed. */
sealOpenSequence(): void {
if (this.openGroup) this.openGroup.sealed = true
this.openGroup = null
this.advanceCursor()
}
/** Reset stale pending work before reconnecting from a known safe cursor. */
reset(initialCursor?: number): void {
this.generation++
this.pending.length = 0
this.openGroup = null
this.confirmedCursor = isValidFirehoseSeq(initialCursor) ? initialCursor : undefined
this.accepting = true
for (const waiter of this.waiters.splice(0)) waiter.resolve(undefined)
}
private isStaleOrRewound(seq: number): boolean {
if (this.confirmedCursor !== undefined && seq <= this.confirmedCursor) return true
const newest = this.pending[this.pending.length - 1]
if (newest && seq < newest.seq) return true
// A sequence which was already sealed cannot be extended later without
// risking a checkpoint regression. Relay delivery is ordered, so it is a
// duplicate/replay and can be safely ignored.
if (newest && newest.seq === seq && this.openGroup !== newest) return true
return false
}
/**
* Returns null only when a new sequence must wait for bounded capacity.
* Callers must validate staleness before this method.
*/
private admit(seq: number, releaseWaiters = true): CursorReservation | undefined | null {
if (this.openGroup && this.openGroup.seq === seq) {
return this.createReservation(this.openGroup)
}
// Seal and advance before evaluating capacity. This is essential at a
// capacity of one: an already-completed trailing sequence must make room
// for the next sequence rather than waiting forever to be sealed.
if (this.openGroup) {
this.openGroup.sealed = true
this.openGroup = null
this.advanceCursor(false)
}
if (this.pending.length >= this.maxPendingEvents) return null
const group: PendingCursorGroup = {
seq,
remaining: 0,
failed: false,
sealed: false,
generation: this.generation,
}
this.pending.push(group)
this.openGroup = group
const reservation = this.createReservation(group)
if (releaseWaiters) this.releaseWaiters()
return reservation
}
private createReservation(group: PendingCursorGroup): CursorReservation {
group.remaining++
return {
seq: group.seq,
complete: () => {
if (group.generation !== this.generation || group.failed || group.remaining < 1) return
group.remaining--
this.advanceCursor()
},
fail: () => {
if (group.generation !== this.generation || group.failed || group.remaining < 1) return
group.remaining--
group.failed = true
},
}
}
private advanceCursor(releaseWaiters = true): void {
while (this.pending.length > 0) {
const first = this.pending[0]
if (!first || first.failed || first.remaining > 0 || !first.sealed) break
// Validation above rejects rewound input. Keep this defensive check so a
// duplicate can never move a confirmed checkpoint backward.
if (this.confirmedCursor === undefined || first.seq > this.confirmedCursor) {
this.confirmedCursor = first.seq
}
this.pending.shift()
}
if (releaseWaiters) this.releaseWaiters()
}
private releaseWaiters(): void {
while (this.accepting && this.waiters.length > 0) {
const waiter = this.waiters[0]
if (!waiter) return
if (!isValidFirehoseSeq(waiter.seq) || this.isStaleOrRewound(waiter.seq)) {
this.waiters.shift()?.resolve(undefined)
continue
}
const reservation = this.admit(waiter.seq, false)
if (reservation === null) return
this.waiters.shift()?.resolve(reservation)
}
}
}