diff --git a/src/remote-device/remote-channel.ts b/src/remote-device/remote-channel.ts index 0f42478d..44878583 100644 --- a/src/remote-device/remote-channel.ts +++ b/src/remote-device/remote-channel.ts @@ -28,6 +28,9 @@ const RECREATE_TIMEOUT_MS = 30000; // resolves/errors well within this window; 3 health ticks of unbroken 'joining' means the // state machine has stalled and only a fresh socket (via recreate) recovers it. const JOINING_WEDGE_TIMEOUT_MS = 30000; +const RECONNECT_BASE_DELAY_MS = 10000; +const RECONNECT_MAX_DELAY_MS = 300000; +const MAX_PROCESSED_CALL_IDS = 1000; export class RemoteChannel { private client: SupabaseClient | null = null; @@ -50,6 +53,9 @@ export class RemoteChannel { private reconnectAttempt = 0; // recreateChannel() attempts since last success private isRecreatingChannel = false; // a recreate is in flight (re-entrancy guard) private joiningSince: number | null = null; // ts the channel entered an unbroken 'joining' run; null when not joining + private nextReconnectAt = 0; + private heartbeatDeviceId: string | null = null; + private processedCallIds = new Set(); private _user: User | null = null; get user(): User | null { return this._user; } @@ -214,7 +220,19 @@ export class RemoteChannel { filter: `user_id=eq.${this.user.id}` }, (payload: any) => { - console.debug('[DEBUG] Realtime event received, payload:', payload?.new?.id); + const callId = payload?.new?.id; + console.debug('[DEBUG] Realtime event received, payload:', callId); + if (typeof callId === 'string' && callId) { + if (this.processedCallIds.has(callId)) { + console.debug('[DEBUG] Ignoring duplicate realtime call:', callId); + return; + } + this.processedCallIds.add(callId); + if (this.processedCallIds.size > MAX_PROCESSED_CALL_IDS) { + const oldest = this.processedCallIds.values().next().value; + if (oldest) this.processedCallIds.delete(oldest); + } + } if (this.onToolCall) { this.onToolCall(payload); } @@ -227,6 +245,7 @@ export class RemoteChannel { if (status === 'SUBSCRIBED') { const recovered = this.reconnectAttempt; this.reconnectAttempt = 0; + this.nextReconnectAt = 0; console.log(`✅ Channel subscribed${recovered > 0 ? ` (recovered after ${recovered} attempt${recovered === 1 ? '' : 's'})` : ''}`); // Update device status on successful connection if (this.deviceId) { @@ -355,6 +374,9 @@ export class RemoteChannel { console.debug('[DEBUG] recreateChannel() skipped - already in progress'); return; } + if (Date.now() < this.nextReconnectAt) { + return; + } this.isRecreatingChannel = true; this.reconnectAttempt++; @@ -384,8 +406,13 @@ export class RemoteChannel { await this.createChannel(); }, RECREATE_TIMEOUT_MS, 'recreateChannel'); } catch (err: any) { + const delay = Math.min( + RECONNECT_BASE_DELAY_MS * (2 ** Math.max(0, this.reconnectAttempt - 1)), + RECONNECT_MAX_DELAY_MS, + ); + this.nextReconnectAt = Date.now() + delay; captureRemote('remote_channel_recreate_error', { errMsg: err?.message, attempt: this.reconnectAttempt }); - console.debug(`[DEBUG] Channel recreation failed: ${err?.message} — ${this.connState()}`); + console.debug(`[DEBUG] Channel recreation failed: ${err?.message}; retry in ${delay}ms — ${this.connState()}`); } finally { this.isRecreatingChannel = false; } @@ -450,6 +477,12 @@ export class RemoteChannel { } startHeartbeat(deviceId: string) { + if (this.heartbeatDeviceId === deviceId && this.connectionCheckInterval && this.heartbeatInterval) { + console.debug('[DEBUG] Heartbeat already active for device:', deviceId); + return; + } + this.stopHeartbeat(); + this.heartbeatDeviceId = deviceId; console.debug('[DEBUG] Starting heartbeat for device:', deviceId); this.connectionCheckInterval = setInterval(() => { this.checkConnectionHealth(); @@ -471,6 +504,7 @@ export class RemoteChannel { clearInterval(this.connectionCheckInterval); this.connectionCheckInterval = null; } + this.heartbeatDeviceId = null; } async setOnlineStatus(deviceId: string, status: 'online' | 'offline') { diff --git a/test/test-remote-channel-reconnect.js b/test/test-remote-channel-reconnect.js index fa939360..550e1cd3 100644 --- a/test/test-remote-channel-reconnect.js +++ b/test/test-remote-channel-reconnect.js @@ -256,6 +256,68 @@ async function goHalfOpenThenDrive(rc, client) { } async function main() { + await test('heartbeat startup is idempotent for the same device', async () => { + const { rc } = makeRemoteChannel(); + const originalSetInterval = globalThis.setInterval; + const originalClearInterval = globalThis.clearInterval; + let created = 0; + globalThis.setInterval = () => ({ id: ++created }); + globalThis.clearInterval = () => {}; + try { + await withQuietLogs(async () => { + rc.sendHeartbeat = async () => {}; + rc.startHeartbeat('device-1'); + rc.startHeartbeat('device-1'); + }); + assert.strictEqual(created, 2, 'expected one connection timer and one heartbeat timer'); + } finally { + rc.stopHeartbeat(); + globalThis.setInterval = originalSetInterval; + globalThis.clearInterval = originalClearInterval; + } + }); + + await test('duplicate realtime call IDs dispatch once', async () => { + const { rc, client } = makeRemoteChannel(); + let realtimeHandler; + let dispatched = 0; + const originalChannel = client.channel.bind(client); + client.channel = (topic) => { + const channel = originalChannel(topic); + channel.on = (_event, _filter, handler) => { + realtimeHandler = handler; + return channel; + }; + return channel; + }; + rc.onToolCall = () => { dispatched++; }; + + await withQuietLogs(async () => { + await rc.createChannel(); + await flush(); + const payload = { new: { id: 'call-1', tool_name: 'read_file', arguments: {} } }; + realtimeHandler(payload); + realtimeHandler(payload); + }); + assert.strictEqual(dispatched, 1, 'same queued call ID should only dispatch once'); + }); + + await test('failed recreates observe bounded backoff before retrying', async () => { + const { rc } = makeRemoteChannel(); + rc.channel = { state: 'errored' }; + rc.client.removeChannel = async () => { rc.channel = null; }; + rc.client.realtime.disconnect = async () => {}; + rc.createChannel = async () => { throw new Error('simulated reconnect failure'); }; + + await withQuietLogs(() => rc.recreateChannel()); + assert.strictEqual(rc.reconnectAttempt, 1); + assert(rc.nextReconnectAt > Date.now(), 'failed recreate should schedule a future retry'); + + rc.channel = { state: 'errored' }; + await withQuietLogs(() => rc.recreateChannel()); + assert.strictEqual(rc.reconnectAttempt, 1, 'retry inside backoff must not execute'); + }); + // CONTROL: prove the harness CAN observe recovery — when the dead socket is // actually torn down (disconnect()), the next recreate re-subscribes. await test('control: recovers when the half-open socket is torn down before recreate', async () => {