diff --git a/apps/server/src/cloud/autoUpdate.test.ts b/apps/server/src/cloud/autoUpdate.test.ts new file mode 100644 index 0000000000000..bf483358b95cb --- /dev/null +++ b/apps/server/src/cloud/autoUpdate.test.ts @@ -0,0 +1,866 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import { + ServerSelfUpdateError, + type ServerSelfUpdateProgressStage, + type ServerSelfUpdateResult, + type TerminalMetadataStreamEvent, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; +import * as TestClock from "effect/testing/TestClock"; + +import { + makeWorkAdmissionGate, + memoryUpdateHoldStore, + UpdateHoldMarkerError, + type UpdateHoldStore, + WORK_ADMISSION_CONSUMERS, + WorkAdmission, + type WorkAdmissionGate, +} from "../workAdmission.ts"; +import { ServiceLauncherRejectedError } from "./serviceLauncherClient.ts"; +import { + activityReason, + afterCheck, + afterFailure, + afterProbe, + AUTO_UPDATE_CHECK_INTERVAL_MS, + AUTO_UPDATE_FAILURE_BACKOFF_MS, + AUTO_UPDATE_FIRST_CHECK_DELAY_MS, + AUTO_UPDATE_MAX_HOLD_MS, + AUTO_UPDATE_MAX_QUIET_WAIT_MS, + AUTO_UPDATE_QUIET_POLL_MS, + decideRelease, + hookReason, + initialAutoUpdateState, + type LauncherRecord, + makeAutoUpdater, + nextAutoUpdateStep, + noActivity, + readLauncherRecord, + runningTerminalCommands, + threadActivity, + AUTO_UPDATE_UNCONFIRMED_GRACE_MS, +} from "./autoUpdate.ts"; +import { SERVICE_LAUNCHER_PROTOCOL } from "./serviceProtocol.ts"; + +const CURRENT = "0.1.9-fork.1"; +const TARGET = "0.1.9-fork.2"; + +const outcome = (status: "committed" | "rolled-back" | "failed", id = "update-1") => ({ + id, + fromVersion: CURRENT, + targetVersion: TARGET, + status, + reason: `${status}-reason`, +}); +const pending = { + id: "update-2", + fromVersion: CURRENT, + targetVersion: TARGET, + dbPath: "/data/state.sqlite", + status: "pending" as const, +}; + +describe("decideRelease", () => { + const decide = (candidate: string | undefined, current = CURRENT) => + decideRelease({ current, candidate, lastUpdate: undefined, backedOffFailureId: undefined }); + + it.each([ + { current: "0.1.9-fork.1", candidate: "0.1.9-fork.2", expected: "0.1.9-fork.2" }, + { current: "0.1.9-fork.3", candidate: "0.1.10-fork.1", expected: "0.1.10-fork.1" }, + { current: "0.1.9", candidate: "0.1.10", expected: "0.1.10" }, + { current: "0.1.9-fork.2", candidate: "0.1.9-fork.2", expected: undefined }, + { current: "0.1.9-fork.2", candidate: "0.1.9-fork.1", expected: undefined }, + { current: "0.1.9", candidate: "0.1.10-nightly.20261112.1", expected: undefined }, + { current: "0.1.9", candidate: "latest", expected: undefined }, + { current: "0.1.9", candidate: undefined, expected: undefined }, + ])("$current -> $candidate", ({ current, candidate, expected }) => { + const decision = decide(candidate, current); + expect(decision.type === "offer" ? decision.version : undefined).toBe(expected); + }); + + it("skips the release the launcher rolled back, but not a newer one", () => { + const lastUpdate = outcome("rolled-back"); + expect( + decideRelease({ current: CURRENT, candidate: TARGET, lastUpdate, backedOffFailureId: "x" }), + ).toEqual({ type: "skip-rolled-back", version: TARGET, reason: "rolled-back-reason" }); + expect( + decideRelease({ + current: CURRENT, + candidate: "0.1.9-fork.3", + lastUpdate, + backedOffFailureId: undefined, + }), + ).toEqual({ type: "offer", version: "0.1.9-fork.3" }); + }); + + it("backs off once from a failed attempt, then offers the release again", () => { + const lastUpdate = outcome("failed", "failed-1"); + expect( + decideRelease({ + current: CURRENT, + candidate: TARGET, + lastUpdate, + backedOffFailureId: undefined, + }), + ).toEqual({ + type: "back-off-failed", + version: TARGET, + updateId: "failed-1", + reason: "failed-reason", + }); + expect( + decideRelease({ + current: CURRENT, + candidate: TARGET, + lastUpdate, + backedOffFailureId: "failed-1", + }), + ).toEqual({ type: "offer", version: TARGET }); + }); + + it("does not offer while the launcher has an update pending", () => { + expect( + decideRelease({ + current: CURRENT, + candidate: TARGET, + lastUpdate: pending, + backedOffFailureId: undefined, + }).type, + ).toBe("launcher-busy"); + }); +}); + +it.layer(NodeServices.layer)("readLauncherRecord", (it) => { + it.effect("reads rolled-back and failed records, and reports a file it cannot use", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* fs.makeTempDirectoryScoped(); + const file = path.join(dir, "service-state.json"); + const write = (update: unknown) => + fs.writeFileString( + file, + JSON.stringify({ protocol: SERVICE_LAUNCHER_PROTOCOL, activeVersion: CURRENT, update }), + ); + + expect(yield* readLauncherRecord(fs, file)).toEqual({ type: "none" }); + yield* write(outcome("rolled-back")); + expect(yield* readLauncherRecord(fs, file)).toEqual({ + type: "update", + update: outcome("rolled-back"), + }); + yield* write(outcome("failed")); + expect(yield* readLauncherRecord(fs, file)).toEqual({ + type: "update", + update: outcome("failed"), + }); + yield* fs.writeFileString(file, "{ not json"); + expect((yield* readLauncherRecord(fs, file)).type).toBe("unreadable"); + }).pipe(Effect.scoped), + ); +}); + +describe("threadActivity", () => { + const run = (status: string, extra: Record = {}) => + ({ id: `run-${status}`, ordinal: 1, status, ...extra }) as never; + const projection = ( + input: { + readonly runs?: ReadonlyArray; + readonly runtimeRequests?: ReadonlyArray; + readonly providerThreads?: ReadonlyArray; + readonly turnItems?: ReadonlyArray; + } = {}, + ) => ({ + thread: { activeProviderThreadId: null } as never, + runs: input.runs ?? [], + runtimeRequests: input.runtimeRequests ?? [], + providerThreads: input.providerThreads ?? [], + turnItems: input.turnItems ?? [], + }); + + it("counts waiting, preparing, starting and running runs as busy", () => { + for (const status of ["preparing", "starting", "running", "waiting"]) { + expect(threadActivity(projection({ runs: [run(status)] })).activeRuns).toBe(1); + } + expect(threadActivity(projection({ runs: [run("completed")] }))).toEqual(noActivity); + expect(threadActivity(projection({ runs: [run("rolled_back")] }))).toEqual(noActivity); + }); + + it("counts unheld queued runs, not held ones", () => { + expect(threadActivity(projection({ runs: [run("queued")] })).queuedRuns).toBe(1); + expect( + threadActivity(projection({ runs: [run("queued", { queueHeld: true })] })).queuedRuns, + ).toBe(0); + }); + + it("counts pending approvals and questions", () => { + const requests = [{ status: "pending" }, { status: "resolved" }] as never[]; + expect(threadActivity(projection({ runtimeRequests: requests })).pendingRequests).toBe(1); + }); + + it("counts background work the way upstream does, for every provider", () => { + // A Codex-style command left running after its turn completed: a turn item, no roster. + const codexCommand = { + id: "item-1", + type: "command_execution", + status: "running", + title: "npm run build", + nativeItemRef: null, + input: "npm run build", + } as never; + expect( + threadActivity(projection({ runs: [run("completed")], turnItems: [codexCommand] })) + .backgroundTasks, + ).toBe(1); + // A Claude-style roster on the provider thread. + const roster = { + id: "pt-1", + pendingBackgroundTasks: [{ taskId: "bg-1", kind: "command" }], + } as never; + expect( + threadActivity(projection({ runs: [run("completed")], providerThreads: [roster] })) + .backgroundTasks, + ).toBe(1); + }); +}); + +describe("activityReason", () => { + it("is null only when nothing is working", () => { + expect(activityReason(noActivity)).toBeNull(); + expect(activityReason({ ...noActivity, activeRuns: 1 })).toBe("1 running turn"); + expect(activityReason({ ...noActivity, pendingRequests: 2, backgroundTasks: 1 })).toBe( + "2 pending approval or questions, 1 background task", + ); + expect(activityReason({ ...noActivity, queuedRuns: 1, terminalCommands: 2 })).toBe( + "1 queued turn, 2 running terminal commands", + ); + }); +}); + +describe("hookReason", () => { + it("allows only a clean exit", () => { + expect(hookReason({ code: 0, stdout: "ok", stderr: "", timedOut: false })).toBeNull(); + expect( + hookReason({ code: 1, stdout: "\n deploy running \nmore", stderr: "", timedOut: false }), + ).toBe("the quiet hook said not now: deploy running"); + expect(hookReason({ code: 3, stdout: "", stderr: "busy", timedOut: false })).toBe( + "the quiet hook said not now: busy", + ); + expect(hookReason({ code: 1, stdout: "", stderr: "", timedOut: false })).toBe( + "the quiet hook said not now", + ); + expect(hookReason({ code: null, stdout: "", stderr: "", timedOut: true })).toBe( + "the quiet hook did not answer in time", + ); + expect(hookReason({ error: "ProcessSpawnError" })).toBe( + "the quiet hook could not run (ProcessSpawnError)", + ); + }); +}); + +describe("auto-update schedule", () => { + const start = 1_000_000; + + it("checks first after the startup delay, then hourly when nothing is newer", () => { + const state = initialAutoUpdateState(start); + expect(nextAutoUpdateStep(state, start)).toEqual({ type: "none" }); + const due = start + AUTO_UPDATE_FIRST_CHECK_DELAY_MS; + expect(nextAutoUpdateStep(state, due)).toEqual({ type: "check" }); + const after = afterCheck(due, undefined); + expect(after).toEqual({ phase: "idle", nextCheckAt: due + AUTO_UPDATE_CHECK_INTERVAL_MS }); + }); + + it("waits for quiet once a minute, then updates", () => { + const waiting = afterCheck(start, TARGET); + expect(nextAutoUpdateStep(waiting, start)).toEqual({ type: "probe", version: TARGET }); + if (waiting.phase !== "waiting") throw new Error("expected waiting"); + const busy = afterProbe(waiting, start, "1 running turn"); + expect(busy.outcome).toBe("wait"); + expect(nextAutoUpdateStep(busy.state, start + 1)).toEqual({ type: "none" }); + expect(nextAutoUpdateStep(busy.state, start + AUTO_UPDATE_QUIET_POLL_MS)).toEqual({ + type: "probe", + version: TARGET, + }); + if (busy.state.phase !== "waiting") throw new Error("expected waiting"); + expect(afterProbe(busy.state, start + AUTO_UPDATE_QUIET_POLL_MS, null).outcome).toBe("update"); + }); + + it("never forces: a bounded wait gives up until the next check", () => { + const waiting = afterCheck(start, TARGET); + if (waiting.phase !== "waiting") throw new Error("expected waiting"); + const late = start + AUTO_UPDATE_MAX_QUIET_WAIT_MS; + const gaveUp = afterProbe(waiting, late, "1 running turn"); + expect(gaveUp.outcome).toBe("give-up"); + expect(gaveUp.state).toEqual({ + phase: "idle", + nextCheckAt: late + AUTO_UPDATE_CHECK_INTERVAL_MS, + }); + // Quiet at the bound still updates. + expect(afterProbe(waiting, late, null).outcome).toBe("update"); + }); + + it("backs off after a failed install, longer than a normal check", () => { + const failed = afterFailure(start); + expect(failed).toEqual({ phase: "idle", nextCheckAt: start + AUTO_UPDATE_FAILURE_BACKOFF_MS }); + expect(AUTO_UPDATE_FAILURE_BACKOFF_MS).toBeGreaterThan(AUTO_UPDATE_CHECK_INTERVAL_MS); + expect(nextAutoUpdateStep(failed, start + AUTO_UPDATE_CHECK_INTERVAL_MS)).toEqual({ + type: "none", + }); + expect(nextAutoUpdateStep(failed, start + AUTO_UPDATE_FAILURE_BACKOFF_MS)).toEqual({ + type: "check", + }); + }); +}); + +/** A gate as the server builds it: a marker store, and every consumer admits through it. */ +const activeGate = Effect.fn("test.activeGate")(function* ( + marker?: UpdateHoldStore, + consumers: ReadonlyArray<(typeof WORK_ADMISSION_CONSUMERS)[number]> = WORK_ADMISSION_CONSUMERS, +) { + const gate = yield* makeWorkAdmissionGate({ marker: marker ?? (yield* memoryUpdateHoldStore) }); + for (const consumer of consumers) yield* gate.register(consumer); + // Startup has started whatever the last update held. + yield* gate.finishHeldWork; + return gate; +}); + +/** A fake server around the real update loop: launcher, thread state, hook and self-update. */ +const makeWorld = Effect.fn("test.makeWorld")(function* ( + options: { + readonly bootService?: boolean; + readonly gate?: WorkAdmissionGate; + readonly marker?: Effect.Success; + /** + * What the launcher does when asked: accept, refuse, or take the request without the answer + * coming back (the 30 s reply timeout), with or without its record then showing it pending. + */ + readonly launcher?: "accepts" | "rejects" | "answer-lost" | "answer-lost-then-pending"; + /** Fails the update before any stage, like the Update button's update already running. */ + readonly refuseBeforeStart?: boolean; + } = {}, +) { + const marker = options.marker ?? (yield* memoryUpdateHoldStore); + const gate = options.gate ?? (yield* activeGate(marker)); + /** Whether the gate was closed each time thread state was read during the handoff. */ + const heldAtInstallRead = yield* Ref.make>([]); + const resets = yield* Ref.make(0); + const activity = yield* Ref.make(null); + /** Thread state seen at the last check, right before the handoff. */ + const activityAtInstall = yield* Ref.make(null); + const hook = yield* Ref.make(null); + const hookAtInstall = yield* Ref.make(null); + const installing = yield* Ref.make(false); + const record = yield* Ref.make({ type: "none" }); + const release = yield* Ref.make(TARGET); + const enabled = yield* Ref.make(true); + /** The setting as read at the last check, right before the handoff. */ + const enabledAtInstall = yield* Ref.make(true); + const calls = yield* Ref.make>([]); + const note = (call: string) => Ref.update(calls, (all) => [...all, call]); + + const update = ( + input: { readonly targetVersion: string }, + report?: (stage: ServerSelfUpdateProgressStage) => Effect.Effect, + onHandoffAccepted?: () => Effect.Effect, + ): Effect.Effect => + Effect.gen(function* () { + yield* note(`update ${input.targetVersion}`); + if (options.refuseBeforeStart) { + return yield* new ServerSelfUpdateError({ + reason: "A server update is already in progress.", + }); + } + yield* report?.("downloading") ?? Effect.void; + yield* Ref.set(installing, true); + yield* report?.("installing") ?? Effect.void; + const launcher = options.launcher ?? "accepts"; + if (launcher === "rejects") { + return yield* new ServerSelfUpdateError({ + reason: "Another server update is pending.", + cause: new ServiceLauncherRejectedError({ + targetVersion: input.targetVersion, + reason: "Another server update is pending.", + }), + }); + } + if (launcher !== "accepts") { + yield* note("launcher asked"); + if (launcher === "answer-lost-then-pending") { + yield* Ref.set(record, { type: "update", update: pending }); + } + return yield* new ServerSelfUpdateError({ + reason: "Could not ask the service launcher to activate the prepared update.", + cause: new Error("timed out waiting for the launcher"), + }); + } + yield* note("handoff accepted"); + yield* Ref.set(record, { type: "update", update: pending }); + yield* onHandoffAccepted?.() ?? Effect.void; + return { targetVersion: input.targetVersion, method: "boot-service" as const }; + }).pipe(Effect.ensuring(Ref.set(installing, false))); + + const updater = yield* makeAutoUpdater({ + currentVersion: CURRENT, + bootService: options.bootService ?? true, + gate, + preferences: Effect.all([ + Ref.get(installing), + Ref.get(enabled), + Ref.get(enabledAtInstall), + ]).pipe( + Effect.map(([staging, enabledNow, enabledLate]) => (staging ? enabledLate : enabledNow)), + Effect.map((automaticUpdates) => ({ + automaticUpdates, + automaticUpdatesQuietCommand: "/usr/local/bin/quiet", + })), + ), + newestRelease: Ref.get(release), + launcherRecord: Ref.get(record), + activity: Effect.gen(function* () { + yield* note("activity"); + if (!(yield* Ref.get(installing))) return yield* Ref.get(activity); + const held = yield* gate.isHeld; + yield* Ref.update(heldAtInstallRead, (all) => [...all, held]); + return yield* Ref.get(activityAtInstall); + }), + askHook: () => + Effect.gen(function* () { + yield* note("hook"); + return (yield* Ref.get(installing)) ? yield* Ref.get(hookAtInstall) : yield* Ref.get(hook); + }), + update, + resetAfterDroppedHandoff: Ref.update(resets, (count) => count + 1), + }); + const at = (ms: number) => TestClock.setTime(ms).pipe(Effect.andThen(updater.tick)); + return { + ...updater, + gate, + marker, + heldAtInstallRead: Ref.get(heldAtInstallRead), + resets: Ref.get(resets), + activity, + activityAtInstall, + hook, + hookAtInstall, + record, + release, + enabled, + enabledAtInstall, + calls: Ref.get(calls), + clearCalls: Ref.set(calls, []), + at, + }; +}); + +const DUE = AUTO_UPDATE_FIRST_CHECK_DELAY_MS; + +describe("auto-update loop", () => { + it.effect( + "updates when quiet, asking the hook first and thread state last, behind the gate", + () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* world.at(DUE); + expect((yield* world.state).phase).toBe("waiting"); + yield* world.clearCalls; + yield* world.at(DUE + 1); + expect(yield* world.calls).toEqual([ + "hook", + "activity", + `update ${TARGET}`, + // The last check: hook, then the gate closes, then thread state. + "hook", + "activity", + "handoff accepted", + ]); + expect(yield* world.state).toEqual({ + phase: "handed-off", + version: TARGET, + since: DUE + 1, + confirmed: true, + }); + expect(yield* world.gate.isHeld).toBe(true); + // Thread state was read once more at the handoff, after the gate had closed. + expect(yield* world.heldAtInstallRead).toEqual([true]); + // The marker tells the next startup which work waited for the restart. + expect(yield* world.marker.current).toEqual({ + heldAt: DateTime.formatIso(DateTime.makeUnsafe(DUE + 1)), + targetVersion: TARGET, + }); + }), + ); + + it.effect("holds new work from the last check until exit", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* world.at(DUE); + yield* world.at(DUE + 1); + const started = yield* Ref.make(false); + const fiber = yield* Effect.forkChild(world.gate.admit(Ref.set(started, true))); + for (let i = 0; i < 10; i++) yield* Effect.yieldNow; + expect(yield* Ref.get(started)).toBe(false); + // Still pending at the launcher: the hold stays, whatever the settings say. + yield* Ref.set(world.enabled, false); + yield* world.at(DUE + AUTO_UPDATE_QUIET_POLL_MS * 5); + for (let i = 0; i < 10; i++) yield* Effect.yieldNow; + expect(yield* Ref.get(started)).toBe(false); + yield* Fiber.interrupt(fiber); + }), + ); + + it.effect( + "opens the gate again when the launcher drops the handoff and the server lives on", + () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect(yield* world.gate.isHeld).toBe(true); + yield* Ref.set(world.record, { type: "update", update: outcome("failed", "update-2") }); + const later = DUE + 2 * AUTO_UPDATE_QUIET_POLL_MS; + yield* world.at(later); + expect(yield* world.gate.isHeld).toBe(false); + expect(yield* world.state).toEqual(afterFailure(later)); + // Work released here starts now, not again after some later restart. + expect(yield* world.marker.current).toBeUndefined(); + // The next update (the button's or an automatic one) may start. + expect(yield* world.resets).toBe(1); + }), + ); + + it.effect("re-checks thread state right before the handoff and waits if work started", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* Ref.set(world.activityAtInstall, "1 running turn"); + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect((yield* world.calls).includes("handoff accepted")).toBe(false); + const state = yield* world.state; + expect(state.phase === "waiting" && state.lastReason).toBe("1 running turn"); + expect(yield* world.gate.isHeld).toBe(false); + }), + ); + + it.effect("re-asks the hook right before the handoff", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* Ref.set(world.hookAtInstall, "the quiet hook said not now: deploy"); + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect((yield* world.calls).includes("handoff accepted")).toBe(false); + expect((yield* world.state).phase).toBe("waiting"); + expect(yield* world.gate.isHeld).toBe(false); + }), + ); + + it.effect("stops at the handoff when the setting was turned off while staging", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* Ref.set(world.enabledAtInstall, false); + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect((yield* world.calls).includes("handoff accepted")).toBe(false); + expect(yield* world.gate.isHeld).toBe(false); + }), + ); + + it.effect("counts work that was admitted before the gate closed", () => + Effect.gen(function* () { + const gate = yield* activeGate(); + const world = yield* makeWorld({ gate }); + const fiber = yield* Effect.forkChild(gate.admit(Effect.never)); + for (let i = 0; i < 5; i++) yield* Effect.yieldNow; + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect((yield* world.calls).includes("handoff accepted")).toBe(false); + const state = yield* world.state; + expect(state.phase === "waiting" && state.lastReason).toBe("1 piece of work starting"); + expect(yield* gate.isHeld).toBe(false); + yield* Fiber.interrupt(fiber); + }), + ); + + it.effect("opens the gate again when the launcher rejects the handoff", () => + Effect.gen(function* () { + const world = yield* makeWorld({ launcher: "rejects" }); + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect(yield* world.gate.isHeld).toBe(false); + expect(yield* world.state).toEqual(afterFailure(DUE + 1)); + expect(yield* world.marker.current).toBeUndefined(); + }), + ); + + it.effect("waits a minute, not six hours, when the Update button's update is running", () => + Effect.gen(function* () { + const world = yield* makeWorld({ refuseBeforeStart: true }); + yield* world.at(DUE); + yield* world.at(DUE + 1); + const state = yield* world.state; + expect(state.phase).toBe("waiting"); + expect(nextAutoUpdateStep(state, DUE + 1 + AUTO_UPDATE_QUIET_POLL_MS).type).toBe("probe"); + }), + ); + + it.effect("waits while thread state is busy and never calls the update", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* Ref.set(world.activity, "1 waiting turn"); + yield* world.at(DUE); + for (let minute = 1; minute <= 5; minute++) { + yield* world.at(DUE + minute * AUTO_UPDATE_QUIET_POLL_MS); + } + expect((yield* world.calls).some((call) => call.startsWith("update"))).toBe(false); + }), + ); + + it.effect("runs only for a background service whose work all goes through the gate", () => + Effect.gen(function* () { + // The default when nothing provides the server's gate. + const passThrough = yield* WorkAdmission; + // All four start on the same clock, so each is due at the same times. + const desktop = yield* makeWorld({ bootService: false }); + const ungated = yield* makeWorld({ gate: passThrough }); + // Recovery built on some other gate: held work would be cancelled at the restart. + const partly = yield* makeWorld({ + gate: yield* activeGate(undefined, ["outbox", "thread-launch"]), + }); + const service = yield* makeWorld(); + for (const world of [desktop, ungated, partly, service]) { + yield* world.at(DUE); + yield* world.at(DUE + 1); + } + expect(yield* desktop.calls).toEqual([]); + expect(yield* ungated.calls).toEqual([]); + expect(yield* partly.calls).toEqual([]); + // The same steps do update a background service with an enforcing gate. + expect(yield* service.calls).toContain(`update ${TARGET}`); + }), + ); + + it.effect("skips a rolled-back release and backs off from a failed one", () => + Effect.gen(function* () { + const rolledBack = yield* makeWorld(); + const failed = yield* makeWorld(); + yield* Ref.set(rolledBack.record, { type: "update", update: outcome("rolled-back") }); + yield* rolledBack.at(DUE); + expect(yield* rolledBack.state).toEqual(afterCheck(DUE, undefined)); + + yield* Ref.set(failed.record, { type: "update", update: outcome("failed") }); + yield* failed.at(DUE); + expect(yield* failed.state).toEqual(afterFailure(DUE)); + // After the back-off the same release is tried again. + yield* failed.at(DUE + AUTO_UPDATE_FAILURE_BACKOFF_MS); + expect((yield* failed.state).phase).toBe("waiting"); + }), + ); + + it.effect("does not update when the launcher state cannot be read", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* Ref.set(world.record, { type: "unreadable", reason: "bad file" }); + yield* world.at(DUE); + expect(yield* world.state).toEqual(afterCheck(DUE, undefined)); + }), + ); + + it.effect("keeps new work waiting while the launcher state cannot be read", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* world.at(DUE); + yield* world.at(DUE + 1); + yield* Ref.set(world.record, { type: "unreadable", reason: "bad file" }); + for (let minute = 2; minute <= 20; minute++) { + yield* world.at(DUE + minute * AUTO_UPDATE_QUIET_POLL_MS); + } + expect(yield* world.gate.isHeld).toBe(true); + expect((yield* world.state).phase).toBe("handed-off"); + expect(yield* world.resets).toBe(0); + }), + ); + + it.effect("keeps holding past the maximum hold while the launcher has this update pending", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* world.at(DUE); + yield* world.at(DUE + 1); + // A slow restart: the launcher's record still shows this update pending, so the restart is + // still coming and work started now would be cut off by it. + for (const extra of [0, 1, 10 * AUTO_UPDATE_QUIET_POLL_MS]) { + yield* world.at(DUE + 1 + AUTO_UPDATE_MAX_HOLD_MS + extra); + expect(yield* world.gate.isHeld).toBe(true); + } + expect((yield* world.state).phase).toBe("handed-off"); + expect(yield* world.resets).toBe(0); + expect(yield* world.marker.current).toMatchObject({ targetVersion: TARGET }); + // Once the record says something else, the cap lets held work start. + yield* Ref.set(world.record, { type: "update", update: outcome("rolled-back") }); + const due = DUE + 1 + AUTO_UPDATE_MAX_HOLD_MS + 11 * AUTO_UPDATE_QUIET_POLL_MS; + yield* world.at(due); + expect(yield* world.gate.isHeld).toBe(false); + expect(yield* world.resets).toBe(1); + }), + ); + + it.effect("starts held work after the maximum hold when the launcher record is unreadable", () => + Effect.gen(function* () { + const world = yield* makeWorld(); + yield* world.at(DUE); + yield* world.at(DUE + 1); + // A hung launcher whose record cannot be read: nothing shows a restart is still coming. + const almost = DUE + 1 + AUTO_UPDATE_MAX_HOLD_MS - 1; + yield* Ref.set(world.record, { type: "unreadable", reason: "bad file" }); + yield* world.at(almost); + expect(yield* world.gate.isHeld).toBe(true); + const due = DUE + 1 + AUTO_UPDATE_MAX_HOLD_MS; + yield* world.at(due); + expect(yield* world.gate.isHeld).toBe(false); + expect(yield* world.state).toEqual(afterFailure(due)); + expect(yield* world.resets).toBe(1); + expect(yield* world.marker.current).toBeUndefined(); + }), + ); + + it.effect("keeps work waiting when the launcher's answer is lost but its record shows it", () => + Effect.gen(function* () { + const world = yield* makeWorld({ launcher: "answer-lost-then-pending" }); + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect(yield* world.gate.isHeld).toBe(true); + expect(yield* world.state).toEqual({ + phase: "handed-off", + version: TARGET, + since: DUE + 1, + confirmed: false, + }); + yield* world.at(DUE + 2 * AUTO_UPDATE_QUIET_POLL_MS); + expect((yield* world.state).phase === "handed-off" && (yield* world.state)).toMatchObject({ + confirmed: true, + }); + // Past the grace period it is confirmed, so it keeps waiting for the restart. + yield* world.at(DUE + 1 + AUTO_UPDATE_UNCONFIRMED_GRACE_MS + AUTO_UPDATE_QUIET_POLL_MS); + expect(yield* world.gate.isHeld).toBe(true); + }), + ); + + it.effect("drops a lost handoff only after the grace period shows nothing pending", () => + Effect.gen(function* () { + const world = yield* makeWorld({ launcher: "answer-lost" }); + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect(yield* world.calls).toContain("launcher asked"); + expect(yield* world.gate.isHeld).toBe(true); + yield* world.at(DUE + 1 + AUTO_UPDATE_UNCONFIRMED_GRACE_MS - 1); + expect(yield* world.gate.isHeld).toBe(true); + const due = DUE + 1 + AUTO_UPDATE_UNCONFIRMED_GRACE_MS; + yield* world.at(due); + expect(yield* world.gate.isHeld).toBe(false); + expect(yield* world.state).toEqual(afterFailure(due)); + expect(yield* world.resets).toBe(1); + }), + ); + + it.effect("does not hold when the update marker cannot be written", () => + Effect.gen(function* () { + const memory = yield* memoryUpdateHoldStore; + const marker = { + ...memory, + write: () => Effect.fail(new UpdateHoldMarkerError({ reason: "disk full" })), + }; + const world = yield* makeWorld({ marker }); + yield* world.at(DUE); + yield* world.at(DUE + 1); + expect((yield* world.calls).includes("handoff accepted")).toBe(false); + const state = yield* world.state; + expect(state.phase === "waiting" && state.lastReason).toContain("could not be recorded"); + expect(yield* world.gate.isHeld).toBe(false); + }), + ); + + it.effect("does not check for updates until work held by the last update has started", () => + Effect.gen(function* () { + const marker = yield* memoryUpdateHoldStore; + const left = { heldAt: DateTime.formatIso(DateTime.makeUnsafe(DUE)), targetVersion: CURRENT }; + yield* marker.write(left); + const gate = yield* makeWorkAdmissionGate({ marker }); + for (const consumer of WORK_ADMISSION_CONSUMERS) yield* gate.register(consumer); + const world = yield* makeWorld({ gate, marker }); + // Startup recovery kept the held work; the update is not committed yet, so none started. + yield* TestClock.setTime(DUE); + expect(yield* gate.heldForUpdate).toBeDefined(); + for (let minute = 0; minute < 5; minute++) { + yield* world.at(DUE + minute * QUIET_TICK); + } + expect(yield* world.state).toEqual(initialAutoUpdateState(0)); + expect(yield* world.calls).toEqual([]); + expect(yield* gate.isHeld).toBe(false); + expect(yield* marker.current).toEqual(left); + // Startup started it: updates go on as usual. + yield* gate.finishHeldWork; + yield* world.at(DUE + 5 * QUIET_TICK); + expect((yield* world.state).phase).toBe("waiting"); + expect(yield* marker.current).toBeUndefined(); + }), + ); +}); + +const QUIET_TICK = AUTO_UPDATE_QUIET_POLL_MS; + +describe("runningTerminalCommands", () => { + const terminal = (terminalId: string, hasRunningSubprocess: boolean) => ({ + threadId: "thread-1", + terminalId, + cwd: "/work", + worktreePath: null, + status: "running" as const, + pid: 10, + exitCode: null, + exitSignal: null, + hasRunningSubprocess, + label: terminalId, + updatedAt: "2026-01-01T00:00:00.000Z", + }); + + it.effect("follows snapshots, upserts and removals", () => + Effect.gen(function* () { + const listeners: Array<(event: TerminalMetadataStreamEvent) => Effect.Effect> = []; + let unsubscribed = 0; + const send = (event: TerminalMetadataStreamEvent) => + Effect.forEach(listeners, (listener) => listener(event), { discard: true }); + yield* Effect.scoped( + Effect.gen(function* () { + const count = yield* runningTerminalCommands({ + subscribeMetadata: (listener) => + Effect.sync(() => { + listeners.push(listener); + return () => { + unsubscribed++; + }; + }), + }); + yield* send({ + type: "snapshot", + terminals: [terminal("a", true), terminal("b", false), terminal("c", true)], + }); + expect(yield* count).toBe(2); + yield* send({ type: "upsert", terminal: terminal("b", true) }); + yield* send({ type: "upsert", terminal: terminal("a", false) }); + expect(yield* count).toBe(2); + yield* send({ type: "remove", threadId: "thread-1", terminalId: "c" }); + expect(yield* count).toBe(1); + // A new snapshot replaces everything seen before. + yield* send({ type: "snapshot", terminals: [terminal("d", false)] }); + expect(yield* count).toBe(0); + }), + ); + expect(unsubscribed).toBe(1); + }), + ); +}); diff --git a/apps/server/src/cloud/autoUpdate.ts b/apps/server/src/cloud/autoUpdate.ts new file mode 100644 index 0000000000000..570117c667fb0 --- /dev/null +++ b/apps/server/src/cloud/autoUpdate.ts @@ -0,0 +1,830 @@ +// Automatic updates for boot-service installs, once threads are quiet and the optional hook agrees. +// The handoff closes the WorkAdmission gate and writes a marker; held work starts once the +// restarted server's update is committed. The gate stays closed until exit or the launcher's +// record says no restart is coming (AUTO_UPDATE_MAX_HOLD_MS applies only while not pending). +import { + ServerSelfUpdateError, + type ServerSelfUpdateOutcome, + type ServerSettingsError, +} from "@t3tools/contracts"; +import { cliReleaseChannelOf } from "@t3tools/shared/cliRelease"; +import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; +import { latestUnheldRun } from "@t3tools/shared/orchestrationV2ThreadError"; +import * as Clock from "effect/Clock"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Predicate from "effect/Predicate"; +import * as Ref from "effect/Ref"; +import { HttpClient } from "effect/http"; + +import packageJson from "../../package.json" with { type: "json" }; +import { resolveMirrorVersion, resolveNewestVersion } from "../cli/update.ts"; +import * as ServerConfig from "../config.ts"; +import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; +import * as ManagedPolicy from "../policy/ManagedPolicy.ts"; +import { managedUpdateMirror } from "../policy/managedUpdates.ts"; +import * as ProcessRunner from "../processRunner.ts"; +import { forkParked } from "../serverActivation.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as TerminalManager from "../terminal/Manager.ts"; +import { WorkAdmission, type WorkAdmissionGate } from "../workAdmission.ts"; +import { ServerSelfUpdate } from "./selfUpdate.ts"; +import * as ServiceLauncherClient from "./serviceLauncherClient.ts"; +import { + compareExactServiceVersions, + isExactServiceVersion, + parseServiceState, + type PendingServiceUpdate, + SERVICE_STATE_FILE, +} from "./serviceProtocol.ts"; + +const MINUTE = 60_000; +/** The first look waits out startup recovery and any restart continuations it starts. */ +export const AUTO_UPDATE_FIRST_CHECK_DELAY_MS = 5 * MINUTE; +export const AUTO_UPDATE_CHECK_INTERVAL_MS = 60 * MINUTE; +export const AUTO_UPDATE_QUIET_POLL_MS = MINUTE; +/** How long one found release waits for quiet before the wait ends until the next check. */ +export const AUTO_UPDATE_MAX_QUIET_WAIT_MS = 2 * 60 * MINUTE; +/** After a failed attempt, so a release that cannot install is not fetched every hour. */ +export const AUTO_UPDATE_FAILURE_BACKOFF_MS = 6 * 60 * MINUTE; +/** New work waits at most this long for the launcher to restart the server, then starts again, + * unless the launcher's record still shows this update pending. */ +export const AUTO_UPDATE_MAX_HOLD_MS = 30 * MINUTE; +/** When the launcher's answer was lost (its 30 s reply timeout), how long its record may take to + * show the update pending before the handoff counts as dropped. */ +export const AUTO_UPDATE_UNCONFIRMED_GRACE_MS = 5 * MINUTE; +const HOOK_TIMEOUT_MS = 30_000; +const HOOK_MAX_OUTPUT_BYTES = 16 * 1024; +const TICK = "1 minute"; + +export type AutoUpdateState = + | { readonly phase: "idle"; readonly nextCheckAt: number } + | { + readonly phase: "waiting"; + readonly version: string; + readonly since: number; + readonly nextQuietCheckAt: number; + readonly lastReason: string | null; + } + | { + readonly phase: "handed-off"; + readonly version: string; + /** When new work started waiting. */ + readonly since: number; + /** The launcher said it accepted, or its record shows this update pending. */ + readonly confirmed: boolean; + }; + +export const initialAutoUpdateState = (nowMs: number): AutoUpdateState => ({ + phase: "idle", + nextCheckAt: nowMs + AUTO_UPDATE_FIRST_CHECK_DELAY_MS, +}); + +/** The launcher's last update record, from `runtime/service-state.json`. */ +export type LauncherRecord = + | { readonly type: "none" } + | { readonly type: "unreadable"; readonly reason: string } + | { readonly type: "update"; readonly update: PendingServiceUpdate | ServerSelfUpdateOutcome }; + +export type ReleaseDecision = + | { readonly type: "none" } + | { readonly type: "offer"; readonly version: string } + /** The launcher rolled this release back: it does not run here, so only a newer one is tried. */ + | { readonly type: "skip-rolled-back"; readonly version: string; readonly reason?: string } + /** The last attempt at this release failed around it (backup, start, state): retry after a wait. */ + | { + readonly type: "back-off-failed"; + readonly version: string; + readonly updateId: string; + readonly reason?: string; + } + | { readonly type: "launcher-busy"; readonly version: string }; + +/** Whether `candidate` is worth installing over `current`: newer, on the same track, not rolled + * back, and not a failure this process has not waited out yet (`backedOffFailureId`). */ +export function decideRelease(input: { + readonly current: string; + readonly candidate: string | undefined; + readonly lastUpdate: PendingServiceUpdate | ServerSelfUpdateOutcome | undefined; + readonly backedOffFailureId: string | undefined; +}): ReleaseDecision { + const { current, candidate, lastUpdate } = input; + if (candidate === undefined || !isExactServiceVersion(candidate)) return { type: "none" }; + if (cliReleaseChannelOf(candidate) !== cliReleaseChannelOf(current)) return { type: "none" }; + if (compareExactServiceVersions(candidate, current) <= 0) return { type: "none" }; + if (lastUpdate?.status === "pending") return { type: "launcher-busy", version: candidate }; + if (lastUpdate !== undefined && lastUpdate.targetVersion === candidate) { + const reason = lastUpdate.reason === undefined ? {} : { reason: lastUpdate.reason }; + if (lastUpdate.status === "rolled-back") { + return { type: "skip-rolled-back", version: candidate, ...reason }; + } + if (lastUpdate.status === "failed" && lastUpdate.id !== input.backedOffFailureId) { + return { type: "back-off-failed", version: candidate, updateId: lastUpdate.id, ...reason }; + } + } + return { type: "offer", version: candidate }; +} + +export interface AutoUpdateActivity { + readonly activeRuns: number; + readonly queuedRuns: number; + readonly pendingRequests: number; + readonly backgroundTasks: number; + readonly terminalCommands: number; +} + +export const noActivity: AutoUpdateActivity = { + activeRuns: 0, + queuedRuns: 0, + pendingRequests: 0, + backgroundTasks: 0, + terminalCommands: 0, +}; + +/** Why thread and terminal state says "not now", or null when nothing is working. Held queued + * runs do not count: they wait for the user, across restarts too. */ +export function activityReason(activity: AutoUpdateActivity): string | null { + const parts = [ + [activity.activeRuns, "running turn"], + [activity.queuedRuns, "queued turn"], + [activity.pendingRequests, "pending approval or question"], + [activity.backgroundTasks, "background task"], + [activity.terminalCommands, "running terminal command"], + ] as const; + const busy = parts + .filter(([count]) => count > 0) + .map(([count, label]) => `${count} ${label}${count === 1 ? "" : "s"}`); + return busy.length === 0 ? null : busy.join(", "); +} + +/** One thread's work, from the same projection startup recovery reads, and counted with the + * background-work rule the thread list and idle compaction use (every provider's tasks and turn + * items). Pull request watches are left out: they survive a restart. */ +export function threadActivity( + projection: Pick< + ProjectionStore.ProjectionRuntimeRecoveryState, + "thread" | "runs" | "runtimeRequests" | "providerThreads" | "turnItems" + >, +): AutoUpdateActivity { + const runs = projection.runs.filter((run) => run.status !== "rolled_back"); + return { + ...noActivity, + activeRuns: runs.filter( + (run) => + run.status === "preparing" || + run.status === "starting" || + run.status === "running" || + run.status === "waiting", + ).length, + queuedRuns: runs.filter((run) => run.status === "queued" && run.queueHeld !== true).length, + pendingRequests: projection.runtimeRequests.filter((request) => request.status === "pending") + .length, + backgroundTasks: derivePendingBackgroundWork({ + latestRun: latestUnheldRun(runs), + runs, + providerThreads: projection.providerThreads, + turnItems: projection.turnItems, + activeProviderThreadId: projection.thread.activeProviderThreadId, + }).length, + }; +} + +const addActivity = (left: AutoUpdateActivity, right: AutoUpdateActivity): AutoUpdateActivity => ({ + activeRuns: left.activeRuns + right.activeRuns, + queuedRuns: left.queuedRuns + right.queuedRuns, + pendingRequests: left.pendingRequests + right.pendingRequests, + backgroundTasks: left.backgroundTasks + right.backgroundTasks, + terminalCommands: left.terminalCommands + right.terminalCommands, +}); + +/** The quiet hook's answer: exit 0 allows the update; anything else is "not now", with the first + * line it printed as the reason. */ +export function hookReason( + result: + | { + readonly code: number | null; + readonly stdout: string; + readonly stderr: string; + readonly timedOut: boolean; + } + | { readonly error: string }, +): string | null { + if ("error" in result) return `the quiet hook could not run (${result.error})`; + if (result.timedOut) return "the quiet hook did not answer in time"; + if (result.code === 0) return null; + const said = `${result.stdout}\n${result.stderr}` + .split(/\r?\n/) + .map((line) => line.trim()) + .find((line) => line !== ""); + return `the quiet hook said not now${said === undefined ? "" : `: ${said.slice(0, 200)}`}`; +} + +export type AutoUpdateStep = + | { readonly type: "none" } + | { readonly type: "check" } + | { readonly type: "probe"; readonly version: string } + | { readonly type: "watch-handoff"; readonly version: string }; + +/** What a tick does in `state` at `nowMs`. */ +export function nextAutoUpdateStep(state: AutoUpdateState, nowMs: number): AutoUpdateStep { + if (state.phase === "idle") + return nowMs >= state.nextCheckAt ? { type: "check" } : { type: "none" }; + if (state.phase === "waiting") { + return nowMs >= state.nextQuietCheckAt + ? { type: "probe", version: state.version } + : { type: "none" }; + } + return { type: "watch-handoff", version: state.version }; +} + +/** The state after a release check found `version`, or none to install. */ +export function afterCheck(nowMs: number, version: string | undefined): AutoUpdateState { + return version === undefined + ? { phase: "idle", nextCheckAt: nowMs + AUTO_UPDATE_CHECK_INTERVAL_MS } + : { phase: "waiting", version, since: nowMs, nextQuietCheckAt: nowMs, lastReason: null }; +} + +/** After a quiet probe: keep waiting (bounded), give up until the next check, or update now. */ +export function afterProbe( + state: Extract, + nowMs: number, + reason: string | null, +): { readonly state: AutoUpdateState; readonly outcome: "update" | "wait" | "give-up" } { + if (reason === null) return { state, outcome: "update" }; + if (nowMs - state.since >= AUTO_UPDATE_MAX_QUIET_WAIT_MS) { + return { + state: { phase: "idle", nextCheckAt: nowMs + AUTO_UPDATE_CHECK_INTERVAL_MS }, + outcome: "give-up", + }; + } + return { + state: { ...state, nextQuietCheckAt: nowMs + AUTO_UPDATE_QUIET_POLL_MS, lastReason: reason }, + outcome: "wait", + }; +} + +/** After an install attempt failed: back off longer than a normal check. */ +export function afterFailure(nowMs: number): AutoUpdateState { + return { phase: "idle", nextCheckAt: nowMs + AUTO_UPDATE_FAILURE_BACKOFF_MS }; +} + +export const readLauncherRecord = ( + fs: FileSystem.FileSystem, + statePath: string, +): Effect.Effect => + fs.readFileString(statePath).pipe( + Effect.map((contents): LauncherRecord => { + const state = parseServiceState(contents); + if (state === undefined) { + return { type: "unreadable", reason: `${SERVICE_STATE_FILE} is not valid launcher state` }; + } + return state.update === undefined + ? { type: "none" } + : { type: "update", update: state.update }; + }), + Effect.catch((error) => + Effect.succeed( + error.reason._tag === "NotFound" + ? { type: "none" } + : { + type: "unreadable", + reason: `${SERVICE_STATE_FILE} could not be read (${error.reason._tag})`, + }, + ), + ), + ); + +/** Everything the update loop reads or does, so tests drive it with fakes. */ +export interface AutoUpdateDeps { + readonly currentVersion: string; + /** Desktop-managed servers update with their app; others have no launcher to hand off to. */ + readonly bootService: boolean; + readonly gate: WorkAdmissionGate; + readonly preferences: Effect.Effect< + { readonly automaticUpdates: boolean; readonly automaticUpdatesQuietCommand: string }, + ServerSettingsError + >; + readonly newestRelease: Effect.Effect; + readonly launcherRecord: Effect.Effect; + /** Null when no thread or terminal is working, else why not; read failures are a reason. */ + readonly activity: Effect.Effect; + /** Null when the quiet hook (if any) allows the update, else why not. */ + readonly askHook: (command: string, targetVersion: string) => Effect.Effect; + readonly update: ServerSelfUpdate["Service"]["update"]; + readonly resetAfterDroppedHandoff: Effect.Effect; +} + +type InstallResult = + /** Unconfirmed: the launcher was asked but its answer was lost, so it may still restart. */ + | { readonly type: "handed-off"; readonly since: number; readonly confirmed: boolean } + /** Work started while staging; the staged runtime is reused by the next attempt. */ + | { readonly type: "busy"; readonly reason: string } + /** Refused before it began, e.g. the Update button's update is running. */ + | { readonly type: "not-started"; readonly reason: string } + | { readonly type: "failed"; readonly reason: string }; + +export const makeAutoUpdater = Effect.fn("cloud.auto_update.make")(function* ( + deps: AutoUpdateDeps, +) { + const state = yield* Ref.make( + initialAutoUpdateState(yield* Clock.currentTimeMillis), + ); + const loggedUnavailable = yield* Ref.make(false); + const loggedUnsettled = yield* Ref.make(false); + const loggedPastCap = yield* Ref.make(false); + const lastUnreadable = yield* Ref.make(null); + const backedOffFailureId = yield* Ref.make(undefined); + const hookCommand = deps.preferences.pipe( + Effect.map((preferences) => preferences.automaticUpdatesQuietCommand), + ); + + const logUnreadable = (reason: string, message: string) => + Ref.getAndSet(lastUnreadable, reason).pipe( + Effect.flatMap((previous) => + previous === reason ? Effect.void : Effect.logWarning(message, { reason }), + ), + ); + + /** Null when the server may restart now, else why not. The hook first, thread state last. */ + const quietReason = Effect.fn("cloud.auto_update.quietReason")(function* (targetVersion: string) { + const command = yield* hookCommand.pipe(Effect.orElseSucceed(() => null)); + if (command === null) return "settings could not be read"; + const hook = yield* deps.askHook(command, targetVersion); + if (hook !== null) return hook; + return yield* deps.activity; + }); + + /** The last check, right before the handoff: closes the gate, and opens it again unless quiet. */ + const holdIfQuiet = Effect.fn("cloud.auto_update.holdIfQuiet")(function* (targetVersion: string) { + const preferences = yield* deps.preferences.pipe(Effect.orElseSucceed(() => null)); + if (preferences === null) return "settings could not be read"; + if (!preferences.automaticUpdates) return "automatic server updates were turned off"; + const command = preferences.automaticUpdatesQuietCommand; + const hook = yield* deps.askHook(command, targetVersion); + if (hook !== null) return hook; + // Anything saved after the gate closes is either counted below (and stops the update) or held, + // and the marker lets startup recovery keep and later start the held part. Recovery decides by + // that order, not by request times; `heldAt` only tells a fresh marker from a stale one. + const heldAt = DateTime.formatIso(yield* DateTime.now); + const starting = yield* deps.gate.hold; + let reason = + starting > 0 + ? `${starting} piece${starting === 1 ? "" : "s"} of work starting` + : yield* deps.activity; + if (reason === null) { + reason = yield* deps.gate.markHeldForUpdate({ heldAt, targetVersion }).pipe( + Effect.as(null), + Effect.catch((error) => + Effect.succeed(`the update hold could not be recorded (${error.reason})`), + ), + ); + } + if (reason !== null) yield* deps.gate.release; + else { + yield* Effect.logInfo( + "Automatic server update: new work waits for the restart and starts after it.", + { targetVersion, heldAt, maxHoldMinutes: AUTO_UPDATE_MAX_HOLD_MS / MINUTE }, + ); + } + return reason; + }); + + const install = Effect.fn("cloud.auto_update.install")(function* (targetVersion: string) { + const lateReason = yield* Ref.make(null); + const started = yield* Ref.make(false); + /** When the gate closed for good and the launcher is being asked. */ + const heldSince = yield* Ref.make(null); + const accepted = yield* Ref.make(false); + return yield* deps + .update( + { targetVersion }, + (stage) => + Ref.set(started, true).pipe( + Effect.andThen( + stage !== "installing" + ? Effect.void + : holdIfQuiet(targetVersion).pipe( + Effect.flatMap((reason) => + reason === null + ? Clock.currentTimeMillis.pipe( + Effect.flatMap((nowMs) => Ref.set(heldSince, nowMs)), + ) + : Ref.set(lateReason, reason).pipe( + Effect.andThen(Effect.fail(new ServerSelfUpdateError({ reason }))), + ), + ), + ), + ), + ), + () => Ref.set(accepted, true), + ) + .pipe( + Effect.andThen( + Effect.gen(function* () { + const since = (yield* Ref.get(heldSince)) ?? (yield* Clock.currentTimeMillis); + return { type: "handed-off", since, confirmed: true } satisfies InstallResult; + }), + ), + Effect.catch((error) => + Effect.gen(function* () { + const late = yield* Ref.get(lateReason); + // holdIfQuiet opened the gate again itself. + if (late !== null) return { type: "busy", reason: late } satisfies InstallResult; + const since = yield* Ref.get(heldSince); + // The launcher's reply timed out or the IPC failed after it was asked: it may still + // restart this server, so new work keeps waiting while watchHandoff follows its record. + if ( + since !== null && + ((yield* Ref.get(accepted)) || + !Predicate.isTagged(error.cause, "ServiceLauncherRejectedError")) + ) { + yield* Effect.logWarning( + "Automatic server update: the launcher's answer was lost; new work keeps waiting while its state is watched.", + { targetVersion, reason: error.reason }, + ); + return { + type: "handed-off", + since, + confirmed: yield* Ref.get(accepted), + } satisfies InstallResult; + } + // Refused, or failed before the launcher was asked: no restart is coming. + yield* deps.gate.release; + return (yield* Ref.get(started)) + ? ({ type: "failed", reason: error.reason } satisfies InstallResult) + : ({ type: "not-started", reason: error.reason } satisfies InstallResult); + }), + ), + // Defects and interruptions: nothing says a restart is coming either. + Effect.onError(() => deps.gate.release), + ); + }); + + const check = Effect.fn("cloud.auto_update.check")(function* (nowMs: number) { + const record = yield* deps.launcherRecord; + if (record.type === "unreadable") { + // Without the record a rolled-back release could be tried again and again. + yield* logUnreadable( + record.reason, + "Automatic server update cannot read the launcher state; not updating.", + ); + yield* Ref.set(state, afterCheck(nowMs, undefined)); + return; + } + yield* Ref.set(lastUnreadable, null); + const candidate = yield* deps.newestRelease.pipe( + Effect.tapError((error) => + Effect.logWarning("Automatic server update could not list releases.", { + reason: error.reason, + }), + ), + Effect.orElseSucceed(() => undefined), + ); + const decision = decideRelease({ + current: deps.currentVersion, + candidate, + lastUpdate: record.type === "update" ? record.update : undefined, + backedOffFailureId: yield* Ref.get(backedOffFailureId), + }); + switch (decision.type) { + case "none": + yield* Ref.set(state, afterCheck(nowMs, undefined)); + return; + case "launcher-busy": + yield* Effect.logInfo( + "Automatic server update waits; the launcher has an update pending.", + { + targetVersion: decision.version, + }, + ); + yield* Ref.set(state, afterCheck(nowMs, undefined)); + return; + case "skip-rolled-back": + yield* Effect.logInfo("Automatic server update skips a release that was rolled back.", { + targetVersion: decision.version, + reason: decision.reason, + }); + yield* Ref.set(state, afterCheck(nowMs, undefined)); + return; + case "back-off-failed": + yield* Effect.logWarning( + "Automatic server update: the last attempt at this release failed; retrying later.", + { targetVersion: decision.version, reason: decision.reason }, + ); + yield* Ref.set(backedOffFailureId, decision.updateId); + yield* Ref.set(state, afterFailure(nowMs)); + return; + case "offer": + yield* Effect.logInfo("Automatic server update found a newer release; waiting for quiet.", { + currentVersion: deps.currentVersion, + targetVersion: decision.version, + }); + yield* Ref.set(state, afterCheck(nowMs, decision.version)); + } + }); + + const dropHandoff = Effect.fn("cloud.auto_update.dropHandoff")(function* ( + current: Extract, + nowMs: number, + why: string, + ) { + yield* deps.gate.release; + yield* deps.resetAfterDroppedHandoff; + yield* Effect.logWarning( + "Automatic server update: the launcher did not restart this server; held work starts now.", + { + targetVersion: current.version, + reason: why, + heldMinutes: Math.round((nowMs - current.since) / MINUTE), + }, + ); + yield* Ref.set(state, afterFailure(nowMs)); + }); + + /** The gate stays closed until exit. It opens again only when the launcher's record shows it + * will not restart this server, or after AUTO_UPDATE_MAX_HOLD_MS unless the record shows this + * update pending (the restart is still coming: work that started now would be cut off by it). + * Before the cap an unreadable record or a lost answer keeps the hold too. */ + const watchHandoff = Effect.fn("cloud.auto_update.watchHandoff")(function* ( + current: Extract, + nowMs: number, + ) { + const heldMs = nowMs - current.since; + const pastCap = heldMs >= AUTO_UPDATE_MAX_HOLD_MS; + const record = yield* deps.launcherRecord; + if ( + record.type === "update" && + record.update.status === "pending" && + record.update.targetVersion === current.version + ) { + yield* Ref.set(lastUnreadable, null); + if (!current.confirmed) yield* Ref.set(state, { ...current, confirmed: true }); + if (pastCap && !(yield* Ref.getAndSet(loggedPastCap, true))) { + yield* Effect.logWarning( + "Automatic server update: the restart is taking long; new work keeps waiting while the launcher has it pending.", + { targetVersion: current.version, heldMinutes: Math.round(heldMs / MINUTE) }, + ); + } + return; + } + if (pastCap) { + return yield* dropHandoff( + current, + nowMs, + `no restart after ${Math.round(heldMs / MINUTE)} minutes`, + ); + } + if (record.type === "unreadable") { + return yield* logUnreadable( + record.reason, + "Automatic server update cannot read the launcher state; new work keeps waiting.", + ); + } + yield* Ref.set(lastUnreadable, null); + if (!current.confirmed && heldMs < AUTO_UPDATE_UNCONFIRMED_GRACE_MS) return; + yield* dropHandoff( + current, + nowMs, + record.type === "update" + ? `the launcher's record says ${record.update.status} for ${record.update.targetVersion}` + : "the launcher has no update pending", + ); + }); + + const probe = Effect.fn("cloud.auto_update.probe")(function* ( + current: Extract, + nowMs: number, + ) { + const version = current.version; + const reason = yield* quietReason(version); + const probed = afterProbe(current, nowMs, reason); + if (probed.outcome === "give-up") { + yield* Effect.logInfo( + "Automatic server update stopped waiting; it retries at the next check.", + { + targetVersion: version, + reason, + waitedMinutes: Math.round((nowMs - current.since) / MINUTE), + }, + ); + yield* Ref.set(state, probed.state); + return; + } + if (probed.outcome === "wait") { + if (reason !== current.lastReason) { + yield* Effect.logInfo("Automatic server update is waiting.", { + targetVersion: version, + reason, + }); + } + yield* Ref.set(state, probed.state); + return; + } + yield* Effect.logInfo("Automatic server update starting; the server is quiet.", { + currentVersion: deps.currentVersion, + targetVersion: version, + }); + const result = yield* install(version); + const after = yield* Clock.currentTimeMillis; + switch (result.type) { + case "handed-off": + yield* Ref.set(state, { + phase: "handed-off", + version, + since: result.since, + confirmed: result.confirmed, + }); + return; + case "busy": + yield* Effect.logInfo("Automatic server update is waiting; not quiet at the handoff.", { + targetVersion: version, + reason: result.reason, + }); + yield* Ref.set(state, afterProbe(current, after, result.reason).state); + return; + case "not-started": + yield* Effect.logInfo("Automatic server update is waiting; another update is running.", { + targetVersion: version, + reason: result.reason, + }); + yield* Ref.set(state, afterProbe(current, after, result.reason).state); + return; + case "failed": + yield* Effect.logWarning("Automatic server update failed; it retries later.", { + targetVersion: version, + reason: result.reason, + }); + yield* Ref.set(state, afterFailure(after)); + } + }); + + const tick = Effect.gen(function* () { + const nowMs = yield* Clock.currentTimeMillis; + const current = yield* Ref.get(state); + // A handoff in progress is followed whatever the settings say now. + if (current.phase === "handed-off") return yield* watchHandoff(current, nowMs); + const preferences = yield* deps.preferences; + if (!preferences.automaticUpdates) { + if (current.phase === "waiting") { + yield* Effect.logInfo("Automatic server update turned off; not waiting any more.", { + targetVersion: current.version, + }); + yield* Ref.set(state, initialAutoUpdateState(nowMs)); + } + return; + } + if (!deps.bootService) { + if (!(yield* Ref.getAndSet(loggedUnavailable, true))) { + yield* Effect.logInfo( + "Automatic server updates need the T3 Code background service; desktop servers update with the app.", + ); + } + return; + } + // Holding a gate that the outbox, thread launches or recovery do not use would let work start + // and then be cancelled by the restart. + const missing = deps.gate.enforced ? yield* deps.gate.missingConsumers : ["shared gate"]; + if (missing.length > 0) { + if (!(yield* Ref.getAndSet(loggedUnavailable, true))) { + yield* Effect.logError( + "Automatic server updates are off: this server cannot hold new work for the restart.", + { notUsingTheWorkGate: missing }, + ); + } + return; + } + // Work an earlier update held must start before this server holds work again; startup starts + // it once its own update is committed. + if (!(yield* deps.gate.heldWorkSettled)) { + if (!(yield* Ref.getAndSet(loggedUnsettled, true))) { + yield* Effect.logInfo( + "Automatic server update waits until work held by the last update has started.", + ); + } + return; + } + const step = nextAutoUpdateStep(current, nowMs); + if (step.type === "check") return yield* check(nowMs); + if (step.type === "probe" && current.phase === "waiting") return yield* probe(current, nowMs); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Automatic server update tick failed.", { cause }), + ), + ); + + return { tick, state: Ref.get(state) }; +}); + +/** How many terminals have a command running, kept current from the metadata stream the UI uses. */ +export const runningTerminalCommands = Effect.fn("cloud.auto_update.runningTerminalCommands")( + function* (terminals: Pick) { + const running = yield* Ref.make(new Map()); + const terminalKey = (terminal: { readonly threadId: string; readonly terminalId: string }) => + `${terminal.threadId}\u0000${terminal.terminalId}`; + const unsubscribe = yield* terminals.subscribeMetadata((event) => + Ref.update(running, (current) => { + const next = event.type === "snapshot" ? new Map() : new Map(current); + if (event.type === "remove") next.delete(terminalKey(event)); + else { + for (const terminal of event.type === "snapshot" ? event.terminals : [event.terminal]) { + next.set(terminalKey(terminal), terminal.hasRunningSubprocess); + } + } + return next; + }), + ); + yield* Effect.addFinalizer(() => Effect.sync(unsubscribe)); + return Ref.get(running).pipe( + Effect.map((current) => [...current.values()].filter(Boolean).length), + ); + }, +); + +const make = Effect.gen(function* () { + const serverConfig = yield* ServerConfig.ServerConfig; + const launcher = yield* ServiceLauncherClient.ServiceLauncherClient; + const settings = yield* ServerSettings.ServerSettingsService; + const selfUpdate = yield* ServerSelfUpdate; + const policy = yield* ManagedPolicy.ManagedPolicy; + const runner = yield* ProcessRunner.ProcessRunner; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const terminals = yield* TerminalManager.TerminalManager; + const gate = yield* WorkAdmission; + const releaseContext = yield* Effect.context(); + const mirror = managedUpdateMirror(policy.document); + const channel = cliReleaseChannelOf(packageJson.version); + const statePath = path.join(serverConfig.baseDir, "runtime", SERVICE_STATE_FILE); + + const terminalCommands = yield* runningTerminalCommands(terminals); + + const threadWork = projections.getRecoveryThreadIds("runtime").pipe( + Effect.flatMap((threadIds) => + Effect.forEach( + threadIds, + (threadId) => + projections.getRuntimeRecoveryProjection(threadId).pipe(Effect.map(threadActivity)), + { concurrency: 2 }, + ), + ), + Effect.map((each) => each.reduce(addActivity, noActivity)), + ); + const activity = Effect.gen(function* () { + const threads = yield* threadWork; + return activityReason({ ...threads, terminalCommands: yield* terminalCommands }); + }).pipe(Effect.orElseSucceed(() => "thread state could not be read")); + + const askHook = (command: string, targetVersion: string) => { + if (command === "") return Effect.succeed(null); + if (!path.isAbsolute(command)) { + return Effect.succeed( + hookReason({ error: "automaticUpdatesQuietCommand must be an absolute path" }), + ); + } + return runner + .run({ + command, + args: [], + cwd: serverConfig.baseDir, + timeout: HOOK_TIMEOUT_MS, + timeoutBehavior: "timedOutResult", + maxOutputBytes: HOOK_MAX_OUTPUT_BYTES, + outputMode: "truncate", + env: { + T3CODE_AUTO_UPDATE_CURRENT_VERSION: packageJson.version, + T3CODE_AUTO_UPDATE_TARGET_VERSION: targetVersion, + }, + }) + .pipe( + Effect.map((result) => hookReason({ ...result, code: result.code ?? null })), + Effect.catch((error) => Effect.succeed(hookReason({ error: error._tag }))), + Effect.withSpan("cloud.auto_update.ask_hook"), + ); + }; + + const updater = yield* makeAutoUpdater({ + currentVersion: packageJson.version, + bootService: serverConfig.mode !== "desktop" && launcher.managed, + gate, + preferences: settings.getSettings, + newestRelease: (mirror === undefined + ? resolveNewestVersion(channel) + : resolveMirrorVersion(mirror, channel) + ).pipe(Effect.provideContext(releaseContext)), + launcherRecord: readLauncherRecord(fs, statePath), + activity, + askHook, + update: selfUpdate.update, + resetAfterDroppedHandoff: selfUpdate.resetAfterDroppedHandoff, + }); + + yield* forkParked(Effect.sleep(TICK).pipe(Effect.andThen(updater.tick), Effect.forever)); +}); + +export const layer = Layer.effectDiscard(make).pipe( + Layer.provide(Layer.merge(ProcessRunner.layer, ProjectionStore.layer)), +); diff --git a/apps/server/src/cloud/autoUpdateSettings.test.ts b/apps/server/src/cloud/autoUpdateSettings.test.ts new file mode 100644 index 0000000000000..149951821ac20 --- /dev/null +++ b/apps/server/src/cloud/autoUpdateSettings.test.ts @@ -0,0 +1,129 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ManagedPolicy from "../policy/ManagedPolicy.ts"; +import * as ServerSettingsModule from "../serverSettings.ts"; +import { renameRetiredAutoUpdateKeys } from "./autoUpdateSettings.ts"; + +// One settings service on a fresh temp base dir per test, as serverSettings.test.ts builds it. +const layerServerSettings = () => + ServerSettingsModule.layer.pipe( + Layer.provide(ServerSecretStore.layer), + Layer.provide(ManagedPolicy.layerTest()), + Layer.provideMerge(Layer.fresh(SqlitePersistence.layerMemory)), + Layer.provideMerge( + Layer.fresh( + ServerConfig.layerTest(process.cwd(), { prefix: "t3code-auto-update-settings-test-" }), + ), + ), + ); + +/** Loads a settings file with these contents (none when undefined) and returns what the server reads. */ +const loadSettingsFile = (contents: string | undefined) => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + if (contents !== undefined) yield* fs.writeFileString(config.settingsPath, contents); + const service = yield* ServerSettingsModule.ServerSettingsService; + return yield* service.getSettings; + }); + +it.layer(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))( + "automatic update settings saved under the old names", + (it) => { + it.effect("loads an old file's choice under the new names, then saves only those", () => + Effect.gen(function* () { + const settings = yield* loadSettingsFile( + JSON.stringify({ + autoUpdateServer: true, + autoUpdateQuietCommand: "/usr/local/bin/quiet", + enableAgentBrowserAccess: false, + }), + ); + assert.isTrue(settings.automaticUpdates); + assert.equal(settings.automaticUpdatesQuietCommand, "/usr/local/bin/quiet"); + assert.isFalse(settings.enableAgentBrowserAccess); + + // Any write saves the new names; the old keys are gone, so a later load reads the same. + const service = yield* ServerSettingsModule.ServerSettingsService; + yield* service.updateSettings({ enableAgentBrowserAccess: true }); + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const persisted = JSON.parse(yield* fs.readFileString(config.settingsPath)); + assert.isTrue(persisted.automaticUpdates); + assert.equal(persisted.automaticUpdatesQuietCommand, "/usr/local/bin/quiet"); + assert.notProperty(persisted, "autoUpdateServer"); + assert.notProperty(persisted, "autoUpdateQuietCommand"); + }).pipe(Effect.provide(layerServerSettings())), + ); + + it.effect("keeps a new name already in the file over the old one", () => + Effect.gen(function* () { + const settings = yield* loadSettingsFile( + JSON.stringify({ + autoUpdateServer: true, + automaticUpdates: false, + autoUpdateQuietCommand: "/old/quiet", + automaticUpdatesQuietCommand: "/new/quiet", + }), + ); + assert.isFalse(settings.automaticUpdates); + assert.equal(settings.automaticUpdatesQuietCommand, "/new/quiet"); + }).pipe(Effect.provide(layerServerSettings())), + ); + + it.effect("drops an old value of the wrong type without resetting the other settings", () => + Effect.gen(function* () { + const settings = yield* loadSettingsFile( + JSON.stringify({ + autoUpdateServer: "yes", + autoUpdateQuietCommand: 7, + enableAgentBrowserAccess: false, + }), + ); + assert.isFalse(settings.automaticUpdates); + assert.equal(settings.automaticUpdatesQuietCommand, ""); + assert.isFalse(settings.enableAgentBrowserAccess); + }).pipe(Effect.provide(layerServerSettings())), + ); + + it.effect("stays off with no settings file", () => + Effect.gen(function* () { + const settings = yield* loadSettingsFile(undefined); + assert.isFalse(settings.automaticUpdates); + assert.equal(settings.automaticUpdatesQuietCommand, ""); + }).pipe(Effect.provide(layerServerSettings())), + ); + + it.effect("stays off with an empty settings object", () => + Effect.gen(function* () { + const settings = yield* loadSettingsFile("{}"); + assert.isFalse(settings.automaticUpdates); + assert.equal(settings.automaticUpdatesQuietCommand, ""); + }).pipe(Effect.provide(layerServerSettings())), + ); + + it.effect("stays off when the file was cut off half-way through an old key", () => + Effect.gen(function* () { + const settings = yield* loadSettingsFile('{ "autoUpdateServer": tr'); + assert.isFalse(settings.automaticUpdates); + assert.equal(settings.automaticUpdatesQuietCommand, ""); + }).pipe(Effect.provide(layerServerSettings())), + ); + }, +); + +it("passes through what is not a settings object, and an object without old keys unchanged", () => { + for (const value of [null, 3, "text", [true]]) { + assert.equal(renameRetiredAutoUpdateKeys(value), value); + } + const current = { automaticUpdates: true }; + assert.strictEqual(renameRetiredAutoUpdateKeys(current), current); +}); diff --git a/apps/server/src/cloud/autoUpdateSettings.ts b/apps/server/src/cloud/autoUpdateSettings.ts new file mode 100644 index 0000000000000..819c85faefaec --- /dev/null +++ b/apps/server/src/cloud/autoUpdateSettings.ts @@ -0,0 +1,26 @@ +// Fork: settings saved before the automatic-update keys took upstream's names still say +// `autoUpdateServer` / `autoUpdateQuietCommand`. Copy them to the new names before decoding, so the +// choice loads without a schema migration; the next write saves only the new names. Delete this +// file and its call in serverSettings.ts when upstream's updater replaces ours. + +const RETIRED_KEYS = [ + { retired: "autoUpdateServer", current: "automaticUpdates", type: "boolean" }, + { retired: "autoUpdateQuietCommand", current: "automaticUpdatesQuietCommand", type: "string" }, +] as const; + +/** + * Copies each retired key's value to its current name. A current key already in the file wins. + * A retired value of the wrong type is not copied: it would fail the whole file's decode, and a + * failed decode resets every setting to its default. + */ +export function renameRetiredAutoUpdateKeys(parsed: unknown): unknown { + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) return parsed; + const stored = parsed as Readonly>; + let renamed: Record | undefined; + for (const { retired, current, type } of RETIRED_KEYS) { + const value = stored[retired]; + if (stored[current] !== undefined || typeof value !== type) continue; + renamed = { ...(renamed ?? stored), [current]: value }; + } + return renamed ?? parsed; +} diff --git a/apps/server/src/cloud/selfUpdate.ts b/apps/server/src/cloud/selfUpdate.ts index 88c9d2d46bd3a..b4228648dbc15 100644 --- a/apps/server/src/cloud/selfUpdate.ts +++ b/apps/server/src/cloud/selfUpdate.ts @@ -72,6 +72,9 @@ export class ServerSelfUpdate extends Context.Service< requestId: string, onHandoffAccepted?: () => Effect.Effect, ) => Effect.Effect; + /** The launcher accepted a handoff but this process is still running past it: allow the + * next update (the button's or an automatic one) to start. */ + readonly resetAfterDroppedHandoff: Effect.Effect; } >()("t3/cloud/selfUpdate/ServerSelfUpdate") {} @@ -349,6 +352,7 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () { return ServerSelfUpdate.of({ update, + resetAfterDroppedHandoff: Ref.set(inFlight, false), commitDesktopUpdate: Effect.fn("ServerSelfUpdate.commitDesktopUpdate")( function (requestId, onHandoffAccepted) { return desktopAppUpdate.commit(requestId, onHandoffAccepted); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index dfb47bd7b3744..c6e76ea36faeb 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -28,6 +28,7 @@ import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationServic import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ServerSettings from "../serverSettings.ts"; import { continueRestartedRun } from "./RestartContinuation.ts"; +import { WorkAdmission } from "../workAdmission.ts"; const isProviderTurnControlError = Schema.is(ProviderTurnControlService.ProviderTurnControlError); @@ -948,6 +949,9 @@ export const runDaemonWithOptions = (options: OrchestrationEffectDaemonOptions = Effect.scoped( Effect.gen(function* () { const worker = yield* OrchestrationEffectWorkerV2; + const admission = yield* WorkAdmission; + // Held effects stay pending, unclaimed: startup recovery decides what they become. + yield* admission.register("outbox"); const requestedConcurrency = options.concurrency ?? DEFAULT_EFFECT_WORKER_CONCURRENCY; const concurrency = Number.isFinite(requestedConcurrency) ? Math.max(1, Math.floor(requestedConcurrency)) @@ -962,7 +966,7 @@ export const runDaemonWithOptions = (options: OrchestrationEffectDaemonOptions = // Notifications wake promptly; availableAt persists retry timing, and liveness polling recovers missed wakeups. const runWorker = Effect.gen(function* () { while (true) { - const outcome = yield* worker.runOnce(claimLimits).pipe( + const outcome = yield* admission.admit(worker.runOnce(claimLimits)).pipe( Effect.map((worked) => (worked ? ("worked" as const) : ("idle" as const))), Effect.catchCause((cause) => Effect.logWarning("Orchestration effect worker failed", cause).pipe( diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 0197a10fae2e2..6cab991950659 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -1,4 +1,4 @@ -import { assert, it, vi } from "@effect/vitest"; +import { assert, describe, it, vi } from "@effect/vitest"; import { MessageId, NodeId, @@ -14,8 +14,10 @@ import { TurnItemId, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as TestClock from "effect/testing/TestClock"; import * as EffectWorker from "./EffectWorker.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; @@ -24,6 +26,15 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts"; import * as ServerSettings from "../serverSettings.ts"; +import { + makeWorkAdmissionGate, + memoryUpdateHoldStore, + UPDATE_HOLD_MAX_AGE_MS, + type UpdateHold, + UpdateHoldMarkerError, + type UpdateHoldStore, + WorkAdmission, +} from "../workAdmission.ts"; it.effect("expires orphaned runtime requests before command readiness", () => { const threadId = ThreadId.make("thread_recovery_requests"); @@ -1538,3 +1549,443 @@ it.effect("leaves delegated tasks to their own child threads after process loss" assert.deepEqual(noted, [`subagent ${nativeSubagentId}`]); }).pipe(Effect.provide(layer)); }); + +describe("work held for an automatic update", () => { + const MINUTE = 60_000; + const HELD_AT = 30 * MINUTE; + type Kind = + | "user message" + | "scheduled run" + | "t3_thread_send" + | "delegated completion" + | "worktree preparation"; + const messageFor = (kind: Kind, id: MessageId) => { + switch (kind) { + case "scheduled run": + return { id, createdBy: "user", creationSource: "web", scheduledTaskId: "task-1" }; + case "t3_thread_send": + return { id, createdBy: "agent", creationSource: "mcp" }; + case "delegated completion": + return { + id, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { childThreadId: "child-thread", status: "completed" }, + }; + default: + return { id, createdBy: "user", creationSource: "web" }; + } + }; + + interface ThreadInput { + readonly kind: Kind; + /** Request time of the unstarted run; "send now", a resumed queue and a worktree retry keep + * one from before the hold. Defaults to a minute after it. */ + readonly requestedAtMs?: number; + readonly queuedBehind?: boolean; + /** The user held this queue: the head run, or the run queued behind it. */ + readonly queueHeld?: "head" | "queued"; + } + + /** A thread with an unstarted run, and optionally a message queued behind it. */ + const threadFor = (input: ThreadInput, index: number) => { + const suffix = `${input.kind.replaceAll(" ", "_")}_${index}`; + const threadId = ThreadId.make(`thread_update_hold_${suffix}`); + const runId = RunId.make(`run_update_hold_${suffix}`); + const queuedRunId = RunId.make(`run_update_hold_queued_${suffix}`); + const attemptId = RunAttemptId.make(`attempt_update_hold_${suffix}`); + const rootNodeId = NodeId.make(`node_update_hold_${suffix}`); + const messageId = MessageId.make(`message_update_hold_${suffix}`); + const queuedMessageId = MessageId.make(`message_update_hold_queued_${suffix}`); + const preparing = input.kind === "worktree preparation"; + const requestedAtMs = input.requestedAtMs ?? HELD_AT + MINUTE; + const projection = { + thread: { id: threadId }, + runtimeRequests: [], + providerSessions: [], + providerThreads: [], + providerTurns: [], + runs: [ + { + id: runId, + status: preparing ? "preparing" : "starting", + startedAt: null, + requestedAt: DateTime.makeUnsafe(requestedAtMs), + activeAttemptId: attemptId, + userMessageId: messageId, + queuePosition: null, + providerInstanceId: ProviderInstanceId.make("codex"), + ...(input.queueHeld === "head" ? { queueHeld: true } : {}), + ...(preparing + ? { + workspacePreparation: { type: "worktree", baseRef: "main", startFromOrigin: false }, + } + : {}), + }, + ...(input.queuedBehind + ? [ + { + id: queuedRunId, + status: "queued", + startedAt: null, + requestedAt: DateTime.makeUnsafe(requestedAtMs + MINUTE), + activeAttemptId: null, + userMessageId: queuedMessageId, + queuePosition: 1, + providerInstanceId: ProviderInstanceId.make("codex"), + ...(input.queueHeld === "queued" ? { queueHeld: true } : {}), + }, + ] + : []), + ], + attempts: [{ id: attemptId, runId, rootNodeId, status: "pending" }], + nodes: [{ id: rootNodeId, runId, status: "pending" }], + subagents: [], + messages: [ + messageFor(input.kind, messageId), + ...(input.queuedBehind ? [messageFor("t3_thread_send", queuedMessageId)] : []), + ], + turnItems: [], + } as unknown as OrchestrationV2ThreadProjection; + return { threadId, runId, queuedRunId, projection }; + }; + + type MarkerState = "none" | "fresh" | "stale" | "invalid" | Partial; + + /** + * The database a held update leaves behind, and processes that start on it. Every process sees + * the same database: what a failed update's rollback restores, or what a crash leaves. + */ + const heldWorld = Effect.fn("test.heldWorld")(function* ( + threads: ReadonlyArray, + options: { readonly marker?: MarkerState } = {}, + ) { + const built = threads.map(threadFor); + const projections = new Map(built.map((thread) => [thread.threadId, thread.projection])); + const commits: Array[0]> = []; + const starts: Array = []; + const prepared: Array = []; + let failStartWrites = false; + yield* TestClock.setTime(HELD_AT + 10 * MINUTE); + const memory = yield* memoryUpdateHoldStore; + let failStartedRecordsAfter = Number.POSITIVE_INFINITY; + let startedRecords = 0; + // The marker file; a write that adds started runs can be made to fail. + const marker: UpdateHoldStore = { + read: memory.read, + clear: memory.clear, + write: (hold) => + Effect.gen(function* () { + const before = yield* memory.current; + const addsStarted = + (hold.started?.length ?? 0) > + (before === undefined || before === "invalid" ? 0 : (before.started?.length ?? 0)); + if (addsStarted && startedRecords++ >= failStartedRecordsAfter) { + return yield* new UpdateHoldMarkerError({ reason: "disk full" }); + } + yield* memory.write(hold); + }), + }; + const fresh = { + heldAt: DateTime.formatIso(DateTime.makeUnsafe(HELD_AT)), + targetVersion: "next", + }; + const state = options.marker ?? "fresh"; + if (state === "fresh") yield* memory.write(fresh); + else if (state === "stale") { + yield* memory.write({ + ...fresh, + heldAt: DateTime.formatIso(DateTime.makeUnsafe(HELD_AT - UPDATE_HOLD_MAX_AGE_MS)), + }); + } else if (state === "invalid") yield* memory.set("invalid"); + else if (state !== "none") yield* memory.write({ ...fresh, ...state }); + + /** One server process: startup recovery, which keeps held work but starts none of it. */ + const startProcess = Effect.fn("test.startProcess")(function* () { + const gate = yield* makeWorkAdmissionGate({ marker }); + yield* gate.setPreparationResumer((preparation) => + Effect.sync(() => { + prepared.push(preparation.runId); + }), + ); + const layer = ProviderRuntimeRecovery.layer.pipe( + Layer.provide(ServerSettings.layerTest()), + Layer.provide( + Layer.mergeAll( + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getRecoveryThreadIds: () => Effect.succeed([...projections.keys()]), + getRuntimeRecoveryProjection: (threadId) => + Effect.succeed(projections.get(threadId)!), + getMessagesById: (threadId, ids) => + Effect.succeed( + new Map( + projections + .get(threadId)! + .messages.filter((message) => ids.includes(message.id)) + .map((message) => [message.id, message]), + ), + ), + getThreadRecords: (threadId) => Effect.succeed(projections.get(threadId)!), + }), + Layer.mock(EventSink.EventSinkV2)({ + commitCommand: (command) => { + commits.push(command); + return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never); + }, + writeWithEffects: (write) => { + if (failStartWrites) return Effect.die("the database went away"); + for (const effect of write.effects) { + if (effect.request.type === "provider-turn.start") { + starts.push(effect.request.runId); + } + } + return Effect.succeed({} as never); + }, + }), + IdAllocator.layer, + Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({ + runRecoveryOnce: Effect.succeed(false), + }), + Layer.mock(EffectOutbox.EffectOutboxV2)({ + reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }), + cancelUnsettled: () => Effect.succeed([]), + signalCancellations: () => Effect.void, + }), + ), + ), + ); + const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService.pipe( + Effect.provide(layer), + Effect.provideService(WorkAdmission, gate), + ); + const recover = recovery.reconcile("startup"); + return { gate, recover, summary: yield* recover }; + }); + + return { + threads: built, + startProcess, + marker: memory.current, + runUpdates: () => + commits.flatMap((command) => + command.events.flatMap((event) => (event.type === "run.updated" ? [event.payload] : [])), + ), + starts: () => [...starts], + prepared: () => [...prepared], + failStartWrites: (fail: boolean) => { + failStartWrites = fail; + }, + failStartedRecordsAfter: (count: number) => { + failStartedRecordsAfter = count; + startedRecords = 0; + }, + }; + }); + + it.effect.each([ + "user message", + "scheduled run", + "t3_thread_send", + "delegated completion", + ] as const)("starts a held %s once the update is committed, not before", (kind) => + Effect.gen(function* () { + const world = yield* heldWorld([{ kind, queuedBehind: true }]); + const [thread] = world.threads; + const process = yield* world.startProcess(); + assert.equal(process.summary.terminalizedRuns, 0); + // Neither cancelled nor put in the queue the user has to release. + assert.deepEqual( + world + .runUpdates() + .filter((update) => update.status === "cancelled" || update.queueHeld === true), + [], + ); + // The launcher has not committed the update yet: a rollback must find nothing started. + assert.deepEqual(world.starts(), []); + assert.deepEqual(yield* world.marker, { + heldAt: DateTime.formatIso(DateTime.makeUnsafe(HELD_AT)), + targetVersion: "next", + runs: [thread!.runId, thread!.queuedRunId], + }); + assert.isFalse(yield* process.gate.heldWorkSettled); + yield* process.gate.startHeldWork; + // A fresh start for the held run; the message queued behind it follows it as usual. + assert.deepEqual(world.starts(), [thread!.runId]); + assert.deepEqual(world.prepared(), []); + assert.isUndefined(yield* world.marker); + assert.isTrue(yield* process.gate.heldWorkSettled); + }), + ); + + it.effect("prepares a held worktree again once the update is committed", () => + Effect.gen(function* () { + const world = yield* heldWorld([{ kind: "worktree preparation" }]); + const process = yield* world.startProcess(); + assert.equal(process.summary.terminalizedRuns, 0); + assert.deepEqual(world.prepared(), []); + yield* process.gate.startHeldWork; + assert.deepEqual(world.starts(), []); + assert.deepEqual(world.prepared(), [world.threads[0]!.runId]); + }), + ); + + it.effect.each(["user message", "worktree preparation"] as const)( + "starts a held %s whatever its request time: send now, a resumed queue or a retry keep an old one", + (kind) => + Effect.gen(function* () { + // Saved after the hold (anything saved before it stopped the update), stamped before it. + const world = yield* heldWorld([{ kind, requestedAtMs: HELD_AT - 10 * MINUTE }]); + const process = yield* world.startProcess(); + assert.equal(process.summary.terminalizedRuns, 0); + yield* process.gate.startHeldWork; + assert.deepEqual([...world.starts(), ...world.prepared()], [world.threads[0]!.runId]); + }), + ); + + it.effect("leaves a queue the user held alone", () => + Effect.gen(function* () { + const head = yield* heldWorld([{ kind: "user message", queueHeld: "head" }]); + yield* (yield* head.startProcess()).gate.startHeldWork; + assert.deepEqual(head.starts(), []); + const queued = yield* heldWorld([ + { kind: "user message", queuedBehind: true, queueHeld: "queued" }, + ]); + const process = yield* queued.startProcess(); + // The held head starts; the queue the user held behind it is not part of the update's work. + assert.deepEqual((yield* queued.marker) as UpdateHold, { + heldAt: DateTime.formatIso(DateTime.makeUnsafe(HELD_AT)), + targetVersion: "next", + runs: [queued.threads[0]!.runId], + }); + yield* process.gate.startHeldWork; + assert.deepEqual(queued.starts(), [queued.threads[0]!.runId]); + assert.isFalse( + queued + .runUpdates() + .some((update) => update.id === queued.threads[0]!.queuedRunId && !update.queueHeld), + ); + }), + ); + + it.effect.each([ + { name: "a failed update's rollback", second: "old version" }, + { name: "a crash between the commit and the start", second: "same version" }, + ])("starts held work exactly once after $name", () => + Effect.gen(function* () { + const world = yield* heldWorld([ + { kind: "user message", queuedBehind: true }, + { kind: "worktree preparation" }, + ]); + // The trial (or the committed process) keeps the work and stops before starting any. + yield* world.startProcess(); + assert.deepEqual([...world.starts(), ...world.prepared()], []); + // The next process finds the same database and the marker's list. + const next = yield* world.startProcess(); + yield* next.gate.startHeldWork; + assert.deepEqual(world.starts(), [world.threads[0]!.runId]); + assert.deepEqual(world.prepared(), [world.threads[1]!.runId]); + assert.isUndefined(yield* world.marker); + // A later startup on that database starts nothing again. + const later = yield* world.startProcess(); + yield* later.gate.startHeldWork; + assert.equal(world.starts().length + world.prepared().length, 2); + }), + ); + + it.effect("starts held work once when recovery and the start run twice in one process", () => + Effect.gen(function* () { + const world = yield* heldWorld([{ kind: "user message", queuedBehind: true }]); + const process = yield* world.startProcess(); + yield* process.recover; + yield* process.gate.startHeldWork; + yield* process.gate.startHeldWork; + assert.deepEqual(world.starts(), [world.threads[0]!.runId]); + }), + ); + + it.effect("keeps the marker when starting fails part-way, and a retry starts only the rest", () => + Effect.gen(function* () { + const world = yield* heldWorld([ + { kind: "user message" }, + { kind: "scheduled run" }, + { kind: "t3_thread_send" }, + ]); + const [first, second, third] = world.threads; + const process = yield* world.startProcess(); + world.failStartedRecordsAfter(1); + yield* process.gate.startHeldWork; + assert.deepEqual(world.starts(), [first!.runId]); + assert.isFalse(yield* process.gate.heldWorkSettled); + assert.deepInclude(yield* world.marker, { started: [first!.runId] }); + // The same process tries again. + world.failStartedRecordsAfter(1); + yield* process.gate.startHeldWork; + assert.deepEqual(world.starts(), [first!.runId, second!.runId]); + // A restart finishes the job. + world.failStartedRecordsAfter(Number.POSITIVE_INFINITY); + const next = yield* world.startProcess(); + yield* next.gate.startHeldWork; + assert.deepEqual(world.starts(), [first!.runId, second!.runId, third!.runId]); + assert.isUndefined(yield* world.marker); + assert.isTrue(yield* next.gate.heldWorkSettled); + }), + ); + + it.effect("never starts a run twice when the process stops while starting it", () => + Effect.gen(function* () { + const world = yield* heldWorld([{ kind: "user message" }]); + const process = yield* world.startProcess(); + // Its start is recorded, then the write is lost with the process. + world.failStartWrites(true); + yield* process.gate.startHeldWork; + assert.deepEqual(world.starts(), []); + world.failStartWrites(false); + const next = yield* world.startProcess(); + yield* next.gate.startHeldWork; + // At most once: it takes the upstream path, the queue the user can release. + assert.deepEqual(world.starts(), []); + assert.isTrue( + world + .runUpdates() + .some((update) => update.id === world.threads[0]!.runId && update.queueHeld === true), + ); + assert.isUndefined(yield* world.marker); + }), + ); + + it.effect("only starts the runs the first startup kept", () => + Effect.gen(function* () { + // The first startup found nothing held; this run was saved after it (and lost by a crash). + const world = yield* heldWorld([{ kind: "user message" }], { marker: { runs: [] } }); + const process = yield* world.startProcess(); + yield* process.gate.startHeldWork; + assert.deepEqual(world.starts(), []); + assert.isTrue(world.runUpdates().some((update) => update.queueHeld === true)); + assert.isUndefined(yield* world.marker); + }), + ); + + it.effect.each(["none", "stale", "invalid"] as const)( + "keeps the upstream restart behaviour with marker %s", + (marker) => + Effect.gen(function* () { + const world = yield* heldWorld( + [{ kind: "user message" }, { kind: "t3_thread_send" }, { kind: "worktree preparation" }], + { marker }, + ); + const process = yield* world.startProcess(); + yield* process.gate.startHeldWork; + assert.deepEqual(world.starts(), []); + assert.deepEqual(world.prepared(), []); + // The user message waits in the held queue; the agent message and preparation stop. + assert.isTrue( + world + .runUpdates() + .some((update) => update.id === world.threads[0]!.runId && update.queueHeld === true), + ); + assert.equal(process.summary.terminalizedRuns, 2); + assert.isUndefined(yield* world.marker); + assert.isTrue(yield* process.gate.heldWorkSettled); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 46b739bec5732..ec3c21b262672 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -3,6 +3,7 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { CommandId, isProviderNativeSubagentThread, + type RunId, type OrchestrationV2DomainEvent, type ProviderThreadId, type OrchestrationV2RestartCancelledBackgroundWork, @@ -16,6 +17,7 @@ import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as EffectOutbox from "./EffectOutbox.ts"; @@ -25,6 +27,7 @@ import * as ProjectionStore from "./ProjectionStore.ts"; import * as ServerSettings from "../serverSettings.ts"; import { restartContinuationRun } from "./RestartContinuation.ts"; import { isBufferedProviderNotification } from "./NotificationMailbox.ts"; +import { WorkAdmission, type HeldForUpdate, type HeldPreparation } from "../workAdmission.ts"; import { queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts"; import { endOrphanedNativeSubagent, @@ -188,6 +191,8 @@ export const make = Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const ids = yield* IdAllocator.IdAllocatorV2; const outbox = yield* EffectOutbox.EffectOutboxV2; + const admission = yield* WorkAdmission; + yield* admission.register("recovery"); type RecoverySession = ProjectionStore.ProjectionRuntimeRecoveryState["providerSessions"][number]; /** Shared sessions stop on their first recovered thread; later threads still decide wakes from pre-recovery liveness. */ const sessionsLiveAtStart = ( @@ -211,6 +216,7 @@ export const make = Effect.gen(function* () { trigger: "startup" | "shutdown", continueAfterRestart: boolean, stoppedThisPass: ReadonlyMap = new Map(), + heldForUpdate: HeldForUpdate | undefined = undefined, ) { const now = yield* DateTime.now; let runs = [] as Array; @@ -364,6 +370,30 @@ export const make = Effect.gen(function* () { const bufferedMessageIds = new Set( queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id), ); + // The gate closed before the update's last quiet check, so every unclaimed start, + // preparation or queued run left at the restart was saved after the hold, whatever its + // request time ("send now", resume and retry keep their old one). The marker narrows this + // to the runs the first startup kept, minus those a startup already started. + const heldSinceUpdate = (run: OrchestrationV2ThreadProjection["runs"][number]) => + heldForUpdate !== undefined && + run.queueHeld !== true && + !bufferedMessageIds.has(run.userMessageId) && + !heldForUpdate.started.has(run.id) && + (heldForUpdate.runs === undefined || heldForUpdate.runs.has(run.id)); + const heldStartsAndPreparations = [ + ...unstartedRuns, + ...runs.filter((run) => { + const status: string = run.status; + return status === "preparing"; + }), + ].filter(heldSinceUpdate); + // Queued runs start behind the held run they wait for. With none on this thread nothing + // would start them, so they take the upstream path (the visible held queue) instead. + const heldForUpdateRuns = + heldStartsAndPreparations.length === 0 + ? [] + : [...heldStartsAndPreparations, ...queuedRuns.filter(heldSinceUpdate)]; + const heldForUpdateIds = new Set(heldForUpdateRuns.map((run) => run.id)); const retryableMessageIds = new Set( queuedMessages .filter( @@ -376,8 +406,8 @@ export const make = Effect.gen(function* () { .map((message) => message.id), ); // A saved user message the provider never accepted stays available through the held queue. - const retryableRuns = unstartedRuns.filter((run) => - retryableMessageIds.has(run.userMessageId), + const retryableRuns = unstartedRuns.filter( + (run) => !heldForUpdateIds.has(run.id) && retryableMessageIds.has(run.userMessageId), ); const retryableRunIds = new Set(retryableRuns.map((run) => run.id)); const retryableItems = @@ -416,10 +446,13 @@ export const make = Effect.gen(function* () { }, }); } - runs = runs.filter((run) => !retryableRunIds.has(run.id)); + runs = runs.filter((run) => !retryableRunIds.has(run.id) && !heldForUpdateIds.has(run.id)); const heldRuns = [ ...retryableRuns, - ...queuedRunsInDeliveryOrder({ runs: queuedRuns, messages: queuedMessages }), + ...queuedRunsInDeliveryOrder({ + runs: queuedRuns.filter((run) => !heldForUpdateIds.has(run.id)), + messages: queuedMessages, + }), ]; for (const [index, run] of heldRuns.entries()) { if (bufferedMessageIds.has(run.userMessageId)) { @@ -905,14 +938,88 @@ export const make = Effect.gen(function* () { closedRequests: requests.length, retiredEffects, cancelledEffectIds, + heldForUpdate: heldForUpdateRuns.map((run) => ({ + runId: run.id, + resume: + run.status === "queued" + ? ("behind-held-run" as const) + : (run.status as string) === "preparing" + ? ("prepare" as const) + : ("start" as const), + })), }; }, ); + /** Work the last startup kept for an automatic update, one entry per thread. */ + interface HeldThreadWork { + readonly threadId: ThreadId; + readonly runIds: ReadonlyArray; + readonly starts: ReadonlyArray; + readonly preparations: ReadonlyArray; + } + const heldWork = yield* Ref.make | undefined>(undefined); + + /** Starts the kept work once the update is committed: new start effects (process loss cancelled + * the old ones) and held preparations again. Runs are saved as started before their start is + * written, so nothing starts twice; a failure keeps the marker so the rest starts later. */ + const startHeldWork = Effect.gen(function* () { + const kept = yield* Ref.get(heldWork); + // Startup recovery did not finish: whatever it kept is decided by the next startup. + if (kept === undefined) return; + let starts = 0; + let preparations = 0; + for (const thread of kept) { + yield* admission.recordHeldRunsStarted(thread.runIds); + yield* Ref.update(heldWork, (current) => current?.filter((entry) => entry !== thread)); + const now = DateTime.formatIso(yield* DateTime.now); + for (const runId of thread.starts) { + const commandId = CommandId.make(`command:update-hold:start:${runId}:${now}`); + yield* eventSink.writeWithEffects({ + commandId, + events: [], + effects: [ + { + id: `effect:${commandId}:provider-turn.start:${runId}`, + commandId, + threadId: thread.threadId, + request: { type: "provider-turn.start", runId }, + }, + ], + }); + starts += 1; + } + for (const preparation of thread.preparations) { + yield* admission.resumePreparation(preparation); + preparations += 1; + } + } + yield* admission.finishHeldWork; + if (starts > 0 || preparations > 0) { + yield* Effect.logInfo("Started work that waited for the automatic update", { + starts, + worktreePreparations: preparations, + }); + } + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logError( + "Work that waited for the automatic update did not all start; the rest starts on the next startup.", + { cause }, + ), + ), + Effect.withSpan("ProviderRuntimeRecoveryService.startHeldWork"), + ); + yield* admission.setHeldWorkStarter(startHeldWork); + const reconcile = Effect.fn("ProviderRuntimeRecoveryService.reconcile")(function* ( trigger: "startup" | "shutdown", ) { const continueAfterRestart = yield* settings.getSettings.pipe(Effect.orElseSucceed(() => null)); + const heldForUpdate = yield* admission.heldForUpdate; + const keptWork: Array = []; const threadIds = yield* projections .getRecoveryThreadIds("runtime") .pipe( @@ -946,7 +1053,33 @@ export const make = Effect.gen(function* () { continueAfterRestart !== null && resolveProjectSettings(continueAfterRestart, projection.thread.projectId).settings .continueThreadsAfterServerUpdate; - const result = yield* reconcileProjection(projection, trigger, enabled, stoppedThisPass); + const result = yield* reconcileProjection( + projection, + trigger, + enabled, + stoppedThisPass, + heldForUpdate, + ); + if (result.heldForUpdate.length > 0) { + keptWork.push({ + threadId, + runIds: result.heldForUpdate.map((held) => held.runId), + starts: result.heldForUpdate.flatMap((held) => + held.resume === "start" ? [held.runId] : [], + ), + preparations: result.heldForUpdate.flatMap((held) => + held.resume === "prepare" + ? [ + { + commandId: CommandId.make(`command:update-hold:prepare:${held.runId}`), + threadId, + runId: held.runId, + }, + ] + : [], + ), + }); + } for (const session of projection.providerSessions) { if (session.status !== "stopped" && session.status !== "error") { stoppedThisPass.set(session.id, session); @@ -974,6 +1107,22 @@ export const make = Effect.gen(function* () { (cause) => new ProviderRuntimeRecoveryError({ operation: "drain-outbox", cause }), ), ); + if (trigger === "startup") { + // Saved before the update commits: a failed update restores the database this startup + // read, so the version that runs next finds the same runs, and later startups after a + // commit consider only these. + if (heldForUpdate !== undefined) { + yield* admission.recordHeldRuns(keptWork.flatMap((thread) => thread.runIds)).pipe( + Effect.catch((error) => + Effect.logWarning("Could not save which runs the update held.", { + reason: error.reason, + }), + ), + ); + } + // Started by startup once the update is committed (see `startHeldWork`). + yield* Ref.set(heldWork, keptWork); + } return { terminalizedRuns, stoppedSessions, diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index a0a438bf97301..e7c3270eaf3d8 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -37,6 +37,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; +import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; @@ -61,6 +62,7 @@ import * as EffectOutbox from "./EffectOutbox.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ThreadLaunch from "./ThreadLaunchService.ts"; +import { makeWorkAdmissionGate, WorkAdmission } from "../workAdmission.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts"; @@ -289,6 +291,8 @@ function makeHarness(options: HarnessOptions = {}) { layerOutbox, layerDatabase, layerExternalServices, + // The held-preparation tests build a second launch service on the same event sink. + layerOrchestrator, ), createWorktree, removeWorktree, @@ -999,6 +1003,88 @@ it.effect("records the checkout's branch on a root launch that names none", () = }).pipe(Effect.provide(harness.layer)); }); +it.effect("waits at a closed work gate to prepare a worktree, then prepares it", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const gate = yield* makeWorkAdmissionGate(); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + // The launch service says it prepares through this gate. + assert.notInclude(yield* gate.missingConsumers, "thread-launch"); + yield* gate.hold; + const launched = yield* launches.launch( + launchInput({ + command: "command:launch:held-preparation", + thread: "thread:launch:held-preparation", + message: "Build it after the update", + workspace: worktreeStrategy, + }), + ); + for (let i = 0; i < 200; i++) yield* Effect.yieldNow; + // Held, not failed: nothing prepared, the run still waits to be prepared. + assert.equal(harness.createWorktree.mock.calls.length, 0); + const held = yield* threads.getThreadProjection(launched.threadId); + assert.equal(held.runs[0]?.status, "preparing"); + yield* gate.release; + const prepared = yield* preparationSettled(launched.threadId); + assert.equal(harness.createWorktree.mock.calls.length, 1); + assert.equal(prepared.runs[0]?.status, "starting"); + }).pipe(Effect.provide(harness.layer), Effect.provideService(WorkAdmission, gate)); + }); +}); + +it.effect("keeps a preparation held when the server stops, and the next server prepares it", () => { + const harness = makeHarness(); + // One thread-launch service per server process, both on the same database. + const launchLayer = Layer.fresh( + ThreadLaunch.layer.pipe( + Layer.provide(Layer.mergeAll(CommandReceiptStore.layer, IdAllocator.layer)), + ), + ); + return Effect.gen(function* () { + const threads = yield* ThreadManagement.ThreadManagementService; + const stopping = yield* makeWorkAdmissionGate(); + const firstProcess = yield* Scope.make(); + const first = yield* Layer.buildWithScope(launchLayer, firstProcess).pipe( + Effect.provideService(WorkAdmission, stopping), + ); + yield* stopping.hold; + const launched = yield* Effect.gen(function* () { + return yield* (yield* ThreadLaunch.ThreadLaunchService).launch( + launchInput({ + command: "command:launch:held-at-stop", + thread: "thread:launch:held-at-stop", + message: "Build it after the update", + workspace: worktreeStrategy, + }), + ); + }).pipe(Effect.provide(first)); + for (let i = 0; i < 200; i++) yield* Effect.yieldNow; + // The server stops while the preparation waits at the gate: held, not failed. + yield* Scope.close(firstProcess, Exit.void); + for (let i = 0; i < 200; i++) yield* Effect.yieldNow; + const held = yield* threads.getThreadProjection(launched.threadId); + assert.equal(held.runs[0]?.status, "preparing"); + assert.equal(harness.createWorktree.mock.calls.length, 0); + // The next server's recovery starts it through its own gate. + const restarted = yield* makeWorkAdmissionGate(); + const secondProcess = yield* Scope.make(); + yield* Layer.buildWithScope(launchLayer, secondProcess).pipe( + Effect.provideService(WorkAdmission, restarted), + ); + yield* restarted.resumePreparation({ + commandId: CommandId.make(`command:update-hold:prepare:${held.runs[0]!.id}`), + threadId: launched.threadId, + runId: held.runs[0]!.id, + }); + const prepared = yield* preparationSettled(launched.threadId); + assert.equal(harness.createWorktree.mock.calls.length, 1); + assert.equal(prepared.runs[0]?.status, "starting"); + yield* Scope.close(secondProcess, Exit.void); + }).pipe(Effect.provide(harness.layer)); +}); + it.effect("launches a worktree request for Scratch at the root, in its own folder", () => { const harness = scratchHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index c65754dece315..1874469b9ecaa 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -55,6 +55,7 @@ import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid"; import { DispatchModeLimit } from "./DispatchModeLimit.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; +import { WorkAdmission } from "../workAdmission.ts"; export type ThreadLaunchWorkspaceStrategy = | { readonly type: "root"; readonly branch?: string | undefined } @@ -245,6 +246,7 @@ const make = Effect.gen(function* () { const setupScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; const providerRegistry = yield* ProviderRegistry.ProviderRegistry; const serverSettings = yield* ServerSettings.ServerSettingsService; + const admission = yield* WorkAdmission; const textGeneration = yield* TextGeneration.TextGeneration; const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; const ids = yield* IdAllocator.IdAllocatorV2; @@ -942,14 +944,20 @@ const make = Effect.gen(function* () { ); yield* Effect.raceFirst(prepare, interruption); }); - yield* preparation.pipe( + // A preparation still waiting at a held gate when the server stops is not a failure: the + // run stays preparing and startup recovery starts it again (WorkAdmission). + const admitted = yield* Ref.make(false); + yield* admission.admit(Ref.set(admitted, true).pipe(Effect.andThen(preparation))).pipe( Effect.onError((cause) => - failPreparedRun( - input, - threadId, - runId, - Cause.hasInterruptsOnly(cause) ? "Worktree setup cancelled." : Cause.squash(cause), - ), + Effect.gen(function* () { + if (Cause.hasInterruptsOnly(cause) && !(yield* Ref.get(admitted))) return; + yield* failPreparedRun( + input, + threadId, + runId, + Cause.hasInterruptsOnly(cause) ? "Worktree setup cancelled." : Cause.squash(cause), + ); + }), ), Effect.ignoreCause, Effect.ensuring(releasePreparation(input.commandId)), @@ -1289,6 +1297,8 @@ const make = Effect.gen(function* () { ); }); + yield* admission.register("thread-launch"); + yield* admission.setPreparationResumer(prepareRun); return ThreadLaunchService.of({ launch, checkWorktreeBase, retryPreparation, prepareRun }); }); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 954f72e2b5e95..f3d0ef41374f4 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -41,9 +41,9 @@ import * as Effect from "effect/Effect"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; -import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; @@ -81,6 +81,7 @@ import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import * as ProjectStore from "./ProjectStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; +import * as WorkAdmission from "../workAdmission.ts"; import { shellStreamItemFromThreadShell } from "./ShellStream.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; @@ -10685,3 +10686,267 @@ it.layer(layerQueueResumeDisabledInstanceTest)( ); }, ); + +it.effect.each([ + { name: "after an automatic update", marker: true }, + { name: "after an ordinary restart", marker: false }, +])("work that waited at the closed gate, then restart: $name", ({ marker }) => + Effect.gen(function* () { + const store = yield* WorkAdmission.memoryUpdateHoldStore; + const gate = yield* WorkAdmission.makeWorkAdmissionGate({ marker: store }); + yield* Effect.gen(function* () { + // Recovery reads the server's gate (the outbox and thread launches register in their own + // tests). + assert.notInclude(yield* gate.missingConsumers, "recovery"); + const orchestrator = yield* Orchestrator.OrchestratorV2; + const recovery = yield* ProviderRuntimeRecoveryService.ProviderRuntimeRecoveryService; + const outbox = yield* EffectOutbox.EffectOutboxV2; + yield* TestClock.setTime(60 * 60_000); + // The final quiet check passed: the gate closes and the marker is written. + yield* gate.hold; + if (marker) { + yield* gate.markHeldForUpdate({ + heldAt: DateTime.formatIso(yield* DateTime.now), + targetVersion: "next", + }); + } + yield* TestClock.adjust("1 seconds"); + const arrive = (name: string, createdBy: "user" | "agent", creationSource: "web" | "mcp") => + Effect.gen(function* () { + const threadId = ThreadId.make(`update-hold-${name}-${marker}`); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: ProjectId.make(`${threadId}:project`), + title: "Arrives during the update window", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy, + creationSource, + commandId: CommandId.make(`${threadId}:message`), + threadId, + messageId: MessageId.make(`${threadId}:message`), + text: `Sent by ${name} while the server waits for its update`, + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + return threadId; + }); + const threads = [ + yield* arrive("user", "user", "web"), + // What t3_thread_send dispatches. + yield* arrive("agent", "agent", "mcp"), + ]; + // The server stops (SIGTERM) and the new one starts: the gate is open again, the marker + // and the database are what the old process left. + yield* TestClock.adjust("30 seconds"); + yield* recovery.reconcile("startup"); + // Startup starts held work once the launcher committed the update. + yield* gate.startHeldWork; + for (const threadId of threads) { + const projection = yield* orchestrator.getThreadProjection(threadId); + const run = projection.runs[0]!; + const restartedAt = DateTime.formatIso(yield* DateTime.now); + const effects = [ + ...(yield* outbox.listByCommandId(CommandId.make(`${threadId}:message`))), + ...(yield* outbox.listByCommandId( + CommandId.make(`command:update-hold:start:${run.id}:${restartedAt}`), + )), + ]; + const pendingStarts = effects.filter( + (effect) => + effect.status === "pending" && + effect.request.type === "provider-turn.start" && + effect.request.runId === run.id, + ); + if (marker) { + // Neither cancelled nor left in the queue the user has to release: it starts. + assert.equal(run.status, "starting", threadId); + assert.notEqual(run.queueHeld, true, threadId); + assert.lengthOf(pendingStarts, 1, threadId); + } else { + // Upstream: a user message waits in the held queue, an agent's message is cancelled. + assert.isTrue( + threadId.includes("-user-") ? run.queueHeld === true : run.status === "cancelled", + `${threadId}: ${run.status}`, + ); + assert.lengthOf(pendingStarts, 0, threadId); + } + } + }).pipe( + Effect.provide(Layer.fresh(layerTest)), + Effect.provideService(WorkAdmission.WorkAdmission, gate), + ); + }), +); + +// The launcher stops the old version, keeps a copy of its database and starts the new version +// on trial. A trial that fails before the commit gets the copy back and the old version runs +// again, so work held for the update must reach a provider exactly once over both processes. +const layerTestAtPath = (dbPath: string) => + Layer.mergeAll( + RuntimeLayer.layer, + layerProjectionReplayTest, + RuntimeLayer.layerEventSink, + ProjectStore.layer, + ProjectionStore.layer, + EffectOutbox.layer, + ThreadCommandExecutor.layer, + ).pipe( + Layer.provide(McpSessionRegistryTestkit.layer), + Layer.provide(McpProviderSessions.layer), + Layer.provideMerge( + SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer)), + ), + Layer.provide(layerCheckpointStoreTest), + Layer.provide(layerServerConfig), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerTestProviderInstanceRegistry), + Layer.provide(layerGitWorkflowTest), + Layer.provide(layerProjectServiceTest), + Layer.provide(layerPlatformTest), + ); + +it.effect.each([ + { name: "the trial commits", trial: "commits" }, + { name: "the trial fails and the launcher restores the database", trial: "rolls back" }, +] as const)("held work starts exactly once when $name", ({ trial }) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "update-hold-" }); + const dbPath = path.join(directory, "state.sqlite"); + const backup = path.join(directory, "backup"); + yield* fs.makeDirectory(backup); + const databaseFiles = ["state.sqlite", "state.sqlite-wal", "state.sqlite-shm"]; + const copyDatabase = (from: string, to: string) => + Effect.forEach(databaseFiles, (file) => + Effect.gen(function* () { + yield* fs.remove(path.join(to, file), { force: true }); + if (yield* fs.exists(path.join(from, file))) { + yield* fs.copyFile(path.join(from, file), path.join(to, file)); + } + }), + ); + // The marker lives outside the database: the launcher does not restore it. + const marker = yield* WorkAdmission.memoryUpdateHoldStore; + const runProcess = ( + body: (gate: WorkAdmission.WorkAdmissionGate) => Effect.Effect, + ) => + Effect.gen(function* () { + const gate = yield* WorkAdmission.makeWorkAdmissionGate({ marker }); + return yield* body(gate).pipe( + Effect.provide(Layer.fresh(layerTestAtPath(dbPath))), + Effect.provideService(WorkAdmission.WorkAdmission, gate), + ); + }); + // Every start written for a held run, whatever happened to it afterwards. + const heldStarts = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ readonly payloadJson: string }>` + SELECT payload_json AS "payloadJson" FROM orchestration_v2_effect_outbox + WHERE effect_type = 'provider-turn.start' + AND command_id LIKE 'command:update-hold:start:%'`; + return rows.map((row) => (JSON.parse(row.payloadJson) as { runId: string }).runId); + }); + const startup = (gate: WorkAdmission.WorkAdmissionGate, committed: boolean) => + Effect.gen(function* () { + const recovery = yield* ProviderRuntimeRecoveryService.ProviderRuntimeRecoveryService; + yield* recovery.reconcile("startup"); + // serverRuntimeStartup: held work starts after `launcher.prepareTrial` succeeded. + if (committed) yield* gate.startHeldWork; + return yield* heldStarts; + }); + + yield* TestClock.setTime(60 * 60_000); + // Old version: the gate closes, two messages arrive, then the marker is written. The + // first message is stamped before the marker's hold time. + const runIds = yield* runProcess((gate) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + yield* gate.hold; + const send = (name: string) => + Effect.gen(function* () { + const threadId = ThreadId.make(`update-rollback-${name}-${trial}`); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: ProjectId.make(`${threadId}:project`), + title: "Arrives during the update window", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:message`), + threadId, + messageId: MessageId.make(`${threadId}:message`), + text: `Sent as ${name} while the server waits for its update`, + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + return (yield* orchestrator.getThreadProjection(threadId)).runs[0]!.id; + }); + const early = yield* send("early"); + yield* TestClock.adjust("1 seconds"); + yield* gate.markHeldForUpdate({ + heldAt: DateTime.formatIso(yield* DateTime.now), + targetVersion: "next", + }); + yield* TestClock.adjust("1 seconds"); + const late = yield* send("late"); + return [early, late] as Array; + }), + ); + yield* copyDatabase(directory, backup); + + yield* TestClock.adjust("30 seconds"); + // New version on trial. + const trialStarts = yield* runProcess((gate) => startup(gate, trial === "commits")); + let laterStarts: ReadonlyArray = []; + if (trial === "rolls back") { + yield* copyDatabase(backup, directory); + yield* TestClock.adjust("30 seconds"); + // Old version again, on the database it stopped with. + laterStarts = yield* runProcess((gate) => startup(gate, true)); + } + // A later restart (or recovery running again) starts nothing new. + yield* TestClock.adjust("30 seconds"); + const finalStarts = yield* runProcess((gate) => startup(gate, true)); + + const reachedProvider = trial === "commits" ? finalStarts : [...trialStarts, ...laterStarts]; + if (trial === "rolls back") { + assert.deepEqual(trialStarts, [], "a trial that is rolled back starts nothing"); + } + assert.deepEqual( + finalStarts, + trial === "commits" ? trialStarts : laterStarts, + "a later restart starts nothing new", + ); + for (const runId of runIds) { + assert.equal( + reachedProvider.filter((started) => started === runId).length, + 1, + `${runId} starts exactly once`, + ); + } + assert.isUndefined(yield* marker.read, "the marker is cleared once the held work started"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index aec0cbd75076d..eec1c535ac9c1 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -180,6 +180,8 @@ import { } from "./cloud/managedTunnelStartup.ts"; import * as CloudCliTokenManager from "./cloud/CliTokenManager.ts"; import * as CloudCliState from "./cloud/CliState.ts"; +import * as ServerAutoUpdate from "./cloud/autoUpdate.ts"; +import * as WorkAdmission from "./workAdmission.ts"; import * as ServerSelfUpdate from "./cloud/selfUpdate.ts"; import * as DesktopAppUpdate from "./desktopUpdate/DesktopAppUpdate.ts"; import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts"; @@ -818,6 +820,8 @@ const layerMakeRoutes = Layer.mergeAll( Layer.effectDiscard( RemoteDelegation.RemoteDelegation.pipe(Effect.flatMap((remote) => remote.start())), ), + // Off unless automaticUpdates is on; shares the Update button's ServerSelfUpdate below. + ServerAutoUpdate.layer, ).pipe( Layer.provide( PullRequestReviewGuideService.layer.pipe(Layer.provide(PullRequestReviewGuides.layer)), @@ -865,6 +869,13 @@ const layerMakeServer = Layer.unwrap( const activation = yield* Deferred.make(); const awaitActivation = Deferred.await(activation); const layerActivation = Layer.succeed(ServerActivation.ServerActivation, awaitActivation); + // One gate for the whole server: automatic updates hold new work until the restart. + const layerWorkAdmission = Layer.succeed( + WorkAdmission.WorkAdmission, + yield* WorkAdmission.makeWorkAdmissionGate({ + marker: yield* WorkAdmission.makeFileUpdateHoldStore(config.baseDir), + }), + ); const runtimeStateParked = yield* Deferred.make(); const tailscaleParked = yield* Deferred.make(); const cloudLinkParked = yield* Deferred.make(); @@ -1221,7 +1232,7 @@ const layerMakeServer = Layer.unwrap( Layer.provide(ServerEnvironment.layer.pipe(Layer.provide(ServerSecretStore.layer))), ), ), - Layer.provide(layerActivation), + Layer.provide(Layer.merge(layerActivation, layerWorkAdmission)), Layer.provideMerge( RelayTracing.layerServerRelayBroker.pipe(Layer.provide(ManagedPolicy.layer)), ), diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 689659ab23f59..43fb2791bcdf3 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -50,6 +50,7 @@ import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import { getServerOwners } from "./telemetry/Identify.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; +import { WorkAdmission } from "./workAdmission.ts"; import { formatHeadlessServeOutput, issueHeadlessServeAccessInfo } from "./startupAccess.ts"; export class ServerRuntimeStartupError extends Schema.TaggedError()( @@ -449,6 +450,7 @@ const make = Effect.fn("serverRuntimeStartup.make")(function* (options?: Startup const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const random = yield* SecureRandom.SecureRandom; const launcher = yield* ServiceLauncherClient.ServiceLauncherClient; + const admission = yield* WorkAdmission; const commandGate = yield* makeCommandGate; const httpListening = yield* Deferred.make(); @@ -634,6 +636,9 @@ const make = Effect.fn("serverRuntimeStartup.make")(function* (options?: Startup yield* runStartupPhase("auxiliary-roots.parked", options?.awaitAuxiliaryParked ?? Effect.void); const updateOutcome = yield* launcher.prepareTrial({ baseDir: serverConfig.baseDir }); + // Work an automatic update held starts only now: a trial that fails before the launcher + // commits restores the database, and the version that runs next starts it instead. + yield* admission.startHeldWork; yield* Effect.logDebug("startup phase: publishing welcome event", { environmentId: environment.environmentId, diff --git a/apps/server/src/serverRuntimeStartup.updateHold.test.ts b/apps/server/src/serverRuntimeStartup.updateHold.test.ts new file mode 100644 index 0000000000000..8288e495b09ad --- /dev/null +++ b/apps/server/src/serverRuntimeStartup.updateHold.test.ts @@ -0,0 +1,185 @@ +// Fork-owned: work an automatic update held starts only once the launcher committed the update. +import { assert, it } from "@effect/vitest"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { DEFAULT_SERVER_SETTINGS, EnvironmentId } from "@t3tools/contracts"; +import { DEFAULT_SIGNAL_EXPORT } from "@t3tools/shared/observability"; +import * as OtelEnvironment from "@t3tools/shared/otelEnvironment"; +import { HttpServer } from "effect/http"; +import * as NetAddress from "effect/net/NetAddress"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; + +import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; +import * as ServerConfig from "./config.ts"; +import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; +import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; +import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts"; +import { SERVICE_LAUNCHER_PROTOCOL } from "./cloud/serviceProtocol.ts"; +import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; +import * as Keybindings from "./keybindings.ts"; +import * as EffectWorker from "./orchestration-v2/EffectWorker.ts"; +import * as LegacyV1ThreadImporter from "./orchestration-v2/legacy/LegacyV1ThreadImporter.ts"; +import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; +import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; +import * as ProviderRuntimeRecovery from "./orchestration-v2/ProviderRuntimeRecoveryService.ts"; +import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; +import * as ThreadLaunch from "./orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; +import * as ExternalLauncher from "./process/externalLauncher.ts"; +import * as ProjectService from "./project/ProjectService.ts"; +import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts"; +import { ServerActivation } from "./serverActivation.ts"; +import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; +import * as ServerSettings from "./serverSettings.ts"; +import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; +import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; +import { makeWorkAdmissionGate, WorkAdmission } from "./workAdmission.ts"; + +it.effect.each([ + { name: "committed", trial: "committed" }, + { name: "failing before the commit", trial: "fails" }, +] as const)("starts held work only after the update trial, $name", ({ trial }) => + Effect.scoped( + Effect.gen(function* () { + const order: Array = []; + const recovery = { + terminalizedRuns: 0, + stoppedSessions: 0, + closedRequests: 0, + retiredEffects: 0, + requeuedEffects: 0, + }; + const imports = { importedThreadCount: 0, importedMessageCount: 0 }; + const gate = yield* makeWorkAdmissionGate(); + yield* gate.setHeldWorkStarter(Effect.sync(() => order.push("start held work"))); + const dependencies = Layer.mergeAll( + NodeCrypto.layer, + Path.layer, + Layer.mock(ServerConfig.ServerConfig)({ + ...(yield* ServerConfig.deriveServerPaths("/update-hold-project", undefined).pipe( + Effect.provide(Path.layer), + )), + baseDir: "/update-hold-project", + logLevel: "Error", + traceMinLevel: "Info", + traceTimingEnabled: false, + traceBatchWindowMs: 200, + traceMaxBytes: 1024, + traceMaxFiles: 1, + otlpTracesUrl: undefined, + otlpMetricsUrl: undefined, + otlpLogsUrl: undefined, + otlpTracesExport: DEFAULT_SIGNAL_EXPORT, + otlpMetricsExport: DEFAULT_SIGNAL_EXPORT, + otlpLogsExport: DEFAULT_SIGNAL_EXPORT, + otelEnvironment: OtelEnvironment.none, + staticDir: undefined, + devAllowedOrigins: [], + desktopBootstrapToken: undefined, + logWebSocketEvents: false, + tailscaleServeEnabled: false, + tailscaleServePort: 443, + mode: "web", + cwd: "/update-hold-project", + host: "localhost", + port: 3773, + devUrl: undefined, + noBrowser: true, + startupPresentation: "browser", + autoBootstrapProjectFromCwd: false, + }), + Layer.mock(Keybindings.Keybindings)({ start: Effect.void }), + Layer.mock(LegacyV1ThreadImporter.LegacyV1ThreadImporter)({ + pendingThreadCount: Effect.succeed(0), + reconcileShells: Effect.succeed(imports), + importPendingTranscripts: Effect.succeed(imports), + }), + Layer.mock(ProviderRuntimeRecovery.ProviderRuntimeRecoveryService)({ + recover: Effect.sync(() => { + order.push("recover"); + return recovery; + }), + prepareForShutdown: Effect.void, + reconcile: () => Effect.succeed(recovery), + }), + Layer.mock(Orchestrator.OrchestratorV2)({ recoverDelegatedTasks: Effect.void }), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ shutdown: Effect.void }), + Layer.mock(AgentAwarenessRelay.AgentAwarenessRelay)({ start: () => Effect.void }), + Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({ runOnce: () => Effect.never }), + Layer.mock(ServerLifecycleEvents.ServerLifecycleEvents)({ + publish: (event) => Effect.succeed({ ...event, sequence: 1 }), + }), + Layer.mock(ServerSettings.ServerSettingsService)({ + start: Effect.void, + getSettings: Effect.succeed(DEFAULT_SERVER_SETTINGS), + }), + Layer.mock(ServerEnvironment.ServerEnvironment)({ + getDescriptor: Effect.succeed({ + environmentId: EnvironmentId.make("update-hold-environment"), + label: "Test environment", + platform: { os: "linux", arch: "x64" }, + serverVersion: "0.0.0-test", + capabilities: { repositoryIdentity: true }, + }), + }), + Layer.mock(ProjectStore.ProjectStoreV2)({}), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(ThreadManagement.ThreadManagementService)({ + countThreads: () => Effect.succeed(0), + listActiveShellIndex: () => Effect.succeed([]), + }), + Layer.mock(ThreadLaunch.ThreadLaunchService)({}), + Layer.mock(ServiceLauncherClient.ServiceLauncherClient)({ + managed: true, + protocol: SERVICE_LAUNCHER_PROTOCOL, + prepareTrial: () => + Effect.suspend(() => { + order.push("prepare trial"); + return trial === "committed" + ? Effect.succeed({ + id: "update-1", + fromVersion: "1.0.0", + targetVersion: "1.0.1", + status: "committed" as const, + }) + : Effect.fail( + new ServiceLauncherClient.ServiceLauncherClientError({ + operation: "disconnect", + }), + ); + }), + }), + Layer.mock(GitVcsDriver.GitVcsDriver)({}), + Layer.mock(AnalyticsService.AnalyticsService)({}), + FileSystem.layerNoop({}), + Layer.mock(ServerSecretStore.ServerSecretStore)({ get: () => Effect.succeedNone }), + Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), + Layer.mock(ExternalLauncher.ExternalLauncher)({}), + Layer.mock(HttpServer.HttpServer)({ + address: NetAddress.inetAddressFromIpStringUnsafe("127.0.0.1", 3773), + }), + ); + yield* Effect.gen(function* () { + const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup; + yield* startup.markHttpListening; + const ready = yield* Effect.exit(startup.awaitCommandReady); + assert.equal(Exit.isSuccess(ready), trial === "committed"); + }).pipe( + Effect.provide(ServerRuntimeStartup.layerWithOptions().pipe(Layer.provide(dependencies))), + Effect.provideService(ServerActivation, Effect.never), + Effect.provideService(WorkAdmission, gate), + ); + // A trial that fails before the commit is rolled back: the version that runs next starts + // the held work, so this one must not have. + assert.deepEqual( + order, + trial === "committed" + ? ["recover", "prepare trial", "start held work"] + : ["recover", "prepare trial"], + ); + }), + ), +); diff --git a/apps/server/src/serverSettings.ts b/apps/server/src/serverSettings.ts index 251becba12f6c..e1c9d53be1a32 100644 --- a/apps/server/src/serverSettings.ts +++ b/apps/server/src/serverSettings.ts @@ -48,6 +48,7 @@ import { resolveSymlinkTarget } from "@t3tools/shared/symlink"; import * as ServerConfig from "./config.ts"; import { type DeepPartial, deepMerge } from "@t3tools/shared/Struct"; import { fromJsonStringPretty, fromLenientJson } from "@t3tools/shared/schemaJson"; +import { renameRetiredAutoUpdateKeys } from "./cloud/autoUpdateSettings.ts"; import { applyServerSettingsPatch, deriveLegacyProjectOverrides, @@ -794,7 +795,9 @@ const make = Effect.gen(function* () { const parsed = decodeSettingsFileJsonExit(raw); const decoded = Exit.isFailure(parsed) ? parsed - : decodePersistedServerSettingsExit(managed.withDefaults(parsed.value)); + : decodePersistedServerSettingsExit( + managed.withDefaults(renameRetiredAutoUpdateKeys(parsed.value)), + ); const legacySettings = decodeLegacyProviderSettingsJsonExit(raw); if (legacySettings._tag === "Success") { legacyProviders = legacySettings.value.providers; diff --git a/apps/server/src/workAdmission.test.ts b/apps/server/src/workAdmission.test.ts new file mode 100644 index 0000000000000..879a39f0f665b --- /dev/null +++ b/apps/server/src/workAdmission.test.ts @@ -0,0 +1,200 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; +import * as TestClock from "effect/testing/TestClock"; + +import * as EffectWorker from "./orchestration-v2/EffectWorker.ts"; +import { + makeFileUpdateHoldStore, + makeWorkAdmissionGate, + memoryUpdateHoldStore, + UPDATE_HOLD_MARKER_FILE, + UPDATE_HOLD_MAX_AGE_MS, + WORK_ADMISSION_CONSUMERS, + WorkAdmission, +} from "./workAdmission.ts"; + +const settle = Effect.gen(function* () { + for (let i = 0; i < 20; i++) yield* Effect.yieldNow; +}); + +describe("WorkAdmission", () => { + it.effect("holds new work until released and counts work already admitted", () => + Effect.gen(function* () { + const gate = yield* makeWorkAdmissionGate(); + const running = yield* Effect.forkChild(gate.admit(Effect.never)); + yield* settle; + expect(yield* gate.hold).toBe(1); + const ran = yield* Ref.make(false); + const waiting = yield* Effect.forkChild(gate.admit(Ref.set(ran, true))); + yield* settle; + expect(yield* Ref.get(ran)).toBe(false); + yield* Fiber.interrupt(running); + expect(yield* gate.inFlight).toBe(0); + yield* gate.release; + yield* Fiber.join(waiting); + expect(yield* Ref.get(ran)).toBe(true); + expect(yield* gate.inFlight).toBe(0); + }), + ); + + it.effect("defaults to a pass-through that says it does not hold", () => + Effect.gen(function* () { + const gate = yield* WorkAdmission; + expect(gate.enforced).toBe(false); + expect(yield* gate.admit(Effect.succeed(1))).toBe(1); + // Nothing uses it, so an automatic update refuses to hold it. + expect(yield* gate.missingConsumers).toEqual(WORK_ADMISSION_CONSUMERS); + }), + ); + + it.effect("stops the outbox daemon from claiming effects while held", () => + Effect.gen(function* () { + const gate = yield* makeWorkAdmissionGate(); + const claims = yield* Ref.make(0); + const worker = EffectWorker.OrchestrationEffectWorkerV2.of({ + awaitWork: Effect.never, + runOnce: () => Ref.update(claims, (count) => count + 1).pipe(Effect.as(false)), + runRecoveryOnce: Effect.succeed(false), + nextClaimableAt: () => Effect.succeedNone, + drain: () => Effect.succeed(0), + }); + yield* gate.hold; + expect(yield* gate.missingConsumers).toContain("outbox"); + const daemon = yield* Effect.forkChild( + EffectWorker.runDaemonWithOptions({ concurrency: 2 }).pipe( + Effect.provideService(EffectWorker.OrchestrationEffectWorkerV2, worker), + Effect.provideService(WorkAdmission, gate), + ), + ); + yield* settle; + expect(yield* Ref.get(claims)).toBe(0); + // The daemon says it claims through this gate. + expect(yield* gate.missingConsumers).not.toContain("outbox"); + yield* gate.release; + yield* settle; + expect(yield* Ref.get(claims)).toBe(2); + yield* Fiber.interrupt(daemon); + }), + ); + + it.effect("tells startup about a fresh update marker only, and release removes it", () => + Effect.gen(function* () { + const marker = yield* memoryUpdateHoldStore; + const gate = yield* makeWorkAdmissionGate({ marker }); + const heldAt = DateTime.makeUnsafe(10 * 60_000); + yield* TestClock.setTime(DateTime.toEpochMillis(heldAt)); + expect(yield* gate.heldForUpdate).toBeUndefined(); + yield* gate.hold; + yield* gate.markHeldForUpdate({ heldAt: DateTime.formatIso(heldAt), targetVersion: "2" }); + yield* TestClock.adjust(UPDATE_HOLD_MAX_AGE_MS); + expect(yield* gate.heldForUpdate).toEqual({ + heldAt, + runs: undefined, + started: new Set(), + }); + // Work released here must not be started again by some later startup. + yield* gate.release; + expect(yield* marker.current).toBeUndefined(); + }), + ); + + it.effect("removes a stale or invalid marker and reads it as none", () => + Effect.gen(function* () { + const marker = yield* memoryUpdateHoldStore; + const gate = yield* makeWorkAdmissionGate({ marker }); + const heldAt = DateTime.makeUnsafe(10 * 60_000); + yield* TestClock.setTime(DateTime.toEpochMillis(heldAt) + UPDATE_HOLD_MAX_AGE_MS + 1); + // Older than the maximum, it is left over from a hold that ended without clearing it. + yield* marker.write({ heldAt: DateTime.formatIso(heldAt), targetVersion: "2" }); + expect(yield* gate.heldForUpdate).toBeUndefined(); + expect(yield* marker.current).toBeUndefined(); + yield* marker.set("invalid"); + expect(yield* gate.heldForUpdate).toBeUndefined(); + expect(yield* marker.current).toBeUndefined(); + yield* marker.write({ heldAt: "not a time", targetVersion: "2" }); + expect(yield* gate.heldForUpdate).toBeUndefined(); + expect(yield* marker.current).toBeUndefined(); + // Nothing to record against once no fresh marker was read. + expect((yield* Effect.flip(gate.recordHeldRunsStarted(["run-1"])))._tag).toBe( + "UpdateHoldMarkerError", + ); + }), + ); + + it.effect("records the kept runs once and what was started, and finishing settles", () => + Effect.gen(function* () { + const marker = yield* memoryUpdateHoldStore; + const gate = yield* makeWorkAdmissionGate({ marker }); + const hold = { heldAt: DateTime.formatIso(yield* DateTime.now), targetVersion: "2" }; + // An old-format marker: no runs, nothing started. + yield* marker.write(hold); + expect(yield* gate.heldWorkSettled).toBe(false); + expect((yield* gate.heldForUpdate)?.runs).toBeUndefined(); + yield* gate.recordHeldRuns(["run-1", "run-2"]); + // A later startup keeps the first list: that is the database the hold left. + yield* gate.recordHeldRuns(["run-1", "run-2", "run-3"]); + yield* gate.recordHeldRunsStarted(["run-1"]); + yield* gate.recordHeldRunsStarted(["run-1"]); + expect(yield* marker.current).toEqual({ + ...hold, + runs: ["run-1", "run-2"], + started: ["run-1"], + }); + // What the next startup reads. + const again = yield* makeWorkAdmissionGate({ marker }); + expect(yield* again.heldForUpdate).toMatchObject({ + runs: new Set(["run-1", "run-2"]), + started: new Set(["run-1"]), + }); + yield* gate.finishHeldWork; + expect(yield* marker.current).toBeUndefined(); + expect(yield* gate.heldWorkSettled).toBe(true); + // The pass-through holds nothing, so nothing waits for it. + expect(yield* (yield* WorkAdmission).heldWorkSettled).toBe(true); + }), + ); + + it.effect("runs the held-work starter that recovery set, and nothing without one", () => + Effect.gen(function* () { + const gate = yield* makeWorkAdmissionGate(); + yield* gate.startHeldWork; + const runs = yield* Ref.make(0); + yield* gate.setHeldWorkStarter(Ref.update(runs, (count) => count + 1)); + yield* gate.startHeldWork; + expect(yield* Ref.get(runs)).toBe(1); + }), + ); + + it.effect("keeps the update marker beside the launcher state and survives a bad file", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const baseDir = yield* fs.makeTempDirectoryScoped(); + const store = yield* makeFileUpdateHoldStore(baseDir); + expect(yield* store.read).toBeUndefined(); + const hold = { heldAt: "2026-01-01T00:00:00.000Z", targetVersion: "2" }; + yield* store.write(hold); + const file = path.join(baseDir, "runtime", UPDATE_HOLD_MARKER_FILE); + expect(yield* fs.exists(file)).toBe(true); + expect(yield* fs.exists(`${file}.tmp`)).toBe(false); + expect(yield* (yield* makeFileUpdateHoldStore(baseDir)).read).toEqual(hold); + const withRuns = { ...hold, runs: ["run-1"], started: [] }; + yield* store.write(withRuns); + expect(yield* store.read).toEqual(withRuns); + // Unknown fields from a newer version do not make it invalid. + yield* fs.writeFileString(file, JSON.stringify({ ...hold, later: 1 })); + expect(yield* store.read).toMatchObject(hold); + yield* fs.writeFileString(file, "{not json"); + expect(yield* store.read).toBe("invalid"); + yield* store.clear; + yield* store.clear; + expect(yield* fs.exists(file)).toBe(false); + }).pipe(Effect.provide(NodeServices.layer)), + ); +}); diff --git a/apps/server/src/workAdmission.ts b/apps/server/src/workAdmission.ts new file mode 100644 index 0000000000000..2d433f3e0a016 --- /dev/null +++ b/apps/server/src/workAdmission.ts @@ -0,0 +1,332 @@ +// Gate in front of outbox claims and worktree preparation. An automatic update closes it before +// its last quiet check and keeps it closed until exit. The marker makes startup keep held work and +// start it once the update is committed, once: it records the held `runs` and those `started` +// (a failed update restores the database). See ProviderRuntimeRecoveryService. +import type { CommandId, RunId, ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Latch from "effect/Latch"; +import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; + +/** The parts of the server that must use one shared gate before an update may hold it. */ +export const WORK_ADMISSION_CONSUMERS = ["outbox", "thread-launch", "recovery"] as const; +export type WorkAdmissionConsumer = (typeof WORK_ADMISSION_CONSUMERS)[number]; + +export const UPDATE_HOLD_MARKER_FILE = "work-held-for-update.json"; +/** An older marker is left over from a hold that ended without clearing it, not this restart. */ +export const UPDATE_HOLD_MAX_AGE_MS = 2 * 60 * 60_000; + +const UpdateHold = Schema.Struct({ + heldAt: Schema.String, + targetVersion: Schema.String, + /** The runs the first startup after the hold kept: later startups consider only these. */ + runs: Schema.optional(Schema.Array(Schema.String)), + /** Runs a startup started (or was about to): they are never started again. */ + started: Schema.optional(Schema.Array(Schema.String)), +}); +export type UpdateHold = typeof UpdateHold.Type; + +/** What a fresh marker tells startup recovery. */ +export interface HeldForUpdate { + readonly heldAt: DateTime.Utc; + readonly runs: ReadonlySet | undefined; + readonly started: ReadonlySet; +} + +export class UpdateHoldMarkerError extends Schema.TaggedError()( + "UpdateHoldMarkerError", + { reason: Schema.String }, +) {} + +/** Where the update marker lives: `/runtime`, beside the launcher's state. */ +export interface UpdateHoldStore { + /** `undefined` when there is no marker or it cannot be read; "invalid" when it is not one. */ + readonly read: Effect.Effect; + readonly write: (hold: UpdateHold) => Effect.Effect; + readonly clear: Effect.Effect; +} + +export const makeFileUpdateHoldStore = Effect.fn("workAdmission.makeFileUpdateHoldStore")( + function* (baseDir: string) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const runtimeDir = path.join(baseDir, "runtime"); + const file = path.join(runtimeDir, UPDATE_HOLD_MARKER_FILE); + const json = Schema.fromJsonString(UpdateHold); + const decode = Schema.decodeUnknownOption(json); + const encode = Schema.encodeEffect(json); + return { + read: fs.readFileString(file).pipe( + Effect.map((contents) => { + const decoded = decode(contents); + return decoded._tag === "Some" ? decoded.value : ("invalid" as const); + }), + Effect.catch((error) => + error.reason._tag === "NotFound" + ? Effect.undefined + : Effect.logWarning("The update hold marker could not be read.", { + reason: error.reason._tag, + }).pipe(Effect.as(undefined)), + ), + ), + write: Effect.fn("workAdmission.writeUpdateHold")( + function* (hold: UpdateHold) { + yield* fs.makeDirectory(runtimeDir, { recursive: true }); + const staging = `${file}.tmp`; + yield* fs.writeFileString(staging, yield* encode(hold)); + yield* fs.rename(staging, file); + }, + Effect.mapError((error) => new UpdateHoldMarkerError({ reason: error.message })), + ), + clear: fs.remove(file, { force: true }).pipe( + Effect.catch((error) => + Effect.logWarning("The update hold marker could not be removed.", { + reason: error.message, + }), + ), + ), + } satisfies UpdateHoldStore; + }, +); + +export const memoryUpdateHoldStore: Effect.Effect< + UpdateHoldStore & { + readonly current: Effect.Effect; + readonly set: (hold: UpdateHold | "invalid" | undefined) => Effect.Effect; + } +> = Effect.gen(function* () { + const marker = yield* Ref.make(undefined); + return { + read: Ref.get(marker), + write: (hold) => Ref.set(marker, hold), + clear: Ref.set(marker, undefined), + current: Ref.get(marker), + set: (hold) => Ref.set(marker, hold), + }; +}); + +/** A worktree preparation that waited at the gate when the last process stopped. */ +export interface HeldPreparation { + readonly commandId: CommandId; + readonly threadId: ThreadId; + readonly runId: RunId; +} + +export interface WorkAdmissionGate { + /** False for the default pass-through: nothing is held, so callers must not rely on `hold`. */ + readonly enforced: boolean; + /** Runs `effect` once the gate is open, counting it as in flight while it runs. */ + readonly admit: (effect: Effect.Effect) => Effect.Effect; + /** Closes the gate. New work waits; returns how many admitted units are still running. */ + readonly hold: Effect.Effect; + /** Opens the gate again, removes the update marker and lets waiting work start. */ + readonly release: Effect.Effect; + readonly inFlight: Effect.Effect; + readonly isHeld: Effect.Effect; + /** Each consumer says it admits its work through this gate. */ + readonly register: (consumer: WorkAdmissionConsumer) => Effect.Effect; + /** Consumers that never registered: an update must not hold a gate they do not use. */ + readonly missingConsumers: Effect.Effect>; + /** Records that new work is held for an update, for the next startup. */ + readonly markHeldForUpdate: (hold: UpdateHold) => Effect.Effect; + /** A fresh marker: work saved while it was held never reached a provider. A stale or invalid + * marker is removed and reads as none. */ + readonly heldForUpdate: Effect.Effect; + /** Saves which runs this startup kept, unless an earlier startup after the hold already did. */ + readonly recordHeldRuns: ( + runIds: ReadonlyArray, + ) => Effect.Effect; + /** Saves that these runs are being started; callers start them only once this succeeded. */ + readonly recordHeldRunsStarted: ( + runIds: ReadonlyArray, + ) => Effect.Effect; + /** Every held run was started: removes the marker and says so through `heldWorkSettled`. */ + readonly finishHeldWork: Effect.Effect; + /** This process has started what an earlier update held, so it may hold work again. */ + readonly heldWorkSettled: Effect.Effect; + /** Startup recovery's way to start the work it kept for an update. */ + readonly setHeldWorkStarter: (start: Effect.Effect) => Effect.Effect; + /** Starts the kept work. Startup calls it only once the update is committed, so a failed + * update that rolls back has started nothing. It never fails. */ + readonly startHeldWork: Effect.Effect; + /** The thread-launch service's way to start a held worktree preparation again. */ + readonly setPreparationResumer: ( + resume: (preparation: HeldPreparation) => Effect.Effect, + ) => Effect.Effect; + readonly resumePreparation: (preparation: HeldPreparation) => Effect.Effect; +} + +interface GateState { + readonly held: boolean; + readonly inFlight: number; +} + +const noMarker: UpdateHoldStore = { + read: Effect.undefined, + write: () => Effect.fail(new UpdateHoldMarkerError({ reason: "this gate keeps no marker" })), + clear: Effect.void, +}; + +const markerSince = (hold: UpdateHold, nowMs: number) => { + const heldAt = DateTime.make(hold.heldAt); + if (heldAt._tag === "None") return undefined; + const ageMs = nowMs - DateTime.toEpochMillis(heldAt.value); + return ageMs >= 0 && ageMs <= UPDATE_HOLD_MAX_AGE_MS ? heldAt.value : undefined; +}; + +export const makeWorkAdmissionGate = Effect.fn("workAdmission.makeWorkAdmissionGate")(function* ( + options: { readonly marker?: UpdateHoldStore } = {}, +) { + const marker = options.marker ?? noMarker; + const state = yield* Ref.make({ held: false, inFlight: 0 }); + const registered = yield* Ref.make(new Set()); + // The fresh marker as this process last read or wrote it. + const freshMarker = yield* Ref.make(undefined); + const settled = yield* Ref.make(false); + const rewriteMarker = Effect.fn("workAdmission.rewriteMarker")(function* ( + change: (hold: UpdateHold) => UpdateHold, + ) { + const current = yield* Ref.get(freshMarker); + if (current === undefined) { + return yield* new UpdateHoldMarkerError({ reason: "no fresh update marker was read" }); + } + const next = change(current); + if (next === current) return; + yield* marker.write(next); + yield* Ref.set(freshMarker, next); + }); + const heldWorkStarter = yield* Ref.make | undefined>(undefined); + const resumer = yield* Ref.make< + ((preparation: HeldPreparation) => Effect.Effect) | undefined + >(undefined); + const open = yield* Latch.make(true); + const admit = (effect: Effect.Effect): Effect.Effect => + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + // The held check and the count change in one step, so a unit admitted before `hold` + // is always counted and none starts after it. + while (true) { + yield* restore(open.await); + const admitted = yield* Ref.modify(state, (current): [boolean, GateState] => + current.held + ? [false, current] + : [true, { ...current, inFlight: current.inFlight + 1 }], + ); + if (admitted) break; + } + return yield* restore(effect).pipe( + Effect.ensuring( + Ref.update(state, (current) => ({ ...current, inFlight: current.inFlight - 1 })), + ), + ); + }), + ); + const gate: WorkAdmissionGate = { + enforced: true, + admit, + hold: Effect.gen(function* () { + const inFlight = yield* Ref.modify(state, (current): [number, GateState] => [ + current.inFlight, + { ...current, held: true }, + ]); + yield* open.close; + return inFlight; + }), + // The marker goes first: work released here must not be restarted by a later startup. + release: marker.clear.pipe( + Effect.andThen(Ref.update(state, (current) => ({ ...current, held: false }))), + Effect.andThen(open.open), + Effect.asVoid, + ), + inFlight: Ref.get(state).pipe(Effect.map((current) => current.inFlight)), + isHeld: Ref.get(state).pipe(Effect.map((current) => current.held)), + register: (consumer) => Ref.update(registered, (current) => new Set([...current, consumer])), + missingConsumers: Ref.get(registered).pipe( + Effect.map((current) => + WORK_ADMISSION_CONSUMERS.filter((consumer) => !current.has(consumer)), + ), + ), + markHeldForUpdate: marker.write, + heldForUpdate: Effect.gen(function* () { + const hold = yield* marker.read; + yield* Ref.set(freshMarker, undefined); + if (hold === undefined) return undefined; + const since = + hold === "invalid" + ? undefined + : markerSince(hold, DateTime.toEpochMillis(yield* DateTime.now)); + if (hold === "invalid" || since === undefined) { + // Left over from a hold that ended without clearing it: none of its work is held now. + yield* Effect.logWarning("Removing a stale or invalid update hold marker.", { + heldAt: hold === "invalid" ? undefined : hold.heldAt, + }); + yield* marker.clear; + return undefined; + } + yield* Ref.set(freshMarker, hold); + return { + heldAt: since, + runs: hold.runs === undefined ? undefined : new Set(hold.runs), + started: new Set(hold.started ?? []), + }; + }), + recordHeldRuns: (runIds) => + rewriteMarker((hold) => (hold.runs !== undefined ? hold : { ...hold, runs: [...runIds] })), + recordHeldRunsStarted: (runIds) => + rewriteMarker((hold) => ({ + ...hold, + started: [...new Set([...(hold.started ?? []), ...runIds])], + })), + finishHeldWork: marker.clear.pipe( + Effect.andThen(Ref.set(freshMarker, undefined)), + Effect.andThen(Ref.set(settled, true)), + ), + heldWorkSettled: Ref.get(settled), + setHeldWorkStarter: (start) => Ref.set(heldWorkStarter, start), + startHeldWork: Ref.get(heldWorkStarter).pipe(Effect.flatMap((start) => start ?? Effect.void)), + setPreparationResumer: (resume) => Ref.set(resumer, resume), + resumePreparation: (preparation) => + Ref.get(resumer).pipe( + Effect.flatMap((resume) => + resume === undefined + ? Effect.logError( + "A held worktree preparation cannot start again: no thread-launch service uses this gate.", + preparation, + ) + : resume(preparation), + ), + ), + }; + return gate; +}); + +const passThrough: WorkAdmissionGate = { + enforced: false, + admit: (effect) => effect, + hold: Effect.succeed(0), + release: Effect.void, + inFlight: Effect.succeed(0), + isHeld: Effect.succeed(false), + register: () => Effect.void, + missingConsumers: Effect.succeed(WORK_ADMISSION_CONSUMERS), + markHeldForUpdate: noMarker.write, + heldForUpdate: Effect.undefined, + recordHeldRuns: () => Effect.void, + recordHeldRunsStarted: () => Effect.void, + finishHeldWork: Effect.void, + heldWorkSettled: Effect.succeed(true), + setHeldWorkStarter: () => Effect.void, + startHeldWork: Effect.void, + setPreparationResumer: () => Effect.void, + resumePreparation: () => Effect.void, +}; + +/** The server provides one shared gate. Tests and tools that do not see the pass-through, which + * holds nothing; automatic updates refuse to run on it (see `enforced` and `missingConsumers`). */ +export class WorkAdmission extends Context.Reference("t3/workAdmission", { + defaultValue: () => passThrough, +}) {} diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 00eeb1797870a..378e423398ad1 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -1371,9 +1371,17 @@ export const ServerSettings = Schema.Struct({ Schema.withDecodingDefault(Effect.succeed("paragraph" as const)), ), enableProviderUpdateChecks: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), + /** + * Fork: a boot-service server installs newer releases by itself once nothing is working. + * Same name and place as upstream's pending automatic-updates setting, whose line replaces this + * one when it lands (upstream defaults it on). A file saying `autoUpdateServer` is renamed on load. + */ + automaticUpdates: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), continueThreadsAfterServerUpdate: Schema.Boolean.pipe( Schema.withDecodingDefault(Effect.succeed(true)), ), + /** Fork: optional executable asked before an automatic update; a non-zero exit means "not now". */ + automaticUpdatesQuietCommand: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), enableAgentBrowserAccess: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), cloudSpeechCredentials: CloudSpeechCredentials.pipe( @@ -1824,7 +1832,9 @@ export const ServerSettingsPatch = Schema.Struct({ fastWorktreeCheckout: Schema.optionalKey(Schema.Boolean), responseStreamingMode: Schema.optionalKey(ResponseStreamingMode), enableProviderUpdateChecks: Schema.optionalKey(Schema.Boolean), + automaticUpdates: Schema.optionalKey(Schema.Boolean), continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean), + automaticUpdatesQuietCommand: Schema.optionalKey(TrimmedString), enableAgentBrowserAccess: Schema.optionalKey(Schema.Boolean), cloudSpeechCredentials: Schema.optionalKey( Schema.Struct({