| 1 | import type { TranscriptFollowResponse, FollowRequest } from "../generated/desktopContract.generated"; |
| 2 | import { addBreadcrumb } from "./breadcrumbs"; |
| 3 | import { |
| 4 | clearTranscriptDiagnostic, |
| 5 | newTranscriptDiagnosticOwner, |
| 6 | publishTranscriptDiagnostic, |
| 7 | type TranscriptDiagnostic, |
| 8 | type TranscriptDiagnosticReason, |
| 9 | type TranscriptDiagnosticStage, |
| 10 | } from "./transcriptDiagnostics"; |
| 11 | |
| 12 | export type TranscriptConnection = "syncing" | "connected" | "disconnected"; |
| 13 | type Change = NonNullable<TranscriptFollowResponse["changes"]>[number]; |
| 14 | type FollowGeneration = { closeSubscription: boolean }; |
| 15 | |
| 16 | export interface FollowConsumer { |
| 17 | install(response: TranscriptFollowResponse): Promise<void> | void; |
| 18 | changes(changes: Change[]): void; |
| 19 | connection(state: TranscriptConnection, error?: string): void; |
| 20 | } |
| 21 | |
| 22 | type TranscriptFollowFailureState = { |
| 23 | stage: TranscriptDiagnosticStage; |
| 24 | reason: TranscriptDiagnosticReason; |
| 25 | firstAt: number; |
| 26 | lastReportedAt: number; |
| 27 | failures: number; |
| 28 | errorType: TranscriptDiagnostic["errorType"]; |
| 29 | }; |
| 30 | |
| 31 | export interface TranscriptFollowClientOptions { |
| 32 | transport?: "local" | "remote"; |
| 33 | now?: () => number; |
| 34 | onDiagnostic?: (event: TranscriptDiagnostic, visible: boolean) => void; |
| 35 | } |
| 36 | |
| 37 | class TranscriptFollowFailure extends Error { |
| 38 | constructor( |
| 39 | readonly stage: TranscriptDiagnosticStage, |
| 40 | readonly reason: TranscriptDiagnosticReason, |
| 41 | readonly errorType: TranscriptDiagnostic["errorType"], |
| 42 | message: string, |
| 43 | ) { |
| 44 | super(message); |
| 45 | } |
| 46 | } |
| 47 | |
| 48 | function failure(stage: TranscriptDiagnosticStage, reason: TranscriptDiagnosticReason, message: string, cause?: unknown): TranscriptFollowFailure { |
| 49 | const errorType = cause instanceof Error ? "error" |
| 50 | : typeof cause === "string" ? "string" |
| 51 | : cause && typeof cause === "object" ? "object" |
| 52 | : cause === undefined ? "classified" : "unknown"; |
| 53 | return new TranscriptFollowFailure(stage, reason, errorType, message); |
| 54 | } |
| 55 | |
| 56 | function classifiedFailure(error: unknown, stage: TranscriptDiagnosticStage, reason: TranscriptDiagnosticReason): TranscriptFollowFailure { |
| 57 | if (error instanceof TranscriptFollowFailure) return error; |
| 58 | const message = error instanceof Error ? error.message : typeof error === "string" ? error : "unknown transcript failure"; |
| 59 | return failure(stage, reason, message, error); |
| 60 | } |
| 61 | |
| 62 | /** One ordered consumer for both transports. Connection failures only request |
| 63 | * another snapshot; this class has no model, submit, stop or retry-turn API. */ |
| 64 | export class TranscriptFollowClient { |
| 65 | private generation: FollowGeneration = { closeSubscription: true }; |
| 66 | private subscription = ""; |
| 67 | private revision = 0; |
| 68 | private coverage = 0; |
| 69 | private identity = ""; |
| 70 | private readonly indexes = new Map<string, number>(); |
| 71 | private readonly attemptMessages = new Map<string, string>(); |
| 72 | private readonly results = new Map<string, number>(); |
| 73 | private readonly diagnosticOwner = newTranscriptDiagnosticOwner(); |
| 74 | private readonly transport: "local" | "remote"; |
| 75 | private readonly now: () => number; |
| 76 | private readonly onDiagnostic?: (event: TranscriptDiagnostic, visible: boolean) => void; |
| 77 | private failureState: TranscriptFollowFailureState | null = null; |
| 78 | |
| 79 | constructor(private readonly read: (request: FollowRequest) => Promise<TranscriptFollowResponse>, options: TranscriptFollowClientOptions = {}) { |
| 80 | this.transport = options.transport ?? "local"; |
| 81 | this.now = options.now ?? (() => Date.now()); |
| 82 | this.onDiagnostic = options.onDiagnostic; |
| 83 | } |
| 84 | |
| 85 | async start(consumer: FollowConsumer): Promise<void> { |
| 86 | this.stop(); |
| 87 | const generation = this.generation; |
| 88 | consumer.connection("syncing"); |
| 89 | try { await this.baseline(generation, consumer); } |
| 90 | catch (error) { |
| 91 | if (generation === this.generation) { |
| 92 | consumer.connection("disconnected", String(error)); |
| 93 | this.noteFailure(classifiedFailure(error, "baseline_read", "unknown")); |
| 94 | } |
| 95 | throw error; |
| 96 | } |
| 97 | if (generation === this.generation) void this.follow(generation, consumer); |
| 98 | } |
| 99 | |
| 100 | stop(closeSubscription = true, reason?: "service_stopping"): void { |
| 101 | const generation = this.generation; |
| 102 | generation.closeSubscription = closeSubscription; |
| 103 | this.generation = { closeSubscription: true }; |
| 104 | const subscription = this.subscription; |
| 105 | this.subscription = ""; |
| 106 | this.closeSubscription(subscription, generation); |
| 107 | if (reason === "service_stopping") { |
| 108 | this.publish("stopped", this.failureState?.stage ?? "none", "service_stopping", true); |
| 109 | this.failureState = null; |
| 110 | } else if (!reason) { |
| 111 | clearTranscriptDiagnostic(this.diagnosticOwner); |
| 112 | this.failureState = null; |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | private closeSubscription(subscription: string | undefined, generation: FollowGeneration): void { |
| 117 | if (subscription && generation.closeSubscription) void this.read({ subscription, close: true }).catch(() => undefined); |
| 118 | } |
| 119 | |
| 120 | private async baseline(generation: FollowGeneration, consumer: FollowConsumer): Promise<void> { |
| 121 | let response: TranscriptFollowResponse; |
| 122 | try { |
| 123 | response = await this.read({}); |
| 124 | } catch (error) { |
| 125 | throw classifiedFailure(error, "baseline_read", "transport_rejected"); |
| 126 | } |
| 127 | if (generation !== this.generation) { |
| 128 | this.closeSubscription(response.subscription, generation); |
| 129 | return; |
| 130 | } |
| 131 | if (response.protocolVersion !== 2) { |
| 132 | this.closeSubscription(response.subscription, generation); |
| 133 | throw failure("baseline_validate", "protocol_version", "Transcript v2 is required. Upgrade Desktop and Serve together."); |
| 134 | } |
| 135 | if (!response.snapshot || !response.subscription) { |
| 136 | this.closeSubscription(response.subscription, generation); |
| 137 | throw failure("baseline_validate", "snapshot_missing", "Transcript v2 snapshot or subscription is missing."); |
| 138 | } |
| 139 | const snapshot = response.snapshot; |
| 140 | const identity = JSON.stringify(snapshot.identity); |
| 141 | if (identity === this.identity && snapshot.projectionRevision < this.revision) { |
| 142 | this.closeSubscription(response.subscription, generation); |
| 143 | throw failure("baseline_validate", "revision_regressed", "transcript snapshot revision regressed"); |
| 144 | } |
| 145 | if (response.history && response.history.status !== "ready") { |
| 146 | this.closeSubscription(response.subscription, generation); |
| 147 | throw failure("baseline_validate", "history_not_ready", `transcript history ${response.history.status}`); |
| 148 | } |
| 149 | try { await consumer.install(response); } catch (error) { |
| 150 | this.closeSubscription(response.subscription, generation); |
| 151 | throw classifiedFailure(error, "snapshot_install", "consumer_error"); |
| 152 | } |
| 153 | if (generation !== this.generation) { |
| 154 | this.closeSubscription(response.subscription, generation); |
| 155 | return; |
| 156 | } |
| 157 | this.identity = identity; |
| 158 | this.subscription = response.subscription; |
| 159 | this.revision = snapshot.projectionRevision; |
| 160 | this.coverage = snapshot.coveredThroughSeq; |
| 161 | this.indexes.clear(); |
| 162 | this.attemptMessages.clear(); |
| 163 | this.results.clear(); |
| 164 | for (const attempt of snapshot.activeAttempts ?? []) { |
| 165 | this.indexes.set(attempt.id, attempt.nextIndex ?? 0); |
| 166 | this.attemptMessages.set(attempt.id, attempt.messageId); |
| 167 | } |
| 168 | if (!this.failureState) addBreadcrumb("transcript.v2", `snapshot epoch=${snapshot.identity.runtimeEpoch} revision=${this.revision} commit=${this.coverage} durable=${snapshot.durableSeq} records=${snapshot.totalRecords} attempts=${this.indexes.size}`); |
| 169 | consumer.connection("connected"); |
| 170 | // Installing a replacement snapshot does not prove that delta following |
| 171 | // recovered. Keep this fault segment until a full follow succeeds. |
| 172 | } |
| 173 | |
| 174 | private validate(changes: Change[]): Change[] { |
| 175 | let revision = this.revision; |
| 176 | let coverage = this.coverage; |
| 177 | const indexes = new Map(this.indexes); |
| 178 | const attempts = new Map(this.attemptMessages); |
| 179 | const results = new Map(this.results); |
| 180 | const accepted: Change[] = []; |
| 181 | // Transport coalescing can reorder frames within a delivered batch. Their |
| 182 | // publisher revisions establish order; an actual missing revision resets. |
| 183 | for (const change of [...changes].sort((a, b) => a.revision - b.revision)) { |
| 184 | if (change.revision <= revision) continue; |
| 185 | if (change.resetRequired || change.revision !== revision + 1) throw failure("delta_validate", "revision_gap", "transcript revision gap"); |
| 186 | if (change.firstSeq) { |
| 187 | if (change.firstSeq !== coverage + 1 || change.commitSeq < change.firstSeq) throw failure("delta_validate", "business_gap", "transcript business gap"); |
| 188 | coverage = change.commitSeq; |
| 189 | } else if (change.commitSeq !== coverage) throw failure("delta_validate", "frame_cut_mismatch", "transcript frame cut mismatch"); |
| 190 | const event = change.event; |
| 191 | for (const record of change.records ?? []) if (record.messageId) results.set(record.messageId, change.commitSeq); |
| 192 | while (results.size > 192) results.delete(results.keys().next().value!); |
| 193 | if (event?.kind === "stream_attempt" && event.streamAttempt?.action === "begin") { |
| 194 | if (!event.messageId) throw failure("delta_validate", "sampling_identity_missing", "transcript sampling identity missing"); |
| 195 | indexes.set(event.streamAttempt.id, 0); |
| 196 | attempts.set(event.streamAttempt.id, event.messageId); |
| 197 | } |
| 198 | if (change.attemptId && !change.resultSeq && event?.kind !== "stream_attempt") { |
| 199 | if (indexes.get(change.attemptId) !== change.index) throw failure("delta_validate", "sampling_gap", "transcript sampling gap"); |
| 200 | indexes.set(change.attemptId, change.index + 1); |
| 201 | } |
| 202 | if (change.resultSeq && (change.resultSeq > coverage || !["message/complete", "message/interrupted"].includes(change.resultKind ?? ""))) throw failure("delta_validate", "settlement_not_committed", "transcript settlement is not committed"); |
| 203 | if (change.resultSeq) { |
| 204 | const message = attempts.get(change.attemptId ?? ""); |
| 205 | if (!message || message !== event?.messageId || (results.has(message) && results.get(message) !== change.resultSeq)) throw failure("delta_validate", "settlement_identity_mismatch", "transcript settlement identity mismatch"); |
| 206 | } |
| 207 | if (event?.kind === "stream_attempt" && event.streamAttempt?.action !== "begin") { |
| 208 | indexes.delete(event.streamAttempt?.id ?? ""); attempts.delete(event.streamAttempt?.id ?? ""); |
| 209 | } |
| 210 | revision = change.revision; |
| 211 | accepted.push(change); |
| 212 | } |
| 213 | this.revision = revision; |
| 214 | this.coverage = coverage; |
| 215 | this.indexes.clear(); |
| 216 | for (const [id, index] of indexes) this.indexes.set(id, index); |
| 217 | this.attemptMessages.clear(); for (const [id, message] of attempts) this.attemptMessages.set(id, message); |
| 218 | this.results.clear(); for (const [id, sequence] of results) this.results.set(id, sequence); |
| 219 | return accepted; |
| 220 | } |
| 221 | |
| 222 | private async follow(generation: FollowGeneration, consumer: FollowConsumer): Promise<void> { |
| 223 | while (generation === this.generation) { |
| 224 | try { |
| 225 | if (!this.subscription) await this.baseline(generation, consumer); |
| 226 | if (generation !== this.generation) return; |
| 227 | let response: TranscriptFollowResponse; |
| 228 | try { |
| 229 | response = await this.read({ subscription: this.subscription, afterRevision: this.revision }); |
| 230 | } catch (error) { |
| 231 | throw classifiedFailure(error, "delta_read", "transport_rejected"); |
| 232 | } |
| 233 | if (generation !== this.generation) return; |
| 234 | if (response.protocolVersion !== 2) throw failure("delta_validate", "protocol_version", "transcript protocol version changed"); |
| 235 | if (response.resetRequired) throw failure("delta_validate", "resync_required", "transcript requires resynchronization"); |
| 236 | const changes = this.validate(response.changes ?? []); |
| 237 | try { |
| 238 | consumer.changes(changes); |
| 239 | } catch (error) { |
| 240 | throw classifiedFailure(error, "delta_apply", "consumer_error"); |
| 241 | } |
| 242 | consumer.connection("connected"); |
| 243 | this.noteRecovery(); |
| 244 | } catch (error) { |
| 245 | if (generation !== this.generation) return; |
| 246 | const transcriptFailure = classifiedFailure(error, "delta_read", "unknown"); |
| 247 | const old = this.subscription; |
| 248 | this.subscription = ""; |
| 249 | this.closeSubscription(old, generation); |
| 250 | consumer.connection("disconnected", String(error)); |
| 251 | this.noteFailure(transcriptFailure); |
| 252 | await new Promise(resolve => setTimeout(resolve, 1000)); |
| 253 | if (generation === this.generation) consumer.connection("syncing"); |
| 254 | } |
| 255 | } |
| 256 | } |
| 257 | |
| 258 | private noteFailure(problem: TranscriptFollowFailure): void { |
| 259 | const now = this.now(); |
| 260 | const previous = this.failureState; |
| 261 | if (!previous) { |
| 262 | this.failureState = { stage: problem.stage, reason: problem.reason, errorType: problem.errorType, firstAt: now, lastReportedAt: now, failures: 1 }; |
| 263 | this.publish("failure", problem.stage, problem.reason, true); |
| 264 | return; |
| 265 | } |
| 266 | if (previous.stage !== problem.stage || previous.reason !== problem.reason) { |
| 267 | this.failureState = { |
| 268 | stage: problem.stage, |
| 269 | reason: problem.reason, |
| 270 | errorType: problem.errorType, |
| 271 | firstAt: previous.firstAt, |
| 272 | lastReportedAt: now, |
| 273 | failures: previous.failures + 1, |
| 274 | }; |
| 275 | this.publish("failure", problem.stage, problem.reason, true); |
| 276 | return; |
| 277 | } |
| 278 | previous.failures++; |
| 279 | previous.errorType = problem.errorType; |
| 280 | const visible = now - previous.lastReportedAt >= 30_000; |
| 281 | if (visible) previous.lastReportedAt = now; |
| 282 | this.publish(visible ? "summary" : "failure", previous.stage, previous.reason, visible); |
| 283 | } |
| 284 | |
| 285 | private noteRecovery(): void { |
| 286 | if (!this.failureState) return; |
| 287 | this.publish("recovered", this.failureState.stage, this.failureState.reason, true); |
| 288 | this.failureState = null; |
| 289 | } |
| 290 | |
| 291 | private publish( |
| 292 | event: TranscriptDiagnostic["event"], |
| 293 | stage: TranscriptDiagnosticStage, |
| 294 | reason: TranscriptDiagnosticReason, |
| 295 | visible: boolean, |
| 296 | ): void { |
| 297 | const state = this.failureState; |
| 298 | const now = this.now(); |
| 299 | const diagnostic: TranscriptDiagnostic = { |
| 300 | kind: "transcript", |
| 301 | event, |
| 302 | stage, |
| 303 | reason, |
| 304 | transport: this.transport, |
| 305 | errorType: state?.errorType ?? "classified", |
| 306 | revision: Math.max(0, this.revision), |
| 307 | commit: Math.max(0, this.coverage), |
| 308 | attempts: this.indexes.size, |
| 309 | failures: state?.failures ?? 0, |
| 310 | durationMs: state ? Math.max(0, now - state.firstAt) : 0, |
| 311 | }; |
| 312 | publishTranscriptDiagnostic(this.diagnosticOwner, diagnostic, visible); |
| 313 | this.onDiagnostic?.(diagnostic, visible); |
| 314 | } |
| 315 | } |
| 316 |