From 4f0564e4eba1613a358542e84c4c3e49bdb3434e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 30 Aug 2026 18:47:40 +0000 Subject: [PATCH 1/8] feat(ai-editing-worker): diagnose sync-init timeouts with socket instrumentation Every prod EditDocument since 2026-08-26 fails with an opaque "initial sync failed: timeout (10000ms)". Add passive WebSocket diagnostics to the worker so the next timeout says how far the connection got: never opened, raw frame arrived but never decoded (e.g. Blob where Bebop expects ArrayBuffer), or decoded frames with no RemoteInitialSync. - new sync-diagnostics recorder wraps the platform WebSocket factory and observes open/message/close/error on the raw socket - decoded-frame kinds counted via the wrapper's public Message event - edit.sync_init span gets sync.ws.* attrs; the timeout error message carries a compact summary into the 502 body - createSyncSocket accepts an optional socket factory (additive) - WebsocketBuilder.withSerializer no longer drops a factory set before it Co-authored-by: teo --- .../collaboration/src/sync-service/socket.ts | 16 +- .../src/websocket/core/websocket-builder.ts | 1 + packages/collaboration/src/websocket/index.ts | 5 + .../ai-editing-worker/src/endpoints/edit.ts | 10 +- services/ai-editing-worker/src/run-edit.ts | 16 +- services/ai-editing-worker/src/sources.ts | 29 +++- .../src/sync-diagnostics.test.ts | 122 ++++++++++++++ .../ai-editing-worker/src/sync-diagnostics.ts | 150 ++++++++++++++++++ 8 files changed, 342 insertions(+), 7 deletions(-) create mode 100644 services/ai-editing-worker/src/sync-diagnostics.test.ts create mode 100644 services/ai-editing-worker/src/sync-diagnostics.ts diff --git a/packages/collaboration/src/sync-service/socket.ts b/packages/collaboration/src/sync-service/socket.ts index c3eb6ea7530..dfc402272c9 100644 --- a/packages/collaboration/src/sync-service/socket.ts +++ b/packages/collaboration/src/sync-service/socket.ts @@ -3,6 +3,7 @@ import { BebopSerializer, ExponentialBackoff, type UrlResolver, + type WebSocketFactory, type Websocket, WebsocketBuilder, type WebsocketConnectionState, @@ -39,11 +40,20 @@ export interface SyncSocket { * heartbeat. The `getUrl` resolver is the only environment-specific bit — the * browser refreshes a permission token on each (re)connect, the worker returns * a static URL. + * + * `factory` optionally overrides how the underlying native socket is created + * (the AI editing worker wraps the default to observe raw frames). */ -export function createSyncSocket(getUrl: UrlResolver): SyncWebsocket { +export function createSyncSocket( + getUrl: UrlResolver, + factory?: WebSocketFactory +): SyncWebsocket { + let builder = new WebsocketBuilder(getUrl).withSerializer( + new BebopSerializer(FromPeer, FromRemote) + ); + if (factory) builder = builder.withFactory(factory); return ( - new WebsocketBuilder(getUrl) - .withSerializer(new BebopSerializer(FromPeer, FromRemote)) + builder // Capped exponential backoff. The scheduler calls next() before the first // retry, so the delays are 250*2^1 = 500ms doubling to a 250*2^5 = 8s // cap; 20 retries ≈ 2 minutes of automatic attempts, after which diff --git a/packages/collaboration/src/websocket/core/websocket-builder.ts b/packages/collaboration/src/websocket/core/websocket-builder.ts index a22c70bbb0e..18c15d71f24 100644 --- a/packages/collaboration/src/websocket/core/websocket-builder.ts +++ b/packages/collaboration/src/websocket/core/websocket-builder.ts @@ -192,6 +192,7 @@ export class WebsocketBuilder { listeners: this._options?.listeners, serializer, buffer: this._options?.buffer, + factory: this._options?.factory, } as unknown as WebsocketOptions; if (serializer.binaryType !== undefined) { diff --git a/packages/collaboration/src/websocket/index.ts b/packages/collaboration/src/websocket/index.ts index c462975f6af..3dfa765acc5 100644 --- a/packages/collaboration/src/websocket/index.ts +++ b/packages/collaboration/src/websocket/index.ts @@ -25,6 +25,11 @@ export { export type { WebsocketOptions } from './core/websocket-options'; export type { WebsocketConnectionRetryOptions } from './core/websocket-retry-options'; export type { UrlResolver } from './core/websocket-url-resolver'; +export { platformWebSocketFactory } from './platform/factory'; +export type { + MinimalWebSocket, + WebSocketFactory, +} from './platform/minimal-websocket'; export { createReconnectEffect, createSocketEffect, diff --git a/services/ai-editing-worker/src/endpoints/edit.ts b/services/ai-editing-worker/src/endpoints/edit.ts index d44676890e1..5ed2c779a42 100644 --- a/services/ai-editing-worker/src/endpoints/edit.ts +++ b/services/ai-editing-worker/src/endpoints/edit.ts @@ -12,6 +12,7 @@ import { type Model, type ResolvedModels, runEditSession } from '../run-edit'; import { runInSandbox } from '../sandbox'; import { watchPresenceSpeed } from '../service-clients'; import { createWorkerSyncSource } from '../sources'; +import { createSyncSocketDiagnostics } from '../sync-diagnostics'; import { renderTraceMarkdown } from '../trace-log'; import { insertEditTrace } from '../traces-db'; @@ -132,7 +133,13 @@ edit.post('/', zValidator('json', EditBody), async (c) => { try { const wsUrl = `${env.SYNC_WS_BASE}/document/${documentId}/connect?token=${documentToken}`; - const source = createWorkerSyncSource(wsUrl, documentId, signal); + const syncDiagnostics = createSyncSocketDiagnostics(); + const source = createWorkerSyncSource( + wsUrl, + documentId, + signal, + syncDiagnostics + ); // Animations play at 1x while a human is watching and speed up to // `unwatchedSpeed` when nobody is, so unseen edits finish faster without @@ -160,6 +167,7 @@ edit.post('/', zValidator('json', EditBody), async (c) => { try { const result = await runEditSession({ source, + syncDiagnostics, documentId, prompt, models: buildModels(env, models, () => modelFallbacks++), diff --git a/services/ai-editing-worker/src/run-edit.ts b/services/ai-editing-worker/src/run-edit.ts index 17965d32e80..70d2c1df066 100644 --- a/services/ai-editing-worker/src/run-edit.ts +++ b/services/ai-editing-worker/src/run-edit.ts @@ -19,6 +19,7 @@ import type { UsageEntry } from './ai-editing/token-tracker'; import type { CoderRunCode, DispatchEditTrace } from './ai-editing/tools'; import { serializeWithXml } from './ai-editing/utils'; import { EditingWorkspace } from './editing-workspace'; +import type { SyncSocketDiagnostics } from './sync-diagnostics'; import { buildTraceSession, type TraceSession } from './trace-log'; export type Model = { @@ -41,6 +42,8 @@ export type ResolvedModels = { export type RunEditArgs = { /** Live sync source, already constructed by the caller (ws in prod). */ source: SyncServiceSource; + /** Diagnostics for the source's socket; reported on the sync-init span. */ + syncDiagnostics?: SyncSocketDiagnostics; documentId: string; prompt: string; models: ResolvedModels; @@ -83,10 +86,21 @@ export async function runEditSession( await Telemetry.span('edit.sync_init', async (span) => { const initialResult = await source.doInitialSync(); + const diagnostics = args.syncDiagnostics; + if (diagnostics) { + for (const [key, value] of Object.entries(diagnostics.attrs())) { + span.setAttr(key, value); + } + } if (initialResult.isErr()) { source.cleanup(); const e = initialResult.error; - const err = new Error(`initial sync failed: ${e.type} (${e.duration}ms)`); + // The summary rides the error into the HTTP 502 body, so callers see + // how far the connection got instead of just "timeout (10000ms)". + const detail = diagnostics ? ` [ws ${diagnostics.summary()}]` : ''; + const err = new Error( + `initial sync failed: ${e.type} (${e.duration}ms)${detail}` + ); span.error(err); throw err; } diff --git a/services/ai-editing-worker/src/sources.ts b/services/ai-editing-worker/src/sources.ts index 42c1b9e39e2..62fb92229cc 100644 --- a/services/ai-editing-worker/src/sources.ts +++ b/services/ai-editing-worker/src/sources.ts @@ -6,21 +6,46 @@ */ import type { Awareness } from '@macro-inc/collaboration/collab/awareness'; +import type { FromRemote } from '@macro-inc/collaboration/sync-service/generated/schema'; import { createSyncSocket } from '@macro-inc/collaboration/sync-service/socket'; import { SyncServiceSource } from '@macro-inc/collaboration/sync-service/source'; +import { WebsocketEvent } from '@macro-inc/collaboration/websocket'; import { EphemeralStore, type PeerID } from 'loro-crdt'; +import type { SyncSocketDiagnostics } from './sync-diagnostics'; + +function remoteMessageKind(message: FromRemote): string { + if (message.isRemoteInitialSync()) return 'initial_sync'; + if (message.isRemoteUpdate()) return 'update'; + if (message.isRemoteAwareness()) return 'awareness'; + if (message.isRemoteSnapshot()) return 'snapshot'; + if (message.isRemoteUpdateAck()) return 'update_ack'; + if (message.isRemoteUpdateSince()) return 'update_since'; + return 'unknown'; +} /** * A {@link SyncServiceSource} bound to a fixed, token-bearing URL. A worker * request is short-lived and online for its whole life, so every (re)connect * resolves to the same URL. Closes the socket on `signal` abort. + * + * When `diagnostics` is given, the underlying socket is created through its + * observing factory and decoded messages are counted, so a sync timeout can + * report how far the connection got. */ export function createWorkerSyncSource( wsUrl: string, documentId: string, - signal?: AbortSignal + signal?: AbortSignal, + diagnostics?: SyncSocketDiagnostics ): SyncServiceSource { - const ws = createSyncSocket(() => wsUrl); + const ws = createSyncSocket(() => wsUrl, diagnostics?.factory); + if (diagnostics) { + // The wrapper only dispatches Message after a successful deserialize, so + // raw frames (from the factory) minus these are frames that failed decode. + ws.addEventListener(WebsocketEvent.Message, (_ws, event) => + diagnostics.recordDecoded(remoteMessageKind(event.data)) + ); + } const source = new SyncServiceSource(ws, documentId); signal?.addEventListener('abort', () => source.cleanup()); return source; diff --git a/services/ai-editing-worker/src/sync-diagnostics.test.ts b/services/ai-editing-worker/src/sync-diagnostics.test.ts new file mode 100644 index 00000000000..b053b89fd07 --- /dev/null +++ b/services/ai-editing-worker/src/sync-diagnostics.test.ts @@ -0,0 +1,122 @@ +import type { MinimalWebSocket } from '@macro-inc/collaboration/websocket'; +import { describe, expect, it } from 'vitest'; +import { createSyncSocketDiagnostics } from './sync-diagnostics'; + +type Listener = (event: never) => void; + +/** Fake native socket: records listeners so the test can fire raw events. */ +class FakeNativeSocket { + binaryType = 'blob'; + readyState = 0; + private readonly listeners = new Map>(); + + addEventListener(type: string, listener: Listener) { + let set = this.listeners.get(type); + if (!set) { + set = new Set(); + this.listeners.set(type, set); + } + set.add(listener); + } + + fire(type: string, event: unknown) { + for (const listener of this.listeners.get(type) ?? []) { + listener(event as never); + } + } +} + +function createDiagnosticsWithSocket() { + const sockets: FakeNativeSocket[] = []; + const diagnostics = createSyncSocketDiagnostics(() => { + const socket = new FakeNativeSocket(); + sockets.push(socket); + return socket as unknown as MinimalWebSocket; + }); + return { diagnostics, sockets }; +} + +describe('createSyncSocketDiagnostics', () => { + it('reports a connection that never opened', () => { + const { diagnostics, sockets } = createDiagnosticsWithSocket(); + diagnostics.factory('wss://example', undefined); + sockets[0].fire('close', { code: 1006, reason: '', wasClean: false }); + + expect(diagnostics.attrs()).toMatchObject({ + 'sync.ws.connect_attempts': 1, + 'sync.ws.opened': false, + 'sync.ws.raw_frames': 0, + 'sync.ws.decoded_frames': 0, + 'sync.ws.close.code': 1006, + 'sync.ws.close.clean': false, + }); + expect(diagnostics.summary()).toBe( + 'attempts=1 opened=false raw_frames=0 decoded=0 close=1006' + ); + }); + + it('reports a raw frame that never decoded (Blob delivery)', () => { + const { diagnostics, sockets } = createDiagnosticsWithSocket(); + diagnostics.factory('wss://example', undefined); + const socket = sockets[0]; + socket.binaryType = 'arraybuffer'; + socket.fire('open', {}); + socket.fire('message', { data: new Blob([new Uint8Array(16)]) }); + + const attrs = diagnostics.attrs(); + expect(attrs).toMatchObject({ + 'sync.ws.opened': true, + 'sync.ws.raw_frames': 1, + 'sync.ws.decoded_frames': 0, + 'sync.ws.first_frame.type': 'Blob', + 'sync.ws.first_frame.bytes': 16, + 'sync.ws.binary_type': 'arraybuffer', + }); + expect(diagnostics.summary()).toBe( + 'attempts=1 opened=true raw_frames=1 decoded=0 first_frame=Blob' + ); + }); + + it('keeps first-frame details while counting later frames and decodes', () => { + const { diagnostics, sockets } = createDiagnosticsWithSocket(); + diagnostics.factory('wss://example', undefined); + const socket = sockets[0]; + socket.binaryType = 'arraybuffer'; + socket.fire('open', {}); + socket.fire('message', { data: new ArrayBuffer(30770) }); + socket.fire('message', { data: new ArrayBuffer(4) }); + diagnostics.recordDecoded('initial_sync'); + diagnostics.recordDecoded('update_ack'); + + expect(diagnostics.attrs()).toMatchObject({ + 'sync.ws.raw_frames': 2, + 'sync.ws.decoded_frames': 2, + 'sync.ws.first_frame.type': 'ArrayBuffer', + 'sync.ws.first_frame.bytes': 30770, + 'sync.ws.decoded_kinds': 'initial_sync,update_ack', + }); + }); + + it('counts retries as separate connect attempts and errors', () => { + const { diagnostics, sockets } = createDiagnosticsWithSocket(); + diagnostics.factory('wss://example', undefined); + sockets[0].fire('error', {}); + sockets[0].fire('close', { + code: 1002, + reason: 'protocol', + wasClean: false, + }); + diagnostics.factory('wss://example', undefined); + sockets[1].fire('error', {}); + + expect(diagnostics.attrs()).toMatchObject({ + 'sync.ws.connect_attempts': 2, + 'sync.ws.error_events': 2, + 'sync.ws.close.code': 1002, + 'sync.ws.close.reason': 'protocol', + }); + expect(diagnostics.summary()).toBe( + 'attempts=2 opened=false raw_frames=0 decoded=0 close=1002 errors=2' + ); + }); +}); diff --git a/services/ai-editing-worker/src/sync-diagnostics.ts b/services/ai-editing-worker/src/sync-diagnostics.ts new file mode 100644 index 00000000000..ca4a4b68f5b --- /dev/null +++ b/services/ai-editing-worker/src/sync-diagnostics.ts @@ -0,0 +1,150 @@ +/** + * Passive diagnostics for the worker's sync WebSocket, built to explain + * `edit.sync_init` timeouts. Every prod timeout so far surfaced as an opaque + * "initial sync failed: timeout (10000ms)"; these counters distinguish the + * three failure modes: the socket never opened, a raw frame arrived but never + * decoded (e.g. Blob delivered where the Bebop serializer expects an + * ArrayBuffer), or frames decoded fine but no RemoteInitialSync was sent. + * + * Observation only: the factory wraps the platform default and merely attaches + * extra listeners to the native socket, so transport behavior is unchanged. + */ + +import { + platformWebSocketFactory, + type WebSocketFactory, +} from '@macro-inc/collaboration/websocket'; +import type { Attribute } from '@macro-inc/observability'; + +const MAX_DECODED_KINDS = 10; + +export type SyncSocketDiagnostics = { + /** Hand to {@link createSyncSocket} so raw socket events are observed. */ + factory: WebSocketFactory; + /** Record a message that made it through deserialization. */ + recordDecoded(kind: string): void; + /** Flat span attributes describing everything observed so far. */ + attrs(): Record; + /** One-line summary for error messages, e.g. "opened=false raw_frames=0". */ + summary(): string; +}; + +/** Raw `MessageEvent.data` type without reading the bytes. */ +function frameType(data: unknown): string { + if (typeof data === 'string') return 'string'; + if (data instanceof ArrayBuffer) return 'ArrayBuffer'; + if (typeof Blob !== 'undefined' && data instanceof Blob) return 'Blob'; + if (ArrayBuffer.isView(data)) return data.constructor.name; + if (data === null || data === undefined) return String(data); + return ( + (data as { constructor?: { name?: string } }).constructor?.name ?? + typeof data + ); +} + +function frameBytes(data: unknown): number | undefined { + if (typeof data === 'string') return data.length; + if (data instanceof ArrayBuffer) return data.byteLength; + if (typeof Blob !== 'undefined' && data instanceof Blob) return data.size; + if (ArrayBuffer.isView(data)) return data.byteLength; + return undefined; +} + +export function createSyncSocketDiagnostics( + create: WebSocketFactory = platformWebSocketFactory +): SyncSocketDiagnostics { + let connectAttempts = 0; + let opened = false; + let errorEvents = 0; + let rawFrames = 0; + let firstFrame: + | { type: string; bytes?: number; binaryType: string } + | undefined; + let lastClose: + | { code: number; reason: string; wasClean: boolean } + | undefined; + let decodedFrames = 0; + const decodedKinds: string[] = []; + + const factory: WebSocketFactory = (url, protocols) => { + const socket = create(url, protocols); + connectAttempts += 1; + socket.addEventListener('open', () => { + opened = true; + }); + socket.addEventListener('error', () => { + errorEvents += 1; + }); + socket.addEventListener('close', (event) => { + lastClose = { + code: event.code, + reason: event.reason, + wasClean: event.wasClean, + }; + }); + socket.addEventListener('message', (event) => { + rawFrames += 1; + firstFrame ??= { + type: frameType(event.data), + bytes: frameBytes(event.data), + // As seen when the frame arrived — the wrapper sets 'arraybuffer', + // so anything else here means the runtime ignored the setter. + binaryType: socket.binaryType, + }; + }); + return socket; + }; + + return { + factory, + recordDecoded(kind: string): void { + decodedFrames += 1; + if (decodedKinds.length < MAX_DECODED_KINDS) decodedKinds.push(kind); + }, + attrs(): Record { + const attrs: Record = { + 'sync.ws.connect_attempts': connectAttempts, + 'sync.ws.opened': opened, + 'sync.ws.error_events': errorEvents, + 'sync.ws.raw_frames': rawFrames, + 'sync.ws.decoded_frames': decodedFrames, + }; + if (firstFrame) { + attrs['sync.ws.first_frame.type'] = firstFrame.type; + attrs['sync.ws.binary_type'] = firstFrame.binaryType; + if (firstFrame.bytes !== undefined) { + attrs['sync.ws.first_frame.bytes'] = firstFrame.bytes; + } + } + if (decodedKinds.length > 0) { + attrs['sync.ws.decoded_kinds'] = decodedKinds.join(','); + } + if (lastClose) { + attrs['sync.ws.close.code'] = lastClose.code; + attrs['sync.ws.close.clean'] = lastClose.wasClean; + if (lastClose.reason) { + attrs['sync.ws.close.reason'] = lastClose.reason.slice(0, 256); + } + } + return attrs; + }, + summary(): string { + const parts = [ + `attempts=${connectAttempts}`, + `opened=${opened}`, + `raw_frames=${rawFrames}`, + `decoded=${decodedFrames}`, + ]; + if (firstFrame) { + parts.push(`first_frame=${firstFrame.type}`); + } + if (lastClose) { + parts.push(`close=${lastClose.code}`); + } + if (errorEvents > 0) { + parts.push(`errors=${errorEvents}`); + } + return parts.join(' '); + }, + }; +} From 106962190178691217d752c6dbd8236f1c55e6f5 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 00:07:30 +0000 Subject: [PATCH 2/8] fix(ai-editing-worker): bootstrap from a requested snapshot when initial sync is lost MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every prod edit has died on 'initial sync failed: timeout (10000ms)' since 2026-08-26 14:02 UTC. Sync-service accepts the worker's /connect and sends the snapshot (span records snapshot.bytes), but the frame never reaches the worker: a workerd hibernation regression (STOR-5552; cloudflare/workerd#7163, follow-up #7147) dropped sends caught in the connect-request handover to the hibernation manager. PeerRequestSnapshot -> RemoteSnapshot is served from the message handler on the manager-owned socket, which does not cross that boundary and demonstrably still works in prod. WorkerSyncSource (worker-only subclass) requests a snapshot when the pushed RemoteInitialSync stays silent for 2s and bootstraps from the reply; first success wins. Browsers keep the plain push-based flow — the shared source only widens initialSyncReceived to protected. Co-authored-by: teo --- bun.lock | 1 + .../collaboration/src/sync-service/source.ts | 5 +- services/ai-editing-worker/package.json | 1 + .../ai-editing-worker/src/sources.test.ts | 183 ++++++++++++++++++ services/ai-editing-worker/src/sources.ts | 108 ++++++++++- 5 files changed, 294 insertions(+), 4 deletions(-) create mode 100644 services/ai-editing-worker/src/sources.test.ts diff --git a/bun.lock b/bun.lock index 43353dc0c01..5fe71bba898 100644 --- a/bun.lock +++ b/bun.lock @@ -298,6 +298,7 @@ "flat": "^6.0.1", "hono": "^4.6.20", "loro-crdt": "catalog:", + "neverthrow": "^8.2.0", "p-queue": "^9.3.0", "quickjs-emscripten": "^0.31.0", "random-js": "^2.1.0", diff --git a/packages/collaboration/src/sync-service/source.ts b/packages/collaboration/src/sync-service/source.ts index 502ebe80d10..98406b6196e 100644 --- a/packages/collaboration/src/sync-service/source.ts +++ b/packages/collaboration/src/sync-service/source.ts @@ -94,7 +94,10 @@ export class SyncServiceSource implements LiveSyncSource { private hasEverHadListener = false; /** Pending request/response waiters, matched against each incoming message. */ private readonly waiters = new Set(); - private initialSyncReceived = false; + /** Gates pushUpdate. Protected so a subclass that bootstraps the initial + * state through another exchange (the AI editing worker's snapshot + * fallback) can unlock updates without duplicating the gate. */ + protected initialSyncReceived = false; private readonly initialSyncPromise: ResultAsync; public constructor( diff --git a/services/ai-editing-worker/package.json b/services/ai-editing-worker/package.json index 6ef0add7a10..2ad39d861d1 100644 --- a/services/ai-editing-worker/package.json +++ b/services/ai-editing-worker/package.json @@ -24,6 +24,7 @@ "envsafe": "^2.0.3", "hono": "^4.6.20", "loro-crdt": "catalog:", + "neverthrow": "^8.2.0", "quickjs-emscripten": "^0.31.0", "yargs": "^18.0.0", "zod": "^4.4.3", diff --git a/services/ai-editing-worker/src/sources.test.ts b/services/ai-editing-worker/src/sources.test.ts new file mode 100644 index 00000000000..5353ee9c486 --- /dev/null +++ b/services/ai-editing-worker/src/sources.test.ts @@ -0,0 +1,183 @@ +import type { + FromRemote, + IFromPeer, +} from '@macro-inc/collaboration/sync-service/generated/schema'; +import type { SyncWebsocket } from '@macro-inc/collaboration/sync-service/socket'; +import { + WebsocketConnectionState, + WebsocketEvent, + type WebsocketEventListener, +} from '@macro-inc/collaboration/websocket'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { WorkerSyncSource } from './sources'; + +type AnyListener = WebsocketEventListener< + WebsocketEvent, + IFromPeer, + FromRemote +>; + +/** Minimal in-memory SyncSocket, mirroring the collaboration source tests. */ +class FakeSocket { + connectionState = WebsocketConnectionState.Connecting; + sent: IFromPeer[] = []; + heartbeats = 0; + closed = false; + private msg = new Set(); + + send(message: IFromPeer) { + this.sent.push(message); + return true; + } + startHeartbeat() { + this.heartbeats++; + } + reconnectIfDisconnected() {} + close() { + this.closed = true; + } + addEventListener(type: K, listener: AnyListener) { + if (type === WebsocketEvent.Message) this.msg.add(listener); + } + removeEventListener( + _type: K, + listener: AnyListener + ) { + this.msg.delete(listener); + } + + deliver(data: FromRemote) { + const event = { data } as Parameters[1]; + for (const listener of [...this.msg]) listener(this as never, event); + } + + asSyncWebsocket(): SyncWebsocket { + return this as unknown as SyncWebsocket; + } +} + +const GUARDS = { + isRemoteInitialSync: () => false, + isRemoteUpdate: () => false, + isRemoteAwareness: () => false, + isRemoteSnapshot: () => false, + isRemoteUpdateAck: () => false, + isRemoteUpdateSince: () => false, +}; + +const remote = { + initialSync: (snapshot: Uint8Array, awareness: Uint8Array) => + ({ + ...GUARDS, + isRemoteInitialSync: () => true, + value: { snapshot, awareness }, + }) as unknown as FromRemote, + snapshot: (snapshot: Uint8Array) => + ({ + ...GUARDS, + isRemoteSnapshot: () => true, + value: { snapshot }, + }) as unknown as FromRemote, + ack: (id: string) => + ({ + ...GUARDS, + isRemoteUpdateAck: () => true, + value: { id }, + }) as unknown as FromRemote, +}; + +const snap = new Uint8Array([1, 2, 3]); +const aw = new Uint8Array([4, 5]); + +// From packages/collaboration/src/sync-service/source.ts. +const INITIAL_SYNC_TIMEOUT_MS = 10_000; +const FALLBACK_MS = 2_000; + +describe('WorkerSyncSource snapshot fallback', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it('resolves from RemoteInitialSync without ever requesting a snapshot', async () => { + const ws = new FakeSocket(); + const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); + const pending = src.doInitialSync(); + ws.deliver(remote.initialSync(snap, aw)); + + expect((await pending)._unsafeUnwrap()).toEqual({ + snapshot: snap, + awareness: aw, + }); + // The fallback timer is canceled; nothing is sent even after its delay. + await vi.advanceTimersByTimeAsync(FALLBACK_MS + 1); + expect(ws.sent).toHaveLength(0); + expect(ws.heartbeats).toBe(1); + }); + + it('bootstraps from a requested snapshot when the push never arrives', async () => { + const ws = new FakeSocket(); + const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); + const pending = src.doInitialSync(); + + expect(ws.sent).toHaveLength(0); + await vi.advanceTimersByTimeAsync(FALLBACK_MS + 1); + expect(ws.sent).toHaveLength(1); // PeerRequestSnapshot went out + + ws.deliver(remote.snapshot(snap)); + expect((await pending)._unsafeUnwrap()).toEqual({ + snapshot: snap, + awareness: new Uint8Array(0), + }); + expect(ws.heartbeats).toBe(1); + }); + + it('unlocks pushUpdate after a fallback bootstrap', async () => { + const ws = new FakeSocket(); + const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); + const pending = src.doInitialSync(); + + await vi.advanceTimersByTimeAsync(FALLBACK_MS + 1); + ws.deliver(remote.snapshot(snap)); + await pending; + + const sentBefore = ws.sent.length; + const acked = src.pushUpdate([new Uint8Array([9])]); + expect(ws.sent.length).toBe(sentBefore + 1); // update went out + const update = ws.sent[ws.sent.length - 1] as unknown as { + value: { id: string }; + }; + ws.deliver(remote.ack(update.value.id)); + expect(await acked).toBe(true); + }); + + it('surfaces the initial-sync timeout when both paths stay silent', async () => { + const ws = new FakeSocket(); + const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); + const pending = src.doInitialSync(); + + // Push times out at 10s; the 2s-delayed snapshot request at 12s. + await vi.advanceTimersByTimeAsync( + FALLBACK_MS + INITIAL_SYNC_TIMEOUT_MS + 1 + ); + + expect((await pending)._unsafeUnwrapErr()).toEqual({ + type: 'timeout', + duration: INITIAL_SYNC_TIMEOUT_MS, + }); + }); + + it('does not request a snapshot after cleanup', async () => { + const ws = new FakeSocket(); + const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); + void src.doInitialSync(); + + src.cleanup(); + await vi.advanceTimersByTimeAsync(FALLBACK_MS + 1); + + expect(ws.sent).toHaveLength(0); + expect(ws.closed).toBe(true); + }); +}); diff --git a/services/ai-editing-worker/src/sources.ts b/services/ai-editing-worker/src/sources.ts index 62fb92229cc..90a3d590c17 100644 --- a/services/ai-editing-worker/src/sources.ts +++ b/services/ai-editing-worker/src/sources.ts @@ -6,11 +6,19 @@ */ import type { Awareness } from '@macro-inc/collaboration/collab/awareness'; +import type { + InitialSync, + TimeoutError, +} from '@macro-inc/collaboration/collab/source'; import type { FromRemote } from '@macro-inc/collaboration/sync-service/generated/schema'; -import { createSyncSocket } from '@macro-inc/collaboration/sync-service/socket'; +import { + createSyncSocket, + type SyncWebsocket, +} from '@macro-inc/collaboration/sync-service/socket'; import { SyncServiceSource } from '@macro-inc/collaboration/sync-service/source'; import { WebsocketEvent } from '@macro-inc/collaboration/websocket'; import { EphemeralStore, type PeerID } from 'loro-crdt'; +import { ResultAsync } from 'neverthrow'; import type { SyncSocketDiagnostics } from './sync-diagnostics'; function remoteMessageKind(message: FromRemote): string { @@ -24,7 +32,101 @@ function remoteMessageKind(message: FromRemote): string { } /** - * A {@link SyncServiceSource} bound to a fixed, token-bearing URL. A worker + * How long the worker waits for the server-pushed `RemoteInitialSync` before + * actively requesting a snapshot over the same socket. The server normally + * sends it well under 100ms after the socket opens, so 2s only triggers when + * the push is genuinely lost. + */ +const INITIAL_SYNC_FALLBACK_MS = 2_000; + +/** + * {@link SyncServiceSource} with a worker-only bootstrap fallback. + * + * The server pushes `RemoteInitialSync` from inside the `/connect` request + * handler. On Cloudflare that send races the hibernation manager reclaiming + * the socket when the request context ends, and a workerd regression + * (STOR-5552; fixed by cloudflare/workerd#7163, follow-up #7147) silently + * dropped sends caught in that handover: the socket stayed open but the + * snapshot never arrived, and every worker edit died on + * "initial sync failed: timeout (10000ms)". + * + * When the push stays silent for {@link INITIAL_SYNC_FALLBACK_MS}, this + * subclass sends `PeerRequestSnapshot` and bootstraps from the + * `RemoteSnapshot` reply (with empty awareness — the worker never reads it). + * That reply is served from the message handler on the manager-owned socket, + * which does not cross the affected boundary. First success wins; browsers + * keep the plain push-based flow. + */ +export class WorkerSyncSource extends SyncServiceSource { + private fallbackTimer: ReturnType | undefined; + + constructor(socket: SyncWebsocket, documentId: string) { + super(socket, documentId); + // Both are assignable instance properties on the base class; capture the + // originals, then wrap. + const withFallback = this.raceWithSnapshotFallback( + this.doInitialSync(), + socket + ); + this.doInitialSync = () => withFallback; + const baseCleanup = this.cleanup; + this.cleanup = () => { + this.cancelFallback(); + baseCleanup(); + }; + } + + private raceWithSnapshotFallback( + initial: ResultAsync, + socket: SyncWebsocket + ): ResultAsync { + const fallback = new Promise((resolve, reject) => { + this.fallbackTimer = setTimeout(() => { + this.fallbackTimer = undefined; + console.warn( + `no initial sync after ${INITIAL_SYNC_FALLBACK_MS}ms; requesting snapshot fallback:`, + this.documentId + ); + this.requestSnapshot().match((snapshot) => { + // The base class only unlocks pushUpdate and starts the heartbeat + // when RemoteInitialSync arrives; the fallback bootstrap must do + // both itself. Both are safe if the push arrives later anyway. + this.initialSyncReceived = true; + socket.startHeartbeat(); + resolve({ snapshot, awareness: new Uint8Array(0) }); + }, reject); + }, INITIAL_SYNC_FALLBACK_MS); + }); + // First success wins. When both fail, surface the first error (the + // original initial-sync timeout), matching the push-only failure shape. + return ResultAsync.fromPromise( + new Promise((resolve, reject) => { + let failures = 0; + let firstError: TimeoutError | undefined; + const fail = (error: TimeoutError) => { + firstError ??= error; + failures += 1; + if (failures === 2) reject(firstError); + }; + void initial.match((sync) => { + this.cancelFallback(); + resolve(sync); + }, fail); + fallback.then(resolve, (error) => fail(error as TimeoutError)); + }), + (error) => error as TimeoutError + ); + } + + private cancelFallback(): void { + if (this.fallbackTimer === undefined) return; + clearTimeout(this.fallbackTimer); + this.fallbackTimer = undefined; + } +} + +/** + * A {@link WorkerSyncSource} bound to a fixed, token-bearing URL. A worker * request is short-lived and online for its whole life, so every (re)connect * resolves to the same URL. Closes the socket on `signal` abort. * @@ -46,7 +148,7 @@ export function createWorkerSyncSource( diagnostics.recordDecoded(remoteMessageKind(event.data)) ); } - const source = new SyncServiceSource(ws, documentId); + const source = new WorkerSyncSource(ws, documentId); signal?.addEventListener('abort', () => source.cleanup()); return source; } From 47975169d7c6d7368486d2b713dd908e322a1598 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 00:12:46 +0000 Subject: [PATCH 3/8] fix(ai-editing-worker): guard diagnostics listeners so they can never break delivery The diagnostics listeners attach to the native socket ahead of the transport wrapper's handlers. A listener exception there could suppress the wrapper's dispatch for that event (or error out the socket under workerd's fail-fast event dispatch), making the instrumentation cause the failure it measures. The bodies are all typeof/instanceof checks that should not throw; swallow exceptions anyway so observation stays observation. Co-authored-by: teo --- .../ai-editing-worker/src/sync-diagnostics.ts | 74 +++++++++++++------ 1 file changed, 51 insertions(+), 23 deletions(-) diff --git a/services/ai-editing-worker/src/sync-diagnostics.ts b/services/ai-editing-worker/src/sync-diagnostics.ts index ca4a4b68f5b..43edd66ddb6 100644 --- a/services/ai-editing-worker/src/sync-diagnostics.ts +++ b/services/ai-editing-worker/src/sync-diagnostics.ts @@ -66,32 +66,60 @@ export function createSyncSocketDiagnostics( let decodedFrames = 0; const decodedKinds: string[] = []; + // These listeners are attached to the native socket ahead of the transport + // wrapper's own handlers. A listener exception here could suppress the + // wrapper's dispatch for that event (or, under workerd's fail-fast event + // dispatch, error out the socket) — the one way observation could break + // delivery. Nothing below should be able to throw; swallow anyway so + // diagnostics can never cause the failure they measure. + const observe = (record: (event: E) => void): ((event: E) => void) => { + return (event) => { + try { + record(event); + } catch { + /* never let diagnostics interfere with delivery */ + } + }; + }; + const factory: WebSocketFactory = (url, protocols) => { const socket = create(url, protocols); connectAttempts += 1; - socket.addEventListener('open', () => { - opened = true; - }); - socket.addEventListener('error', () => { - errorEvents += 1; - }); - socket.addEventListener('close', (event) => { - lastClose = { - code: event.code, - reason: event.reason, - wasClean: event.wasClean, - }; - }); - socket.addEventListener('message', (event) => { - rawFrames += 1; - firstFrame ??= { - type: frameType(event.data), - bytes: frameBytes(event.data), - // As seen when the frame arrived — the wrapper sets 'arraybuffer', - // so anything else here means the runtime ignored the setter. - binaryType: socket.binaryType, - }; - }); + socket.addEventListener( + 'open', + observe(() => { + opened = true; + }) + ); + socket.addEventListener( + 'error', + observe(() => { + errorEvents += 1; + }) + ); + socket.addEventListener( + 'close', + observe((event) => { + lastClose = { + code: event.code, + reason: event.reason, + wasClean: event.wasClean, + }; + }) + ); + socket.addEventListener( + 'message', + observe((event) => { + rawFrames += 1; + firstFrame ??= { + type: frameType(event.data), + bytes: frameBytes(event.data), + // As seen when the frame arrived — the wrapper sets 'arraybuffer', + // so anything else here means the runtime ignored the setter. + binaryType: socket.binaryType, + }; + }) + ); return socket; }; From 6a92af20603fc81ea30b45e6dcd90c3221b8e0e6 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 00:53:14 +0000 Subject: [PATCH 4/8] fix(ai-editing-worker): survive poison frames; probe first-frame bytes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dev deploy of the fallback still failed, and its diagnostics finally showed why: the initial sync ARRIVES (785-byte ArrayBuffer, matching the 771-byte snapshot the sync service recorded sending) but fails to deserialize, and the deserialize throw escapes the wrapper's message listener into the runtime's event dispatch, which errors out the socket without a close. The zombie socket never retries and silently swallowed the snapshot fallback request. Two changes: - Tolerant serializer (worker-only): deserialize failures report to diagnostics and yield an inert message instead of throwing, so the socket survives and the snapshot fallback still runs. createSyncSocket gains an optional serializer parameter (additive; browser flow unchanged). - Diagnostics v2: hex prefix + independent probe decode (with exception text) of the first frame, ordered event log with ms offsets, and decode-failure counters — the next dev run will say exactly what the bytes are. Co-authored-by: teo --- .../collaboration/src/sync-service/socket.ts | 9 +- packages/collaboration/src/websocket/index.ts | 1 + .../ai-editing-worker/src/sources.test.ts | 42 ++++++- services/ai-editing-worker/src/sources.ts | 57 ++++++++- .../src/sync-diagnostics.test.ts | 32 +++++ .../ai-editing-worker/src/sync-diagnostics.ts | 112 ++++++++++++++++-- 6 files changed, 237 insertions(+), 16 deletions(-) diff --git a/packages/collaboration/src/sync-service/socket.ts b/packages/collaboration/src/sync-service/socket.ts index dfc402272c9..b6b2c46f1fe 100644 --- a/packages/collaboration/src/sync-service/socket.ts +++ b/packages/collaboration/src/sync-service/socket.ts @@ -9,6 +9,7 @@ import { type WebsocketConnectionState, type WebsocketEvent, type WebsocketEventListener, + type WebsocketSerializer, } from '../websocket'; import { FromPeer, FromRemote, type IFromPeer } from './generated/schema'; @@ -43,13 +44,17 @@ export interface SyncSocket { * * `factory` optionally overrides how the underlying native socket is created * (the AI editing worker wraps the default to observe raw frames). + * + * `serializer` optionally overrides the default Bebop serializer (the AI + * editing worker wraps it to survive frames that fail to deserialize). */ export function createSyncSocket( getUrl: UrlResolver, - factory?: WebSocketFactory + factory?: WebSocketFactory, + serializer?: WebsocketSerializer ): SyncWebsocket { let builder = new WebsocketBuilder(getUrl).withSerializer( - new BebopSerializer(FromPeer, FromRemote) + serializer ?? new BebopSerializer(FromPeer, FromRemote) ); if (factory) builder = builder.withFactory(factory); return ( diff --git a/packages/collaboration/src/websocket/index.ts b/packages/collaboration/src/websocket/index.ts index 3dfa765acc5..41d5464be66 100644 --- a/packages/collaboration/src/websocket/index.ts +++ b/packages/collaboration/src/websocket/index.ts @@ -24,6 +24,7 @@ export { } from './core/websocket-event'; export type { WebsocketOptions } from './core/websocket-options'; export type { WebsocketConnectionRetryOptions } from './core/websocket-retry-options'; +export type { WebsocketSerializer } from './core/websocket-serializer'; export type { UrlResolver } from './core/websocket-url-resolver'; export { platformWebSocketFactory } from './platform/factory'; export type { diff --git a/services/ai-editing-worker/src/sources.test.ts b/services/ai-editing-worker/src/sources.test.ts index 5353ee9c486..1fd0aed463b 100644 --- a/services/ai-editing-worker/src/sources.test.ts +++ b/services/ai-editing-worker/src/sources.test.ts @@ -1,6 +1,7 @@ -import type { - FromRemote, - IFromPeer, +import { + FromRemote as FromRemoteSchema, + type FromRemote, + type IFromPeer, } from '@macro-inc/collaboration/sync-service/generated/schema'; import type { SyncWebsocket } from '@macro-inc/collaboration/sync-service/socket'; import { @@ -9,7 +10,7 @@ import { type WebsocketEventListener, } from '@macro-inc/collaboration/websocket'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { WorkerSyncSource } from './sources'; +import { createTolerantSerializer, WorkerSyncSource } from './sources'; type AnyListener = WebsocketEventListener< WebsocketEvent, @@ -181,3 +182,36 @@ describe('WorkerSyncSource snapshot fallback', () => { expect(ws.closed).toBe(true); }); }); + +describe('createTolerantSerializer', () => { + it('decodes a valid frame', () => { + const onFailure = vi.fn(); + const serializer = createTolerantSerializer(onFailure); + const encoded = FromRemoteSchema.encode( + FromRemoteSchema.fromRemoteSnapshot({ snapshot: new Uint8Array([7, 8]) }) + ); + const buffer = encoded.buffer.slice( + encoded.byteOffset, + encoded.byteOffset + encoded.byteLength + ) as ArrayBuffer; + + const message = serializer.deserialize(buffer); + expect(message.isRemoteSnapshot()).toBe(true); + expect(onFailure).not.toHaveBeenCalled(); + }); + + it('returns an inert message instead of throwing on garbage bytes', () => { + const onFailure = vi.fn(); + const serializer = createTolerantSerializer(onFailure); + const garbage = new Uint8Array([0xff, 0xff, 0xff, 0xff, 0x01]).buffer; + + const message = serializer.deserialize(garbage); + expect(onFailure).toHaveBeenCalledTimes(1); + expect(message.isRemoteInitialSync()).toBe(false); + expect(message.isRemoteUpdate()).toBe(false); + expect(message.isRemoteAwareness()).toBe(false); + expect(message.isRemoteSnapshot()).toBe(false); + expect(message.isRemoteUpdateAck()).toBe(false); + expect(message.isRemoteUpdateSince()).toBe(false); + }); +}); diff --git a/services/ai-editing-worker/src/sources.ts b/services/ai-editing-worker/src/sources.ts index 90a3d590c17..4780151318c 100644 --- a/services/ai-editing-worker/src/sources.ts +++ b/services/ai-editing-worker/src/sources.ts @@ -10,13 +10,21 @@ import type { InitialSync, TimeoutError, } from '@macro-inc/collaboration/collab/source'; -import type { FromRemote } from '@macro-inc/collaboration/sync-service/generated/schema'; +import { + FromPeer, + FromRemote, + type IFromPeer, +} from '@macro-inc/collaboration/sync-service/generated/schema'; import { createSyncSocket, type SyncWebsocket, } from '@macro-inc/collaboration/sync-service/socket'; import { SyncServiceSource } from '@macro-inc/collaboration/sync-service/source'; -import { WebsocketEvent } from '@macro-inc/collaboration/websocket'; +import { + BebopSerializer, + WebsocketEvent, + type WebsocketSerializer, +} from '@macro-inc/collaboration/websocket'; import { EphemeralStore, type PeerID } from 'loro-crdt'; import { ResultAsync } from 'neverthrow'; import type { SyncSocketDiagnostics } from './sync-diagnostics'; @@ -39,6 +47,46 @@ function remoteMessageKind(message: FromRemote): string { */ const INITIAL_SYNC_FALLBACK_MS = 2_000; +/** Inert message: every guard is false, so listeners and waiters ignore it. */ +const POISON_FRAME: FromRemote = { + isRemoteInitialSync: () => false, + isRemoteUpdate: () => false, + isRemoteAwareness: () => false, + isRemoteSnapshot: () => false, + isRemoteUpdateAck: () => false, + isRemoteUpdateSince: () => false, +} as unknown as FromRemote; + +/** + * Bebop serializer whose deserialize failures don't take the socket down. + * + * A deserialize exception escapes the transport wrapper's message listener + * into the runtime's event dispatch, which errors out the WebSocket without a + * close event — leaving a zombie socket that never retries and can't carry + * the snapshot fallback (observed on prod/dev as `raw_frames=1 decoded=0 + * errors=1`). Swallow the failure, report it to diagnostics, and hand the + * wrapper an inert message instead, so the socket stays usable and the + * fallback still runs. + */ +export function createTolerantSerializer( + onFailure: (error: unknown) => void +): WebsocketSerializer { + const base = new BebopSerializer(FromPeer, FromRemote); + return { + serialize: (data) => base.serialize(data), + deserialize: (data) => { + try { + return base.deserialize(data as ArrayBuffer); + } catch (error) { + console.error('sync frame failed to deserialize; dropping it:', error); + onFailure(error); + return POISON_FRAME; + } + }, + binaryType: base.binaryType, + }; +} + /** * {@link SyncServiceSource} with a worker-only bootstrap fallback. * @@ -140,7 +188,10 @@ export function createWorkerSyncSource( signal?: AbortSignal, diagnostics?: SyncSocketDiagnostics ): SyncServiceSource { - const ws = createSyncSocket(() => wsUrl, diagnostics?.factory); + const serializer = createTolerantSerializer((error) => + diagnostics?.recordDecodeFailure(error) + ); + const ws = createSyncSocket(() => wsUrl, diagnostics?.factory, serializer); if (diagnostics) { // The wrapper only dispatches Message after a successful deserialize, so // raw frames (from the factory) minus these are frames that failed decode. diff --git a/services/ai-editing-worker/src/sync-diagnostics.test.ts b/services/ai-editing-worker/src/sync-diagnostics.test.ts index b053b89fd07..73766fc3e55 100644 --- a/services/ai-editing-worker/src/sync-diagnostics.test.ts +++ b/services/ai-editing-worker/src/sync-diagnostics.test.ts @@ -97,6 +97,38 @@ describe('createSyncSocketDiagnostics', () => { }); }); + it('records hex prefix and probe decode for the first binary frame', () => { + const { diagnostics, sockets } = createDiagnosticsWithSocket(); + diagnostics.factory('wss://example', undefined); + const socket = sockets[0]; + socket.binaryType = 'arraybuffer'; + socket.fire('open', {}); + socket.fire('message', { + data: new Uint8Array([0xff, 0xfe, 0xfd]).buffer, + }); + + const attrs = diagnostics.attrs(); + expect(attrs['sync.ws.first_frame.hex']).toBe('fffefd'); + expect(String(attrs['sync.ws.first_frame.probe'])).toMatch(/^err:/); + expect(String(attrs['sync.ws.events'])).toMatch( + /^connect\+\d+ms,open\+\d+ms,frame\(3\)\+\d+ms$/ + ); + expect(diagnostics.summary()).toContain('probe=err:'); + }); + + it('tracks deserialize failures reported by the serializer', () => { + const { diagnostics, sockets } = createDiagnosticsWithSocket(); + diagnostics.factory('wss://example', undefined); + sockets[0].fire('open', {}); + diagnostics.recordDecodeFailure(new Error('unexpected end of buffer')); + + expect(diagnostics.attrs()).toMatchObject({ + 'sync.ws.decode_failures': 1, + 'sync.ws.first_decode_error': 'unexpected end of buffer', + }); + expect(diagnostics.summary()).toContain('decode_failures=1'); + }); + it('counts retries as separate connect attempts and errors', () => { const { diagnostics, sockets } = createDiagnosticsWithSocket(); diagnostics.factory('wss://example', undefined); diff --git a/services/ai-editing-worker/src/sync-diagnostics.ts b/services/ai-editing-worker/src/sync-diagnostics.ts index 43edd66ddb6..25be6eeaf02 100644 --- a/services/ai-editing-worker/src/sync-diagnostics.ts +++ b/services/ai-editing-worker/src/sync-diagnostics.ts @@ -10,6 +10,7 @@ * extra listeners to the native socket, so transport behavior is unchanged. */ +import { FromRemote } from '@macro-inc/collaboration/sync-service/generated/schema'; import { platformWebSocketFactory, type WebSocketFactory, @@ -17,12 +18,16 @@ import { import type { Attribute } from '@macro-inc/observability'; const MAX_DECODED_KINDS = 10; +const MAX_EVENTS = 12; +const HEX_PREFIX_BYTES = 24; export type SyncSocketDiagnostics = { /** Hand to {@link createSyncSocket} so raw socket events are observed. */ factory: WebSocketFactory; /** Record a message that made it through deserialization. */ recordDecoded(kind: string): void; + /** Record a frame the serializer failed to deserialize. */ + recordDecodeFailure(error: unknown): void; /** Flat span attributes describing everything observed so far. */ attrs(): Record; /** One-line summary for error messages, e.g. "opened=false raw_frames=0". */ @@ -50,6 +55,48 @@ function frameBytes(data: unknown): number | undefined { return undefined; } +/** Synchronous byte view of frame data; undefined for Blob (async-only). */ +function frameView(data: unknown): Uint8Array | undefined { + if (data instanceof ArrayBuffer) return new Uint8Array(data); + if (ArrayBuffer.isView(data)) { + return new Uint8Array(data.buffer, data.byteOffset, data.byteLength); + } + return undefined; +} + +function hexPrefix(view: Uint8Array): string { + let out = ''; + const n = Math.min(view.length, HEX_PREFIX_BYTES); + for (let i = 0; i < n; i++) { + out += view[i].toString(16).padStart(2, '0'); + } + return out; +} + +function errorText(error: unknown): string { + return (error instanceof Error ? error.message : String(error)).slice(0, 160); +} + +/** + * What the first frame's bytes actually are: an independent decode attempt + * against the Bebop schema, recorded so a decode failure in the transport + * shows up with its exception text instead of as a silent `decoded=0`. + */ +function probeDecode(view: Uint8Array): string { + try { + const message = FromRemote.decode(view); + if (message.isRemoteInitialSync()) return 'ok:initial_sync'; + if (message.isRemoteUpdate()) return 'ok:update'; + if (message.isRemoteAwareness()) return 'ok:awareness'; + if (message.isRemoteSnapshot()) return 'ok:snapshot'; + if (message.isRemoteUpdateAck()) return 'ok:update_ack'; + if (message.isRemoteUpdateSince()) return 'ok:update_since'; + return 'ok:unknown'; + } catch (error) { + return `err:${errorText(error)}`; + } +} + export function createSyncSocketDiagnostics( create: WebSocketFactory = platformWebSocketFactory ): SyncSocketDiagnostics { @@ -65,6 +112,19 @@ export function createSyncSocketDiagnostics( | undefined; let decodedFrames = 0; const decodedKinds: string[] = []; + let decodeFailures = 0; + let firstDecodeError: string | undefined; + let firstFrameHex: string | undefined; + let firstFrameProbe: string | undefined; + // Ordered event log with ms offsets from diagnostics creation, so "did the + // error fire before or after the frame" is answerable from the span. + const startedAt = Date.now(); + const events: string[] = []; + const logEvent = (name: string) => { + if (events.length < MAX_EVENTS) { + events.push(`${name}+${Date.now() - startedAt}ms`); + } + }; // These listeners are attached to the native socket ahead of the transport // wrapper's own handlers. A listener exception here could suppress the @@ -85,16 +145,19 @@ export function createSyncSocketDiagnostics( const factory: WebSocketFactory = (url, protocols) => { const socket = create(url, protocols); connectAttempts += 1; + logEvent('connect'); socket.addEventListener( 'open', observe(() => { opened = true; + logEvent('open'); }) ); socket.addEventListener( 'error', observe(() => { errorEvents += 1; + logEvent('error'); }) ); socket.addEventListener( @@ -105,19 +168,28 @@ export function createSyncSocketDiagnostics( reason: event.reason, wasClean: event.wasClean, }; + logEvent(`close(${event.code})`); }) ); socket.addEventListener( 'message', observe((event) => { rawFrames += 1; - firstFrame ??= { - type: frameType(event.data), - bytes: frameBytes(event.data), - // As seen when the frame arrived — the wrapper sets 'arraybuffer', - // so anything else here means the runtime ignored the setter. - binaryType: socket.binaryType, - }; + logEvent(`frame(${frameBytes(event.data) ?? '?'})`); + if (firstFrame === undefined) { + firstFrame = { + type: frameType(event.data), + bytes: frameBytes(event.data), + // As seen when the frame arrived — the wrapper sets 'arraybuffer', + // so anything else here means the runtime ignored the setter. + binaryType: socket.binaryType, + }; + const view = frameView(event.data); + if (view) { + firstFrameHex = hexPrefix(view); + firstFrameProbe = probeDecode(view); + } + } }) ); return socket; @@ -129,6 +201,11 @@ export function createSyncSocketDiagnostics( decodedFrames += 1; if (decodedKinds.length < MAX_DECODED_KINDS) decodedKinds.push(kind); }, + recordDecodeFailure(error: unknown): void { + decodeFailures += 1; + firstDecodeError ??= errorText(error); + logEvent('decode_failure'); + }, attrs(): Record { const attrs: Record = { 'sync.ws.connect_attempts': connectAttempts, @@ -143,10 +220,25 @@ export function createSyncSocketDiagnostics( if (firstFrame.bytes !== undefined) { attrs['sync.ws.first_frame.bytes'] = firstFrame.bytes; } + if (firstFrameHex !== undefined) { + attrs['sync.ws.first_frame.hex'] = firstFrameHex; + } + if (firstFrameProbe !== undefined) { + attrs['sync.ws.first_frame.probe'] = firstFrameProbe; + } + } + if (decodeFailures > 0) { + attrs['sync.ws.decode_failures'] = decodeFailures; + if (firstDecodeError !== undefined) { + attrs['sync.ws.first_decode_error'] = firstDecodeError; + } } if (decodedKinds.length > 0) { attrs['sync.ws.decoded_kinds'] = decodedKinds.join(','); } + if (events.length > 0) { + attrs['sync.ws.events'] = events.join(','); + } if (lastClose) { attrs['sync.ws.close.code'] = lastClose.code; attrs['sync.ws.close.clean'] = lastClose.wasClean; @@ -166,6 +258,12 @@ export function createSyncSocketDiagnostics( if (firstFrame) { parts.push(`first_frame=${firstFrame.type}`); } + if (firstFrameProbe !== undefined) { + parts.push(`probe=${firstFrameProbe}`); + } + if (decodeFailures > 0) { + parts.push(`decode_failures=${decodeFailures}`); + } if (lastClose) { parts.push(`close=${lastClose.code}`); } From 6ea5b85c3a4e27c37e1ddb3e5fbe40c6cfb63499 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 01:03:24 +0000 Subject: [PATCH 5/8] fix(ai-editing-worker): contain native listener throws; harden MessageEvent copy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The probe build on dev showed the frame is VALID (probe=ok:initial_sync, hex has the Bebop envelope + loro magic) yet decoded=0 with no deserialize attempt and an error event in the same ms as the frame: dispatch dies between the first-registered native listener and the wrapper's handler. - Wrapper: copying origin/lastEventId/source into the re-dispatched MessageEvent can throw on runtimes with stricter Event internals; fall back to a data-only event (consumers only read data). - Diagnostics: route every native-socket listener through a guard that contains exceptions and records a per-listener start/ok/threw trail plus the first error with stack line (sync.ws.first_listener_error) — the next run distinguishes 'listener threw' from 'runtime aborted dispatch'. Co-authored-by: teo --- .../src/websocket/core/websocket.ts | 22 +++-- .../ai-editing-worker/src/sync-diagnostics.ts | 91 ++++++++++++++++++- 2 files changed, 106 insertions(+), 7 deletions(-) diff --git a/packages/collaboration/src/websocket/core/websocket.ts b/packages/collaboration/src/websocket/core/websocket.ts index 14e5e77d921..2f89ecb6af6 100644 --- a/packages/collaboration/src/websocket/core/websocket.ts +++ b/packages/collaboration/src/websocket/core/websocket.ts @@ -589,12 +589,22 @@ export class Websocket { }) .with({ type: WebsocketEvent.Message }, ({ event }) => { const data = deserializeIfNeeded(event.data, this._options.serializer); - const newEvent = new MessageEvent('message', { - data, - origin: event.origin, - lastEventId: event.lastEventId, - source: event.source, - }); + // Copying origin/lastEventId/source can throw on runtimes with + // stricter Event internals (Cloudflare Workers): the accessors or the + // MessageEvent init validation reject non-browser shapes, and a throw + // here escapes into the native dispatch and errors out the socket. + // Consumers only read `data`, so fall back to a data-only event. + let newEvent: MessageEvent; + try { + newEvent = new MessageEvent('message', { + data, + origin: event.origin, + lastEventId: event.lastEventId, + source: event.source, + }); + } catch { + newEvent = new MessageEvent('message', { data }); + } this.dispatchEvent(WebsocketEvent.Message, newEvent); }) .with( diff --git a/services/ai-editing-worker/src/sync-diagnostics.ts b/services/ai-editing-worker/src/sync-diagnostics.ts index 25be6eeaf02..ef60cfc8ed1 100644 --- a/services/ai-editing-worker/src/sync-diagnostics.ts +++ b/services/ai-editing-worker/src/sync-diagnostics.ts @@ -18,7 +18,7 @@ import { import type { Attribute } from '@macro-inc/observability'; const MAX_DECODED_KINDS = 10; -const MAX_EVENTS = 12; +const MAX_EVENTS = 24; const HEX_PREFIX_BYTES = 24; export type SyncSocketDiagnostics = { @@ -116,6 +116,8 @@ export function createSyncSocketDiagnostics( let firstDecodeError: string | undefined; let firstFrameHex: string | undefined; let firstFrameProbe: string | undefined; + let listenerErrors = 0; + let firstListenerError: string | undefined; // Ordered event log with ms offsets from diagnostics creation, so "did the // error fire before or after the frame" is answerable from the span. const startedAt = Date.now(); @@ -142,10 +144,85 @@ export function createSyncSocketDiagnostics( }; }; + /** + * Route every listener registered on the native socket (the transport + * wrapper's included) through a guard that contains and records exceptions. + * On prod/dev the message dispatch dies between our first-registered + * listener and the wrapper's handler (`frame → error` in the same ms, + * `decoded=0`, no deserialize attempt): either a later listener throws on + * the runtime's stricter Event objects, or the runtime aborts multi-listener + * dispatch. The guard's per-listener start/ok/threw trail distinguishes + * those, and containment keeps a throwing listener from erroring the socket. + */ + const instrumentListeners = (socket: { + addEventListener: (...args: never[]) => void; + removeEventListener: (...args: never[]) => void; + }) => { + type AnyListener = (event: unknown) => void; + const originalAdd = socket.addEventListener.bind( + socket + ) as unknown as (type: string, listener: AnyListener, options?: unknown) => void; + const originalRemove = socket.removeEventListener.bind( + socket + ) as unknown as (type: string, listener: AnyListener, options?: unknown) => void; + const guards = new WeakMap(); + let nextId = 0; + + socket.addEventListener = (( + type: string, + listener: AnyListener, + options?: unknown + ) => { + const id = nextId++; + const guard: AnyListener = (event) => { + // Trail only the first frame's message dispatch to keep events short. + const trail = type === 'message' && rawFrames <= 1; + if (trail) logEvent(`l${id}.msg`); + try { + listener.call(socket, event as never); + if (trail) logEvent(`l${id}.ok`); + } catch (error) { + listenerErrors += 1; + const stackLine = + error instanceof Error && error.stack + ? (error.stack.split('\n')[1] ?? '').trim() + : ''; + firstListenerError ??= + `${type} listener: ${errorText(error)}${stackLine ? ` @ ${stackLine}` : ''}`.slice( + 0, + 300 + ); + logEvent(`l${id}.threw(${type})`); + console.error( + `sync socket ${type} listener threw (contained):`, + error + ); + } + }; + guards.set(listener, guard); + originalAdd(type, guard, options); + }) as never; + + socket.removeEventListener = (( + type: string, + listener: AnyListener, + options?: unknown + ) => { + originalRemove(type, guards.get(listener) ?? listener, options); + }) as never; + }; + const factory: WebSocketFactory = (url, protocols) => { const socket = create(url, protocols); connectAttempts += 1; logEvent('connect'); + try { + instrumentListeners( + socket as unknown as Parameters[0] + ); + } catch { + /* patching is best-effort; an unpatched socket still works */ + } socket.addEventListener( 'open', observe(() => { @@ -233,6 +310,12 @@ export function createSyncSocketDiagnostics( attrs['sync.ws.first_decode_error'] = firstDecodeError; } } + if (listenerErrors > 0) { + attrs['sync.ws.listener_errors'] = listenerErrors; + if (firstListenerError !== undefined) { + attrs['sync.ws.first_listener_error'] = firstListenerError; + } + } if (decodedKinds.length > 0) { attrs['sync.ws.decoded_kinds'] = decodedKinds.join(','); } @@ -264,6 +347,12 @@ export function createSyncSocketDiagnostics( if (decodeFailures > 0) { parts.push(`decode_failures=${decodeFailures}`); } + if (listenerErrors > 0) { + parts.push(`listener_errors=${listenerErrors}`); + } + if (firstListenerError !== undefined) { + parts.push(`listener_error=${firstListenerError}`); + } if (lastClose) { parts.push(`close=${lastClose.code}`); } From 9bbc51515b5dbb3f621d7e85b7afa456ccfc5e63 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 01:09:36 +0000 Subject: [PATCH 6/8] revert(ai-editing-worker): drop the snapshot fallback; keep diagnostics + dispatch fix Scope the PR to instrumentation and the dispatch fix per review: remove WorkerSyncSource, the tolerant serializer, createSyncSocket's serializer param, the protected widening of initialSyncReceived, and the worker's neverthrow dependency. Deserialize throws are still contained and recorded by the diagnostics listener guards (listener_errors / first_listener_error), so a poison frame degrades to a fully-described timeout instead of a zombie socket. Co-authored-by: teo --- bun.lock | 1 - .../collaboration/src/sync-service/socket.ts | 9 +- .../collaboration/src/sync-service/source.ts | 5 +- packages/collaboration/src/websocket/index.ts | 1 - services/ai-editing-worker/package.json | 1 - .../ai-editing-worker/src/sources.test.ts | 217 ------------------ services/ai-editing-worker/src/sources.ts | 165 +------------ .../src/sync-diagnostics.test.ts | 33 ++- .../ai-editing-worker/src/sync-diagnostics.ts | 18 -- 9 files changed, 33 insertions(+), 417 deletions(-) delete mode 100644 services/ai-editing-worker/src/sources.test.ts diff --git a/bun.lock b/bun.lock index 5fe71bba898..43353dc0c01 100644 --- a/bun.lock +++ b/bun.lock @@ -298,7 +298,6 @@ "flat": "^6.0.1", "hono": "^4.6.20", "loro-crdt": "catalog:", - "neverthrow": "^8.2.0", "p-queue": "^9.3.0", "quickjs-emscripten": "^0.31.0", "random-js": "^2.1.0", diff --git a/packages/collaboration/src/sync-service/socket.ts b/packages/collaboration/src/sync-service/socket.ts index b6b2c46f1fe..dfc402272c9 100644 --- a/packages/collaboration/src/sync-service/socket.ts +++ b/packages/collaboration/src/sync-service/socket.ts @@ -9,7 +9,6 @@ import { type WebsocketConnectionState, type WebsocketEvent, type WebsocketEventListener, - type WebsocketSerializer, } from '../websocket'; import { FromPeer, FromRemote, type IFromPeer } from './generated/schema'; @@ -44,17 +43,13 @@ export interface SyncSocket { * * `factory` optionally overrides how the underlying native socket is created * (the AI editing worker wraps the default to observe raw frames). - * - * `serializer` optionally overrides the default Bebop serializer (the AI - * editing worker wraps it to survive frames that fail to deserialize). */ export function createSyncSocket( getUrl: UrlResolver, - factory?: WebSocketFactory, - serializer?: WebsocketSerializer + factory?: WebSocketFactory ): SyncWebsocket { let builder = new WebsocketBuilder(getUrl).withSerializer( - serializer ?? new BebopSerializer(FromPeer, FromRemote) + new BebopSerializer(FromPeer, FromRemote) ); if (factory) builder = builder.withFactory(factory); return ( diff --git a/packages/collaboration/src/sync-service/source.ts b/packages/collaboration/src/sync-service/source.ts index 98406b6196e..502ebe80d10 100644 --- a/packages/collaboration/src/sync-service/source.ts +++ b/packages/collaboration/src/sync-service/source.ts @@ -94,10 +94,7 @@ export class SyncServiceSource implements LiveSyncSource { private hasEverHadListener = false; /** Pending request/response waiters, matched against each incoming message. */ private readonly waiters = new Set(); - /** Gates pushUpdate. Protected so a subclass that bootstraps the initial - * state through another exchange (the AI editing worker's snapshot - * fallback) can unlock updates without duplicating the gate. */ - protected initialSyncReceived = false; + private initialSyncReceived = false; private readonly initialSyncPromise: ResultAsync; public constructor( diff --git a/packages/collaboration/src/websocket/index.ts b/packages/collaboration/src/websocket/index.ts index 41d5464be66..3dfa765acc5 100644 --- a/packages/collaboration/src/websocket/index.ts +++ b/packages/collaboration/src/websocket/index.ts @@ -24,7 +24,6 @@ export { } from './core/websocket-event'; export type { WebsocketOptions } from './core/websocket-options'; export type { WebsocketConnectionRetryOptions } from './core/websocket-retry-options'; -export type { WebsocketSerializer } from './core/websocket-serializer'; export type { UrlResolver } from './core/websocket-url-resolver'; export { platformWebSocketFactory } from './platform/factory'; export type { diff --git a/services/ai-editing-worker/package.json b/services/ai-editing-worker/package.json index 2ad39d861d1..6ef0add7a10 100644 --- a/services/ai-editing-worker/package.json +++ b/services/ai-editing-worker/package.json @@ -24,7 +24,6 @@ "envsafe": "^2.0.3", "hono": "^4.6.20", "loro-crdt": "catalog:", - "neverthrow": "^8.2.0", "quickjs-emscripten": "^0.31.0", "yargs": "^18.0.0", "zod": "^4.4.3", diff --git a/services/ai-editing-worker/src/sources.test.ts b/services/ai-editing-worker/src/sources.test.ts deleted file mode 100644 index 1fd0aed463b..00000000000 --- a/services/ai-editing-worker/src/sources.test.ts +++ /dev/null @@ -1,217 +0,0 @@ -import { - FromRemote as FromRemoteSchema, - type FromRemote, - type IFromPeer, -} from '@macro-inc/collaboration/sync-service/generated/schema'; -import type { SyncWebsocket } from '@macro-inc/collaboration/sync-service/socket'; -import { - WebsocketConnectionState, - WebsocketEvent, - type WebsocketEventListener, -} from '@macro-inc/collaboration/websocket'; -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { createTolerantSerializer, WorkerSyncSource } from './sources'; - -type AnyListener = WebsocketEventListener< - WebsocketEvent, - IFromPeer, - FromRemote ->; - -/** Minimal in-memory SyncSocket, mirroring the collaboration source tests. */ -class FakeSocket { - connectionState = WebsocketConnectionState.Connecting; - sent: IFromPeer[] = []; - heartbeats = 0; - closed = false; - private msg = new Set(); - - send(message: IFromPeer) { - this.sent.push(message); - return true; - } - startHeartbeat() { - this.heartbeats++; - } - reconnectIfDisconnected() {} - close() { - this.closed = true; - } - addEventListener(type: K, listener: AnyListener) { - if (type === WebsocketEvent.Message) this.msg.add(listener); - } - removeEventListener( - _type: K, - listener: AnyListener - ) { - this.msg.delete(listener); - } - - deliver(data: FromRemote) { - const event = { data } as Parameters[1]; - for (const listener of [...this.msg]) listener(this as never, event); - } - - asSyncWebsocket(): SyncWebsocket { - return this as unknown as SyncWebsocket; - } -} - -const GUARDS = { - isRemoteInitialSync: () => false, - isRemoteUpdate: () => false, - isRemoteAwareness: () => false, - isRemoteSnapshot: () => false, - isRemoteUpdateAck: () => false, - isRemoteUpdateSince: () => false, -}; - -const remote = { - initialSync: (snapshot: Uint8Array, awareness: Uint8Array) => - ({ - ...GUARDS, - isRemoteInitialSync: () => true, - value: { snapshot, awareness }, - }) as unknown as FromRemote, - snapshot: (snapshot: Uint8Array) => - ({ - ...GUARDS, - isRemoteSnapshot: () => true, - value: { snapshot }, - }) as unknown as FromRemote, - ack: (id: string) => - ({ - ...GUARDS, - isRemoteUpdateAck: () => true, - value: { id }, - }) as unknown as FromRemote, -}; - -const snap = new Uint8Array([1, 2, 3]); -const aw = new Uint8Array([4, 5]); - -// From packages/collaboration/src/sync-service/source.ts. -const INITIAL_SYNC_TIMEOUT_MS = 10_000; -const FALLBACK_MS = 2_000; - -describe('WorkerSyncSource snapshot fallback', () => { - beforeEach(() => { - vi.useFakeTimers(); - }); - afterEach(() => { - vi.useRealTimers(); - }); - - it('resolves from RemoteInitialSync without ever requesting a snapshot', async () => { - const ws = new FakeSocket(); - const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); - const pending = src.doInitialSync(); - ws.deliver(remote.initialSync(snap, aw)); - - expect((await pending)._unsafeUnwrap()).toEqual({ - snapshot: snap, - awareness: aw, - }); - // The fallback timer is canceled; nothing is sent even after its delay. - await vi.advanceTimersByTimeAsync(FALLBACK_MS + 1); - expect(ws.sent).toHaveLength(0); - expect(ws.heartbeats).toBe(1); - }); - - it('bootstraps from a requested snapshot when the push never arrives', async () => { - const ws = new FakeSocket(); - const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); - const pending = src.doInitialSync(); - - expect(ws.sent).toHaveLength(0); - await vi.advanceTimersByTimeAsync(FALLBACK_MS + 1); - expect(ws.sent).toHaveLength(1); // PeerRequestSnapshot went out - - ws.deliver(remote.snapshot(snap)); - expect((await pending)._unsafeUnwrap()).toEqual({ - snapshot: snap, - awareness: new Uint8Array(0), - }); - expect(ws.heartbeats).toBe(1); - }); - - it('unlocks pushUpdate after a fallback bootstrap', async () => { - const ws = new FakeSocket(); - const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); - const pending = src.doInitialSync(); - - await vi.advanceTimersByTimeAsync(FALLBACK_MS + 1); - ws.deliver(remote.snapshot(snap)); - await pending; - - const sentBefore = ws.sent.length; - const acked = src.pushUpdate([new Uint8Array([9])]); - expect(ws.sent.length).toBe(sentBefore + 1); // update went out - const update = ws.sent[ws.sent.length - 1] as unknown as { - value: { id: string }; - }; - ws.deliver(remote.ack(update.value.id)); - expect(await acked).toBe(true); - }); - - it('surfaces the initial-sync timeout when both paths stay silent', async () => { - const ws = new FakeSocket(); - const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); - const pending = src.doInitialSync(); - - // Push times out at 10s; the 2s-delayed snapshot request at 12s. - await vi.advanceTimersByTimeAsync( - FALLBACK_MS + INITIAL_SYNC_TIMEOUT_MS + 1 - ); - - expect((await pending)._unsafeUnwrapErr()).toEqual({ - type: 'timeout', - duration: INITIAL_SYNC_TIMEOUT_MS, - }); - }); - - it('does not request a snapshot after cleanup', async () => { - const ws = new FakeSocket(); - const src = new WorkerSyncSource(ws.asSyncWebsocket(), 'doc1'); - void src.doInitialSync(); - - src.cleanup(); - await vi.advanceTimersByTimeAsync(FALLBACK_MS + 1); - - expect(ws.sent).toHaveLength(0); - expect(ws.closed).toBe(true); - }); -}); - -describe('createTolerantSerializer', () => { - it('decodes a valid frame', () => { - const onFailure = vi.fn(); - const serializer = createTolerantSerializer(onFailure); - const encoded = FromRemoteSchema.encode( - FromRemoteSchema.fromRemoteSnapshot({ snapshot: new Uint8Array([7, 8]) }) - ); - const buffer = encoded.buffer.slice( - encoded.byteOffset, - encoded.byteOffset + encoded.byteLength - ) as ArrayBuffer; - - const message = serializer.deserialize(buffer); - expect(message.isRemoteSnapshot()).toBe(true); - expect(onFailure).not.toHaveBeenCalled(); - }); - - it('returns an inert message instead of throwing on garbage bytes', () => { - const onFailure = vi.fn(); - const serializer = createTolerantSerializer(onFailure); - const garbage = new Uint8Array([0xff, 0xff, 0xff, 0xff, 0x01]).buffer; - - const message = serializer.deserialize(garbage); - expect(onFailure).toHaveBeenCalledTimes(1); - expect(message.isRemoteInitialSync()).toBe(false); - expect(message.isRemoteUpdate()).toBe(false); - expect(message.isRemoteAwareness()).toBe(false); - expect(message.isRemoteSnapshot()).toBe(false); - expect(message.isRemoteUpdateAck()).toBe(false); - expect(message.isRemoteUpdateSince()).toBe(false); - }); -}); diff --git a/services/ai-editing-worker/src/sources.ts b/services/ai-editing-worker/src/sources.ts index 4780151318c..62fb92229cc 100644 --- a/services/ai-editing-worker/src/sources.ts +++ b/services/ai-editing-worker/src/sources.ts @@ -6,27 +6,11 @@ */ import type { Awareness } from '@macro-inc/collaboration/collab/awareness'; -import type { - InitialSync, - TimeoutError, -} from '@macro-inc/collaboration/collab/source'; -import { - FromPeer, - FromRemote, - type IFromPeer, -} from '@macro-inc/collaboration/sync-service/generated/schema'; -import { - createSyncSocket, - type SyncWebsocket, -} from '@macro-inc/collaboration/sync-service/socket'; +import type { FromRemote } from '@macro-inc/collaboration/sync-service/generated/schema'; +import { createSyncSocket } from '@macro-inc/collaboration/sync-service/socket'; import { SyncServiceSource } from '@macro-inc/collaboration/sync-service/source'; -import { - BebopSerializer, - WebsocketEvent, - type WebsocketSerializer, -} from '@macro-inc/collaboration/websocket'; +import { WebsocketEvent } from '@macro-inc/collaboration/websocket'; import { EphemeralStore, type PeerID } from 'loro-crdt'; -import { ResultAsync } from 'neverthrow'; import type { SyncSocketDiagnostics } from './sync-diagnostics'; function remoteMessageKind(message: FromRemote): string { @@ -40,141 +24,7 @@ function remoteMessageKind(message: FromRemote): string { } /** - * How long the worker waits for the server-pushed `RemoteInitialSync` before - * actively requesting a snapshot over the same socket. The server normally - * sends it well under 100ms after the socket opens, so 2s only triggers when - * the push is genuinely lost. - */ -const INITIAL_SYNC_FALLBACK_MS = 2_000; - -/** Inert message: every guard is false, so listeners and waiters ignore it. */ -const POISON_FRAME: FromRemote = { - isRemoteInitialSync: () => false, - isRemoteUpdate: () => false, - isRemoteAwareness: () => false, - isRemoteSnapshot: () => false, - isRemoteUpdateAck: () => false, - isRemoteUpdateSince: () => false, -} as unknown as FromRemote; - -/** - * Bebop serializer whose deserialize failures don't take the socket down. - * - * A deserialize exception escapes the transport wrapper's message listener - * into the runtime's event dispatch, which errors out the WebSocket without a - * close event — leaving a zombie socket that never retries and can't carry - * the snapshot fallback (observed on prod/dev as `raw_frames=1 decoded=0 - * errors=1`). Swallow the failure, report it to diagnostics, and hand the - * wrapper an inert message instead, so the socket stays usable and the - * fallback still runs. - */ -export function createTolerantSerializer( - onFailure: (error: unknown) => void -): WebsocketSerializer { - const base = new BebopSerializer(FromPeer, FromRemote); - return { - serialize: (data) => base.serialize(data), - deserialize: (data) => { - try { - return base.deserialize(data as ArrayBuffer); - } catch (error) { - console.error('sync frame failed to deserialize; dropping it:', error); - onFailure(error); - return POISON_FRAME; - } - }, - binaryType: base.binaryType, - }; -} - -/** - * {@link SyncServiceSource} with a worker-only bootstrap fallback. - * - * The server pushes `RemoteInitialSync` from inside the `/connect` request - * handler. On Cloudflare that send races the hibernation manager reclaiming - * the socket when the request context ends, and a workerd regression - * (STOR-5552; fixed by cloudflare/workerd#7163, follow-up #7147) silently - * dropped sends caught in that handover: the socket stayed open but the - * snapshot never arrived, and every worker edit died on - * "initial sync failed: timeout (10000ms)". - * - * When the push stays silent for {@link INITIAL_SYNC_FALLBACK_MS}, this - * subclass sends `PeerRequestSnapshot` and bootstraps from the - * `RemoteSnapshot` reply (with empty awareness — the worker never reads it). - * That reply is served from the message handler on the manager-owned socket, - * which does not cross the affected boundary. First success wins; browsers - * keep the plain push-based flow. - */ -export class WorkerSyncSource extends SyncServiceSource { - private fallbackTimer: ReturnType | undefined; - - constructor(socket: SyncWebsocket, documentId: string) { - super(socket, documentId); - // Both are assignable instance properties on the base class; capture the - // originals, then wrap. - const withFallback = this.raceWithSnapshotFallback( - this.doInitialSync(), - socket - ); - this.doInitialSync = () => withFallback; - const baseCleanup = this.cleanup; - this.cleanup = () => { - this.cancelFallback(); - baseCleanup(); - }; - } - - private raceWithSnapshotFallback( - initial: ResultAsync, - socket: SyncWebsocket - ): ResultAsync { - const fallback = new Promise((resolve, reject) => { - this.fallbackTimer = setTimeout(() => { - this.fallbackTimer = undefined; - console.warn( - `no initial sync after ${INITIAL_SYNC_FALLBACK_MS}ms; requesting snapshot fallback:`, - this.documentId - ); - this.requestSnapshot().match((snapshot) => { - // The base class only unlocks pushUpdate and starts the heartbeat - // when RemoteInitialSync arrives; the fallback bootstrap must do - // both itself. Both are safe if the push arrives later anyway. - this.initialSyncReceived = true; - socket.startHeartbeat(); - resolve({ snapshot, awareness: new Uint8Array(0) }); - }, reject); - }, INITIAL_SYNC_FALLBACK_MS); - }); - // First success wins. When both fail, surface the first error (the - // original initial-sync timeout), matching the push-only failure shape. - return ResultAsync.fromPromise( - new Promise((resolve, reject) => { - let failures = 0; - let firstError: TimeoutError | undefined; - const fail = (error: TimeoutError) => { - firstError ??= error; - failures += 1; - if (failures === 2) reject(firstError); - }; - void initial.match((sync) => { - this.cancelFallback(); - resolve(sync); - }, fail); - fallback.then(resolve, (error) => fail(error as TimeoutError)); - }), - (error) => error as TimeoutError - ); - } - - private cancelFallback(): void { - if (this.fallbackTimer === undefined) return; - clearTimeout(this.fallbackTimer); - this.fallbackTimer = undefined; - } -} - -/** - * A {@link WorkerSyncSource} bound to a fixed, token-bearing URL. A worker + * A {@link SyncServiceSource} bound to a fixed, token-bearing URL. A worker * request is short-lived and online for its whole life, so every (re)connect * resolves to the same URL. Closes the socket on `signal` abort. * @@ -188,10 +38,7 @@ export function createWorkerSyncSource( signal?: AbortSignal, diagnostics?: SyncSocketDiagnostics ): SyncServiceSource { - const serializer = createTolerantSerializer((error) => - diagnostics?.recordDecodeFailure(error) - ); - const ws = createSyncSocket(() => wsUrl, diagnostics?.factory, serializer); + const ws = createSyncSocket(() => wsUrl, diagnostics?.factory); if (diagnostics) { // The wrapper only dispatches Message after a successful deserialize, so // raw frames (from the factory) minus these are frames that failed decode. @@ -199,7 +46,7 @@ export function createWorkerSyncSource( diagnostics.recordDecoded(remoteMessageKind(event.data)) ); } - const source = new WorkerSyncSource(ws, documentId); + const source = new SyncServiceSource(ws, documentId); signal?.addEventListener('abort', () => source.cleanup()); return source; } diff --git a/services/ai-editing-worker/src/sync-diagnostics.test.ts b/services/ai-editing-worker/src/sync-diagnostics.test.ts index 73766fc3e55..7f145b66ea8 100644 --- a/services/ai-editing-worker/src/sync-diagnostics.test.ts +++ b/services/ai-editing-worker/src/sync-diagnostics.test.ts @@ -19,6 +19,10 @@ class FakeNativeSocket { set.add(listener); } + removeEventListener(type: string, listener: Listener) { + this.listeners.get(type)?.delete(listener); + } + fire(type: string, event: unknown) { for (const listener of this.listeners.get(type) ?? []) { listener(event as never); @@ -110,23 +114,34 @@ describe('createSyncSocketDiagnostics', () => { const attrs = diagnostics.attrs(); expect(attrs['sync.ws.first_frame.hex']).toBe('fffefd'); expect(String(attrs['sync.ws.first_frame.probe'])).toMatch(/^err:/); + // The trail includes per-listener entries (lN.msg / lN.ok) around the + // frame observation from the diagnostics' own message listener. expect(String(attrs['sync.ws.events'])).toMatch( - /^connect\+\d+ms,open\+\d+ms,frame\(3\)\+\d+ms$/ + /^connect\+\d+ms,open\+\d+ms,l\d+\.msg\+\d+ms,frame\(3\)\+\d+ms,l\d+\.ok\+\d+ms$/ ); expect(diagnostics.summary()).toContain('probe=err:'); }); - it('tracks deserialize failures reported by the serializer', () => { + it('contains and records a throwing native listener', () => { const { diagnostics, sockets } = createDiagnosticsWithSocket(); diagnostics.factory('wss://example', undefined); - sockets[0].fire('open', {}); - diagnostics.recordDecodeFailure(new Error('unexpected end of buffer')); - - expect(diagnostics.attrs()).toMatchObject({ - 'sync.ws.decode_failures': 1, - 'sync.ws.first_decode_error': 'unexpected end of buffer', + const socket = sockets[0]; + // Registered through the patched addEventListener, like the wrapper's. + (socket as unknown as MinimalWebSocket).addEventListener('message', () => { + throw new Error('boom in listener'); }); - expect(diagnostics.summary()).toContain('decode_failures=1'); + + socket.fire('open', {}); + socket.fire('message', { data: new ArrayBuffer(4) }); + + const attrs = diagnostics.attrs(); + expect(attrs['sync.ws.listener_errors']).toBe(1); + expect(String(attrs['sync.ws.first_listener_error'])).toContain( + 'boom in listener' + ); + // The frame was still observed despite the throwing listener. + expect(attrs['sync.ws.raw_frames']).toBe(1); + expect(diagnostics.summary()).toContain('listener_errors=1'); }); it('counts retries as separate connect attempts and errors', () => { diff --git a/services/ai-editing-worker/src/sync-diagnostics.ts b/services/ai-editing-worker/src/sync-diagnostics.ts index ef60cfc8ed1..f3ec6dfc507 100644 --- a/services/ai-editing-worker/src/sync-diagnostics.ts +++ b/services/ai-editing-worker/src/sync-diagnostics.ts @@ -26,8 +26,6 @@ export type SyncSocketDiagnostics = { factory: WebSocketFactory; /** Record a message that made it through deserialization. */ recordDecoded(kind: string): void; - /** Record a frame the serializer failed to deserialize. */ - recordDecodeFailure(error: unknown): void; /** Flat span attributes describing everything observed so far. */ attrs(): Record; /** One-line summary for error messages, e.g. "opened=false raw_frames=0". */ @@ -112,8 +110,6 @@ export function createSyncSocketDiagnostics( | undefined; let decodedFrames = 0; const decodedKinds: string[] = []; - let decodeFailures = 0; - let firstDecodeError: string | undefined; let firstFrameHex: string | undefined; let firstFrameProbe: string | undefined; let listenerErrors = 0; @@ -278,11 +274,6 @@ export function createSyncSocketDiagnostics( decodedFrames += 1; if (decodedKinds.length < MAX_DECODED_KINDS) decodedKinds.push(kind); }, - recordDecodeFailure(error: unknown): void { - decodeFailures += 1; - firstDecodeError ??= errorText(error); - logEvent('decode_failure'); - }, attrs(): Record { const attrs: Record = { 'sync.ws.connect_attempts': connectAttempts, @@ -304,12 +295,6 @@ export function createSyncSocketDiagnostics( attrs['sync.ws.first_frame.probe'] = firstFrameProbe; } } - if (decodeFailures > 0) { - attrs['sync.ws.decode_failures'] = decodeFailures; - if (firstDecodeError !== undefined) { - attrs['sync.ws.first_decode_error'] = firstDecodeError; - } - } if (listenerErrors > 0) { attrs['sync.ws.listener_errors'] = listenerErrors; if (firstListenerError !== undefined) { @@ -344,9 +329,6 @@ export function createSyncSocketDiagnostics( if (firstFrameProbe !== undefined) { parts.push(`probe=${firstFrameProbe}`); } - if (decodeFailures > 0) { - parts.push(`decode_failures=${decodeFailures}`); - } if (listenerErrors > 0) { parts.push(`listener_errors=${listenerErrors}`); } From 5c2697a2139a566be5a2b9585a058e25909ca9fb Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 01:16:59 +0000 Subject: [PATCH 7/8] revert(ai-editing-worker): drop all sync-socket instrumentation; keep dispatch fix + decode-failure log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Net change vs main is now a single file. The wrapper's message path: - logs a clear, token-free error when an incoming frame fails to deserialize (frame shape + socket path + exception), so an undecodable initial sync is distinguishable from one that never arrived — previously the throw died inside the runtime's native dispatch with no trace; - constructs the re-dispatched MessageEvent with a data-only fallback, since copying origin/lastEventId/source can throw on runtimes with stricter Event internals (Cloudflare Workers) and error out the socket. Co-authored-by: teo --- .../collaboration/src/sync-service/socket.ts | 16 +- .../src/websocket/core/websocket-builder.ts | 1 - .../src/websocket/core/websocket.ts | 36 +- packages/collaboration/src/websocket/index.ts | 5 - .../ai-editing-worker/src/endpoints/edit.ts | 10 +- services/ai-editing-worker/src/run-edit.ts | 16 +- services/ai-editing-worker/src/sources.ts | 29 +- .../src/sync-diagnostics.test.ts | 169 --------- .../ai-editing-worker/src/sync-diagnostics.ts | 347 ------------------ 9 files changed, 42 insertions(+), 587 deletions(-) delete mode 100644 services/ai-editing-worker/src/sync-diagnostics.test.ts delete mode 100644 services/ai-editing-worker/src/sync-diagnostics.ts diff --git a/packages/collaboration/src/sync-service/socket.ts b/packages/collaboration/src/sync-service/socket.ts index dfc402272c9..c3eb6ea7530 100644 --- a/packages/collaboration/src/sync-service/socket.ts +++ b/packages/collaboration/src/sync-service/socket.ts @@ -3,7 +3,6 @@ import { BebopSerializer, ExponentialBackoff, type UrlResolver, - type WebSocketFactory, type Websocket, WebsocketBuilder, type WebsocketConnectionState, @@ -40,20 +39,11 @@ export interface SyncSocket { * heartbeat. The `getUrl` resolver is the only environment-specific bit — the * browser refreshes a permission token on each (re)connect, the worker returns * a static URL. - * - * `factory` optionally overrides how the underlying native socket is created - * (the AI editing worker wraps the default to observe raw frames). */ -export function createSyncSocket( - getUrl: UrlResolver, - factory?: WebSocketFactory -): SyncWebsocket { - let builder = new WebsocketBuilder(getUrl).withSerializer( - new BebopSerializer(FromPeer, FromRemote) - ); - if (factory) builder = builder.withFactory(factory); +export function createSyncSocket(getUrl: UrlResolver): SyncWebsocket { return ( - builder + new WebsocketBuilder(getUrl) + .withSerializer(new BebopSerializer(FromPeer, FromRemote)) // Capped exponential backoff. The scheduler calls next() before the first // retry, so the delays are 250*2^1 = 500ms doubling to a 250*2^5 = 8s // cap; 20 retries ≈ 2 minutes of automatic attempts, after which diff --git a/packages/collaboration/src/websocket/core/websocket-builder.ts b/packages/collaboration/src/websocket/core/websocket-builder.ts index 18c15d71f24..a22c70bbb0e 100644 --- a/packages/collaboration/src/websocket/core/websocket-builder.ts +++ b/packages/collaboration/src/websocket/core/websocket-builder.ts @@ -192,7 +192,6 @@ export class WebsocketBuilder { listeners: this._options?.listeners, serializer, buffer: this._options?.buffer, - factory: this._options?.factory, } as unknown as WebsocketOptions; if (serializer.binaryType !== undefined) { diff --git a/packages/collaboration/src/websocket/core/websocket.ts b/packages/collaboration/src/websocket/core/websocket.ts index 2f89ecb6af6..c6c540d1846 100644 --- a/packages/collaboration/src/websocket/core/websocket.ts +++ b/packages/collaboration/src/websocket/core/websocket.ts @@ -31,6 +31,26 @@ import { type UrlResolver, } from './websocket-url-resolver'; +/** Frame shape for the deserialize-failure log, without reading the bytes. */ +function describeFrameData(data: unknown): string { + if (typeof data === 'string') return `string of ${data.length} chars`; + if (data instanceof ArrayBuffer) + return `ArrayBuffer of ${data.byteLength} bytes`; + if (typeof Blob !== 'undefined' && data instanceof Blob) { + return `Blob of ${data.size} bytes`; + } + if (ArrayBuffer.isView(data)) { + return `${data.constructor.name} of ${data.byteLength} bytes`; + } + return typeof data; +} + +/** Strips the query (auth tokens live there) so the URL is safe to log. */ +function redactUrl(url: string): string { + const queryStart = url.indexOf('?'); + return queryStart === -1 ? url : url.slice(0, queryStart); +} + /** * A websocket wrapper that can be configured to reconnect automatically and buffer messages when the websocket is not connected. */ @@ -588,7 +608,21 @@ export class Websocket { this.dispatchEvent(type, event); }) .with({ type: WebsocketEvent.Message }, ({ event }) => { - const data = deserializeIfNeeded(event.data, this._options.serializer); + let data: Receive | WebsocketData; + try { + data = deserializeIfNeeded(event.data, this._options.serializer); + } catch (error) { + // Without this log a deserialize failure dies inside the runtime's + // native event dispatch with no trace — on Cloudflare Workers it + // errors out the socket. Log it loudly first so an undecodable + // frame (e.g. a poisoned initial sync) is distinguishable from a + // frame that never arrived. + console.error( + `websocket: failed to deserialize incoming message (${describeFrameData(event.data)}) from ${redactUrl(this._url)}:`, + error + ); + throw error; + } // Copying origin/lastEventId/source can throw on runtimes with // stricter Event internals (Cloudflare Workers): the accessors or the // MessageEvent init validation reject non-browser shapes, and a throw diff --git a/packages/collaboration/src/websocket/index.ts b/packages/collaboration/src/websocket/index.ts index 3dfa765acc5..c462975f6af 100644 --- a/packages/collaboration/src/websocket/index.ts +++ b/packages/collaboration/src/websocket/index.ts @@ -25,11 +25,6 @@ export { export type { WebsocketOptions } from './core/websocket-options'; export type { WebsocketConnectionRetryOptions } from './core/websocket-retry-options'; export type { UrlResolver } from './core/websocket-url-resolver'; -export { platformWebSocketFactory } from './platform/factory'; -export type { - MinimalWebSocket, - WebSocketFactory, -} from './platform/minimal-websocket'; export { createReconnectEffect, createSocketEffect, diff --git a/services/ai-editing-worker/src/endpoints/edit.ts b/services/ai-editing-worker/src/endpoints/edit.ts index 5ed2c779a42..d44676890e1 100644 --- a/services/ai-editing-worker/src/endpoints/edit.ts +++ b/services/ai-editing-worker/src/endpoints/edit.ts @@ -12,7 +12,6 @@ import { type Model, type ResolvedModels, runEditSession } from '../run-edit'; import { runInSandbox } from '../sandbox'; import { watchPresenceSpeed } from '../service-clients'; import { createWorkerSyncSource } from '../sources'; -import { createSyncSocketDiagnostics } from '../sync-diagnostics'; import { renderTraceMarkdown } from '../trace-log'; import { insertEditTrace } from '../traces-db'; @@ -133,13 +132,7 @@ edit.post('/', zValidator('json', EditBody), async (c) => { try { const wsUrl = `${env.SYNC_WS_BASE}/document/${documentId}/connect?token=${documentToken}`; - const syncDiagnostics = createSyncSocketDiagnostics(); - const source = createWorkerSyncSource( - wsUrl, - documentId, - signal, - syncDiagnostics - ); + const source = createWorkerSyncSource(wsUrl, documentId, signal); // Animations play at 1x while a human is watching and speed up to // `unwatchedSpeed` when nobody is, so unseen edits finish faster without @@ -167,7 +160,6 @@ edit.post('/', zValidator('json', EditBody), async (c) => { try { const result = await runEditSession({ source, - syncDiagnostics, documentId, prompt, models: buildModels(env, models, () => modelFallbacks++), diff --git a/services/ai-editing-worker/src/run-edit.ts b/services/ai-editing-worker/src/run-edit.ts index 70d2c1df066..17965d32e80 100644 --- a/services/ai-editing-worker/src/run-edit.ts +++ b/services/ai-editing-worker/src/run-edit.ts @@ -19,7 +19,6 @@ import type { UsageEntry } from './ai-editing/token-tracker'; import type { CoderRunCode, DispatchEditTrace } from './ai-editing/tools'; import { serializeWithXml } from './ai-editing/utils'; import { EditingWorkspace } from './editing-workspace'; -import type { SyncSocketDiagnostics } from './sync-diagnostics'; import { buildTraceSession, type TraceSession } from './trace-log'; export type Model = { @@ -42,8 +41,6 @@ export type ResolvedModels = { export type RunEditArgs = { /** Live sync source, already constructed by the caller (ws in prod). */ source: SyncServiceSource; - /** Diagnostics for the source's socket; reported on the sync-init span. */ - syncDiagnostics?: SyncSocketDiagnostics; documentId: string; prompt: string; models: ResolvedModels; @@ -86,21 +83,10 @@ export async function runEditSession( await Telemetry.span('edit.sync_init', async (span) => { const initialResult = await source.doInitialSync(); - const diagnostics = args.syncDiagnostics; - if (diagnostics) { - for (const [key, value] of Object.entries(diagnostics.attrs())) { - span.setAttr(key, value); - } - } if (initialResult.isErr()) { source.cleanup(); const e = initialResult.error; - // The summary rides the error into the HTTP 502 body, so callers see - // how far the connection got instead of just "timeout (10000ms)". - const detail = diagnostics ? ` [ws ${diagnostics.summary()}]` : ''; - const err = new Error( - `initial sync failed: ${e.type} (${e.duration}ms)${detail}` - ); + const err = new Error(`initial sync failed: ${e.type} (${e.duration}ms)`); span.error(err); throw err; } diff --git a/services/ai-editing-worker/src/sources.ts b/services/ai-editing-worker/src/sources.ts index 62fb92229cc..42c1b9e39e2 100644 --- a/services/ai-editing-worker/src/sources.ts +++ b/services/ai-editing-worker/src/sources.ts @@ -6,46 +6,21 @@ */ import type { Awareness } from '@macro-inc/collaboration/collab/awareness'; -import type { FromRemote } from '@macro-inc/collaboration/sync-service/generated/schema'; import { createSyncSocket } from '@macro-inc/collaboration/sync-service/socket'; import { SyncServiceSource } from '@macro-inc/collaboration/sync-service/source'; -import { WebsocketEvent } from '@macro-inc/collaboration/websocket'; import { EphemeralStore, type PeerID } from 'loro-crdt'; -import type { SyncSocketDiagnostics } from './sync-diagnostics'; - -function remoteMessageKind(message: FromRemote): string { - if (message.isRemoteInitialSync()) return 'initial_sync'; - if (message.isRemoteUpdate()) return 'update'; - if (message.isRemoteAwareness()) return 'awareness'; - if (message.isRemoteSnapshot()) return 'snapshot'; - if (message.isRemoteUpdateAck()) return 'update_ack'; - if (message.isRemoteUpdateSince()) return 'update_since'; - return 'unknown'; -} /** * A {@link SyncServiceSource} bound to a fixed, token-bearing URL. A worker * request is short-lived and online for its whole life, so every (re)connect * resolves to the same URL. Closes the socket on `signal` abort. - * - * When `diagnostics` is given, the underlying socket is created through its - * observing factory and decoded messages are counted, so a sync timeout can - * report how far the connection got. */ export function createWorkerSyncSource( wsUrl: string, documentId: string, - signal?: AbortSignal, - diagnostics?: SyncSocketDiagnostics + signal?: AbortSignal ): SyncServiceSource { - const ws = createSyncSocket(() => wsUrl, diagnostics?.factory); - if (diagnostics) { - // The wrapper only dispatches Message after a successful deserialize, so - // raw frames (from the factory) minus these are frames that failed decode. - ws.addEventListener(WebsocketEvent.Message, (_ws, event) => - diagnostics.recordDecoded(remoteMessageKind(event.data)) - ); - } + const ws = createSyncSocket(() => wsUrl); const source = new SyncServiceSource(ws, documentId); signal?.addEventListener('abort', () => source.cleanup()); return source; diff --git a/services/ai-editing-worker/src/sync-diagnostics.test.ts b/services/ai-editing-worker/src/sync-diagnostics.test.ts deleted file mode 100644 index 7f145b66ea8..00000000000 --- a/services/ai-editing-worker/src/sync-diagnostics.test.ts +++ /dev/null @@ -1,169 +0,0 @@ -import type { MinimalWebSocket } from '@macro-inc/collaboration/websocket'; -import { describe, expect, it } from 'vitest'; -import { createSyncSocketDiagnostics } from './sync-diagnostics'; - -type Listener = (event: never) => void; - -/** Fake native socket: records listeners so the test can fire raw events. */ -class FakeNativeSocket { - binaryType = 'blob'; - readyState = 0; - private readonly listeners = new Map>(); - - addEventListener(type: string, listener: Listener) { - let set = this.listeners.get(type); - if (!set) { - set = new Set(); - this.listeners.set(type, set); - } - set.add(listener); - } - - removeEventListener(type: string, listener: Listener) { - this.listeners.get(type)?.delete(listener); - } - - fire(type: string, event: unknown) { - for (const listener of this.listeners.get(type) ?? []) { - listener(event as never); - } - } -} - -function createDiagnosticsWithSocket() { - const sockets: FakeNativeSocket[] = []; - const diagnostics = createSyncSocketDiagnostics(() => { - const socket = new FakeNativeSocket(); - sockets.push(socket); - return socket as unknown as MinimalWebSocket; - }); - return { diagnostics, sockets }; -} - -describe('createSyncSocketDiagnostics', () => { - it('reports a connection that never opened', () => { - const { diagnostics, sockets } = createDiagnosticsWithSocket(); - diagnostics.factory('wss://example', undefined); - sockets[0].fire('close', { code: 1006, reason: '', wasClean: false }); - - expect(diagnostics.attrs()).toMatchObject({ - 'sync.ws.connect_attempts': 1, - 'sync.ws.opened': false, - 'sync.ws.raw_frames': 0, - 'sync.ws.decoded_frames': 0, - 'sync.ws.close.code': 1006, - 'sync.ws.close.clean': false, - }); - expect(diagnostics.summary()).toBe( - 'attempts=1 opened=false raw_frames=0 decoded=0 close=1006' - ); - }); - - it('reports a raw frame that never decoded (Blob delivery)', () => { - const { diagnostics, sockets } = createDiagnosticsWithSocket(); - diagnostics.factory('wss://example', undefined); - const socket = sockets[0]; - socket.binaryType = 'arraybuffer'; - socket.fire('open', {}); - socket.fire('message', { data: new Blob([new Uint8Array(16)]) }); - - const attrs = diagnostics.attrs(); - expect(attrs).toMatchObject({ - 'sync.ws.opened': true, - 'sync.ws.raw_frames': 1, - 'sync.ws.decoded_frames': 0, - 'sync.ws.first_frame.type': 'Blob', - 'sync.ws.first_frame.bytes': 16, - 'sync.ws.binary_type': 'arraybuffer', - }); - expect(diagnostics.summary()).toBe( - 'attempts=1 opened=true raw_frames=1 decoded=0 first_frame=Blob' - ); - }); - - it('keeps first-frame details while counting later frames and decodes', () => { - const { diagnostics, sockets } = createDiagnosticsWithSocket(); - diagnostics.factory('wss://example', undefined); - const socket = sockets[0]; - socket.binaryType = 'arraybuffer'; - socket.fire('open', {}); - socket.fire('message', { data: new ArrayBuffer(30770) }); - socket.fire('message', { data: new ArrayBuffer(4) }); - diagnostics.recordDecoded('initial_sync'); - diagnostics.recordDecoded('update_ack'); - - expect(diagnostics.attrs()).toMatchObject({ - 'sync.ws.raw_frames': 2, - 'sync.ws.decoded_frames': 2, - 'sync.ws.first_frame.type': 'ArrayBuffer', - 'sync.ws.first_frame.bytes': 30770, - 'sync.ws.decoded_kinds': 'initial_sync,update_ack', - }); - }); - - it('records hex prefix and probe decode for the first binary frame', () => { - const { diagnostics, sockets } = createDiagnosticsWithSocket(); - diagnostics.factory('wss://example', undefined); - const socket = sockets[0]; - socket.binaryType = 'arraybuffer'; - socket.fire('open', {}); - socket.fire('message', { - data: new Uint8Array([0xff, 0xfe, 0xfd]).buffer, - }); - - const attrs = diagnostics.attrs(); - expect(attrs['sync.ws.first_frame.hex']).toBe('fffefd'); - expect(String(attrs['sync.ws.first_frame.probe'])).toMatch(/^err:/); - // The trail includes per-listener entries (lN.msg / lN.ok) around the - // frame observation from the diagnostics' own message listener. - expect(String(attrs['sync.ws.events'])).toMatch( - /^connect\+\d+ms,open\+\d+ms,l\d+\.msg\+\d+ms,frame\(3\)\+\d+ms,l\d+\.ok\+\d+ms$/ - ); - expect(diagnostics.summary()).toContain('probe=err:'); - }); - - it('contains and records a throwing native listener', () => { - const { diagnostics, sockets } = createDiagnosticsWithSocket(); - diagnostics.factory('wss://example', undefined); - const socket = sockets[0]; - // Registered through the patched addEventListener, like the wrapper's. - (socket as unknown as MinimalWebSocket).addEventListener('message', () => { - throw new Error('boom in listener'); - }); - - socket.fire('open', {}); - socket.fire('message', { data: new ArrayBuffer(4) }); - - const attrs = diagnostics.attrs(); - expect(attrs['sync.ws.listener_errors']).toBe(1); - expect(String(attrs['sync.ws.first_listener_error'])).toContain( - 'boom in listener' - ); - // The frame was still observed despite the throwing listener. - expect(attrs['sync.ws.raw_frames']).toBe(1); - expect(diagnostics.summary()).toContain('listener_errors=1'); - }); - - it('counts retries as separate connect attempts and errors', () => { - const { diagnostics, sockets } = createDiagnosticsWithSocket(); - diagnostics.factory('wss://example', undefined); - sockets[0].fire('error', {}); - sockets[0].fire('close', { - code: 1002, - reason: 'protocol', - wasClean: false, - }); - diagnostics.factory('wss://example', undefined); - sockets[1].fire('error', {}); - - expect(diagnostics.attrs()).toMatchObject({ - 'sync.ws.connect_attempts': 2, - 'sync.ws.error_events': 2, - 'sync.ws.close.code': 1002, - 'sync.ws.close.reason': 'protocol', - }); - expect(diagnostics.summary()).toBe( - 'attempts=2 opened=false raw_frames=0 decoded=0 close=1002 errors=2' - ); - }); -}); diff --git a/services/ai-editing-worker/src/sync-diagnostics.ts b/services/ai-editing-worker/src/sync-diagnostics.ts deleted file mode 100644 index f3ec6dfc507..00000000000 --- a/services/ai-editing-worker/src/sync-diagnostics.ts +++ /dev/null @@ -1,347 +0,0 @@ -/** - * Passive diagnostics for the worker's sync WebSocket, built to explain - * `edit.sync_init` timeouts. Every prod timeout so far surfaced as an opaque - * "initial sync failed: timeout (10000ms)"; these counters distinguish the - * three failure modes: the socket never opened, a raw frame arrived but never - * decoded (e.g. Blob delivered where the Bebop serializer expects an - * ArrayBuffer), or frames decoded fine but no RemoteInitialSync was sent. - * - * Observation only: the factory wraps the platform default and merely attaches - * extra listeners to the native socket, so transport behavior is unchanged. - */ - -import { FromRemote } from '@macro-inc/collaboration/sync-service/generated/schema'; -import { - platformWebSocketFactory, - type WebSocketFactory, -} from '@macro-inc/collaboration/websocket'; -import type { Attribute } from '@macro-inc/observability'; - -const MAX_DECODED_KINDS = 10; -const MAX_EVENTS = 24; -const HEX_PREFIX_BYTES = 24; - -export type SyncSocketDiagnostics = { - /** Hand to {@link createSyncSocket} so raw socket events are observed. */ - factory: WebSocketFactory; - /** Record a message that made it through deserialization. */ - recordDecoded(kind: string): void; - /** Flat span attributes describing everything observed so far. */ - attrs(): Record; - /** One-line summary for error messages, e.g. "opened=false raw_frames=0". */ - summary(): string; -}; - -/** Raw `MessageEvent.data` type without reading the bytes. */ -function frameType(data: unknown): string { - if (typeof data === 'string') return 'string'; - if (data instanceof ArrayBuffer) return 'ArrayBuffer'; - if (typeof Blob !== 'undefined' && data instanceof Blob) return 'Blob'; - if (ArrayBuffer.isView(data)) return data.constructor.name; - if (data === null || data === undefined) return String(data); - return ( - (data as { constructor?: { name?: string } }).constructor?.name ?? - typeof data - ); -} - -function frameBytes(data: unknown): number | undefined { - if (typeof data === 'string') return data.length; - if (data instanceof ArrayBuffer) return data.byteLength; - if (typeof Blob !== 'undefined' && data instanceof Blob) return data.size; - if (ArrayBuffer.isView(data)) return data.byteLength; - return undefined; -} - -/** Synchronous byte view of frame data; undefined for Blob (async-only). */ -function frameView(data: unknown): Uint8Array | undefined { - if (data instanceof ArrayBuffer) return new Uint8Array(data); - if (ArrayBuffer.isView(data)) { - return new Uint8Array(data.buffer, data.byteOffset, data.byteLength); - } - return undefined; -} - -function hexPrefix(view: Uint8Array): string { - let out = ''; - const n = Math.min(view.length, HEX_PREFIX_BYTES); - for (let i = 0; i < n; i++) { - out += view[i].toString(16).padStart(2, '0'); - } - return out; -} - -function errorText(error: unknown): string { - return (error instanceof Error ? error.message : String(error)).slice(0, 160); -} - -/** - * What the first frame's bytes actually are: an independent decode attempt - * against the Bebop schema, recorded so a decode failure in the transport - * shows up with its exception text instead of as a silent `decoded=0`. - */ -function probeDecode(view: Uint8Array): string { - try { - const message = FromRemote.decode(view); - if (message.isRemoteInitialSync()) return 'ok:initial_sync'; - if (message.isRemoteUpdate()) return 'ok:update'; - if (message.isRemoteAwareness()) return 'ok:awareness'; - if (message.isRemoteSnapshot()) return 'ok:snapshot'; - if (message.isRemoteUpdateAck()) return 'ok:update_ack'; - if (message.isRemoteUpdateSince()) return 'ok:update_since'; - return 'ok:unknown'; - } catch (error) { - return `err:${errorText(error)}`; - } -} - -export function createSyncSocketDiagnostics( - create: WebSocketFactory = platformWebSocketFactory -): SyncSocketDiagnostics { - let connectAttempts = 0; - let opened = false; - let errorEvents = 0; - let rawFrames = 0; - let firstFrame: - | { type: string; bytes?: number; binaryType: string } - | undefined; - let lastClose: - | { code: number; reason: string; wasClean: boolean } - | undefined; - let decodedFrames = 0; - const decodedKinds: string[] = []; - let firstFrameHex: string | undefined; - let firstFrameProbe: string | undefined; - let listenerErrors = 0; - let firstListenerError: string | undefined; - // Ordered event log with ms offsets from diagnostics creation, so "did the - // error fire before or after the frame" is answerable from the span. - const startedAt = Date.now(); - const events: string[] = []; - const logEvent = (name: string) => { - if (events.length < MAX_EVENTS) { - events.push(`${name}+${Date.now() - startedAt}ms`); - } - }; - - // These listeners are attached to the native socket ahead of the transport - // wrapper's own handlers. A listener exception here could suppress the - // wrapper's dispatch for that event (or, under workerd's fail-fast event - // dispatch, error out the socket) — the one way observation could break - // delivery. Nothing below should be able to throw; swallow anyway so - // diagnostics can never cause the failure they measure. - const observe = (record: (event: E) => void): ((event: E) => void) => { - return (event) => { - try { - record(event); - } catch { - /* never let diagnostics interfere with delivery */ - } - }; - }; - - /** - * Route every listener registered on the native socket (the transport - * wrapper's included) through a guard that contains and records exceptions. - * On prod/dev the message dispatch dies between our first-registered - * listener and the wrapper's handler (`frame → error` in the same ms, - * `decoded=0`, no deserialize attempt): either a later listener throws on - * the runtime's stricter Event objects, or the runtime aborts multi-listener - * dispatch. The guard's per-listener start/ok/threw trail distinguishes - * those, and containment keeps a throwing listener from erroring the socket. - */ - const instrumentListeners = (socket: { - addEventListener: (...args: never[]) => void; - removeEventListener: (...args: never[]) => void; - }) => { - type AnyListener = (event: unknown) => void; - const originalAdd = socket.addEventListener.bind( - socket - ) as unknown as (type: string, listener: AnyListener, options?: unknown) => void; - const originalRemove = socket.removeEventListener.bind( - socket - ) as unknown as (type: string, listener: AnyListener, options?: unknown) => void; - const guards = new WeakMap(); - let nextId = 0; - - socket.addEventListener = (( - type: string, - listener: AnyListener, - options?: unknown - ) => { - const id = nextId++; - const guard: AnyListener = (event) => { - // Trail only the first frame's message dispatch to keep events short. - const trail = type === 'message' && rawFrames <= 1; - if (trail) logEvent(`l${id}.msg`); - try { - listener.call(socket, event as never); - if (trail) logEvent(`l${id}.ok`); - } catch (error) { - listenerErrors += 1; - const stackLine = - error instanceof Error && error.stack - ? (error.stack.split('\n')[1] ?? '').trim() - : ''; - firstListenerError ??= - `${type} listener: ${errorText(error)}${stackLine ? ` @ ${stackLine}` : ''}`.slice( - 0, - 300 - ); - logEvent(`l${id}.threw(${type})`); - console.error( - `sync socket ${type} listener threw (contained):`, - error - ); - } - }; - guards.set(listener, guard); - originalAdd(type, guard, options); - }) as never; - - socket.removeEventListener = (( - type: string, - listener: AnyListener, - options?: unknown - ) => { - originalRemove(type, guards.get(listener) ?? listener, options); - }) as never; - }; - - const factory: WebSocketFactory = (url, protocols) => { - const socket = create(url, protocols); - connectAttempts += 1; - logEvent('connect'); - try { - instrumentListeners( - socket as unknown as Parameters[0] - ); - } catch { - /* patching is best-effort; an unpatched socket still works */ - } - socket.addEventListener( - 'open', - observe(() => { - opened = true; - logEvent('open'); - }) - ); - socket.addEventListener( - 'error', - observe(() => { - errorEvents += 1; - logEvent('error'); - }) - ); - socket.addEventListener( - 'close', - observe((event) => { - lastClose = { - code: event.code, - reason: event.reason, - wasClean: event.wasClean, - }; - logEvent(`close(${event.code})`); - }) - ); - socket.addEventListener( - 'message', - observe((event) => { - rawFrames += 1; - logEvent(`frame(${frameBytes(event.data) ?? '?'})`); - if (firstFrame === undefined) { - firstFrame = { - type: frameType(event.data), - bytes: frameBytes(event.data), - // As seen when the frame arrived — the wrapper sets 'arraybuffer', - // so anything else here means the runtime ignored the setter. - binaryType: socket.binaryType, - }; - const view = frameView(event.data); - if (view) { - firstFrameHex = hexPrefix(view); - firstFrameProbe = probeDecode(view); - } - } - }) - ); - return socket; - }; - - return { - factory, - recordDecoded(kind: string): void { - decodedFrames += 1; - if (decodedKinds.length < MAX_DECODED_KINDS) decodedKinds.push(kind); - }, - attrs(): Record { - const attrs: Record = { - 'sync.ws.connect_attempts': connectAttempts, - 'sync.ws.opened': opened, - 'sync.ws.error_events': errorEvents, - 'sync.ws.raw_frames': rawFrames, - 'sync.ws.decoded_frames': decodedFrames, - }; - if (firstFrame) { - attrs['sync.ws.first_frame.type'] = firstFrame.type; - attrs['sync.ws.binary_type'] = firstFrame.binaryType; - if (firstFrame.bytes !== undefined) { - attrs['sync.ws.first_frame.bytes'] = firstFrame.bytes; - } - if (firstFrameHex !== undefined) { - attrs['sync.ws.first_frame.hex'] = firstFrameHex; - } - if (firstFrameProbe !== undefined) { - attrs['sync.ws.first_frame.probe'] = firstFrameProbe; - } - } - if (listenerErrors > 0) { - attrs['sync.ws.listener_errors'] = listenerErrors; - if (firstListenerError !== undefined) { - attrs['sync.ws.first_listener_error'] = firstListenerError; - } - } - if (decodedKinds.length > 0) { - attrs['sync.ws.decoded_kinds'] = decodedKinds.join(','); - } - if (events.length > 0) { - attrs['sync.ws.events'] = events.join(','); - } - if (lastClose) { - attrs['sync.ws.close.code'] = lastClose.code; - attrs['sync.ws.close.clean'] = lastClose.wasClean; - if (lastClose.reason) { - attrs['sync.ws.close.reason'] = lastClose.reason.slice(0, 256); - } - } - return attrs; - }, - summary(): string { - const parts = [ - `attempts=${connectAttempts}`, - `opened=${opened}`, - `raw_frames=${rawFrames}`, - `decoded=${decodedFrames}`, - ]; - if (firstFrame) { - parts.push(`first_frame=${firstFrame.type}`); - } - if (firstFrameProbe !== undefined) { - parts.push(`probe=${firstFrameProbe}`); - } - if (listenerErrors > 0) { - parts.push(`listener_errors=${listenerErrors}`); - } - if (firstListenerError !== undefined) { - parts.push(`listener_error=${firstListenerError}`); - } - if (lastClose) { - parts.push(`close=${lastClose.code}`); - } - if (errorEvents > 0) { - parts.push(`errors=${errorEvents}`); - } - return parts.join(' '); - }, - }; -} From 7c18cc6f9059f26ee11fa7b1d97a3dc77e73c961 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 01:19:01 +0000 Subject: [PATCH 8/8] chore(collaboration): drop redundant comment above the deserialize-failure log Co-authored-by: teo --- packages/collaboration/src/websocket/core/websocket.ts | 5 ----- 1 file changed, 5 deletions(-) diff --git a/packages/collaboration/src/websocket/core/websocket.ts b/packages/collaboration/src/websocket/core/websocket.ts index c6c540d1846..e9bcfc73a3e 100644 --- a/packages/collaboration/src/websocket/core/websocket.ts +++ b/packages/collaboration/src/websocket/core/websocket.ts @@ -612,11 +612,6 @@ export class Websocket { try { data = deserializeIfNeeded(event.data, this._options.serializer); } catch (error) { - // Without this log a deserialize failure dies inside the runtime's - // native event dispatch with no trace — on Cloudflare Workers it - // errors out the socket. Log it loudly first so an undecodable - // frame (e.g. a poisoned initial sync) is distinguishable from a - // frame that never arrived. console.error( `websocket: failed to deserialize incoming message (${describeFrameData(event.data)}) from ${redactUrl(this._url)}:`, error