fix(sandbox): stop displaced gateway reconnect loops
This commit is contained in:
parent
f78b1421dd
commit
a20f1a4ab6
@ -7,6 +7,11 @@
|
||||
-- request(). Both run as separate coroutines under the same Lua state, so they
|
||||
-- share the single os event queue.
|
||||
|
||||
-- Server-side application close code emitted when a newer registration displaces an
|
||||
-- older one (see trap-sandbox registry). Treated as terminal: the loser yields rather
|
||||
-- than reconnecting, so two daemons sharing a traposId collapse to one (no eviction loop).
|
||||
local DISPLACED_CLOSE_CODE = 4000;
|
||||
|
||||
local function defaultOs()
|
||||
return os;
|
||||
end
|
||||
@ -184,11 +189,15 @@ local function createSandbox()
|
||||
local reconnectDelay = opts.reconnectDelay or 5;
|
||||
local helloTimeout = opts.helloTimeout or 10;
|
||||
local secret = opts.secret;
|
||||
-- Lifecycle logger: log(level, msg, fields?). Defaults to a no-op so the daemon
|
||||
-- stays silent unless the caller wires a sink (e.g. a /logs file appender).
|
||||
local log = opts.log or function() end;
|
||||
|
||||
local url = buildConnectUrl(baseUrl);
|
||||
local traposId = api.traposId(osLike);
|
||||
|
||||
-- state: connecting | awaiting_hello | ready | unauthorized
|
||||
-- state: connecting | awaiting_hello | ready | unauthorized | displaced
|
||||
-- (unauthorized and displaced are terminal: no reconnect.)
|
||||
local state = 'connecting';
|
||||
local lastError = nil;
|
||||
local activeWs = nil;
|
||||
@ -209,15 +218,17 @@ local function createSandbox()
|
||||
|
||||
local function connect()
|
||||
state = 'connecting';
|
||||
log('info', 'connecting', { url = url });
|
||||
httpLike.websocketAsync(url);
|
||||
end
|
||||
|
||||
local function scheduleReconnect()
|
||||
activeWs = nil;
|
||||
if state == 'unauthorized' then
|
||||
if state == 'unauthorized' or state == 'displaced' then
|
||||
return;
|
||||
end
|
||||
state = 'connecting';
|
||||
log('info', 'reconnect scheduled', { delaySeconds = reconnectDelay });
|
||||
el.setTimeout(connect, reconnectDelay);
|
||||
end
|
||||
|
||||
@ -227,10 +238,12 @@ local function createSandbox()
|
||||
end
|
||||
if ok then
|
||||
state = 'ready';
|
||||
log('info', 'connected', { traposId = traposId });
|
||||
queueEvent('trapos_sandbox_connected', traposId);
|
||||
else
|
||||
state = 'unauthorized';
|
||||
lastError = (type(err) == 'table' and err.code) or 'unauthorized';
|
||||
log('warn', 'hello rejected; yielding', { code = lastError });
|
||||
closeActive();
|
||||
end
|
||||
end
|
||||
@ -239,6 +252,7 @@ local function createSandbox()
|
||||
if eventUrl ~= url then return; end
|
||||
activeWs = ws;
|
||||
state = 'awaiting_hello';
|
||||
log('info', 'socket open; sending hello', { traposId = traposId });
|
||||
sendFrame(api.buildHello(traposId, secret, makeUuid(opts.random)));
|
||||
-- hello-ack watchdog: state-guarded, no stored cancel. If hello already
|
||||
-- resolved this is a harmless no-op; closing triggers reconnect.
|
||||
@ -262,11 +276,23 @@ local function createSandbox()
|
||||
|
||||
el.register('websocket_failure', function(eventUrl)
|
||||
if eventUrl ~= url then return; end
|
||||
log('warn', 'connection failed', { url = url });
|
||||
scheduleReconnect();
|
||||
end);
|
||||
|
||||
el.register('websocket_closed', function(eventUrl)
|
||||
-- CC:T passes (url, reason, code). A displacement close (code 4000) is terminal:
|
||||
-- another computer registered with our traposId, so we yield rather than reconnect
|
||||
-- (reconnecting would just displace it back, an endless ping-pong).
|
||||
el.register('websocket_closed', function(eventUrl, reason, code)
|
||||
if eventUrl ~= url then return; end
|
||||
if code == DISPLACED_CLOSE_CODE or reason == 'displaced' then
|
||||
activeWs = nil;
|
||||
state = 'displaced';
|
||||
lastError = 'displaced';
|
||||
log('warn', 'displaced by a newer registration; yielding', { traposId = traposId });
|
||||
return;
|
||||
end
|
||||
log('info', 'connection closed', { code = code, reason = reason });
|
||||
scheduleReconnect();
|
||||
end);
|
||||
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "TrapOS",
|
||||
"version": "0.9.0",
|
||||
"version": "0.9.1",
|
||||
"branch": "next",
|
||||
"packages": [
|
||||
"trapos"
|
||||
|
||||
@ -6,8 +6,8 @@
|
||||
"trapos-net": "0.3.0",
|
||||
"trapos-ui": "0.2.2",
|
||||
"trapos-ai": "0.7.0",
|
||||
"trapos-sandbox": "0.2.3",
|
||||
"trapos-sandbox": "0.2.4",
|
||||
"trapos-sandbox-legacy": "0.2.2",
|
||||
"trapos": "0.9.0"
|
||||
"trapos": "0.9.1"
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "trapos-sandbox",
|
||||
"version": "0.2.3",
|
||||
"version": "0.2.4",
|
||||
"description": "TrapOS sandbox gateway client: WS daemon (servers/sandbox) + sandbox command",
|
||||
"dependencies": ["trapos-core"],
|
||||
"files": [
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "trapos",
|
||||
"version": "0.9.0",
|
||||
"version": "0.9.1",
|
||||
"description": "TrapOS full install meta-package",
|
||||
"dependencies": [
|
||||
"trapos-boot",
|
||||
|
||||
@ -3,6 +3,15 @@ local createVersion = require('/apis/libversion');
|
||||
|
||||
local URL_SETTING = 'sandbox.url';
|
||||
local PASSWORD_SETTING = 'sandbox.password';
|
||||
local LOG_PATH = '/logs/sandbox.log';
|
||||
|
||||
-- Single-instance guard: a clean reboot resets _G, so this is a no-op then. It only
|
||||
-- bites if boot ever re-runs in a VM whose previous eventloop survived, which would
|
||||
-- otherwise spawn a second daemon that fights the first over the shared traposId.
|
||||
if _G.__trapos_sandbox_daemon then
|
||||
print('sandbox: daemon already running, skipping.');
|
||||
return;
|
||||
end
|
||||
|
||||
local url = settings.get(URL_SETTING);
|
||||
if type(url) ~= 'string' or url == '' then
|
||||
@ -15,13 +24,31 @@ if not http or not http.websocket then
|
||||
return;
|
||||
end
|
||||
|
||||
-- Daemon diagnostics go to a file, never the terminal: the daemon runs headless under
|
||||
-- the boot event loop, so terminal output would corrupt the shell. log(level, msg, fields).
|
||||
local function fileLog(level, msg, fields)
|
||||
local f = fs.open(LOG_PATH, 'a');
|
||||
if not f then return; end
|
||||
local line = '[' .. os.date('%H:%M:%S') .. '] ' .. string.upper(level) .. ': ' .. msg;
|
||||
if type(fields) == 'table' then
|
||||
for k, v in pairs(fields) do
|
||||
line = line .. ' ' .. tostring(k) .. '=' .. tostring(v);
|
||||
end
|
||||
end
|
||||
f.writeLine(line);
|
||||
f.close();
|
||||
end
|
||||
|
||||
local secret = settings.get(PASSWORD_SETTING);
|
||||
|
||||
_G.__trapos_sandbox_daemon = true;
|
||||
|
||||
createSandbox().startSession({
|
||||
eventloop = _G.bootEventLoop,
|
||||
url = url,
|
||||
secret = secret,
|
||||
os = os,
|
||||
log = fileLog,
|
||||
});
|
||||
|
||||
print('sandbox v' .. createVersion().forSelf() .. ' started (' .. url .. ').');
|
||||
print('sandbox v' .. createVersion().forSelf() .. ' started (' .. url .. '), logging to ' .. LOG_PATH .. '.');
|
||||
|
||||
@ -223,6 +223,47 @@ testlib.test('startSession reconnects on a network drop', function()
|
||||
testlib.assertEquals(#ctx.http.connects, 2);
|
||||
end);
|
||||
|
||||
testlib.test('startSession reconnects on an abnormal close with a non-displaced code', function()
|
||||
local ctx = startSession();
|
||||
local ws = fakes.fakeWs();
|
||||
ctx.el.fire('websocket_success', ctx.url, ws);
|
||||
testlib.assertEquals(#ctx.http.connects, 1);
|
||||
|
||||
ctx.el.fire('websocket_closed', ctx.url, '', 1006); -- abnormal close, not a displacement
|
||||
ctx.el.fireTimer(#ctx.el.timers);
|
||||
testlib.assertEquals(#ctx.http.connects, 2);
|
||||
end);
|
||||
|
||||
testlib.test('startSession treats a displacement close (4000) as terminal', function()
|
||||
local ctx = startSession();
|
||||
local ws = fakes.fakeWs();
|
||||
ctx.el.fire('websocket_success', ctx.url, ws);
|
||||
ctx.el.fire('websocket_message', ctx.url, encode({ event = 'gateway_response', type = 'hello', ok = true, payload = {} }));
|
||||
testlib.assertEquals(ctx.session.isReady(), true);
|
||||
local timersBefore = #ctx.el.timers;
|
||||
|
||||
ctx.el.fire('websocket_closed', ctx.url, 'displaced', 4000);
|
||||
|
||||
testlib.assertEquals(ctx.session.isReady(), false);
|
||||
testlib.assertEquals(ctx.session.lastError(), 'displaced');
|
||||
-- No reconnect timer scheduled, and no further connect attempt even if one fired.
|
||||
testlib.assertEquals(#ctx.el.timers, timersBefore);
|
||||
testlib.assertEquals(#ctx.http.connects, 1);
|
||||
end);
|
||||
|
||||
testlib.test('startSession treats a displacement close by reason as terminal', function()
|
||||
local ctx = startSession();
|
||||
local ws = fakes.fakeWs();
|
||||
ctx.el.fire('websocket_success', ctx.url, ws);
|
||||
local timersBefore = #ctx.el.timers;
|
||||
|
||||
ctx.el.fire('websocket_closed', ctx.url, 'displaced'); -- reason only, code absent
|
||||
|
||||
testlib.assertEquals(ctx.session.lastError(), 'displaced');
|
||||
testlib.assertEquals(#ctx.el.timers, timersBefore);
|
||||
testlib.assertEquals(#ctx.http.connects, 1);
|
||||
end);
|
||||
|
||||
testlib.test('hello watchdog closes a socket stuck awaiting ack', function()
|
||||
local ctx = startSession();
|
||||
local ws = fakes.fakeWs();
|
||||
|
||||
@ -26,6 +26,10 @@ type Entry = {
|
||||
|
||||
const DEFAULT_REQUEST_TIMEOUT_MS = 600_000;
|
||||
|
||||
// Application close code (3000-4999 range) used when a newer registration displaces
|
||||
// an older one. The Lua client treats this as terminal (yield, no reconnect).
|
||||
export const DISPLACED_CLOSE_CODE = 4000;
|
||||
|
||||
function isErrorCode(code: string): code is ErrorCode {
|
||||
return (ERROR_CODES as readonly string[]).includes(code);
|
||||
}
|
||||
@ -49,13 +53,15 @@ export class GatewayRegistry {
|
||||
const k = this.key(accountId, traposId);
|
||||
const existing = this.entries.get(k);
|
||||
if (existing) {
|
||||
this.log.debug(
|
||||
this.log.info(
|
||||
{ accountId, traposId, reason: "duplicate_hello" },
|
||||
"evicting previous connection (newest-wins)",
|
||||
);
|
||||
this.rejectAll(existing, new GatewayError("not_connected", "connection displaced by newer registration"));
|
||||
try {
|
||||
existing.socket.close();
|
||||
// Distinct close code so the displaced client can tell a displacement apart
|
||||
// from a normal drop and yield instead of reconnecting into an eviction loop.
|
||||
existing.socket.close(DISPLACED_CLOSE_CODE, "displaced");
|
||||
} catch {
|
||||
// already closed / errored - nothing to do
|
||||
}
|
||||
|
||||
73
tools/trap-sandbox/test-integration/displaced.test.ts
Normal file
73
tools/trap-sandbox/test-integration/displaced.test.ts
Normal file
@ -0,0 +1,73 @@
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
import { formatFailure, startCraftos, startSandbox, waitFor } from "./harness.js";
|
||||
|
||||
const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
// Regression for the reconnect ping-pong: two live daemons sharing a traposId used to
|
||||
// thrash forever — each newest-wins eviction (close 1000) made the displaced client
|
||||
// reconnect, which evicted the other, every `reconnectDelay`. The fix closes the evicted
|
||||
// socket with code 4000 ("displaced"); the Lua client treats that as terminal and yields.
|
||||
//
|
||||
// registry.count() can't see the churn (newest-wins pins it at 1), so we count server-side
|
||||
// "gateway connection opened" log lines. With the fix the loser yields, so opens stay at the
|
||||
// two initial connects; pre-fix this number climbed once per reconnectDelay.
|
||||
test("displaced daemon yields instead of reconnect-looping", async () => {
|
||||
const sandbox = await startSandbox({ captureLogs: true });
|
||||
const TRAPOS_ID = "9:dup";
|
||||
// Aggressive reconnect so the pre-fix loop would be obvious within the observation window.
|
||||
const RECONNECT_DELAY = "1";
|
||||
|
||||
const first = startCraftos("gateway-client.lua", {
|
||||
computerId: 9,
|
||||
computerLabel: "dup",
|
||||
shellArgs: [sandbox.baseUrl, "-", "idle", RECONNECT_DELAY],
|
||||
});
|
||||
|
||||
let second: ReturnType<typeof startCraftos> | undefined;
|
||||
try {
|
||||
await waitFor(() => first.snapshot().includes("__SANDBOX_READY__"), 12_000, "first connected");
|
||||
await waitFor(() => sandbox.registry.has("local", TRAPOS_ID), 2_000, "first registered server-side");
|
||||
|
||||
const secondHandle = startCraftos("gateway-client.lua", {
|
||||
computerId: 9,
|
||||
computerLabel: "dup",
|
||||
shellArgs: [sandbox.baseUrl, "-", "idle", RECONNECT_DELAY],
|
||||
});
|
||||
second = secondHandle;
|
||||
await waitFor(() => secondHandle.snapshot().includes("__SANDBOX_READY__"), 12_000, "second connected");
|
||||
|
||||
const openedAfterBoth = sandbox.countLog("gateway connection opened");
|
||||
|
||||
// Let several reconnect intervals elapse. If the displaced daemon reconnected, we would
|
||||
// see a fresh "gateway connection opened" every ~1s here.
|
||||
await sleep(5_000);
|
||||
|
||||
const openedAfterWait = sandbox.countLog("gateway connection opened");
|
||||
assert.equal(
|
||||
openedAfterWait,
|
||||
openedAfterBoth,
|
||||
`no new connections after both daemons settled (was ${openedAfterBoth}, now ${openedAfterWait})`,
|
||||
);
|
||||
// The two initial connects, and nothing more.
|
||||
assert.ok(openedAfterWait <= 3, `bounded connection count, got ${openedAfterWait}`);
|
||||
assert.equal(sandbox.registry.count(), 1, "exactly one registration survives");
|
||||
|
||||
// The surviving (newest) daemon still answers a gateway->computer ping.
|
||||
const payload = await sandbox.registry.request("local", TRAPOS_ID, "ping", {}, 2_000);
|
||||
assert.deepEqual(payload, { traposId: TRAPOS_ID }, "survivor answers reverse ping");
|
||||
} catch (error) {
|
||||
throw new Error(
|
||||
formatFailure(
|
||||
error instanceof Error ? error.message : String(error),
|
||||
`first:\n${first.snapshot()}\nsecond:\n${second?.snapshot() ?? "(not started)"}`,
|
||||
),
|
||||
{ cause: error },
|
||||
);
|
||||
} finally {
|
||||
first.abort();
|
||||
second?.abort();
|
||||
await Promise.all([first.done.catch(() => undefined), second?.done.catch(() => undefined)]);
|
||||
await sandbox.close();
|
||||
}
|
||||
});
|
||||
@ -19,6 +19,10 @@ export type Sandbox = {
|
||||
baseUrl: string;
|
||||
// Full WS endpoint (origin + `/gateway`), for direct ws clients.
|
||||
socketUrl: string;
|
||||
// Count captured server log lines whose `msg` contains `substring` (0 unless the
|
||||
// sandbox was started with `captureLogs: true`). Lets a test observe connection
|
||||
// churn that registry.count() hides (newest-wins keeps the count pinned at 1).
|
||||
countLog: (substring: string) => number;
|
||||
close: () => Promise<void>;
|
||||
};
|
||||
|
||||
@ -26,11 +30,18 @@ export type SandboxOptions = {
|
||||
sandboxPassword?: string;
|
||||
opencodeUrl?: string;
|
||||
requestTimeoutMs?: number;
|
||||
// Capture server logs (at info level) so tests can assert on message frequency.
|
||||
captureLogs?: boolean;
|
||||
};
|
||||
|
||||
// Real Fastify gateway on an ephemeral port. Mirrors index.ts's wiring but with a
|
||||
// silent logger and test-supplied config (never reads process.env).
|
||||
export async function startSandbox(opts: SandboxOptions = {}): Promise<Sandbox> {
|
||||
const logLines: string[] = [];
|
||||
const logger: Config["logger"] = opts.captureLogs
|
||||
? { level: "info", stream: { write: (line: string) => void logLines.push(line) } }
|
||||
: { level: "silent" };
|
||||
|
||||
const config: Config = {
|
||||
host: "127.0.0.1",
|
||||
port: 0,
|
||||
@ -39,7 +50,7 @@ export async function startSandbox(opts: SandboxOptions = {}): Promise<Sandbox>
|
||||
opencodeUrl: opts.opencodeUrl,
|
||||
opencodeUsername: "opencode",
|
||||
opencodeServerPassword: undefined,
|
||||
logger: { level: "silent" },
|
||||
logger,
|
||||
};
|
||||
|
||||
const { app, registry } = await createApp(config);
|
||||
@ -51,6 +62,7 @@ export async function startSandbox(opts: SandboxOptions = {}): Promise<Sandbox>
|
||||
registry,
|
||||
baseUrl,
|
||||
socketUrl: `${baseUrl}/gateway`,
|
||||
countLog: (substring) => logLines.filter((line) => line.includes(substring)).length,
|
||||
close: () => app.close(),
|
||||
};
|
||||
}
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "node:test";
|
||||
import { GatewayRegistry } from "../src/modules/trapos-cloud-gateway/registry.js";
|
||||
import { DISPLACED_CLOSE_CODE, GatewayRegistry } from "../src/modules/trapos-cloud-gateway/registry.js";
|
||||
import { GatewayError, type ComputerResponse } from "../src/modules/trapos-cloud-gateway/protocol.js";
|
||||
import { FakeSocket, silentLog } from "./helpers.js";
|
||||
|
||||
@ -30,6 +30,8 @@ test("newest-wins evicts previous socket for same key", () => {
|
||||
reg.register("local", "7:", a);
|
||||
reg.register("local", "7:", b);
|
||||
assert.equal(a.closed, true);
|
||||
// Distinct close code lets the displaced Lua client yield instead of reconnect-looping.
|
||||
assert.equal(a.closeCode, DISPLACED_CLOSE_CODE);
|
||||
assert.equal(b.closed, false);
|
||||
assert.equal(reg.count(), 1);
|
||||
});
|
||||
|
||||
Loading…
Reference in New Issue
Block a user