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 11 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
374 changes: 30 additions & 344 deletions packages/client/src/__tests__/sse-connection.test.ts
View file
Open in desktop

Large diffs are not rendered by default.

132 changes: 43 additions & 89 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,8 @@ 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;
/** Sessions that need a reconnect POST — set when reconnect fails, retried on next welcome. */
private _pendingReconnectSessions: Array<{ sessionId: string; lastSeq: number }> | null = null;
private boundOnVisibility: (() => void) | null = null;
private boundOnPageShow: ((e: PageTransitionEvent) => void) | null = null;
private boundOnPageHide: (() => void) | null = null;
Expand All @@ -47,7 +47,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 +59,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 All @@ -88,12 +83,12 @@ export class SseConnection implements ChatConnection {
if (!endpoint) return false;

if (this._connected && this._connectionId) {
this.doPost(endpoint, msg);
this.doPost(endpoint, msg).catch(() => {});

@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.

🔵 style: Bare .catch(() => {}) on four doPost calls silently swallows errors. Now that doPost logs a console.warn before throwing, the catch is functionally correct, but the pattern is fragile — a future caller might assume doPost doesn't throw. Consider having doPost not throw (log-only) for fire-and-forget calls, and a separate doPostOrThrow for the reconnect path that needs error discrimination.

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 @@ -150,7 +145,7 @@ export class SseConnection implements ChatConnection {

// Try POST first
if (this._connected && this._connectionId) {
this.doPost('suspend', { type: 'session_suspend', sessions });
this.doPost('suspend', { type: 'session_suspend', sessions }).catch(() => {});
return;
}

Expand All @@ -175,7 +170,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 +187,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 +206,38 @@ 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;
if (this._isReconnect && this.seqBySession.size > 0) {
this.doReconnectPost(welcomeConnectionId, welcomeEs);
// Fire reconnect POST if reconnecting with sessions, or retry a
// previously failed reconnect. handleSendV2 handles ownership on first
// message, and replayed events arrive via SSE regardless.
const sessions =

@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: _pendingReconnectSessions is used as a retry payload but is never reconciled with seqBySession. If a session is cleared (clearSession) while _pendingReconnectSessions holds a reference to it, the next welcome retries a reconnect POST for a session the client no longer tracks. The stale entry will cause unnecessary server work and log noise. Consider filtering _pendingReconnectSessions against current seqBySession keys before retrying. [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.

🔵 style: The session-selection logic (lines 217-227) is dense: a null-coalescing chain of filtered/mapped pending sessions falling back to a ternary for fresh sessions. Consider extracting to a private method like getReconnectSessions() for readability. [fixable]

this._pendingReconnectSessions ??
(this._isReconnect && this.seqBySession.size > 0
? Array.from(this.seqBySession.entries()).map(([sessionId, lastSeq]) => ({
sessionId,
lastSeq,
}))
: null);
this._connected = true;

@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: Setting _connected = true before the reconnect POST completes means send() (line 86-88) will fire user messages immediately via doPost(), potentially arriving at the server before the reconnect POST. The comment at ws-handler-v2.ts:563-569 documents this is safe due to client-side seq dedup, but there is no server-side ordering guarantee. If a user message arrives before handleReconnect sets up the cursor, broadcast() events may be delivered at cursor=0 until the reconnect POST arrives. This is self-healing but could cause a burst of duplicate events.

if (sessions) {
this._pendingReconnectSessions = sessions;
this.doPost('reconnect', { type: 'reconnect', sessions }).then(

@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: The reconnect POST's .then() callback clears _pendingReconnectSessions and flushes sends without checking whether _connectionId has changed since the POST was initiated. If a new welcome arrives while the POST is in-flight (EventSource auto-reconnect), the new welcome captures _pendingReconnectSessions (line 213), starts a second POST, and then the stale .then() clears the flag. This is benign only if the server handles duplicate reconnect POSTs with different connection IDs idempotently — which it does (handleReconnect is stateless per call) — but the double flushPendingSends() could cause sends to fire twice. Consider capturing _connectionId before the POST and comparing in the callback. [fixable]

() => {
this._pendingReconnectSessions = null;
// Flush pending sends AFTER reconnect so the server processes
// handleReconnect (cursor reset, replay) before user messages.
this.flushPendingSends();
},
() => {
// doPost already logs the warning. Keep _pendingReconnectSessions
// so the next EventSource reconnect retries automatically.
// Still flush — handleSendV2 handles ownership independently.
this.flushPendingSends();
},
);
} else {
this._connected = true;
this.flushPendingSends();
this.listener?.({ type: '_open' });
}
this.listener?.({ type: '_open' });
this._isReconnect = true;
});

Expand Down Expand Up @@ -262,66 +271,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 All @@ -333,9 +282,14 @@ export class SseConnection implements ChatConnection {
},
body: JSON.stringify(body),
});
} catch {
} catch (err) {
// POST failures are non-fatal — the server may be temporarily
// unreachable. The SSE stream will reconnect and replay missed events.
// unreachable. SSE EventSource auto-reconnects and replays missed events
// from the EventStore. However, a failed reconnect POST means the server
// won't reset the cursor or re-send boot context until the next
// reconnect cycle. Client-side seq dedup prevents duplicate delivery.
console.warn(`[mitzo] ${endpoint} POST failed:`, err instanceof Error ? err.message : err);
throw err;
}
}

Expand All @@ -344,7 +298,7 @@ export class SseConnection implements ChatConnection {
const toFlush = this.pendingSends;
this.pendingSends = [];
for (const { endpoint, body } of toFlush) {
this.doPost(endpoint, body);
this.doPost(endpoint, body).catch(() => {});
}
}

Expand Down
Loading
Loading