chore(sandbox): improve gateway logging
This commit is contained in:
parent
dbb056c9c6
commit
f78b1421dd
@ -24,7 +24,7 @@ const MAX_PAYLOAD_BYTES = 1024 * 1024; // ~1MB
|
||||
export async function createApp(
|
||||
config: Config,
|
||||
): Promise<{ app: FastifyInstance; registry: GatewayRegistry }> {
|
||||
const app = Fastify({ logger: config.logger });
|
||||
const app = Fastify({ logger: config.logger, disableRequestLogging: true });
|
||||
|
||||
const registry = new GatewayRegistry(app.log, config.requestTimeoutMs);
|
||||
const probe = createOpencodeProbe({
|
||||
|
||||
@ -20,7 +20,7 @@ export type ConnectionDeps = {
|
||||
|
||||
export type Connection = {
|
||||
onMessage(raw: unknown): Promise<void>;
|
||||
onClose(): void;
|
||||
onClose(code?: number, reason?: string): void;
|
||||
};
|
||||
|
||||
type State = { accountId: AccountId; traposId: string };
|
||||
@ -69,6 +69,7 @@ export function createConnection(deps: ConnectionDeps): Connection {
|
||||
const { traposId, secret } = parsed.data;
|
||||
const accountId = resolveAccount(secret);
|
||||
if (accountId === null) {
|
||||
log.warn({ traposId, reason: "unauthorized" }, "gateway hello rejected");
|
||||
reply(req, false, undefined, { code: "unauthorized", message: "invalid secret" });
|
||||
socket.close();
|
||||
return;
|
||||
@ -76,10 +77,12 @@ export function createConnection(deps: ConnectionDeps): Connection {
|
||||
|
||||
registry.register(accountId, traposId, socket);
|
||||
state = { accountId, traposId };
|
||||
log.info({ accountId, traposId }, "gateway hello accepted");
|
||||
reply(req, true, { accountId });
|
||||
}
|
||||
|
||||
async function handleProbeGateway(req: ComputerRequest): Promise<void> {
|
||||
log.debug({ accountId: state?.accountId, traposId: state?.traposId, messageId: req.messageId }, "probing gateway dependencies");
|
||||
const result = await probe.probe();
|
||||
const payload: Record<string, unknown> = { sandboxOk: true, opencodeOk: result.ok };
|
||||
if (!result.ok && result.detail !== undefined) {
|
||||
@ -105,13 +108,14 @@ export function createConnection(deps: ConnectionDeps): Connection {
|
||||
async function dispatchPostHello(frame: Frame, current: State): Promise<void> {
|
||||
if (frame.event === "computer_request") {
|
||||
if (frame.type === "hello") {
|
||||
log.debug({ reason: "duplicate_hello" }, "ignoring duplicate hello on live connection");
|
||||
log.debug({ accountId: current.accountId, traposId: current.traposId, reason: "duplicate_hello" }, "ignoring duplicate hello on live connection");
|
||||
return;
|
||||
}
|
||||
if (frame.type === "probe_gateway") {
|
||||
await handleProbeGateway(frame);
|
||||
return;
|
||||
}
|
||||
log.debug({ accountId: current.accountId, traposId: current.traposId, messageId: frame.messageId, type: frame.type, reason: "unknown_type" }, "rejecting unknown computer_request type");
|
||||
reply(frame, false, undefined, { code: "unknown_type", message: `unknown type: ${frame.type}` });
|
||||
return;
|
||||
}
|
||||
@ -143,10 +147,14 @@ export function createConnection(deps: ConnectionDeps): Connection {
|
||||
await dispatchPostHello(frame, state);
|
||||
}
|
||||
|
||||
function onClose(): void {
|
||||
function onClose(code?: number, reason?: string): void {
|
||||
if (state) {
|
||||
log.info({ accountId: state.accountId, traposId: state.traposId, closeCode: code, closeReason: reason }, "gateway connection closed");
|
||||
registry.unregister(state.accountId, state.traposId, socket);
|
||||
state = null;
|
||||
return;
|
||||
}
|
||||
log.info({ closeCode: code, closeReason: reason }, "gateway connection closed before hello");
|
||||
}
|
||||
|
||||
return { onMessage, onClose };
|
||||
|
||||
@ -5,6 +5,8 @@ import type { GatewayRegistry, GatewaySocket } from "./registry.js";
|
||||
import type { OpencodeProbe } from "../../opencode.js";
|
||||
import type { AccountId } from "../../auth.js";
|
||||
|
||||
let nextConnectionId = 1;
|
||||
|
||||
export type GatewayModuleOptions = {
|
||||
registry: GatewayRegistry;
|
||||
probe: OpencodeProbe;
|
||||
@ -18,27 +20,32 @@ const gatewayModule: FastifyPluginAsync<GatewayModuleOptions> = (fastify, opts)
|
||||
// Open liveness check for the HTTP service itself — no auth, no opencode probe.
|
||||
fastify.get("/health", () => ({ ok: true }));
|
||||
|
||||
fastify.get("/gateway", { websocket: true }, (socket: WebSocket) => {
|
||||
fastify.get("/gateway", { websocket: true }, (socket: WebSocket, req) => {
|
||||
const connId = `gw-${nextConnectionId++}`;
|
||||
const log = fastify.log.child({ connId });
|
||||
const gatewaySocket: GatewaySocket = {
|
||||
send: (data) => socket.send(data),
|
||||
close: (code, reason) => socket.close(code, reason),
|
||||
};
|
||||
|
||||
log.info({ remoteAddress: req.ip }, "gateway connection opened");
|
||||
|
||||
const connection = createConnection({
|
||||
socket: gatewaySocket,
|
||||
registry: opts.registry,
|
||||
probe: opts.probe,
|
||||
resolveAccount: opts.resolveAccount,
|
||||
log: fastify.log,
|
||||
log,
|
||||
});
|
||||
|
||||
socket.on("message", (raw) => {
|
||||
void connection.onMessage(raw);
|
||||
});
|
||||
socket.on("close", () => {
|
||||
connection.onClose();
|
||||
socket.on("close", (code, reason) => {
|
||||
connection.onClose(code, reason.toString());
|
||||
});
|
||||
socket.on("error", () => {
|
||||
socket.on("error", (err) => {
|
||||
log.warn({ err }, "gateway socket error");
|
||||
connection.onClose();
|
||||
});
|
||||
});
|
||||
|
||||
@ -9,12 +9,14 @@ export type GatewaySocket = {
|
||||
};
|
||||
|
||||
type LogFn = (obj: object, msg?: string) => void;
|
||||
export type Logger = { debug: LogFn; warn: LogFn };
|
||||
export type Logger = { debug: LogFn; info: LogFn; warn: LogFn };
|
||||
|
||||
type Pending = {
|
||||
resolve: (payload: Record<string, unknown>) => void;
|
||||
reject: (err: GatewayError) => void;
|
||||
timer: ReturnType<typeof setTimeout>;
|
||||
type: string;
|
||||
startedAt: number;
|
||||
};
|
||||
|
||||
type Entry = {
|
||||
@ -110,16 +112,29 @@ export class GatewayRegistry {
|
||||
return new Promise<Record<string, unknown>>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
entry.pending.delete(messageId);
|
||||
this.log.warn(
|
||||
{ accountId, traposId, messageId, type, timeoutMs, durationMs: Date.now() - startedAt, pendingCount: entry.pending.size },
|
||||
"gateway request timed out",
|
||||
);
|
||||
reject(new GatewayError("timeout", `request "${type}" timed out after ${timeoutMs}ms`));
|
||||
}, timeoutMs);
|
||||
|
||||
entry.pending.set(messageId, { resolve, reject, timer });
|
||||
const startedAt = Date.now();
|
||||
entry.pending.set(messageId, { resolve, reject, timer, type, startedAt });
|
||||
|
||||
try {
|
||||
entry.socket.send(JSON.stringify(frame));
|
||||
this.log.debug(
|
||||
{ accountId, traposId, messageId, type, timeoutMs, pendingCount: entry.pending.size },
|
||||
"gateway request sent",
|
||||
);
|
||||
} catch (err) {
|
||||
clearTimeout(timer);
|
||||
entry.pending.delete(messageId);
|
||||
this.log.warn(
|
||||
{ accountId, traposId, messageId, type, pendingCount: entry.pending.size, err },
|
||||
"gateway request send failed",
|
||||
);
|
||||
reject(new GatewayError("internal", err instanceof Error ? err.message : String(err)));
|
||||
}
|
||||
});
|
||||
@ -140,6 +155,18 @@ export class GatewayRegistry {
|
||||
}
|
||||
entry.pending.delete(frame.messageId);
|
||||
clearTimeout(pending.timer);
|
||||
this.log.debug(
|
||||
{
|
||||
accountId,
|
||||
traposId,
|
||||
messageId: frame.messageId,
|
||||
type: pending.type,
|
||||
ok: frame.ok,
|
||||
durationMs: Date.now() - pending.startedAt,
|
||||
pendingCount: entry.pending.size,
|
||||
},
|
||||
"gateway response received",
|
||||
);
|
||||
|
||||
if (frame.ok) {
|
||||
pending.resolve(frame.payload);
|
||||
|
||||
@ -30,6 +30,7 @@ export class FakeSocket implements GatewaySocket {
|
||||
// on drop reasons).
|
||||
export const silentLog: Logger = {
|
||||
debug: () => undefined,
|
||||
info: () => undefined,
|
||||
warn: () => undefined,
|
||||
};
|
||||
|
||||
@ -38,6 +39,7 @@ export function captureLog(): Logger & { entries: { obj: object; msg?: string }[
|
||||
return {
|
||||
entries,
|
||||
debug: (obj, msg) => entries.push({ obj, msg }),
|
||||
info: (obj, msg) => entries.push({ obj, msg }),
|
||||
warn: (obj, msg) => entries.push({ obj, msg }),
|
||||
};
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user