| 1 | import { asArray } from "./array"; |
| 2 | import { app } from "./bridge"; |
| 3 | import { recordFrontendDiagnostic } from "./frontendDiagnosticBridge"; |
| 4 | import type { TurnEventEnvelope, TurnEventReplayView, WireEvent } from "./types"; |
| 5 | |
| 6 | type WireHandler = (event: WireEvent) => void; |
| 7 | type ResetHandler = (tabId: string, replay: TurnEventReplayView) => Promise<boolean>; |
| 8 | |
| 9 | const MAX_REPLAY_PAGES = 32; |
| 10 | const MAX_QUEUED_EVENTS = 1024; |
| 11 | const MAX_QUEUED_BYTES = 8 * 1024 * 1024; |
| 12 | |
| 13 | export interface TurnEventTransport { |
| 14 | replay(tabId: string, afterSeq: number, identity?: TranscriptIdentity): Promise<TurnEventReplayView>; |
| 15 | } |
| 16 | |
| 17 | export interface TranscriptIdentity { |
| 18 | sessionId: string; |
| 19 | headId: string; |
| 20 | rewriteEpoch: number; |
| 21 | runtimeEpoch: string; |
| 22 | } |
| 23 | |
| 24 | export interface TranscriptSnapshotBoundary { |
| 25 | protocolVersion: 1; |
| 26 | snapshotId: string; |
| 27 | identity: TranscriptIdentity; |
| 28 | projectionRevision: number; |
| 29 | coveredThroughSeq: number; |
| 30 | } |
| 31 | |
| 32 | export interface SnapshotRequest { |
| 33 | readonly tabId: string; |
| 34 | readonly generation: number; |
| 35 | readonly identity?: Readonly<TranscriptIdentity>; |
| 36 | } |
| 37 | |
| 38 | function sameIdentity(a: Readonly<TranscriptIdentity>, b: Readonly<TranscriptIdentity>): boolean { |
| 39 | return a.sessionId === b.sessionId && a.headId === b.headId && |
| 40 | a.rewriteEpoch === b.rewriteEpoch && a.runtimeEpoch === b.runtimeEpoch; |
| 41 | } |
| 42 | |
| 43 | // TurnEventProjector is the per-tab ordered projection boundary. While a gap or |
| 44 | // checkpoint reset is being repaired, live events are held and applied only |
| 45 | // after the durable page and transcript prefix agree. |
| 46 | export class TurnEventProjector { |
| 47 | private readonly sequenceByTab = new Map<string, number>(); |
| 48 | private readonly repairByTab = new Map<string, Promise<void>>(); |
| 49 | private readonly pendingRepairByTab = new Map<string, { afterSeq: number; runtimeEpoch?: string }>(); |
| 50 | private readonly gapQueueByTab = new Map<string, WireEvent[]>(); |
| 51 | private readonly queuedBytesByTab = new Map<string, number>(); |
| 52 | private readonly receivedThroughByTab = new Map<string, number>(); |
| 53 | private readonly receivedByEpoch = new Map<string, Map<string, number>>(); |
| 54 | private readonly epochByTab = new Map<string, string>(); |
| 55 | private readonly generationByTab = new Map<string, number>(); |
| 56 | private handler?: WireHandler; |
| 57 | private resetHandler?: ResetHandler; |
| 58 | private readonly snapshotRequestByTab = new Map<string, SnapshotRequest>(); |
| 59 | private readonly snapshotBoundaryByTab = new Map<string, TranscriptSnapshotBoundary>(); |
| 60 | |
| 61 | constructor(private readonly transport?: TurnEventTransport) {} |
| 62 | |
| 63 | bind(handler: WireHandler) { this.handler = handler; } |
| 64 | unbind(handler: WireHandler) { if (this.handler === handler) this.handler = undefined; } |
| 65 | bindReset(handler: ResetHandler) { this.resetHandler = handler; } |
| 66 | unbindReset(handler: ResetHandler) { if (this.resetHandler === handler) this.resetHandler = undefined; } |
| 67 | |
| 68 | release(tabId: string) { |
| 69 | this.generationByTab.set(tabId, (this.generationByTab.get(tabId) ?? 0) + 1); |
| 70 | this.sequenceByTab.delete(tabId); |
| 71 | this.gapQueueByTab.delete(tabId); |
| 72 | this.queuedBytesByTab.delete(tabId); |
| 73 | this.receivedThroughByTab.delete(tabId); |
| 74 | this.receivedByEpoch.delete(tabId); |
| 75 | this.epochByTab.delete(tabId); |
| 76 | this.pendingRepairByTab.delete(tabId); |
| 77 | this.repairByTab.delete(tabId); |
| 78 | this.snapshotRequestByTab.delete(tabId); |
| 79 | this.snapshotBoundaryByTab.delete(tabId); |
| 80 | } |
| 81 | |
| 82 | /** Suspend admission before issuing a snapshot request. The immutable lease |
| 83 | * fences both its response and the live suffix to one session generation. */ |
| 84 | beginSnapshot(tabId: string, identity?: TranscriptIdentity): SnapshotRequest { |
| 85 | const previous = this.snapshotBoundaryByTab.get(tabId)?.identity ?? this.snapshotRequestByTab.get(tabId)?.identity; |
| 86 | if (previous && identity && !sameIdentity(previous, identity)) this.release(tabId); |
| 87 | const generation = (this.generationByTab.get(tabId) ?? 0) + 1; |
| 88 | this.generationByTab.set(tabId, generation); |
| 89 | this.pendingRepairByTab.delete(tabId); |
| 90 | this.repairByTab.delete(tabId); |
| 91 | if (identity) this.epochByTab.set(tabId, identity.runtimeEpoch); |
| 92 | const request = Object.freeze({ tabId, generation, identity: identity ? Object.freeze({ ...identity }) : undefined }); |
| 93 | this.snapshotRequestByTab.set(tabId, request); |
| 94 | return request; |
| 95 | } |
| 96 | |
| 97 | abortSnapshot(request: SnapshotRequest) { |
| 98 | if (this.snapshotRequestByTab.get(request.tabId) !== request) return; |
| 99 | // With no installed prefix, keep admission suspended until an explicit |
| 100 | // retry supplies one. A failed fetch is not a new empty transcript. |
| 101 | if (!this.snapshotBoundaryByTab.has(request.tabId)) return; |
| 102 | this.snapshotRequestByTab.delete(request.tabId); |
| 103 | } |
| 104 | |
| 105 | /** The caller commits rows and runtime in one synchronous store transaction. |
| 106 | * Coverage changes only after that transaction returns successfully. */ |
| 107 | installSnapshot(request: SnapshotRequest, snapshot: TranscriptSnapshotBoundary, commit: () => void): boolean { |
| 108 | const { tabId, generation } = request; |
| 109 | if (this.snapshotRequestByTab.get(tabId) !== request || this.generationByTab.get(tabId) !== generation) return false; |
| 110 | if (snapshot.protocolVersion !== 1 || !snapshot.snapshotId || (request.identity && !sameIdentity(request.identity, snapshot.identity)) || |
| 111 | !Number.isSafeInteger(snapshot.coveredThroughSeq) || snapshot.coveredThroughSeq < 0 || |
| 112 | !Number.isSafeInteger(snapshot.projectionRevision) || snapshot.projectionRevision < 0) { |
| 113 | throw new Error("invalid transcript snapshot boundary"); |
| 114 | } |
| 115 | const previous = this.snapshotBoundaryByTab.get(tabId); |
| 116 | if (previous && sameIdentity(previous.identity, snapshot.identity) && |
| 117 | (snapshot.coveredThroughSeq < (this.sequenceByTab.get(tabId) ?? 0) || snapshot.projectionRevision < previous.projectionRevision)) { |
| 118 | throw new Error("transcript snapshot would roll back committed coverage"); |
| 119 | } |
| 120 | commit(); |
| 121 | if (this.snapshotRequestByTab.get(tabId) !== request || this.generationByTab.get(tabId) !== generation) return false; |
| 122 | this.snapshotBoundaryByTab.set(tabId, { ...snapshot, identity: { ...snapshot.identity } }); |
| 123 | this.epochByTab.set(tabId, snapshot.identity.runtimeEpoch); |
| 124 | this.sequenceByTab.set(tabId, snapshot.coveredThroughSeq); |
| 125 | this.snapshotRequestByTab.delete(tabId); |
| 126 | const remaining = asArray(this.gapQueueByTab.get(tabId)).filter((event) => |
| 127 | (event.seq ?? 0) > snapshot.coveredThroughSeq && |
| 128 | (!event.runtimeEpoch || event.runtimeEpoch === snapshot.identity.runtimeEpoch) && |
| 129 | (!event.sessionId || event.sessionId === snapshot.identity.sessionId)); |
| 130 | this.gapQueueByTab.set(tabId, remaining); |
| 131 | this.queuedBytesByTab.set(tabId, remaining.reduce((sum, event) => sum + JSON.stringify(event).length * 2, 0)); |
| 132 | this.receivedThroughByTab.set(tabId, Math.max(snapshot.coveredThroughSeq, |
| 133 | this.receivedByEpoch.get(tabId)?.get(snapshot.identity.runtimeEpoch) ?? 0, |
| 134 | ...remaining.map((event) => event.seq ?? 0))); |
| 135 | if ((this.receivedThroughByTab.get(tabId) ?? 0) > snapshot.coveredThroughSeq) { |
| 136 | this.requestReplay(tabId, snapshot.coveredThroughSeq, snapshot.identity.runtimeEpoch); |
| 137 | } |
| 138 | return true; |
| 139 | } |
| 140 | |
| 141 | snapshotBoundary(tabId: string): TranscriptSnapshotBoundary | undefined { |
| 142 | const boundary = this.snapshotBoundaryByTab.get(tabId); |
| 143 | return boundary ? { ...boundary, identity: { ...boundary.identity } } : undefined; |
| 144 | } |
| 145 | |
| 146 | refresh(tabId: string) { |
| 147 | if (!this.snapshotRequestByTab.has(tabId)) this.requestReplay(tabId, this.sequenceByTab.get(tabId) ?? 0, this.epochByTab.get(tabId)); |
| 148 | } |
| 149 | |
| 150 | observeRuntime(tabId: string, runtimeEpoch: string | undefined, latest: number, replayAfter: number | undefined, active: boolean) { |
| 151 | // Modern snapshots own identity and initial coverage. Status polling is |
| 152 | // only a high-water hint; it cannot install a different transcript prefix. |
| 153 | if (this.snapshotRequestByTab.has(tabId) || this.snapshotBoundaryByTab.has(tabId)) { |
| 154 | if (runtimeEpoch && runtimeEpoch !== this.epochByTab.get(tabId)) return; |
| 155 | this.receivedThroughByTab.set(tabId, Math.max(this.receivedThroughByTab.get(tabId) ?? 0, latest)); |
| 156 | const projected = this.sequenceByTab.get(tabId) ?? 0; |
| 157 | if (!this.snapshotRequestByTab.has(tabId) && latest > projected) this.requestReplay(tabId, projected, runtimeEpoch); |
| 158 | return; |
| 159 | } |
| 160 | if (runtimeEpoch && runtimeEpoch !== this.epochByTab.get(tabId)) { |
| 161 | this.generationByTab.set(tabId, (this.generationByTab.get(tabId) ?? 0) + 1); |
| 162 | this.epochByTab.set(tabId, runtimeEpoch); |
| 163 | this.sequenceByTab.delete(tabId); |
| 164 | this.gapQueueByTab.delete(tabId); |
| 165 | this.queuedBytesByTab.delete(tabId); |
| 166 | this.receivedThroughByTab.delete(tabId); |
| 167 | this.pendingRepairByTab.delete(tabId); |
| 168 | this.repairByTab.delete(tabId); |
| 169 | } |
| 170 | let projected = this.sequenceByTab.get(tabId); |
| 171 | if (projected === undefined) { |
| 172 | projected = active ? Math.min(replayAfter ?? latest, latest) : latest; |
| 173 | this.sequenceByTab.set(tabId, projected); |
| 174 | } |
| 175 | if (latest > projected) this.requestReplay(tabId, projected, runtimeEpoch); |
| 176 | } |
| 177 | |
| 178 | receiveLive(tabId: string, event: WireEvent, runtimeEpoch?: string): boolean { |
| 179 | const pendingSnapshot = this.snapshotRequestByTab.get(tabId); |
| 180 | const unboundSnapshot = pendingSnapshot && !pendingSnapshot.identity; |
| 181 | const epoch = this.epochByTab.get(tabId) ?? runtimeEpoch; |
| 182 | if (!unboundSnapshot && epoch && event.runtimeEpoch && event.runtimeEpoch !== epoch) return false; |
| 183 | if (typeof event.seq !== "number" || event.seq <= 0) { |
| 184 | if (!this.handler) throw new Error("turn event projection has no commit handler"); |
| 185 | this.handler({ ...event, tabId }); |
| 186 | return true; |
| 187 | } |
| 188 | if (!Number.isSafeInteger(event.seq)) throw new Error("invalid live event sequence"); |
| 189 | const last = this.sequenceByTab.get(tabId) ?? 0; |
| 190 | if (!unboundSnapshot && event.seq <= last) return false; |
| 191 | const eventEpoch = event.runtimeEpoch ?? runtimeEpoch ?? epoch ?? ""; |
| 192 | const watermarks = this.receivedByEpoch.get(tabId) ?? new Map<string, number>(); |
| 193 | watermarks.set(eventEpoch, Math.max(watermarks.get(eventEpoch) ?? 0, event.seq)); |
| 194 | if (watermarks.size > 8) watermarks.delete(watermarks.keys().next().value!); |
| 195 | this.receivedByEpoch.set(tabId, watermarks); |
| 196 | this.receivedThroughByTab.set(tabId, Math.max(this.receivedThroughByTab.get(tabId) ?? 0, event.seq)); |
| 197 | if (this.snapshotRequestByTab.has(tabId) || this.repairByTab.has(tabId) || event.seq > last + 1) { |
| 198 | const queued = this.gapQueueByTab.get(tabId) ?? []; |
| 199 | if (!queued.some((pending) => pending.seq === event.seq)) { |
| 200 | const bytes = JSON.stringify(event).length * 2; |
| 201 | const priorBytes = this.queuedBytesByTab.get(tabId) ?? 0; |
| 202 | if (queued.length >= MAX_QUEUED_EVENTS || priorBytes + bytes > MAX_QUEUED_BYTES) { |
| 203 | // The ledger owns the source of truth. Releasing this redundant |
| 204 | // buffer never advances coverage; replay will retrieve the suffix. |
| 205 | queued.length = 0; |
| 206 | this.queuedBytesByTab.set(tabId, 0); |
| 207 | recordFrontendDiagnostic("runtime", "turn-events-live-buffer-overflow", {}); |
| 208 | } |
| 209 | if (bytes <= MAX_QUEUED_BYTES) { |
| 210 | queued.push({ ...event, tabId }); |
| 211 | this.queuedBytesByTab.set(tabId, (this.queuedBytesByTab.get(tabId) ?? 0) + bytes); |
| 212 | } |
| 213 | } |
| 214 | this.gapQueueByTab.set(tabId, queued); |
| 215 | if (!this.snapshotRequestByTab.has(tabId) && !this.repairByTab.has(tabId)) { |
| 216 | this.requestReplay(tabId, last, event.runtimeEpoch ?? runtimeEpoch); |
| 217 | } |
| 218 | return false; |
| 219 | } |
| 220 | this.applyOrdered(tabId, event); |
| 221 | return true; |
| 222 | } |
| 223 | |
| 224 | /** The sole commit entry: callers must have established ordering first. */ |
| 225 | private applyOrdered(tabId: string, event: WireEvent) { |
| 226 | const generation = this.generationByTab.get(tabId) ?? 0; |
| 227 | if (!this.handler) throw new Error("turn event projection has no commit handler"); |
| 228 | this.handler({ ...event, tabId }); |
| 229 | if ((this.generationByTab.get(tabId) ?? 0) === generation && typeof event.seq === "number" && event.seq > 0) { |
| 230 | this.sequenceByTab.set(tabId, event.seq); |
| 231 | } |
| 232 | } |
| 233 | |
| 234 | private requestReplay(tabId: string, afterSeq: number, runtimeEpoch?: string) { |
| 235 | if (!this.transport && typeof app.TurnEventsForTab !== "function" && typeof app.TranscriptReplayForTab !== "function") return; |
| 236 | if (this.repairByTab.has(tabId)) { |
| 237 | this.pendingRepairByTab.set(tabId, { afterSeq, runtimeEpoch }); |
| 238 | return; |
| 239 | } |
| 240 | const generation = this.generationByTab.get(tabId) ?? 0; |
| 241 | const repair = this.replayGap(tabId, afterSeq, runtimeEpoch, generation) |
| 242 | .catch((error) => recordFrontendDiagnostic("runtime", "turn-events-gap-repair-failed", { |
| 243 | afterSeq: this.sequenceByTab.get(tabId) ?? afterSeq, |
| 244 | error: error instanceof Error ? error.message : String(error), |
| 245 | })) |
| 246 | .finally(() => { |
| 247 | if (this.repairByTab.get(tabId) !== repair) return; |
| 248 | this.repairByTab.delete(tabId); |
| 249 | const pending = this.pendingRepairByTab.get(tabId); |
| 250 | if (!pending) return; |
| 251 | this.pendingRepairByTab.delete(tabId); |
| 252 | this.requestReplay(tabId, pending.afterSeq, pending.runtimeEpoch); |
| 253 | }); |
| 254 | this.repairByTab.set(tabId, repair); |
| 255 | } |
| 256 | |
| 257 | private async replayGap(tabId: string, afterSeq: number, requestedEpoch: string | undefined, generation: number) { |
| 258 | let cursor = afterSeq; |
| 259 | for (let page = 0; page < MAX_REPLAY_PAGES; page += 1) { |
| 260 | if ((this.generationByTab.get(tabId) ?? 0) !== generation) return; |
| 261 | const identity = this.snapshotBoundaryByTab.get(tabId)?.identity; |
| 262 | const replay = await (this.transport ? this.transport.replay(tabId, cursor, identity) : |
| 263 | identity && app.TranscriptReplayForTab ? app.TranscriptReplayForTab(tabId, { identity, after: cursor }) : app.TurnEventsForTab!(tabId, cursor)); |
| 264 | if ((this.generationByTab.get(tabId) ?? 0) !== generation) return; |
| 265 | const currentEpoch = this.epochByTab.get(tabId); |
| 266 | if ((requestedEpoch && currentEpoch && requestedEpoch !== currentEpoch) || |
| 267 | (!replay.resetRequired && replay.runtimeEpoch && currentEpoch && replay.runtimeEpoch !== currentEpoch)) { |
| 268 | return; |
| 269 | } |
| 270 | |
| 271 | if (replay.resetRequired) { |
| 272 | const modern = this.snapshotBoundaryByTab.has(tabId); |
| 273 | if (!this.resetHandler || !(await this.resetHandler(tabId, replay))) { |
| 274 | throw new Error("turn event checkpoint reset could not hydrate the transcript"); |
| 275 | } |
| 276 | if ((this.generationByTab.get(tabId) ?? 0) !== generation) return; |
| 277 | if (modern) throw new Error("transcript reset did not install an authoritative snapshot"); |
| 278 | cursor = Math.max(0, replay.floorSeq - 1); |
| 279 | this.sequenceByTab.set(tabId, cursor); |
| 280 | } |
| 281 | |
| 282 | const envelopes = asArray(replay.events).slice().sort((a, b) => a.seq - b.seq); |
| 283 | for (const envelope of envelopes) { |
| 284 | if (envelope.seq <= cursor) continue; |
| 285 | if (envelope.seq !== cursor + 1) throw new Error(`turn event replay gap after ${cursor}`); |
| 286 | this.projectEnvelope(tabId, envelope, requestedEpoch); |
| 287 | if ((this.generationByTab.get(tabId) ?? 0) !== generation) return; |
| 288 | cursor = envelope.seq; |
| 289 | } |
| 290 | if (replay.hasMore) { |
| 291 | const next = replay.nextAfterSeq; |
| 292 | if (next !== cursor) throw new Error(`turn event replay cursor mismatch: projected ${cursor}, backend ${next}`); |
| 293 | if (envelopes.length === 0) throw new Error("turn event replay made no progress"); |
| 294 | continue; |
| 295 | } |
| 296 | |
| 297 | const pending = asArray(this.gapQueueByTab.get(tabId)).slice().sort((a, b) => (a.seq ?? 0) - (b.seq ?? 0)); |
| 298 | for (const live of pending) { |
| 299 | if (typeof live.seq !== "number" || live.seq <= 0) { |
| 300 | this.applyOrdered(tabId, live); |
| 301 | continue; |
| 302 | } |
| 303 | if (live.seq <= cursor) continue; |
| 304 | if (live.seq !== cursor + 1) { |
| 305 | break; |
| 306 | } |
| 307 | this.applyOrdered(tabId, live); |
| 308 | if ((this.generationByTab.get(tabId) ?? 0) !== generation) return; |
| 309 | cursor = live.seq; |
| 310 | } |
| 311 | // Keep ownership of the queue until every successful commit is known. |
| 312 | // A throwing reducer must leave the uncommitted suffix available, and a |
| 313 | // reentrant arrival must not be overwritten by the page's older copy. |
| 314 | const remaining = asArray(this.gapQueueByTab.get(tabId)).filter((live) => (live.seq ?? 0) > cursor); |
| 315 | this.gapQueueByTab.set(tabId, remaining); |
| 316 | this.queuedBytesByTab.set(tabId, remaining.reduce((total, live) => total + JSON.stringify(live).length * 2, 0)); |
| 317 | if (remaining.length === 0 && cursor >= (this.receivedThroughByTab.get(tabId) ?? 0)) return; |
| 318 | } |
| 319 | recordFrontendDiagnostic("runtime", "turn-events-gap-repair-incomplete", { |
| 320 | afterSeq: this.sequenceByTab.get(tabId) ?? cursor, |
| 321 | }); |
| 322 | } |
| 323 | |
| 324 | private projectEnvelope(tabId: string, envelope: TurnEventEnvelope, runtimeEpoch?: string) { |
| 325 | const durable = envelope?.event; |
| 326 | if (!durable || !Number.isSafeInteger(envelope.seq) || envelope.seq <= 0) throw new Error("invalid replay envelope"); |
| 327 | const epoch = this.epochByTab.get(tabId) ?? runtimeEpoch; |
| 328 | if (envelope.runtimeEpoch && epoch && envelope.runtimeEpoch !== epoch) throw new Error("replay envelope belongs to another runtime"); |
| 329 | const identity = this.snapshotBoundaryByTab.get(tabId)?.identity; |
| 330 | if (identity && envelope.sessionId && envelope.sessionId !== identity.sessionId) throw new Error("replay envelope belongs to another session"); |
| 331 | this.applyOrdered(tabId, { |
| 332 | ...durable, |
| 333 | sessionId: envelope.sessionId || durable.sessionId, |
| 334 | submissionId: envelope.submissionId || durable.submissionId, |
| 335 | turnId: envelope.turnId || durable.turnId, |
| 336 | seq: envelope.seq, |
| 337 | status: (envelope.status || durable.status) as WireEvent["status"], |
| 338 | tabId, |
| 339 | runtimeEpoch: envelope.runtimeEpoch ?? runtimeEpoch, |
| 340 | }); |
| 341 | } |
| 342 | } |
| 343 |