mirror of
https://github.com/HeyPuter/puter.git
synced 2026-11-05 21:46:51 +00:00
fix(puter-js): end the conversation on an oversized event mid-reconnect
A server event over the 4 MiB limit closes the socket with 4009, which ends a connected conversation. During a reconnect attempt the same close was reported as `session_closed`, which is retried, so the conversation tried ten more times and finally closed with the original drop code. A replacement session opened after a restart notice was reopened again and again for the same reason. A 4009 close now ends the conversation during a reconnect too, reporting 4009, and a replacement session that hits it isn't reopened. Tests cover both, a multibyte event of exactly 4 MiB, and a whole retrieved item through the relay. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
6a67c561b9
commit
6986a34bc8
5 files changed
+98
-23
No files matched your search
@@ -1,6 +1,10 @@
|
||||
import { fetchUrl } from '../../lib/networkUtils.js';
|
||||
import { promptIfUpgradeRequired } from '../../lib/upgradePrompt.js';
|
||||
import { RealtimeSession, createRealtimeSocket } from './convoTransport.js';
|
||||
import {
|
||||
CLOSE_MESSAGE_TOO_LARGE,
|
||||
RealtimeSession,
|
||||
createRealtimeSocket,
|
||||
} from './convoTransport.js';
|
||||
import { RealtimeEvents } from './realtimeEvents.js';
|
||||
import { ConvoMicrophone, ConvoSpeaker } from './convoMedia.js';
|
||||
import { ConvoHistory } from './convoHistory.js';
|
||||
@@ -45,6 +49,17 @@ const SESSION_TIMEOUT_MS = 20_000;
|
||||
// Close codes after which a dropped connection is replaced: going away,
|
||||
// abnormal closure, server error, service restart, try again later.
|
||||
const RECONNECT_CODES = new Set([1001, 1006, 1011, 1012, 1013]);
|
||||
/**
|
||||
* Why a replacement session ended: one that hit the size limit is refused for
|
||||
* good, anything else can be retried.
|
||||
*
|
||||
* @param {{ type: string; code?: number }} event
|
||||
* @returns {string}
|
||||
*/
|
||||
const closeFailure = (event) =>
|
||||
event.type === 'close' && event.code === CLOSE_MESSAGE_TOO_LARGE
|
||||
? 'message_too_large'
|
||||
: 'session_closed';
|
||||
// Failures while reconnecting that are worth another attempt.
|
||||
const RETRYABLE_ERRORS = new Set([
|
||||
'session_closed',
|
||||
@@ -779,7 +794,7 @@ export class Conversation extends RealtimeEvents {
|
||||
code: String(event.error?.code ?? 'upstream_error'),
|
||||
});
|
||||
else if (event.type === 'close')
|
||||
done({ code: 'session_closed' });
|
||||
done({ code: closeFailure(event) });
|
||||
});
|
||||
handover.stopProbe = () => done({});
|
||||
});
|
||||
@@ -799,7 +814,7 @@ export class Conversation extends RealtimeEvents {
|
||||
handover.probe === session &&
|
||||
['close', 'error', RECONNECT_NOTICE].includes(event.type)
|
||||
)
|
||||
this.#replacementFailed(handover, 'session_closed');
|
||||
this.#replacementFailed(handover, closeFailure(event));
|
||||
});
|
||||
this.#scheduleHandover();
|
||||
}
|
||||
@@ -1280,7 +1295,12 @@ export class Conversation extends RealtimeEvents {
|
||||
code: event.code,
|
||||
reason: event.reason,
|
||||
};
|
||||
if (this.#state === 'connecting') {
|
||||
// An event over the size limit would only arrive again, so it ends
|
||||
// the conversation even mid-reconnect.
|
||||
if (
|
||||
this.#state === 'connecting' &&
|
||||
event.code !== CLOSE_MESSAGE_TOO_LARGE
|
||||
) {
|
||||
// `connect()` or the reconnect loop handles the failure.
|
||||
this.#rejectPending(
|
||||
failure(
|
||||
|
||||
@@ -1289,6 +1289,23 @@ describe('restart notice', () => {
|
||||
expect(c.state).toBe('connected');
|
||||
});
|
||||
|
||||
it('stops opening replacements after one hits the size limit', async () => {
|
||||
const c = await connected();
|
||||
fakeTime();
|
||||
const first = socket();
|
||||
first.deliver(notice);
|
||||
await vi.advanceTimersByTimeAsync(SPREAD);
|
||||
socket().deliver({
|
||||
type: 'session.created',
|
||||
session: { instructions: 'x'.repeat(4 * 1024 * 1024) },
|
||||
});
|
||||
const sockets = FakeWebSocket.instances.length;
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(FakeWebSocket.instances).toHaveLength(sockets);
|
||||
expect(c.state).toBe('connected');
|
||||
expect(first.readyState).toBe(1);
|
||||
});
|
||||
|
||||
it('drops a held replacement that gets its own notice or an error', async () => {
|
||||
const c = await connected();
|
||||
fakeTime();
|
||||
@@ -2020,6 +2037,26 @@ describe('reconnect retries', () => {
|
||||
expect(FakeWebSocket.instances.length).toBe(attempts);
|
||||
});
|
||||
|
||||
it('ends on an event over the size limit during a reconnect, without more attempts', async () => {
|
||||
const c = await connected();
|
||||
fakeTime();
|
||||
const closed = vi.fn();
|
||||
c.on('close', closed);
|
||||
socket().close(1006);
|
||||
await vi.advanceTimersByTimeAsync(500);
|
||||
const attempts = FakeWebSocket.instances.length;
|
||||
socket().deliver({
|
||||
type: 'session.created',
|
||||
session: { instructions: 'x'.repeat(4 * 1024 * 1024) },
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(c.state).toBe('closed');
|
||||
expect(closed).toHaveBeenCalledExactlyOnceWith(
|
||||
expect.objectContaining({ type: 'close', code: 4009 }),
|
||||
);
|
||||
expect(FakeWebSocket.instances.length).toBe(attempts);
|
||||
});
|
||||
|
||||
describe('while the browser is offline', () => {
|
||||
let target;
|
||||
beforeEach(() => {
|
||||
|
||||
@@ -7,6 +7,12 @@ const encoder = new TextEncoder();
|
||||
// The relay's cap on provider messages, which can carry whole items. What the
|
||||
// SDK sends is capped separately, and much lower.
|
||||
const MAX_SERVER_MESSAGE_BYTES = 4 * 1024 * 1024;
|
||||
/**
|
||||
* Close code for a server event over the limit; browsers only accept 1000 and
|
||||
* 3000-4999 from the client.
|
||||
* @internal
|
||||
*/
|
||||
export const CLOSE_MESSAGE_TOO_LARGE = 4009;
|
||||
|
||||
/**
|
||||
* Whether `data` encodes to more than `limit` UTF-8 bytes. A UTF-16 code unit
|
||||
@@ -39,8 +45,7 @@ export class RealtimeSession extends RealtimeEvents {
|
||||
const data =
|
||||
typeof msg.data === 'string' ? msg.data : String(msg.data);
|
||||
if (exceedsBytes(data, MAX_SERVER_MESSAGE_BYTES)) {
|
||||
// Browsers only accept 1000 and 3000-4999 here.
|
||||
socket.close(4009, 'message too large');
|
||||
socket.close(CLOSE_MESSAGE_TOO_LARGE, 'message too large');
|
||||
return;
|
||||
}
|
||||
event = JSON.parse(data);
|
||||
|
||||
@@ -90,6 +90,17 @@ describe('RealtimeSession', () => {
|
||||
expect(socket.close).toHaveBeenCalledExactlyOnceWith(4009, 'message too large');
|
||||
});
|
||||
|
||||
it('accepts a multibyte event of exactly 4 MiB', () => {
|
||||
const socket = fakeSocket();
|
||||
const { typed } = listen(new RealtimeSession(/** @type {any} */ (socket)));
|
||||
const data = eventOfBytes(LIMIT, 'é');
|
||||
expect(new TextEncoder().encode(data).byteLength).toBe(LIMIT);
|
||||
expect(data.length * 3).toBeGreaterThan(LIMIT);
|
||||
socket.deliverRaw(data);
|
||||
expect(typed).toHaveBeenCalledOnce();
|
||||
expect(socket.close).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('counts encoded bytes, not characters', () => {
|
||||
const socket = fakeSocket();
|
||||
const { typed } = listen(new RealtimeSession(/** @type {any} */ (socket)));
|
||||
|
||||
@@ -139,22 +139,21 @@ describe('convo built SDK and browser media', () => {
|
||||
type: 'conversation.item.added',
|
||||
item: { ...item, status: 'in_progress', content: [] },
|
||||
});
|
||||
send({
|
||||
type: 'conversation.item.done',
|
||||
item: {
|
||||
...item,
|
||||
status: 'completed',
|
||||
content: [
|
||||
{
|
||||
type: 'output_audio',
|
||||
audio: Buffer.alloc(1536 * 1024).toString(
|
||||
'base64',
|
||||
),
|
||||
transcript: 'Hello there',
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
const done = {
|
||||
...item,
|
||||
status: 'completed',
|
||||
content: [
|
||||
{
|
||||
type: 'output_audio',
|
||||
audio: Buffer.alloc(1536 * 1024).toString(
|
||||
'base64',
|
||||
),
|
||||
transcript: 'Hello there',
|
||||
},
|
||||
],
|
||||
};
|
||||
send({ type: 'conversation.item.done', item: done });
|
||||
send({ type: 'conversation.item.retrieved', item: done });
|
||||
}
|
||||
if (event.type === 'conversation.item.truncate')
|
||||
socket.send(
|
||||
@@ -508,7 +507,10 @@ describe('convo built SDK and browser media', () => {
|
||||
try {
|
||||
await conversation.connect();
|
||||
conversation.send({ type: 'test.large' });
|
||||
await log.until(() => log.count('conversation.item.done') === 1);
|
||||
await log.until(
|
||||
() => log.count('conversation.item.retrieved') === 1,
|
||||
);
|
||||
expect(log.count('conversation.item.done')).toBe(1);
|
||||
const update = log.events.find(
|
||||
(event) =>
|
||||
event.type === 'session.updated' &&
|
||||
|
||||
Reference in new issue
Block a user