feat: feat(cluster-protocol): publish the TypeScript cluster client with full v1 parity - #801
feat: feat(cluster-protocol): publish the TypeScript cluster client with full v1 parity#801tomdps wants to merge 1 commit into
Conversation
… cluster client with full v1 parity
Greptile SummaryAdds and publishes a TypeScript v1 cluster-protocol client.
Confidence Score: 2/5The reconnect and queue lifecycle defects need to be fixed before merging because they can leak live sockets and subscriptions or leave consumers permanently pending. Durable reconnects are neither serialized nor coordinated with close and do not clean up failed setup, while the bounded queue stores only one pending receiver and loses earlier concurrent waiters. Files Needing Attention: src/cluster/durable-watch.ts, src/cluster/multiplexed-transport.ts
|
| Filename | Overview |
|---|---|
| src/cluster/durable-watch.ts | Implements full-socket watch recovery, but reconnect failure, concurrent reads, and close races can leak or revive transports. |
| src/cluster/multiplexed-transport.ts | Adds request demultiplexing and bounded subscription queues, but concurrent queue receives can permanently strand a consumer. |
| src/cluster/websocket-transport.ts | Binds the multiplexed transport to browser-compatible and Node WebSocket implementations. |
| src/cluster/subscription-stream.ts | Adds common cursorless subscription establishment, notification parsing, iteration, and cancellation. |
| src/cluster/watch-subscription.ts | Adds cursor-aware watch subscriptions, deduplication, and same-transport reconnection. |
| src/cluster/cluster-client.ts | Adds typed unary cluster methods with response validation and AbortSignal cancellation. |
| package.json | Publishes the cluster subpath and integrates generation, build, and package checks into release workflows. |
Sequence Diagram
sequenceDiagram
participant Consumer
participant DurableWatch
participant OldSocket
participant FreshSocket
Consumer->>DurableWatch: next()
OldSocket-->>DurableWatch: stream ended
DurableWatch->>FreshSocket: connect()
DurableWatch->>FreshSocket: "get({})"
DurableWatch->>FreshSocket: watch(runId, fromCursor)
DurableWatch->>OldSocket: close()
DurableWatch-->>Consumer: resumed event stream
Reviews (1): Last reviewed commit: "feat: implement #748 - feat(cluster-prot..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| const freshTransport = await WebSocketTransport.connect(this.url, this.options); | ||
| await new ClusterClient(freshTransport).get({}); | ||
| const { stream } = await new WatchSubscriptionClient(freshTransport).watchWithDedup( | ||
| { runId, fromCursor }, | ||
| dedup | ||
| ); |
There was a problem hiding this comment.
When the fresh WebSocket connects but get() or watchWithDedup() rejects, this method propagates the error without closing freshTransport, causing repeated reconnect failures to leak sockets and any subscription established before the failure.
| this.transport.close(); | ||
| this.transport = freshTransport; | ||
| this.stream = stream; |
There was a problem hiding this comment.
When close() completes while reconnect setup is awaiting connection, snapshot retrieval, or watch establishment, this code unconditionally installs the fresh transport afterward, leaving a closed client with a live WebSocket and server-side subscription that it will not consume or close.
| if (outcome !== null) return outcome; | ||
| if (this.closing) return null; | ||
| await this.reconnectFullSocket(); |
There was a problem hiding this comment.
When two consumers call next() after the old queue has ended, both calls start reconnectFullSocket() without sharing an in-flight attempt. Each opens a socket and subscription, and the later assignment overwrites the earlier transport, leaving the first reconnect's socket and subscription unowned.
| recv(): Promise<string | null> { | ||
| if (this.buffer.length > 0) return Promise.resolve(this.buffer.shift() ?? null); | ||
| if (this.closed) return Promise.resolve(null); |
There was a problem hiding this comment.
When a subscription stream receives two concurrent next() calls while its queue is empty, the second recv() overwrites resolveWaiting. A later event or queue shutdown resolves only the second promise, leaving the first consumer permanently pending.
|
External review found four unresolved P1 lifecycle races: failed reconnect leaks the fresh transport, close racing reconnect can revive a closed durable watch, concurrent reads spawn duplicate reconnects and orphan a transport, and the queue's single waiting receiver leaves an earlier concurrent next() pending forever. Closing this attempt; the replacement contract must serialize reconnect, coordinate close, clean failed setup, and support/reject concurrent receivers deterministically. |
1 participant
Closes #748