fix(server): supervise background service updates

apps/server/src/cli/service.ts

@@ -2,11 +2,14 @@ import * as Console from "effect/Console";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Terminal from "effect/Terminal";
-import { Command, GlobalFlag, Prompt } from "effect/unstable/cli";
+import { Command, Flag, GlobalFlag, Prompt } from "effect/unstable/cli";
+import { FetchHttpClient } from "effect/unstable/http";
 
 import packageJson from "../../package.json" with { type: "json" };
 import * as BootService from "../cloud/bootService.ts";
+import { applySystemdSelfUpdatePlan } from "../cloud/systemdSelfUpdate.ts";
 import type * as ServerConfig from "../config.ts";
+import { validateAutomaticUpdateMigrationFrontier } from "../persistence/Migrations.ts";
 import * as ProcessRunner from "../processRunner.ts";
 import { projectLocationFlags, resolveCliAuthConfig } from "./config.ts";
 
@@ -142,6 +145,38 @@ const serviceStatusCommand = Command.make("status", projectLocationFlags).pipe(
   ),
 );
 
+/** Internal entrypoint launched by systemd-run outside t3code.service. */
+const serviceApplyUpdateCommand = Command.make("_apply-update", {
+  plan: Flag.string("plan"),
+}).pipe(
+  Command.withHidden,
+  Command.withHandler(({ plan }) =>
+    applySystemdSelfUpdatePlan(plan).pipe(
+      Effect.provide(Layer.mergeAll(ProcessRunner.layer, FetchHttpClient.layer)),
+    ),
+  ),
+);
+
+/** Target-artifact preflight used before the running server is stopped. */
+const serviceValidateUpdateCommand = Command.make("_validate-update", {
+  database: Flag.string("database"),
+}).pipe(
+  Command.withHidden,
+  Command.withHandler(({ database }) =>
+    Effect.gen(function* () {
+      // Keep node:sqlite behind the same lazy bundle boundary as normal server
+      // startup; most service commands never open the database.
+      const NodeSqliteClient = yield* Effect.promise(
+        () => import("../persistence/NodeSqliteClient.ts"),
+      );
+      yield* validateAutomaticUpdateMigrationFrontier().pipe(
+        Effect.scoped,
+        Effect.provide(NodeSqliteClient.layer({ filename: database, readonly: true })),
+      );
+    }),
+  ),
+);
+
 export const offerServiceDuringOnboarding = Effect.gen(function* () {
   const service = yield* BootService.BootService;
   const { supported, installed, current } = yield* service.status;
@@ -195,5 +230,7 @@ export const serviceCommand = Command.make("service").pipe(
     serviceUninstallCommand,
     serviceUpdateCommand,
     serviceStatusCommand,
+    serviceApplyUpdateCommand,
+    serviceValidateUpdateCommand,
   ]),
 );

apps/server/src/cloud/selfUpdate.test.ts

@@ -5,6 +5,7 @@ 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 Schema from "effect/Schema";
 import * as TestClock from "effect/testing/TestClock";
 import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
 
@@ -23,36 +24,15 @@ import {
   renderBootServiceUnit,
 } from "./bootService.ts";
 import * as SelfUpdate from "./selfUpdate.ts";
+import {
+  SYSTEMD_SELF_UPDATE_DIRECTORY,
+  SYSTEMD_SELF_UPDATE_PLAN_FILE,
+  SYSTEMD_SELF_UPDATE_UNIT,
+  SystemdSelfUpdatePlan,
+} from "./systemdSelfUpdate.ts";
 
 const NODE_PATH = "/usr/local/bin/node";
 
-const eventuallyFileString = Effect.fn("test.eventuallyFileString")(function* (
-  filePath: string,
-  expected: string,
-) {
-  const fs = yield* FileSystem.FileSystem;
-  for (let iteration = 0; iteration < 1_000; iteration += 1) {
-    const contents = yield* fs.readFileString(filePath);
-    if (contents === expected) {
-      return;
-    }
-    // The rollback performs real filesystem I/O on a detached fiber, which
-    // advancing TestClock does not await.
-    yield* Effect.yieldNow;
-  }
-  return yield* Effect.die(new Error(`Expected file contents were not observed at ${filePath}.`));
-});
-
-const eventuallyTrue = Effect.fn("test.eventuallyTrue")(function* (predicate: () => boolean) {
-  for (let iteration = 0; iteration < 1_000; iteration += 1) {
-    if (predicate()) {
-      return;
-    }
-    yield* Effect.yieldNow;
-  }
-  return yield* Effect.die(new Error("Expected condition was not observed."));
-});
-
 interface RecordedCommand {
   readonly command: string;
   readonly args: ReadonlyArray<string>;
@@ -445,7 +425,7 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => {
       assert.deepEqual(result, { targetVersion: "0.0.29", method: "respawn" });
       assert.deepEqual(
         context.commands.map((entry) => entry.command),
-        [NODE_PATH, "npm", NODE_PATH],
+        [NODE_PATH, "npm", NODE_PATH, NODE_PATH],
       );
     }).pipe(Effect.provide(TestClock.layer())),
   );
