Skip to content

refactor(transport): SSOT phase 3 — clean reconnect #440

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dimakis wants to merge 15 commits into
base: main
Choose a base branch
from
Open

refactor(transport): SSOT phase 3 — clean reconnect #440

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
53d1571
refactor(server): remove ownership dance from handleReconnect (P3)
dimakis Jul 4, 2026
9c1cbb6
refactor(harness): remove periodic sync from ConnectionRegistry (P3)
dimakis Jul 4, 2026
6c91ae9
refactor(client): fire-and-forget reconnect POST in SseConnection (P3)
dimakis Jul 4, 2026
1f94b7a
refactor(protocol): remove ReconnectMessage from WS union (P3)
dimakis Jul 4, 2026
74e7d46
test(ws-handler-v2): fix 2 failing tests from P3 reconnect simplifica…
dimakis Jul 4, 2026
fe40257
fix(test): update mocks and schema test for P3 reconnect changes
dimakis Jul 25, 2026
263f7bc
fix(server): remove dead reconnect case from WS message handler
dimakis Jul 25, 2026
031c537
fix(server): guard queryInstance.interrupt() against ProcessTransport…
dimakis Jul 25, 2026
66ef2cf
fix(transport): address Centaur review — restore WS reconnect, docume…
dimakis Jul 25, 2026
13153d6
fix(transport): address second Centaur review — all 6 findings
dimakis Jul 25, 2026
1737898
style(server): fix prettier formatting in ws-handler-v2
dimakis Jul 25, 2026
76c4040
fix(transport): address third Centaur review — 4 findings
dimakis Jul 25, 2026
46c3750
fix(transport): address fourth Centaur review — 3 findings
dimakis Jul 25, 2026
d50296f
fix(transport): address fifth Centaur review — 8 findings
dimakis Jul 25, 2026
c72a27a
fix(transport): address sixth Centaur review — 5 findings
dimakis Jul 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
376 changes: 11 additions & 365 deletions packages/client/src/__tests__/sse-connection.test.ts
View file
Open in desktop

Large diffs are not rendered by default.

