fix(relay): avoid stale client timer resetting control socket
This commit is contained in:
@@ -0,0 +1,96 @@
|
|||||||
|
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||||
|
import { RelayDurableObject } from "./cloudflare-adapter.js";
|
||||||
|
|
||||||
|
type MockSocket = WebSocket & {
|
||||||
|
send: ReturnType<typeof vi.fn>;
|
||||||
|
close: ReturnType<typeof vi.fn>;
|
||||||
|
serializeAttachment: ReturnType<typeof vi.fn>;
|
||||||
|
deserializeAttachment: ReturnType<typeof vi.fn>;
|
||||||
|
};
|
||||||
|
|
||||||
|
function createMockSocket(attachment: unknown = null): MockSocket {
|
||||||
|
let storedAttachment = attachment;
|
||||||
|
return {
|
||||||
|
send: vi.fn(),
|
||||||
|
close: vi.fn(),
|
||||||
|
serializeAttachment: vi.fn((value: unknown) => {
|
||||||
|
storedAttachment = value;
|
||||||
|
}),
|
||||||
|
deserializeAttachment: vi.fn(() => storedAttachment),
|
||||||
|
} as unknown as MockSocket;
|
||||||
|
}
|
||||||
|
|
||||||
|
function createMockState() {
|
||||||
|
const socketsByTag = new Map<string, WebSocket[]>();
|
||||||
|
const state = {
|
||||||
|
acceptWebSocket: vi.fn(),
|
||||||
|
getWebSockets: vi.fn((tag?: string): WebSocket[] => {
|
||||||
|
if (!tag) {
|
||||||
|
const out: WebSocket[] = [];
|
||||||
|
for (const sockets of socketsByTag.values()) out.push(...sockets);
|
||||||
|
return out;
|
||||||
|
}
|
||||||
|
return socketsByTag.get(tag) ?? [];
|
||||||
|
}),
|
||||||
|
};
|
||||||
|
|
||||||
|
return {
|
||||||
|
state,
|
||||||
|
setTagSockets: (tag: string, sockets: WebSocket[]) => {
|
||||||
|
socketsByTag.set(tag, sockets);
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("RelayDurableObject control nudge/reset behavior", () => {
|
||||||
|
afterEach(() => {
|
||||||
|
vi.useRealTimers();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not nudge or reset control after the client already disconnected", () => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
const clientId = "clt_stale_timer";
|
||||||
|
const control = createMockSocket();
|
||||||
|
const { state, setTagSockets } = createMockState();
|
||||||
|
|
||||||
|
setTagSockets("server-control", [control]);
|
||||||
|
setTagSockets("client", []);
|
||||||
|
setTagSockets(`client:${clientId}`, []);
|
||||||
|
setTagSockets(`server:${clientId}`, []);
|
||||||
|
|
||||||
|
const relay = new RelayDurableObject(state as any);
|
||||||
|
(relay as any).nudgeOrResetControlForClient(clientId);
|
||||||
|
|
||||||
|
vi.advanceTimersByTime(15_000);
|
||||||
|
|
||||||
|
expect(control.send).not.toHaveBeenCalled();
|
||||||
|
expect(control.close).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("resets control when the client remains connected but no server-data socket appears", () => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
const clientId = "clt_waiting_for_daemon";
|
||||||
|
const control = createMockSocket();
|
||||||
|
const client = createMockSocket({
|
||||||
|
role: "client",
|
||||||
|
clientId,
|
||||||
|
serverId: "srv_test",
|
||||||
|
createdAt: Date.now(),
|
||||||
|
});
|
||||||
|
const { state, setTagSockets } = createMockState();
|
||||||
|
|
||||||
|
setTagSockets("server-control", [control]);
|
||||||
|
setTagSockets("client", [client]);
|
||||||
|
setTagSockets(`client:${clientId}`, [client]);
|
||||||
|
setTagSockets(`server:${clientId}`, []);
|
||||||
|
|
||||||
|
const relay = new RelayDurableObject(state as any);
|
||||||
|
(relay as any).nudgeOrResetControlForClient(clientId);
|
||||||
|
|
||||||
|
vi.advanceTimersByTime(10_000);
|
||||||
|
expect(control.send).toHaveBeenCalledTimes(1);
|
||||||
|
|
||||||
|
vi.advanceTimersByTime(5_000);
|
||||||
|
expect(control.close).toHaveBeenCalledWith(1011, "Control unresponsive");
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -83,6 +83,14 @@ export class RelayDurableObject {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private hasClientSocket(clientId: string): boolean {
|
||||||
|
try {
|
||||||
|
return this.state.getWebSockets(`client:${clientId}`).length > 0;
|
||||||
|
} catch {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private nudgeOrResetControlForClient(clientId: string): void {
|
private nudgeOrResetControlForClient(clientId: string): void {
|
||||||
// If the daemon's control WS becomes half-open, the DO can't reliably detect it via ws.send errors
|
// If the daemon's control WS becomes half-open, the DO can't reliably detect it via ws.send errors
|
||||||
// (Cloudflare may accept writes even if the other side is no longer reading).
|
// (Cloudflare may accept writes even if the other side is no longer reading).
|
||||||
@@ -94,12 +102,14 @@ export class RelayDurableObject {
|
|||||||
const secondDelayMs = 5_000;
|
const secondDelayMs = 5_000;
|
||||||
|
|
||||||
setTimeout(() => {
|
setTimeout(() => {
|
||||||
|
if (!this.hasClientSocket(clientId)) return;
|
||||||
if (this.hasServerDataSocket(clientId)) return;
|
if (this.hasServerDataSocket(clientId)) return;
|
||||||
|
|
||||||
// First nudge: send a full sync list.
|
// First nudge: send a full sync list.
|
||||||
this.notifyControls({ type: "sync", clientIds: this.listConnectedClientIds() });
|
this.notifyControls({ type: "sync", clientIds: this.listConnectedClientIds() });
|
||||||
|
|
||||||
setTimeout(() => {
|
setTimeout(() => {
|
||||||
|
if (!this.hasClientSocket(clientId)) return;
|
||||||
if (this.hasServerDataSocket(clientId)) return;
|
if (this.hasServerDataSocket(clientId)) return;
|
||||||
|
|
||||||
// Still nothing: assume control is stuck and force a reconnect.
|
// Still nothing: assume control is stuck and force a reconnect.
|
||||||
|
|||||||
Reference in New Issue
Block a user