@@ -506,6 +486,7 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => {
         [
           `npm install --prefix ${context.path.join(context.baseDir, "runtime/versions/0.0.29")} --no-fund --no-audit t3@0.0.29`,
           `${NODE_PATH} ${pinnedEntry} --version`,
+          `${NODE_PATH} ${pinnedEntry} service _validate-update --database ${context.path.join(context.baseDir, "userdata/state.sqlite")}`,
         ],
       );
 
@@ -521,56 +502,74 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => {
     }).pipe(Effect.provide(TestClock.layer())),
   );
 
-  it.effect("rewrites the systemd unit and restarts the boot service", () =>
+  it.effect("stages the systemd unit and hands activation to a transient service", () =>
     Effect.gen(function* () {
       const context = yield* makeContext({ bootService: true });
+      const unitPath = context.path.join(
+        context.home,
+        ".config",
+        "systemd",
+        "user",
+        BOOT_SERVICE_UNIT_FILE,
+      );
+      const previousUnit = yield* context.fs.readFileString(unitPath);
       const result = yield* context.service.update({ targetVersion: "0.0.29" });
       assert.deepEqual(result, { targetVersion: "0.0.29", method: "boot-service" });
 
       const pinnedEntry = context.path.join(
         context.baseDir,
         "runtime/versions/0.0.29/node_modules/t3/dist/bin.mjs",
       );
-      const unit = yield* context.fs.readFileString(
-        context.path.join(context.home, ".config", "systemd", "user", "t3code.service"),
+      // The RPC process does not rewrite the unit it is currently running
+      // under. The independent transient service applies the staged unit.
+      assert.equal(yield* context.fs.readFileString(unitPath), previousUnit);
+      const planPath = context.path.join(
+        context.baseDir,
+        "userdata",
+        SYSTEMD_SELF_UPDATE_DIRECTORY,
+        SYSTEMD_SELF_UPDATE_PLAN_FILE,
+      );
+      const plan = yield* context.fs
+        .readFileString(planPath)
+        .pipe(
+          Effect.flatMap(Schema.decodeUnknownEffect(Schema.fromJsonString(SystemdSelfUpdatePlan))),
+        );
+      assert.equal(plan.previousUnit, previousUnit);
+      assert.equal(plan.serviceUnit, BOOT_SERVICE_UNIT_FILE);
+      assert.include(plan.nextUnit, `ExecStart=${NODE_PATH} ${pinnedEntry} serve`);
+      assert.equal(
+        plan.runtimeStatePath,
+        context.path.join(context.baseDir, "userdata/server-runtime.json"),
       );
-      assert.include(unit, `ExecStart=${NODE_PATH} ${pinnedEntry} serve`);
       assert.deepEqual(
         context.commands.map((entry) => entry.command),
-        ["npm", NODE_PATH, "systemctl"],
+        ["npm", NODE_PATH, NODE_PATH, "systemd-run"],
       );
-      assert.deepEqual(context.commands[2]?.args, ["--user", "daemon-reload"]);
-
-      // Restart waits until after the update acknowledgement can flush.
-      yield* TestClock.adjust(Duration.seconds(10));
-      assert.deepEqual(context.commands[3], {
-        command: "systemctl",
-        args: ["--user", "restart", "--no-block", "t3code.service"],
-      });
+      assert.include(context.commands[3]?.args ?? [], `--unit=${SYSTEMD_SELF_UPDATE_UNIT}`);
+      assert.include(context.commands[3]?.args ?? [], planPath);
       assert.lengthOf(context.spawns, 0);
-      // systemd replaces the process; the server must not exit itself.
       assert.equal(context.exitCount(), 0);
 
-      // The queued restart returns while this process is still shutting
-      // down; the lock must stay held so a second update cannot rewrite the
-      // unit mid-teardown.
+      // The lock stays held after a successful handoff because the transient
+      // service is now responsible for replacing this process.
       const concurrentError = yield* context.service
         .update({ targetVersion: "0.0.30" })
         .pipe(Effect.flip);
       assert.include(concurrentError.reason, "already in progress");
-    }).pipe(Effect.provide(TestClock.layer())),
+    }),
   );
 
