diff --git a/packages/civ7-direct-control/src/index.ts b/packages/civ7-direct-control/src/index.ts index 8e7477677e..f10e884b16 100644 --- a/packages/civ7-direct-control/src/index.ts +++ b/packages/civ7-direct-control/src/index.ts @@ -382,6 +382,7 @@ export type { Civ7DirectControlEndpoint, Civ7DirectControlHealth, Civ7DirectControlOptions, + Civ7DirectControlSessionStats, Civ7TunerState, Civ7TunerStateRole, Civ7TunerStateSelection, diff --git a/packages/civ7-direct-control/src/session/session.ts b/packages/civ7-direct-control/src/session/session.ts index a6c9828978..d898579c49 100644 --- a/packages/civ7-direct-control/src/session/session.ts +++ b/packages/civ7-direct-control/src/session/session.ts @@ -17,10 +17,14 @@ import type { Civ7CommandResult, Civ7DirectControlEndpoint, Civ7DirectControlOptions, + Civ7DirectControlSessionStats, Civ7TunerState, Civ7TunerStateSelection, } from "./types.js"; +/** How long a graceful FIN gets before `close()` falls back to `destroy()`. */ +const GRACEFUL_CLOSE_TIMEOUT_MS = 1_000; + type PendingCiv7TunerRequest = { resolve: (frame: Civ7TunerFrame) => void; reject: (err: Error) => void; @@ -36,6 +40,7 @@ export class Civ7DirectControlSession { private endpointValue: Civ7DirectControlEndpoint | undefined; private buffer = Buffer.alloc(0); private readonly pending = new Map(); + private consecutiveResponseTimeouts = 0; constructor(options: Civ7DirectControlOptions = {}) { this.config = resolveCiv7DirectControlConfig(options); @@ -45,6 +50,16 @@ export class Civ7DirectControlSession { return this.endpointValue; } + /** + * Health counters observed on this socket — the one vantage point that + * sees ALL traffic on a shared session (every consumer's requests). A + * sustained run of response-timeouts is the wedged/busy-tuner signature + * the studio's backoff gate keys on. + */ + get stats(): Civ7DirectControlSessionStats { + return { consecutiveResponseTimeouts: this.consecutiveResponseTimeouts }; + } + async connect(): Promise { if (this.socket && !this.socket.destroyed && this.endpointValue) { return this.endpointValue; @@ -84,13 +99,30 @@ export class Civ7DirectControlSession { ); } + /** + * Graceful close: FIN first (`socket.end()`), so the game can release its + * descriptor cleanly — abrupt `destroy()` teardown is the suspected driver + * of the game-side fd leak that wedges the tuner after long sessions. The + * destroy fallback only fires if the peer never completes the handshake. + */ async close(): Promise { const socket = this.socket; this.socket = undefined; this.endpointValue = undefined; this.buffer = Buffer.alloc(0); this.rejectPending(new Civ7DirectControlError("socket-closed", "Civ7 tuner socket closed")); - if (socket && !socket.destroyed) socket.destroy(); + if (!socket || socket.destroyed) return; + await new Promise((resolve) => { + const timer = setTimeout(() => socket.destroy(), GRACEFUL_CLOSE_TIMEOUT_MS); + socket.once("close", () => { + clearTimeout(timer); + resolve(); + }); + // Post-end errors (peer reset during the handshake) must not surface as + // unhandled; "close" always follows. + socket.on("error", () => {}); + socket.end(); + }); } async queryStates(options: { timeoutMs?: number } = {}): Promise> { @@ -132,6 +164,7 @@ export class Civ7DirectControlSession { const response = new Promise((resolve, reject) => { const timer = setTimeout(() => { this.pending.delete(listenerId); + this.consecutiveResponseTimeouts += 1; reject( new Civ7DirectControlError( "response-timeout", @@ -155,6 +188,7 @@ export class Civ7DirectControlSession { if (!pending) continue; clearTimeout(pending.timer); this.pending.delete(parsed.frame.listenerId); + this.consecutiveResponseTimeouts = 0; pending.resolve(parsed.frame); } } @@ -180,6 +214,11 @@ export async function withCiv7DirectControlSession( options: Civ7DirectControlOptions, run: (session: Civ7DirectControlSession) => Promise, ): Promise { + // Caller-owned shared session: reuse, never close — the owner (e.g. the + // studio daemon's Effect-scoped service) manages acquisition/release. + if (options.session) { + return await run(options.session); + } const session = new Civ7DirectControlSession(options); try { return await run(session); diff --git a/packages/civ7-direct-control/src/session/types.ts b/packages/civ7-direct-control/src/session/types.ts index 9dbebc1bcf..4c45f37b03 100644 --- a/packages/civ7-direct-control/src/session/types.ts +++ b/packages/civ7-direct-control/src/session/types.ts @@ -1,4 +1,5 @@ import type { Civ7DirectControlError } from "../direct-control-error.js"; +import type { Civ7DirectControlSession } from "./session.js"; export type Civ7TunerState = Readonly<{ id: string; @@ -26,6 +27,20 @@ export type Civ7DirectControlOptions = Readonly<{ port?: number; timeoutMs?: number; env?: NodeJS.ProcessEnv; + /** + * Caller-owned shared session. When present, procedures reuse it (the + * connection is multiplexed by listenerId, so concurrent calls are fine) + * and NEVER close it — lifecycle belongs to the owner (e.g. the studio + * daemon's Effect-scoped `Civ7TunerSession`). When absent, behavior is + * unchanged: a fresh session per call, closed in `finally`. + */ + session?: Civ7DirectControlSession; +}>; + +/** Read-only health counters observed on a session's socket (all traffic). */ +export type Civ7DirectControlSessionStats = Readonly<{ + /** Response-timeouts since the last successfully resolved frame. */ + consecutiveResponseTimeouts: number; }>; export type Civ7CommandResult = Readonly<{ diff --git a/packages/civ7-direct-control/test/shared-session.test.ts b/packages/civ7-direct-control/test/shared-session.test.ts new file mode 100644 index 0000000000..a01f500fcb --- /dev/null +++ b/packages/civ7-direct-control/test/shared-session.test.ts @@ -0,0 +1,200 @@ +import { createServer, type Socket } from "node:net"; +import { afterEach, describe, expect, test } from "vitest"; + +import { Civ7DirectControlSession, executeCiv7Command } from "../src/index"; + +// Pins for the shared-session seam (mapgen-studio-tuner-session workstream): +// a caller-owned session is reused across procedure calls and never closed by +// the callee; `close()` is a graceful FIN (the abrupt-destroy teardown is the +// suspected driver of the game-side fd leak); `stats` tracks consecutive +// response-timeouts and resets on success. + +type SharedSessionServer = Readonly<{ + port: number; + connections: () => number; + finReceived: () => boolean; + close: () => Promise; +}>; + +const openServers: Array<() => Promise> = []; + +afterEach(async () => { + await Promise.all(openServers.splice(0).map((close) => close())); +}); + +describe("caller-owned shared tuner session", () => { + test("procedures reuse the injected session over one connection and do not close it", async () => { + const server = await startSharedSessionServer(); + const session = new Civ7DirectControlSession({ host: "127.0.0.1", port: server.port }); + try { + const first = await executeCiv7Command({ + port: server.port, + session, + command: "first", + timeoutMs: 1_000, + }); + const second = await executeCiv7Command({ + port: server.port, + session, + command: "second", + timeoutMs: 1_000, + }); + + expect(first.output).toEqual(["null"]); + expect(second.output).toEqual(["null"]); + expect(server.connections()).toBe(1); + // The wrapper must NOT have closed the caller-owned session. + expect(session.endpoint).toEqual({ host: "127.0.0.1", port: server.port }); + } finally { + await session.close(); + } + }); + + test("without an injected session, each call opens and closes its own connection", async () => { + const server = await startSharedSessionServer(); + await executeCiv7Command({ + host: "127.0.0.1", + port: server.port, + command: "first", + timeoutMs: 1_000, + }); + await executeCiv7Command({ + host: "127.0.0.1", + port: server.port, + command: "second", + timeoutMs: 1_000, + }); + expect(server.connections()).toBe(2); + }); + + test("close() delivers a FIN handshake, not an abrupt teardown", async () => { + const server = await startSharedSessionServer(); + const session = new Civ7DirectControlSession({ host: "127.0.0.1", port: server.port }); + await session.queryStates({ timeoutMs: 1_000 }); + + await session.close(); + // Give the server's event loop a beat to observe the FIN. + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(server.finReceived()).toBe(true); + }); + + test("stats: consecutive response-timeouts accumulate and reset on success", async () => { + const server = await startSharedSessionServer({ silentCommands: ["slow"] }); + const session = new Civ7DirectControlSession({ host: "127.0.0.1", port: server.port }); + try { + expect(session.stats.consecutiveResponseTimeouts).toBe(0); + + await expect( + executeCiv7Command({ port: server.port, session, command: "slow", timeoutMs: 50 }), + ).rejects.toMatchObject({ code: "response-timeout" }); + await expect( + executeCiv7Command({ port: server.port, session, command: "slow", timeoutMs: 50 }), + ).rejects.toMatchObject({ code: "response-timeout" }); + + // Each failed call performs LSQ (succeeds → reset) then CMD (times out), + // so the counter reflects the trailing run of timeouts on this socket. + expect(session.stats.consecutiveResponseTimeouts).toBe(1); + + const ok = await executeCiv7Command({ + port: server.port, + session, + command: "fast", + timeoutMs: 1_000, + }); + expect(ok.output).toEqual(["null"]); + expect(session.stats.consecutiveResponseTimeouts).toBe(0); + } finally { + await session.close(); + } + }); + + test("stats: an unresponsive tuner accumulates across consecutive requests", async () => { + const server = await startSharedSessionServer({ silent: true }); + const session = new Civ7DirectControlSession({ host: "127.0.0.1", port: server.port }); + try { + await expect(session.queryStates({ timeoutMs: 50 })).rejects.toMatchObject({ + code: "response-timeout", + }); + await expect(session.queryStates({ timeoutMs: 50 })).rejects.toMatchObject({ + code: "response-timeout", + }); + await expect(session.queryStates({ timeoutMs: 50 })).rejects.toMatchObject({ + code: "response-timeout", + }); + expect(session.stats.consecutiveResponseTimeouts).toBe(3); + } finally { + await session.close(); + } + }); +}); + +async function startSharedSessionServer( + options: Readonly<{ silent?: boolean; silentCommands?: readonly string[] }> = {}, +): Promise { + let connections = 0; + let finReceived = false; + const sockets = new Set(); + + const server = createServer((socket) => { + connections += 1; + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + socket.on("end", () => { + finReceived = true; + }); + socket.on("error", () => {}); + let buffer = Buffer.alloc(0); + socket.on("data", (chunk) => { + buffer = Buffer.concat([buffer, chunk]); + for (;;) { + if (buffer.length < 8) return; + const messageLength = buffer.readUInt32LE(0); + const bytesRead = 8 + messageLength; + if (buffer.length < bytesRead) return; + const listenerId = buffer.readUInt32LE(4); + const message = buffer.subarray(8, bytesRead).toString("utf8").replace(/\0$/, ""); + buffer = buffer.subarray(bytesRead); + + if (options.silent) continue; + const isSilent = options.silentCommands?.some((command) => + message.includes(`:${command}`), + ); + if (isSilent) continue; + if (message === "LSQ:") { + socket.write(encodeResponse(listenerId, ["65535", "App UI", "1", "Tuner"])); + } else { + socket.write(encodeResponse(listenerId, ["null"])); + } + } + }); + }); + + await new Promise((resolve, reject) => { + server.listen(0, "127.0.0.1", () => resolve()); + server.on("error", reject); + }); + + const close = () => + new Promise((resolve) => { + for (const socket of sockets) socket.destroy(); + server.close(() => resolve()); + }); + openServers.push(close); + + return { + port: (server.address() as { port: number }).port, + connections: () => connections, + finReceived: () => finReceived, + close, + }; +} + +function encodeResponse(listenerId: number, parts: readonly string[]): Buffer { + const message = Buffer.from(`${parts.join("\0")}\0`, "utf8"); + const frame = Buffer.alloc(8 + message.length); + frame.writeUInt32LE(message.length, 0); + frame.writeUInt32LE(listenerId, 4); + message.copy(frame, 8); + return frame; +}