Skip to content

fix: wait for ReconnectResponse to arrive before declaring signal rec… #2082

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
lukasIO wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix: wait for ReconnectResponse to arrive before declaring signal rec… #2082

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
88e322c
fix: wait for ReconnectResponse to arrive before declaring signal rec…
lukasIO Sep 2, 2026
f0387f6
clarify
lukasIO Sep 2, 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
4 changes: 1 addition & 3 deletions src/api/SignalClient.e2e.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,9 +174,7 @@ describe.skipIf(!!unavailable)('SignalClient e2e', () => {

it('rejects a reconnect when a leave arrives as the first message', async () => {
// Reconnecting into a room that sends leave-first exercises the client's
// first-message validation while in RECONNECTING (path-independent: it
// doesn't rely on the mock detecting reconnect, which v1 hides inside the
// gzipped join_request the mock ignores).
// first-message validation while in RECONNECTING.
await join('happy');
const token = await createToken({ signal: 'leave_first_message' });
const err = await client.reconnect(serverUrl, token, 'RM_session').then(
Expand Down
24 changes: 14 additions & 10 deletions src/api/SignalClient.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ describe('SignalClient.connect', () => {
expect(signalClient.currentState).toBe(SignalConnectionState.CONNECTED);
});

it('should handle reconnect with non-reconnect message (edge case)', async () => {
it('drops stray messages during reconnect and waits for the reconnect response', async () => {
// First, initial connection
const joinResponse = createJoinResponse();
const joinSignalResponse = createSignalResponse('join', joinResponse);
Expand All @@ -247,19 +247,24 @@ describe('SignalClient.connect', () => {

await signalClient.join('wss://test.livekit.io', 'test-token', defaultOptions);

// Setup reconnect with non-reconnect message (e.g., participant update)
// Server sends a stray update first, then the actual reconnect response.
// The stray should be dropped with a warning; the reconnect response should resolve.
const updateSignalResponse = createSignalResponse('update', { participants: [] });
const reconnectMockReadable = createMockReadableStream([updateSignalResponse]);
const reconnectResponse = new ReconnectResponse({ iceServers: [] });
const reconnectSignalResponse = createSignalResponse('reconnect', reconnectResponse);
const reconnectMockReadable = createMockReadableStream([
updateSignalResponse,
reconnectSignalResponse,
]);
const reconnectMockConnection = createMockConnection(reconnectMockReadable);

mockWebSocketStream({ connection: reconnectMockConnection });

const result = await signalClient.reconnect('wss://test.livekit.io', 'test-token', 'sid-123');

// This is an edge case: reconnect resolves with undefined when non-reconnect message is received
expect(result).toBeUndefined();
expect(result).toEqual(reconnectResponse);
expect(signalClient.currentState).toBe(SignalConnectionState.CONNECTED);
}, 1000);
});
});

describe('Failure Case - Timeout', () => {
Expand Down Expand Up @@ -1026,7 +1031,7 @@ describe('SignalClient.validateFirstMessage', () => {
}
});

it('should accept non-reconnect message during reconnecting state', async () => {
it('should reject non-reconnect message during reconnecting state', async () => {
// First establish a connection
const joinResponse = createJoinResponse();
const joinSignalResponse = createSignalResponse('join', joinResponse);
Expand All @@ -1044,9 +1049,8 @@ describe('SignalClient.validateFirstMessage', () => {
const validateMethod = (signalClient as any).validateFirstMessage;
if (validateMethod) {
const result = validateMethod.call(signalClient, updateSignalResponse, true);
expect(result.isValid).toBe(true);
expect(result.response).toBeUndefined();
expect(result.shouldProcessFirstMessage).toBe(true);
expect(result.isValid).toBe(false);
expect(result.error).toBeInstanceOf(ConnectionError);
}
});

Expand Down
107 changes: 54 additions & 53 deletions src/api/SignalClient.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -581,38 +581,61 @@ export class SignalClient {
this.streamWriter = connection.writable.getWriter();

// wsTimeout only guarded the upgrade; guard the first-message read with
// its own timeout so a silent server can't hang join() forever.
let firstMessage: ReadableStreamReadResult<string | ArrayBuffer>;
let firstMessageTimeout: ReturnType<typeof setTimeout> | undefined;
// its own deadline so a silent server can't hang join() forever.
// During reconnect, drop any stray messages that arrive before the
// ReconnectResponse so we do not declare the channel reconnected on the
// wrong signal.
let firstSignalResponse: SignalResponse;
try {
firstMessage = await Promise.race([
signalReader.read(),
new Promise<never>((_, rejectRead) => {
firstMessageTimeout = setTimeout(() => {
rejectRead(
ConnectionError.timeout(
'signal connection timed out while waiting for the first message',
),
);
}, JOIN_RESPONSE_TIMEOUT);
}),
]);
const deadline = Date.now() + JOIN_RESPONSE_TIMEOUT;
// eslint-disable-next-line no-constant-condition
while (true) {
const remaining = deadline - Date.now();
if (remaining <= 0) {
throw ConnectionError.timeout(
'signal connection timed out while waiting for the first message',
);
}
let readTimeout: ReturnType<typeof setTimeout> | undefined;
const readResult = await Promise.race([
signalReader.read(),
new Promise<never>((_, rejectRead) => {
readTimeout = setTimeout(() => {
rejectRead(
ConnectionError.timeout(
'signal connection timed out while waiting for the first message',
),
);
}, remaining);
}),
]);
clearTimeout(readTimeout);
if (!readResult.value) {
throw ConnectionError.internal('no message received as first message');
}
const parsed = parseSignalResponse(readResult.value);
if (
this.lifecycleState === 'reconnecting' &&
parsed.message?.case !== 'reconnect' &&
parsed.message?.case !== 'leave'
) {
this.log.warn('dropping signal message while awaiting reconnect response', {
messageCase: parsed.message?.case,
});
continue;
Comment on lines +617 to +625

@devin-ai-integration devin-ai-integration Bot Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

🟡 Reconnect discards valid room updates

When parsed.message precedes the reconnect response, the client discards participant, track, room, and token updates. Room state can remain stale after resume.

Prompt for agents 
During SignalClient.connect reconnect handling in src/api/SignalClient.ts, messages received before ReconnectResponse are currently consumed from the ordered stream and discarded. These messages can be legitimate server updates and will never appear in startReadingLoop. Preserve them while waiting for ReconnectResponse, then process them through handleSignalResponse after the reconnect has been validated and the lifecycle transition succeeds. Keep Leave handling immediate and ensure buffered messages from an abandoned or superseded attempt are not delivered into a later session. Add coverage proving an update before ReconnectResponse reaches its callback exactly once and in order.
Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

@1egoman 1egoman Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I think this is also my main question reading this, do you want to replay these updates prior to the ReconnectResponse after the ReconnectResponse is received? I would think so but I might be missing something.

@lukasIO lukasIO Sep 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

the server guarantees the ordering now. Any relevant updates that happen in between should be replayed by the server.

@1egoman 1egoman Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Sounds good - maybe it's worth adding a comment inline mentioning this? That's not necessarily initially intuitive behavior from the client end IMO.

}
firstSignalResponse = parsed;
break;
}
} catch (e) {
// No first message in time: release the reader and tear down the ws
// No usable first message in time: release the reader and tear down the ws
// so we surface the timeout instead of leaking an open connection.
signalReader.releaseLock();
reject(e);
this.close();
return;
} finally {
clearTimeout(firstMessageTimeout);
}
signalReader.releaseLock();
if (!firstMessage.value) {
throw ConnectionError.internal('no message received as first message');
}

const firstSignalResponse = parseSignalResponse(firstMessage.value);

// Validate the first message
const validation = this.validateFirstMessage(
Expand Down Expand Up @@ -643,11 +666,7 @@ export class SignalClient {
}
}

// Handle successful connection
const firstMessageToProcess = validation.shouldProcessFirstMessage
? firstSignalResponse
: undefined;
this.handleSignalConnected(connection, wsTimeout, attemptId, firstMessageToProcess);
this.handleSignalConnected(connection, wsTimeout, attemptId);
resolve(validation.response);
} catch (e) {
reject(e);
Expand All @@ -660,13 +679,7 @@ export class SignalClient {
});
}

async startReadingLoop(
signalReader: ReadableStreamDefaultReader<string | ArrayBuffer>,
firstMessage?: SignalResponse,
) {
if (firstMessage) {
this.handleSignalResponse(firstMessage);
}
async startReadingLoop(signalReader: ReadableStreamDefaultReader<string | ArrayBuffer>) {
const attemptId = this.attemptId;
while (true) {
if (this.signalLatency) {
Expand Down Expand Up @@ -1184,7 +1197,6 @@ export class SignalClient {
connection: WebSocketConnection,
timeoutHandle: ReturnType<typeof setTimeout>,
attemptId: number,
firstMessage?: SignalResponse,
) {
clearTimeout(timeoutHandle);
const established = this.sendLifecycleInput(
Expand All @@ -1205,7 +1217,7 @@ export class SignalClient {
}
this.log.info('signal connected');
this.startPingInterval();
this.startReadingLoop(connection.readable.getReader(), firstMessage);
this.startReadingLoop(connection.readable.getReader());
}

/**
Expand All @@ -1222,7 +1234,6 @@ export class SignalClient {
isValid: boolean;
response?: JoinResponse | ReconnectResponse;
error?: ConnectionError;
shouldProcessFirstMessage?: boolean;
} {
if (firstSignalResponse.message?.case === 'join') {
return {
Expand All @@ -1231,22 +1242,12 @@ export class SignalClient {
};
} else if (
this.lifecycleState === 'reconnecting' &&
firstSignalResponse.message?.case !== 'leave'
firstSignalResponse.message?.case === 'reconnect'
) {
if (firstSignalResponse.message?.case === 'reconnect') {
return {
isValid: true,
response: firstSignalResponse.message.value,
};
} else {
// in reconnecting, any message received means signal reconnected and we still need to process it
this.log.debug('declaring signal reconnected without reconnect response received');
return {
isValid: true,
response: undefined,
shouldProcessFirstMessage: true,
};
}
return {
isValid: true,
response: firstSignalResponse.message.value,
};
} else if (this.isEstablishingConnection && firstSignalResponse.message?.case === 'leave') {
return {
isValid: false,
Expand Down
Loading