-  it.effect("restores the previous unit and permits a retry when systemd restart fails", () =>
+  it.effect("leaves the live unit untouched and permits a retry when the handoff fails", () =>
     Effect.gen(function* () {
-      let failRestart = true;
+      let failHandoff = true;
       const context = yield* makeContext({
         bootService: true,
         failWhen: (command, args) => {
-          if (command !== "systemctl" || args[1] !== "restart" || !failRestart) {
+          if (command !== "systemd-run" || !failHandoff) {
             return false;
           }
-          failRestart = false;
+          assert.include(args, `--unit=${SYSTEMD_SELF_UPDATE_UNIT}`);
+          failHandoff = false;
           return true;
         },
       });
@@ -583,29 +582,46 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => {
       );
       const previousUnit = yield* context.fs.readFileString(unitPath);
 
+      const first = yield* context.service.update({ targetVersion: "0.0.29" }).pipe(Effect.flip);
+      assert.include(first.reason, "Systemd rejected the update handoff");
+      assert.equal(yield* context.fs.readFileString(unitPath), previousUnit);
+
+      const retry = yield* context.service.update({ targetVersion: "0.0.30" });
+      assert.deepEqual(retry, { targetVersion: "0.0.30", method: "boot-service" });
+    }),
+  );
+
+  it.effect("permits a retry when the transient helper stops before replacing this process", () =>
+    Effect.gen(function* () {
+      let helperActive = true;
+      const context = yield* makeContext({
+        bootService: true,
+        failWhen: (command, args) =>
+          command === "systemctl" && args.includes("is-active") && !helperActive,
+      });
+
       const first = yield* context.service.update({ targetVersion: "0.0.29" });
       assert.deepEqual(first, { targetVersion: "0.0.29", method: "boot-service" });
-      yield* TestClock.adjust(Duration.seconds(10));
-      yield* eventuallyFileString(unitPath, previousUnit);
-      yield* eventuallyTrue(() => context.commands.at(-1)?.args[1] === "daemon-reload");
-      assert.deepEqual(
-        context.commands.slice(-2).map((entry) => entry.args),
-        [
-          ["--user", "restart", "--no-block", BOOT_SERVICE_UNIT_FILE],
-          ["--user", "daemon-reload"],
-        ],
-      );
+
+      helperActive = false;
+      yield* TestClock.adjust(Duration.seconds(2));
+      yield* Effect.yieldNow;
 
       const retry = yield* context.service.update({ targetVersion: "0.0.30" });
       assert.deepEqual(retry, { targetVersion: "0.0.30", method: "boot-service" });
+      assert.isTrue(
+        context.commands.some(
+          ({ command, args }) => command === "systemctl" && args.includes("is-active"),
+        ),
+      );
     }).pipe(Effect.provide(TestClock.layer())),
   );
 
-  it.effect("restores the previous systemd unit when daemon-reload fails", () =>
+  it.effect("rejects a target that cannot validate the current database", () =>
     Effect.gen(function* () {
       const context = yield* makeContext({
         bootService: true,
-        failWhen: (command) => command === "systemctl",
+        failWhen: (command, args) => command === NODE_PATH && args.includes("_validate-update"),
       });
       const unitPath = context.path.join(
         context.home,
@@ -617,16 +633,15 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => {
       const previousUnit = yield* context.fs.readFileString(unitPath);
 
       const error = yield* context.service.update({ targetVersion: "0.0.29" }).pipe(Effect.flip);
-      assert.include(error.reason, "Reloading systemd units failed");
+      assert.include(error.reason, "not compatible with this server's database");
       assert.equal(yield* context.fs.readFileString(unitPath), previousUnit);
       assert.deepEqual(
         context.commands.map((entry) => entry.command),
-        ["npm", NODE_PATH, "systemctl", "systemctl"],
+        ["npm", NODE_PATH, NODE_PATH],
       );
 
-      yield* TestClock.adjust(Duration.seconds(10));
       assert.lengthOf(context.spawns, 0);
       assert.equal(context.exitCount(), 0);
-    }).pipe(Effect.provide(TestClock.layer())),
+    }),
   );
 });