From 1796ee07e1c90c70b0377f0fe27b5c680503a420 Mon Sep 17 00:00:00 2001 From: Matei Date: Fri, 12 Jun 2026 02:19:40 -0400 Subject: [PATCH] =?UTF-8?q?feat(studio-server):=20Civ7TunerSession=20?= =?UTF-8?q?=E2=80=94=20Effect-scoped=20owner=20of=20the=20shared=20tuner?= =?UTF-8?q?=20connection=20+=20backoff=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Layer.scoped + acquireRelease around ONE Civ7DirectControlSession (lazy connect, self-healing, listenerId-multiplexed); release = graceful FIN on runtime.dispose(). use(run) gates on the session's consecutive response-timeout counter: past the threshold (4) reads fail fast with typed Civ7TunerBackoffError for one cooldown (15s), half-open after, reset on success — connection-refused stays un-gated so readiness keeps reporting fast. Civ7TunerClient routes every tuner read through the shared session (savedConfigurations is filesystem-only and bypasses it). StudioRpcHandle gains tuner.{session,health} ports + dispose() — nothing closed the runtime scope before, so finalizers never ran. Also dedupes effect to 3.21.3 across the workspace (the app/control-orpc pinned 3.21.2 → mixed-runtime warning). Gates: studio-server tsc + build; studio app tsc + 204 tests (3 new pins: one shared connection + FIN on dispose, gate fail-fast/half-open/reset, wedge-suspicion health). Co-Authored-By: Claude Fable 5 --- apps/mapgen-studio/package.json | 2 +- .../test/server/tunerSession.test.ts | 210 ++++++++++++++++++ bun.lock | 10 +- packages/civ7-control-orpc/package.json | 2 +- packages/studio-server/src/handler.ts | 36 ++- packages/studio-server/src/index.ts | 19 +- packages/studio-server/src/runtime.ts | 20 +- .../src/services/Civ7TunerClient.ts | 56 ++--- .../src/services/Civ7TunerSession.ts | 148 ++++++++++++ 9 files changed, 458 insertions(+), 45 deletions(-) create mode 100644 apps/mapgen-studio/test/server/tunerSession.test.ts create mode 100644 packages/studio-server/src/services/Civ7TunerSession.ts diff --git a/apps/mapgen-studio/package.json b/apps/mapgen-studio/package.json index bc3f00744a..3652a5fac2 100644 --- a/apps/mapgen-studio/package.json +++ b/apps/mapgen-studio/package.json @@ -51,7 +51,7 @@ "@swooper/mapgen-viz": "workspace:*", "class-variance-authority": "latest", "clsx": "latest", - "effect": "3.21.2", + "effect": "3.21.3", "effect-orpc": "^0.2.2", "lucide-react": "0.522.0", "mod-swooper-maps": "workspace:*", diff --git a/apps/mapgen-studio/test/server/tunerSession.test.ts b/apps/mapgen-studio/test/server/tunerSession.test.ts new file mode 100644 index 0000000000..214482e8c1 --- /dev/null +++ b/apps/mapgen-studio/test/server/tunerSession.test.ts @@ -0,0 +1,210 @@ +import { createServer, type Socket } from "node:net"; +import { ManagedRuntime, Effect } from "effect"; +import { afterEach, describe, expect, test } from "vitest"; + +import { executeCiv7Command } from "@civ7/direct-control"; +import { + Civ7TunerSession, + makeCiv7TunerSessionLayer, + type Civ7TunerSessionApi, +} from "@civ7/studio-server"; + +// Pins for the Effect-scoped shared tuner session (mapgen-studio-tuner-session): +// one connection shared across uses; backoff gate opens after the threshold of +// consecutive response-timeouts, fails fast during cooldown (no socket +// traffic), half-opens after, and resets on success; runtime dispose runs the +// release finalizer (graceful close → FIN observed by the peer). + +type FakeTuner = Readonly<{ + port: number; + connections: () => number; + framesReceived: () => number; + finReceived: () => boolean; + setSilent: (silent: boolean) => void; + close: () => Promise; +}>; + +const cleanups: Array<() => Promise> = []; + +afterEach(async () => { + await Promise.all(cleanups.splice(0).map((fn) => fn())); +}); + +describe("Civ7TunerSession (Effect scoped shared session)", () => { + test("shares one connection across uses and disposes with a FIN", async () => { + const tuner = await startFakeTuner(); + const { runtime, service } = await makeRuntime(tuner.port); + + const first = await runtime.runPromise( + service.use((o) => + executeCiv7Command({ port: tuner.port, command: "first", timeoutMs: 1_000, ...o }), + ), + ); + const second = await runtime.runPromise( + service.use((o) => + executeCiv7Command({ port: tuner.port, command: "second", timeoutMs: 1_000, ...o }), + ), + ); + + expect(first.output).toEqual(["null"]); + expect(second.output).toEqual(["null"]); + expect(tuner.connections()).toBe(1); + + await runtime.dispose(); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(tuner.finReceived()).toBe(true); + }); + + test("gate: fails fast after the threshold, half-opens after cooldown, resets on success", async () => { + const tuner = await startFakeTuner(); + const { runtime, service } = await makeRuntime(tuner.port, { + gate: { threshold: 2, cooldownMs: 250 }, + }); + cleanups.push(() => runtime.dispose()); + + const read = (command: string, timeoutMs: number) => + runtime.runPromiseExit( + service.use((o) => + executeCiv7Command({ port: tuner.port, command, timeoutMs, ...o }), + ), + ); + + // Warm the connection, then go silent: accumulate response-timeouts. + await read("warm", 1_000); + tuner.setSilent(true); + expect((await read("t1", 40))._tag).toBe("Failure"); + expect((await read("t2", 40))._tag).toBe("Failure"); + + // Threshold (2) crossed → gate open: fail fast WITHOUT touching the socket. + const framesBefore = tuner.framesReceived(); + const gated = await read("gated", 1_000); + expect(gated._tag).toBe("Failure"); + expect(JSON.stringify(gated)).toContain("Civ7TunerBackoffError"); + expect(tuner.framesReceived()).toBe(framesBefore); + + // Half-open after cooldown: the next read flows; tuner answers again → + // counter resets and the gate stays closed. + await new Promise((resolve) => setTimeout(resolve, 300)); + tuner.setSilent(false); + const recovered = await read("recovered", 1_000); + expect(recovered._tag).toBe("Success"); + const health = await runtime.runPromise(service.health); + expect(health.consecutiveResponseTimeouts).toBe(0); + expect(health.gateOpenUntil).toBeNull(); + expect(health.wedgeSuspected).toBe(false); + }); + + test("health reports wedge suspicion while the tuner is silent", async () => { + const tuner = await startFakeTuner(); + const { runtime, service } = await makeRuntime(tuner.port, { + gate: { threshold: 2, cooldownMs: 10_000 }, + }); + cleanups.push(() => runtime.dispose()); + + await runtime.runPromise( + service.use((o) => + executeCiv7Command({ port: tuner.port, command: "warm", timeoutMs: 1_000, ...o }), + ), + ); + tuner.setSilent(true); + for (let i = 0; i < 2; i += 1) { + await runtime.runPromiseExit( + service.use((o) => + executeCiv7Command({ port: tuner.port, command: `t${i}`, timeoutMs: 40, ...o }), + ), + ); + } + + const health = await runtime.runPromise(service.health); + expect(health.consecutiveResponseTimeouts).toBe(2); + expect(health.wedgeSuspected).toBe(true); + expect(health.gateOpenUntil).not.toBeNull(); + }); +}); + +async function makeRuntime( + port: number, + options: Parameters[0] = {}, +): Promise<{ + runtime: ManagedRuntime.ManagedRuntime; + service: Civ7TunerSessionApi; +}> { + const runtime = ManagedRuntime.make( + makeCiv7TunerSessionLayer({ host: "127.0.0.1", port, env: {}, ...options }), + ); + const service = await runtime.runPromise( + Effect.map(Civ7TunerSession, (s) => s), + ); + return { runtime, service }; +} + +async function startFakeTuner(): Promise { + let connections = 0; + let framesReceived = 0; + let finReceived = false; + let silent = 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); + framesReceived += 1; + + if (silent) 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()); + }); + cleanups.push(close); + + return { + port: (server.address() as { port: number }).port, + connections: () => connections, + framesReceived: () => framesReceived, + finReceived: () => finReceived, + setSilent: (value) => { + silent = value; + }, + 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; +} diff --git a/bun.lock b/bun.lock index 4aeb757f63..15747c1f58 100644 --- a/bun.lock +++ b/bun.lock @@ -68,7 +68,7 @@ "@swooper/mapgen-viz": "workspace:*", "class-variance-authority": "latest", "clsx": "latest", - "effect": "3.21.2", + "effect": "3.21.3", "effect-orpc": "^0.2.2", "lucide-react": "0.522.0", "mod-swooper-maps": "workspace:*", @@ -172,7 +172,7 @@ "@orpc/server": "^1.14.4", "@orpc/shared": "^1.14.4", "@standard-schema/spec": "^1.1.0", - "effect": "3.21.2", + "effect": "3.21.3", "effect-orpc": "^0.2.2", "typebox": "^1.0.80", }, @@ -1669,7 +1669,7 @@ "ee-first": ["ee-first@1.1.1", "", {}, "sha512-WMwm9LhRUo+WUaRN+vRuETqG89IgZphVSNkdFgeb6sS/E4OrDIN7t48CAewSHXc6C8lefD8KKfr5vY61brQlow=="], - "effect": ["effect@3.21.2", "", { "dependencies": { "@standard-schema/spec": "^1.0.0", "fast-check": "^3.23.1" } }, "sha512-rXd2FGDM8KdjSIrc+mqEELo7ScW7xTVxEf1iInmPSpIde9/nyGuFM710cjTo7/EreGXiUX2MOonPpprbz2XHCg=="], + "effect": ["effect@3.21.3", "", { "dependencies": { "@standard-schema/spec": "^1.0.0", "fast-check": "^3.23.1" } }, "sha512-RqwU7WnJ6CqYhyjpOVJA5vh1Sgkn6eVECO6mnD0EjlbWcC2M3LJaPglXXr13Rdo/Y+B+wTEPzGRYFNL2xKxNeQ=="], "effect-orpc": ["effect-orpc@0.2.2", "", { "peerDependencies": { "@orpc/client": ">=1.13.0", "@orpc/contract": ">=1.13.0", "@orpc/server": ">=1.13.0", "@orpc/shared": ">=1.13.0", "effect": ">=3.18.0", "typescript": "^5" } }, "sha512-wQJgXzhHWHEuF/HmjvGncSPDosBuVQS5Ichz/0HwDzYfOC7Qg5jGqIQHbPY0/6JChjz3E22XNgPCfUrmtZRuPA=="], @@ -3185,8 +3185,6 @@ "@babel/helper-compilation-targets/semver": ["semver@6.3.1", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-BR7VvDCVHO+q2xBEWskxS6DJE1qRnb7DxzUrogb71CWoSficBxYsiAGd+Kl0mmq/MprG9yArRkyrQxTO6XjMzA=="], - "@civ7/studio-server/effect": ["effect@3.21.3", "", { "dependencies": { "@standard-schema/spec": "^1.0.0", "fast-check": "^3.23.1" } }, "sha512-RqwU7WnJ6CqYhyjpOVJA5vh1Sgkn6eVECO6mnD0EjlbWcC2M3LJaPglXXr13Rdo/Y+B+wTEPzGRYFNL2xKxNeQ=="], - "@eslint-community/eslint-utils/eslint-visitor-keys": ["eslint-visitor-keys@3.4.3", "", {}, "sha512-wpc+LXeiyiisxPlEkUzU6svyS1frIO3Mgxj1fdy7Pm8Ygzguax2N3Fa/D/ag1WqbOprdI+uY6wMUl8/a2G+iag=="], "@eslint/eslintrc/globals": ["globals@14.0.0", "", {}, "sha512-oahGvuMGQlPw/ivIYBjVSrWAfWLBeku5tpPE2fOPLi+WHffIWbuh2tCjhyQhTBPMf5E9jDEH4FOmTYgYwbKwtQ=="], @@ -3215,6 +3213,8 @@ "@inquirer/select/@inquirer/type": ["@inquirer/type@1.5.5", "", { "dependencies": { "mute-stream": "^1.0.0" } }, "sha512-MzICLu4yS7V8AA61sANROZ9vT1H3ooca5dSmI1FjZkzq7o/koMsRfQSzRtFo+F3Ao4Sf1C0bpLKejpKB/+j6MA=="], + "@mateicanavra/civ7-cli/effect": ["effect@3.21.2", "", { "dependencies": { "@standard-schema/spec": "^1.0.0", "fast-check": "^3.23.1" } }, "sha512-rXd2FGDM8KdjSIrc+mqEELo7ScW7xTVxEf1iInmPSpIde9/nyGuFM710cjTo7/EreGXiUX2MOonPpprbz2XHCg=="], + "@mintlify/cli/@inquirer/prompts": ["@inquirer/prompts@7.9.0", "", { "dependencies": { "@inquirer/checkbox": "^4.3.0", "@inquirer/confirm": "^5.1.19", "@inquirer/editor": "^4.2.21", "@inquirer/expand": "^4.0.21", "@inquirer/input": "^4.2.5", "@inquirer/number": "^3.0.21", "@inquirer/password": "^4.0.21", "@inquirer/rawlist": "^4.1.9", "@inquirer/search": "^3.2.0", "@inquirer/select": "^4.4.0" }, "peerDependencies": { "@types/node": ">=18" }, "optionalPeers": ["@types/node"] }, "sha512-X7/+dG9SLpSzRkwgG5/xiIzW0oMrV3C0HOa7YHG1WnrLK+vCQHfte4k/T80059YBdei29RBC3s+pSMvPJDU9/A=="], "@mintlify/cli/chalk": ["chalk@5.2.0", "", {}, "sha512-ree3Gqw/nazQAPuJJEy+avdl7QfZMcUvmHIKgEZkGL+xOBzRvup5Hxo6LHuMceSxOabuJLJm5Yp/92R9eMmMvA=="], diff --git a/packages/civ7-control-orpc/package.json b/packages/civ7-control-orpc/package.json index 6fb17a5cec..25163e5da5 100644 --- a/packages/civ7-control-orpc/package.json +++ b/packages/civ7-control-orpc/package.json @@ -54,7 +54,7 @@ "@orpc/server": "^1.14.4", "@orpc/shared": "^1.14.4", "@standard-schema/spec": "^1.1.0", - "effect": "3.21.2", + "effect": "3.21.3", "effect-orpc": "^0.2.2", "typebox": "^1.0.80" }, diff --git a/packages/studio-server/src/handler.ts b/packages/studio-server/src/handler.ts index 036f8a506f..30d8992528 100644 --- a/packages/studio-server/src/handler.ts +++ b/packages/studio-server/src/handler.ts @@ -1,9 +1,16 @@ +import { Effect } from "effect"; import { onError } from "@orpc/server"; import { RPCHandler } from "@orpc/server/fetch"; +import type { Civ7DirectControlSession } from "@civ7/direct-control"; + import type { StudioServerContext } from "./context.js"; import { createStudioRouter, type StudioRouter } from "./router/index.js"; import { makeStudioRuntime } from "./runtime.js"; +import { + Civ7TunerSession, + type Civ7TunerSessionHealth, +} from "./services/Civ7TunerSession.js"; /** * `createStudioRpcHandler(context)` — the host entrypoint. @@ -11,13 +18,20 @@ import { makeStudioRuntime } from "./runtime.js"; * Builds the per-host `ManagedRuntime` (injecting {@link StudioServerContext}), * implements the effect-orpc router, and wraps it in an oRPC `RPCHandler` (fetch * adapter — `Request`/`Response`, works under Bun and Node/Vite). The host mounts - * `handle(request, { prefix })` at `/rpc` (the Vite dev middleware this run; a Bun - * server later). `RPCHandler` returns `{ matched, response }`; `matched: false` - * means no procedure matched the path → the host falls through (`next()`). + * `handle(request, { prefix })` at `/rpc`. `RPCHandler` returns + * `{ matched, response }`; `matched: false` means no procedure matched the path + * → the host falls through (`next()`). + * + * The handle also carries the shared-tuner-session ports: + * - `tuner.session()` — the ONE `Civ7DirectControlSession` for injection into + * non-Effect consumers (`Civ7DirectControlOptions.session`, e.g. the + * control-oRPC mount's `endpointDefaults`). Lifecycle stays with the runtime. + * - `tuner.health()` — consecutive response-timeouts + backoff-gate state. + * - `dispose()` — closes the runtime scope (graceful FIN to the game). The + * host MUST call this on shutdown or the release finalizer never runs. * * `StrictGetMethodPlugin` is on by default (GET CSRF hardening) — left enabled. - * CORS is omitted: `/rpc` is same-origin (served from the Vite dev server / the - * app's own host), so no cross-origin plugin is needed. + * CORS is omitted: `/rpc` is same-origin, so no cross-origin plugin is needed. */ export interface StudioRpcHandle { readonly router: StudioRouter; @@ -25,6 +39,11 @@ export interface StudioRpcHandle { request: Request, options?: { prefix?: `/${string}`; context?: Record }, ): Promise<{ matched: boolean; response?: Response }>; + readonly tuner: { + session(): Promise; + health(): Promise; + }; + dispose(): Promise; } export function createStudioRpcHandler(context: StudioServerContext): StudioRpcHandle { @@ -47,5 +66,12 @@ export function createStudioRpcHandler(context: StudioServerContext): StudioRpcH prefix: options?.prefix ?? "/rpc", context: options?.context ?? {}, }), + tuner: { + session: () => + runtime.runPromise(Effect.map(Civ7TunerSession, (tuner) => tuner.session)), + health: () => + runtime.runPromise(Effect.flatMap(Civ7TunerSession, (tuner) => tuner.health)), + }, + dispose: () => runtime.dispose(), }; } diff --git a/packages/studio-server/src/index.ts b/packages/studio-server/src/index.ts index 7973a955c6..f3f0871e0e 100644 --- a/packages/studio-server/src/index.ts +++ b/packages/studio-server/src/index.ts @@ -2,11 +2,13 @@ * `@civ7/studio-server` — public entrypoint. * * - Contract surface (slice A1): zod I/O for all 16 endpoints. - * - Effect services (A2): `Civ7TunerClient`, `StudioConfig`. + * - Effect services (A2): `Civ7TunerClient`, `StudioConfig`; `Civ7TunerSession` + * (tuner-session workstream) — the scoped owner of the ONE shared FireTuner + * connection, with the backoff gate and the host injection/health ports. * - effect-orpc router (A3): `createStudioRouter` + non-uniform error mapping. * - Host handler (A4-lite): `createStudioRpcHandler(context)` → an `RPCHandler` - * the host mounts at `/rpc` (the Vite dev middleware this run; a Bun server - * later). A standalone Bun process is DEFERRED (FRAME §4.7). + * the host (the studio Bun daemon) mounts at `/rpc`, plus `tuner.*` ports and + * the `dispose()` shutdown obligation. * * The host supplies a {@link StudioServerContext} carrying the process singletons, * the catalog loader, and the three stateful engine fns (shared queue + dual-store @@ -26,4 +28,15 @@ export { createStudioRpcHandler, type StudioRpcHandle } from "./handler.js"; export { createStudioRouter, type StudioRouter } from "./router/index.js"; export { makeStudioRuntime, type StudioRuntime } from "./runtime.js"; export { Civ7TunerClient } from "./services/Civ7TunerClient.js"; +export { + CIV7_TUNER_GATE_COOLDOWN_MS, + CIV7_TUNER_GATE_THRESHOLD, + Civ7TunerBackoffError, + Civ7TunerSession, + Civ7TunerSessionLive, + makeCiv7TunerSessionLayer, + type Civ7TunerSessionApi, + type Civ7TunerSessionHealth, + type Civ7TunerSessionOptions, +} from "./services/Civ7TunerSession.js"; export { StudioConfig } from "./services/StudioConfig.js"; diff --git a/packages/studio-server/src/runtime.ts b/packages/studio-server/src/runtime.ts index 99fb9cfcf5..401928d52b 100644 --- a/packages/studio-server/src/runtime.ts +++ b/packages/studio-server/src/runtime.ts @@ -2,23 +2,33 @@ import { Layer, ManagedRuntime } from "effect"; import type { StudioServerContext } from "./context.js"; import { Civ7TunerClient } from "./services/Civ7TunerClient.js"; +import { Civ7TunerSession, Civ7TunerSessionLive } from "./services/Civ7TunerSession.js"; import { StudioConfig } from "./services/StudioConfig.js"; /** * Builds the `ManagedRuntime` backing the effect-orpc router for one host. * - * `Civ7TunerClient` is self-contained (wraps `@civ7/direct-control`). `StudioConfig` - * carries the host-supplied {@link StudioServerContext} (process singletons + - * catalog loader + stateful engine fns), so the runtime is constructed per host - * with that context injected as a `Layer`. + * `Civ7TunerSession` is the scoped owner of the ONE shared FireTuner + * connection; `Civ7TunerClient` consumes it, and layer memoization guarantees + * both graphs see the same instance (the SAME `Civ7TunerSessionLive` + * reference appears in the merge and in the client's dependencies). + * `StudioConfig` carries the host-supplied {@link StudioServerContext} + * (process singletons + catalog loader + stateful engine fns), so the runtime + * is constructed per host with that context injected as a `Layer`. + * + * Lifecycle obligation: the host MUST call `runtime.dispose()` on shutdown — + * that closes the runtime scope and runs the session's release finalizer + * (graceful FIN to the game). `createStudioRpcHandler` exposes this as + * `handle.dispose()`. */ export type StudioRuntime = ManagedRuntime.ManagedRuntime< - Civ7TunerClient | StudioConfig, + Civ7TunerClient | Civ7TunerSession | StudioConfig, never >; export function makeStudioRuntime(context: StudioServerContext): StudioRuntime { const layer = Layer.mergeAll( + Civ7TunerSessionLive, Civ7TunerClient.Default, Layer.succeed(StudioConfig, context), ); diff --git a/packages/studio-server/src/services/Civ7TunerClient.ts b/packages/studio-server/src/services/Civ7TunerClient.ts index 87f3f18343..f65801e345 100644 --- a/packages/studio-server/src/services/Civ7TunerClient.ts +++ b/packages/studio-server/src/services/Civ7TunerClient.ts @@ -14,70 +14,76 @@ import { } from "@civ7/direct-control"; import { Effect } from "effect"; +import { Civ7TunerSession, Civ7TunerSessionLive } from "./Civ7TunerSession.js"; + /** * `Civ7TunerClient` — Effect service wrapping the `@civ7/direct-control` FireTuner * socket + filesystem reads used by the studio's read surface. * - * Each accessor lifts a direct-control promise into an Effect via - * `Effect.tryPromise`, surfacing the rejection in the Effect error channel so the - * procedure layer (router/*) can map it to the correct legacy status code. The - * direct-control *call shapes* (args, timeout, includeAreaRegionCounts, clamps) - * are lifted verbatim from the `/api/*` handlers in - * `apps/mapgen-studio/vite.config.ts` — no semantic change. + * Every tuner read routes through the shared `Civ7TunerSession` (`use` + + * `options.session` injection): one multiplexed connection for the whole + * polling surface instead of connect-per-request — the churn that wedged the + * game — plus the session's backoff gate when the tuner stops answering. The + * direct-control *call shapes* (args, timeout, includeAreaRegionCounts, + * clamps) are unchanged from the legacy handlers — no semantic change to the + * read surface. `savedConfigurations` is a filesystem read (no socket) and + * deliberately bypasses the session/gate. * - * Parity note: the studio remains direct-control-backed for these live reads this - * run (FRAME §4.7). The control-oRPC seam (architecture/12) is designed-toward, - * not yet bound. No FireTuner reads are added beyond the existing handler set. + * Parity note: the studio remains direct-control-backed for these live reads + * this run. The control-oRPC seam (architecture/12) is designed-toward, not + * yet bound. No FireTuner reads are added beyond the existing handler set. */ export class Civ7TunerClient extends Effect.Service()( "@civ7/studio-server/Civ7TunerClient", { accessors: true, - sync: () => { + dependencies: [Civ7TunerSessionLive], + effect: Effect.gen(function* () { + const tuner = yield* Civ7TunerSession; const timeoutMs = DEFAULT_CIV7_TUNER_TIMEOUT_MS; return { // #1 status — getCiv7PlayableStatus({ timeoutMs }) playableStatus: () => - Effect.tryPromise(() => getCiv7PlayableStatus({ timeoutMs })), + tuner.use((o) => getCiv7PlayableStatus({ timeoutMs, ...o })), // #2 mapSummary — includeAreaRegionCounts: true mapSummary: () => - Effect.tryPromise(() => - getCiv7MapSummary({ timeoutMs, includeAreaRegionCounts: true }), + tuner.use((o) => + getCiv7MapSummary({ timeoutMs, includeAreaRegionCounts: true, ...o }), ), // live.status field read — includeAreaRegionCounts: false liveMapSummary: () => - Effect.tryPromise(() => - getCiv7MapSummary({ timeoutMs, includeAreaRegionCounts: false }), + tuner.use((o) => + getCiv7MapSummary({ timeoutMs, includeAreaRegionCounts: false, ...o }), ), - appUiSnapshot: () => Effect.tryPromise(() => getCiv7AppUiSnapshot({ timeoutMs })), + appUiSnapshot: () => tuner.use((o) => getCiv7AppUiSnapshot({ timeoutMs, ...o })), - autoplayStatus: () => Effect.tryPromise(() => getCiv7AutoplayStatus({ timeoutMs })), + autoplayStatus: () => tuner.use((o) => getCiv7AutoplayStatus({ timeoutMs, ...o })), // #3 / live.gameInfo — getCiv7GameInfoRows({ table, limit }, { timeoutMs }) gameInfoRows: (table: string, limit: number) => - Effect.tryPromise(() => getCiv7GameInfoRows({ table, limit }, { timeoutMs })), + tuner.use((o) => getCiv7GameInfoRows({ table, limit }, { timeoutMs, ...o })), // #10 setupConfig — getCiv7SetupSnapshot({ timeoutMs }) - setupSnapshot: () => Effect.tryPromise(() => getCiv7SetupSnapshot({ timeoutMs })), + setupSnapshot: () => tuner.use((o) => getCiv7SetupSnapshot({ timeoutMs, ...o })), - // #11 savedConfigs — listCiv7SavedGameConfigurations() + // #11 savedConfigs — filesystem read; no tuner socket, no gate. savedConfigurations: () => Effect.tryPromise(() => listCiv7SavedGameConfigurations()), // #5 live.snapshot — getCiv7MapGrid(input, { timeoutMs }) mapGrid: (input: Parameters[0]) => - Effect.tryPromise(() => getCiv7MapGrid(input, { timeoutMs })), + tuner.use((o) => getCiv7MapGrid(input, { timeoutMs, ...o })), // #6 live.entities — player/unit/city summaries playerSummary: (input: Parameters[0]) => - Effect.tryPromise(() => getCiv7PlayerSummary(input, { timeoutMs })), + tuner.use((o) => getCiv7PlayerSummary(input, { timeoutMs, ...o })), unitSummary: (input: Parameters[0]) => - Effect.tryPromise(() => getCiv7UnitSummary(input, { timeoutMs })), + tuner.use((o) => getCiv7UnitSummary(input, { timeoutMs, ...o })), citySummary: (input: Parameters[0]) => - Effect.tryPromise(() => getCiv7CitySummary(input, { timeoutMs })), + tuner.use((o) => getCiv7CitySummary(input, { timeoutMs, ...o })), }; - }, + }), }, ) {} diff --git a/packages/studio-server/src/services/Civ7TunerSession.ts b/packages/studio-server/src/services/Civ7TunerSession.ts new file mode 100644 index 0000000000..e584b425f9 --- /dev/null +++ b/packages/studio-server/src/services/Civ7TunerSession.ts @@ -0,0 +1,148 @@ +import { + Civ7DirectControlSession, + type Civ7DirectControlOptions, +} from "@civ7/direct-control"; +import { Clock, Context, Data, Effect, Layer, Ref, type Scope } from "effect"; +import type { UnknownException } from "effect/Cause"; + +/** + * `Civ7TunerSession` — the ONE Effect-scoped owner of the shared FireTuner + * connection (mapgen-studio-tuner-session workstream). + * + * Why: connect-per-request churn leaks descriptors inside the game process + * until the tuner wedges (Bun-server workstream Phase 4 addendum: 187 leaked + * fds, all reads timing out). The session protocol multiplexes concurrent + * requests over one socket (listenerIds) and `connect()` is reuse-idempotent + * (a dropped socket reconnects on the next request), so one instance serves + * every polling consumer for the runtime's whole life: + * + * - acquisition/release: `Layer.scoped` + `Effect.acquireRelease` — the + * session is created (not yet connected) when the layer builds and closed + * with a graceful FIN when the host disposes the `ManagedRuntime`. + * - `use(run)`: gated execution. The gate keys on the session's OWN + * consecutive-response-timeout counter (`session.stats` — it observes all + * traffic on the shared socket, including the control-oRPC mount's), so a + * busy/wedged tuner stops being hammered: past the threshold, reads fail + * fast with `Civ7TunerBackoffError` for one cooldown, then half-open (the + * next read flows; success resets the counter, a timeout re-opens). + * - `session`: exposed for host injection into non-Effect consumers (the + * control-oRPC `endpointDefaults.session` field). Lifecycle stays HERE. + * + * Deliberately NOT a pool/RcRef: the instance self-heals and never needs + * replacement (proportional complexity). + */ + +export interface Civ7TunerSessionGateOptions { + /** Consecutive response-timeouts that open the gate. */ + readonly threshold?: number; + /** How long reads fail fast before the gate half-opens. */ + readonly cooldownMs?: number; +} + +export type Civ7TunerSessionOptions = Civ7DirectControlOptions & { + readonly gate?: Civ7TunerSessionGateOptions; +}; + +export const CIV7_TUNER_GATE_THRESHOLD = 4; +export const CIV7_TUNER_GATE_COOLDOWN_MS = 15_000; + +/** Fail-fast error while the backoff gate is open (tuner not answering). */ +export class Civ7TunerBackoffError extends Data.TaggedError("Civ7TunerBackoffError")<{ + readonly consecutiveResponseTimeouts: number; + readonly retryAtMs: number; +}> { + override get message(): string { + return ( + `Civ7 tuner is not answering (${this.consecutiveResponseTimeouts} consecutive timeouts); ` + + `backing off until ${new Date(this.retryAtMs).toISOString()}` + ); + } +} + +export interface Civ7TunerSessionHealth { + readonly consecutiveResponseTimeouts: number; + /** ISO timestamp while the gate is open, null otherwise. */ + readonly gateOpenUntil: string | null; + /** Threshold crossed: the game process may be alive but the tuner silent. */ + readonly wedgeSuspected: boolean; +} + +export interface Civ7TunerSessionApi { + /** The shared session, for host injection into `Civ7DirectControlOptions.session`. */ + readonly session: Civ7DirectControlSession; + /** Run a direct-control promise against the shared session, behind the gate. */ + readonly use: ( + run: (options: { readonly session: Civ7DirectControlSession }) => Promise, + ) => Effect.Effect; + readonly health: Effect.Effect; +} + +export class Civ7TunerSession extends Context.Tag("@civ7/studio-server/Civ7TunerSession")< + Civ7TunerSession, + Civ7TunerSessionApi +>() {} + +const make = ( + options: Civ7TunerSessionOptions, +): Effect.Effect => + Effect.gen(function* () { + const threshold = options.gate?.threshold ?? CIV7_TUNER_GATE_THRESHOLD; + const cooldownMs = options.gate?.cooldownMs ?? CIV7_TUNER_GATE_COOLDOWN_MS; + const { gate: _gate, ...directControlOptions } = options; + + const session = yield* Effect.acquireRelease( + Effect.sync(() => new Civ7DirectControlSession(directControlOptions)), + (s) => Effect.promise(() => s.close()), + ); + const gateOpenUntil = yield* Ref.make(0); + + const use = ( + run: (o: { readonly session: Civ7DirectControlSession }) => Promise, + ) => + Effect.gen(function* () { + const now = yield* Clock.currentTimeMillis; + const openUntil = yield* Ref.get(gateOpenUntil); + if (now < openUntil) { + return yield* new Civ7TunerBackoffError({ + consecutiveResponseTimeouts: session.stats.consecutiveResponseTimeouts, + retryAtMs: openUntil, + }); + } + return yield* Effect.tryPromise(() => run({ session })).pipe( + Effect.tapError(() => + // The session's counter only moves on response-timeouts (the + // wedge/busy signature) — connection-refused (game not running) + // stays un-gated so readiness keeps reporting fast. + session.stats.consecutiveResponseTimeouts >= threshold + ? Effect.flatMap(Clock.currentTimeMillis, (at) => + Ref.set(gateOpenUntil, at + cooldownMs), + ) + : Effect.void, + ), + ); + }); + + const health = Effect.gen(function* () { + const openUntil = yield* Ref.get(gateOpenUntil); + const now = yield* Clock.currentTimeMillis; + const stats = session.stats; + return { + consecutiveResponseTimeouts: stats.consecutiveResponseTimeouts, + gateOpenUntil: openUntil > now ? new Date(openUntil).toISOString() : null, + wedgeSuspected: stats.consecutiveResponseTimeouts >= threshold, + } satisfies Civ7TunerSessionHealth; + }); + + return { session, use, health }; + }); + +/** Parameterized layer (tests inject a fake tuner endpoint + short gate timings). */ +export function makeCiv7TunerSessionLayer( + options: Civ7TunerSessionOptions = {}, +): Layer.Layer { + return Layer.scoped(Civ7TunerSession, make(options)); +} + +/** Production layer: env-resolved endpoint, default gate policy. */ +export const Civ7TunerSessionLive: Layer.Layer = + makeCiv7TunerSessionLayer();