100 changes: 15 additions & 85 deletions packages/client/src/sse-connection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ export interface SseConnectionConfig {
fetch: (url: string, init?: RequestInit) => Promise<Response>;
/** Factory for EventSource — allows injection for testing. */
createEventSource?: (url: string) => EventSource;
reconnectDelayMs?: number;
/** URL for the sendBeacon suspend fallback. */
suspendUrl?: string;
}
Expand All @@ -38,7 +37,6 @@ export class SseConnection implements ChatConnection {
private listener: ConnectionListener | null = null;
private seqBySession = new Map<string, number>();
private pendingSends: Array<{ endpoint: string; body: Record<string, unknown> }> = [];
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
private boundOnVisibility: (() => void) | null = null;
private boundOnPageShow: ((e: PageTransitionEvent) => void) | null = null;
private boundOnPageHide: (() => void) | null = null;
Expand All @@ -47,7 +45,6 @@ export class SseConnection implements ChatConnection {
constructor(config: SseConnectionConfig) {
this.config = {
createEventSource: (url: string) => new EventSource(url),
reconnectDelayMs: 500,
suspendUrl: '',
...config,
};
Expand All @@ -60,10 +57,6 @@ export class SseConnection implements ChatConnection {

disconnect(): void {

@dimakis dimakis Jul 25, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 bugs: disconnect() does not clear _pendingReconnectSessions. If a reconnect POST failed (setting _pendingReconnectSessions), then the user disconnects and later reconnects, the stale sessions array will be retried on the next welcome — even if those sessions no longer exist locally (cleared via clearSession). Add this._pendingReconnectSessions = null; in disconnect(). [fixable]

@dimakis dimakis Jul 25, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 bugs: disconnect() does not clear _pendingReconnectSessions. If disconnect() is called while a reconnect POST is in-flight, the .then() callback can still fire and call flushPendingSends() after the connection is torn down. Additionally, the stale sessions array persists in memory. Add this._pendingReconnectSessions = null; to disconnect(). [fixable]

this.removeBrowserListeners();
if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
if (this.es) {
this.es.close();
this.es = null;
Expand Down Expand Up @@ -92,8 +85,8 @@ export class SseConnection implements ChatConnection {
return true;
}

// Queue if reconnecting
if (this.reconnectTimer || this.es) {
// Queue if EventSource exists (reconnecting)
if (this.es) {
if (this.pendingSends.length >= MAX_PENDING_SENDS) {
this.pendingSends.shift();
}
Expand Down Expand Up @@ -175,7 +168,6 @@ export class SseConnection implements ChatConnection {
*/
checkAndReconnect(force = false): void {
if (!force && this._connected) return;
if (this.reconnectTimer) return;
if (this.es) {
this.es.close();
this.es = null;
Expand All @@ -193,11 +185,6 @@ export class SseConnection implements ChatConnection {
private doConnect(): void {
if (this.es) return;

if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}

// Always use the base URL — reconnect sessions are sent via POST in the
// welcome handler. This avoids the bug where EventSource auto-reconnect
// reuses the original URL (missing ?sessions=), and eliminates double
Expand All @@ -217,18 +204,21 @@ export class SseConnection implements ChatConnection {
}
this._connectionId = msg.connectionId as string;

// _connected deferred until doReconnectPost succeeds — prevents
// external send() from bypassing the pending queue mid-reconnect.
// Capture both connectionId and ES instance for the staleness guard.
const welcomeConnectionId = this._connectionId;
const welcomeEs = this.es;
// Fire reconnect POST (fire-and-forget) if reconnecting with sessions.
// No need to defer _connected — handleSendV2 handles ownership on first
// message, and replayed events arrive via SSE regardless.
if (this._isReconnect && this.seqBySession.size > 0) {
this.doReconnectPost(welcomeConnectionId, welcomeEs);
} else {
this._connected = true;
this.flushPendingSends();
this.listener?.({ type: '_open' });
this.doPost('reconnect', {

@dimakis dimakis Jul 25, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 unsafe_assumptions: Fire-and-forget reconnect POST means sends can arrive at the server before handleReconnect runs. handleSendV2 calls watch() (line 541 of ws-handler-v2.ts) but does NOT call resetCursor(), so the cursor starts at 0 for that connection+session. If a broadcast fires during the race window (between send and reconnect POST processing), the cursor advances. When handleReconnect then calls resetCursor(lastSeq), it could roll the cursor backward, potentially causing duplicate event delivery on the next reconnect. Consider adding a resetCursor call in the handleSendV2 takeover/reattach path, or documenting why this is acceptable. [fixable]

type: 'reconnect',
sessions: Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({
sessionId,
lastSeq,
})),
});
}
this._connected = true;
this.flushPendingSends();
this.listener?.({ type: '_open' });
this._isReconnect = true;
});

Expand Down Expand Up @@ -262,66 +252,6 @@ export class SseConnection implements ChatConnection {
// but we wait for the 'welcome' event before marking as connected.
}

/**
* Send the reconnect POST and only mark connected on success.
*
* On failure the client stays disconnected — the next EventSource
* auto-reconnect will trigger a fresh welcome + retry. This prevents
* flushing pending sends into the void when the server never ran
* handleReconnect (no watch, no reattach, no replay).
*/
private async doReconnectPost(
welcomeConnectionId: string,
welcomeEs: EventSource | null,
): Promise<void> {
try {
const res = await this.config.fetch(`${this.config.baseUrl}/api/chat/reconnect`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-Connection-ID': welcomeConnectionId,
},
body: JSON.stringify({
type: 'reconnect',
sessions: Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({
sessionId,
lastSeq,
})),
}),
});

// Guard: bail if disconnect() was called, a newer welcome arrived,
// or checkAndReconnect replaced the EventSource while in-flight.
if (!this.es || this.es !== welcomeEs || this._connectionId !== welcomeConnectionId) return;

if (res.ok) {
this._connected = true;
this.flushPendingSends();
this.listener?.({ type: '_open' });
} else {
console.warn('[SseConnection] reconnect POST returned', res.status);
this.scheduleReconnect();
}
} catch (err) {
if (!this.es || this.es !== welcomeEs || this._connectionId !== welcomeConnectionId) return;
console.warn('[SseConnection] reconnect POST failed', err);
this.scheduleReconnect();
}
}

/** Tear down and reconnect after a delay to avoid tight retry loops. */
private scheduleReconnect(): void {
if (this.reconnectTimer) return;
if (this.es) {
this.es.close();
this.es = null;
}
this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null;
this.doConnect();
}, this.config.reconnectDelayMs);
}

private async doPost(endpoint: string, body: Record<string, unknown>): Promise<void> {

@dimakis dimakis Jul 25, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 bugs: doPost does not check res.ok. fetch() only rejects on network errors — an HTTP 500 resolves the promise successfully. The reconnect POST .then() handler (line 224) clears _pendingReconnectSessions on resolve, so a server 500 is treated as success: cursor is never reset, replay never happens, and the retry mechanism is silently disarmed. The old doReconnectPost explicitly checked res.ok and called scheduleReconnect() on non-ok responses. Fix: check res.ok in doPost and throw on non-ok, or check it at the reconnect call site. [fixable]

if (!this._connectionId) return;

@dimakis dimakis Jul 25, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 bugs: doPost silently resolves (doesn't throw) when _connectionId is null. When called from the reconnect .then() chain (line 236), if _connectionId were somehow cleared between the welcome event and the fetch, the success handler would fire without the POST ever being sent — marking _connected=true and flushing sends into the void. In practice this path is unreachable (connectionId is set at line 208 before the POST), but the contract is misleading: callers assume doPost either succeeds or throws, yet this early return is a silent success. [fixable]

try {
Expand Down
Loading
Loading