History

fix(server): supervise background service updates

Theo Browne committed 3 months ago

Commit 90f0df51 parent 0ad91b611 files · +895 −188Browse files

Unified · Split

11 files
  1. apps/server/src/cli/service.ts +38 −1
  2. apps/server/src/cloud/selfUpdate.test.ts +57 −69
  3. apps/server/src/cloud/selfUpdate.ts +83 −88
  4. apps/server/src/cloud/systemdSelfUpdate.test.ts +165 −0
  5. apps/server/src/cloud/systemdSelfUpdate.ts +310 −0
  6. apps/server/src/persistence/Migrations.test.ts +53 −0
  7. apps/server/src/persistence/Migrations.ts +56 −0
  8. docs/internals/server-updates.md +26 −13
  9. docs/user/updating.md +4 −0
  10. packages/client-runtime/src/state/server.test.ts +23 −0
  11. packages/client-runtime/src/state/server.ts +80 −17
apps/server/src/cli/service.ts+38−1
@@ -2,11 +2,14 @@ import * as Console from "effect/Console"; · 1 unmodified line
2import * as Effect from "effect/Effect";2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";3import * as Layer from "effect/Layer";
4import * as Terminal from "effect/Terminal";4import * as Terminal from "effect/Terminal";
5import { Command, GlobalFlag, Prompt } from "effect/unstable/cli";5import { Command, Flag, GlobalFlag, Prompt } from "effect/unstable/cli";
6import { FetchHttpClient } from "effect/unstable/http";
6 7
7import packageJson from "../../package.json" with { type: "json" };8import packageJson from "../../package.json" with { type: "json" };
8import * as BootService from "../cloud/bootService.ts";9import * as BootService from "../cloud/bootService.ts";
10import { applySystemdSelfUpdatePlan } from "../cloud/systemdSelfUpdate.ts";
9import type * as ServerConfig from "../config.ts";11import type * as ServerConfig from "../config.ts";
12import { validateMigrationIdentities } from "../persistence/Migrations.ts";
10import * as ProcessRunner from "../processRunner.ts";13import * as ProcessRunner from "../processRunner.ts";
11import { projectLocationFlags, resolveCliAuthConfig } from "./config.ts";14import { projectLocationFlags, resolveCliAuthConfig } from "./config.ts";
12 15
@@ -142,6 +145,38 @@ const serviceStatusCommand = Command.make("status", projectLocationFlags).pipe( · 129 unmodified lines
142 ),145 ),
143);146);
144 147
148/** Internal entrypoint launched by systemd-run outside t3code.service. */
149const serviceApplyUpdateCommand = Command.make("_apply-update", {
150 plan: Flag.string("plan"),
151}).pipe(
152 Command.withHidden,
153 Command.withHandler(({ plan }) =>
154 applySystemdSelfUpdatePlan(plan).pipe(
155 Effect.provide(Layer.mergeAll(ProcessRunner.layer, FetchHttpClient.layer)),
156 ),
157 ),
158);
159
160/** Target-artifact preflight used before the running server is stopped. */
161const serviceValidateUpdateCommand = Command.make("_validate-update", {
162 database: Flag.string("database"),
163}).pipe(
164 Command.withHidden,
165 Command.withHandler(({ database }) =>
166 Effect.gen(function* () {
167 // Keep node:sqlite behind the same lazy bundle boundary as normal server
168 // startup; most service commands never open the database.
169 const NodeSqliteClient = yield* Effect.promise(
170 () => import("../persistence/NodeSqliteClient.ts"),
171 );
172 yield* validateMigrationIdentities().pipe(
173 Effect.scoped,
174 Effect.provide(NodeSqliteClient.layer({ filename: database, readonly: true })),
175 );
176 }),
177 ),
178);
179
145export const offerServiceDuringOnboarding = Effect.gen(function* () {180export const offerServiceDuringOnboarding = Effect.gen(function* () {
146 const service = yield* BootService.BootService;181 const service = yield* BootService.BootService;
147 const { supported, installed, current } = yield* service.status;182 const { supported, installed, current } = yield* service.status;
@@ -195,5 +230,7 @@ export const serviceCommand = Command.make("service").pipe( · 47 unmodified lines
195 serviceUninstallCommand,230 serviceUninstallCommand,
196 serviceUpdateCommand,231 serviceUpdateCommand,
197 serviceStatusCommand,232 serviceStatusCommand,
233 serviceApplyUpdateCommand,
234 serviceValidateUpdateCommand,
198 ]),235 ]),
199);236);
apps/server/src/cloud/selfUpdate.test.ts+57−69
@@ -5,6 +5,7 @@ import * as Effect from "effect/Effect"; · 4 unmodified lines
5import * as FileSystem from "effect/FileSystem";5import * as FileSystem from "effect/FileSystem";
6import * as Layer from "effect/Layer";6import * as Layer from "effect/Layer";
7import * as Path from "effect/Path";7import * as Path from "effect/Path";
8import * as Schema from "effect/Schema";
8import * as TestClock from "effect/testing/TestClock";9import * as TestClock from "effect/testing/TestClock";
9import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";10import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
10 11
@@ -23,36 +24,15 @@ import { · 12 unmodified lines
23 renderBootServiceUnit,24 renderBootServiceUnit,
24} from "./bootService.ts";25} from "./bootService.ts";
25import * as SelfUpdate from "./selfUpdate.ts";26import * as SelfUpdate from "./selfUpdate.ts";
27import {
28 SYSTEMD_SELF_UPDATE_DIRECTORY,
29 SYSTEMD_SELF_UPDATE_PLAN_FILE,
30 SYSTEMD_SELF_UPDATE_UNIT,
31 SystemdSelfUpdatePlan,
32} from "./systemdSelfUpdate.ts";
26 33
27const NODE_PATH = "/usr/local/bin/node";34const NODE_PATH = "/usr/local/bin/node";
28 35
29const eventuallyFileString = Effect.fn("test.eventuallyFileString")(function* (
30 filePath: string,
31 expected: string,
32) {
33 const fs = yield* FileSystem.FileSystem;
34 for (let iteration = 0; iteration < 1_000; iteration += 1) {
35 const contents = yield* fs.readFileString(filePath);
36 if (contents === expected) {
37 return;
38 }
39 // The rollback performs real filesystem I/O on a detached fiber, which
40 // advancing TestClock does not await.
41 yield* Effect.yieldNow;
42 }
43 return yield* Effect.die(new Error(`Expected file contents were not observed at ${filePath}.`));
44});
45
46const eventuallyTrue = Effect.fn("test.eventuallyTrue")(function* (predicate: () => boolean) {
47 for (let iteration = 0; iteration < 1_000; iteration += 1) {
48 if (predicate()) {
49 return;
50 }
51 yield* Effect.yieldNow;
52 }
53 return yield* Effect.die(new Error("Expected condition was not observed."));
54});
55
56interface RecordedCommand {36interface RecordedCommand {
57 readonly command: string;37 readonly command: string;
58 readonly args: ReadonlyArray<string>;38 readonly args: ReadonlyArray<string>;
@@ -445,7 +425,7 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { · 386 unmodified lines
445 assert.deepEqual(result, { targetVersion: "0.0.29", method: "respawn" });425 assert.deepEqual(result, { targetVersion: "0.0.29", method: "respawn" });
446 assert.deepEqual(426 assert.deepEqual(
447 context.commands.map((entry) => entry.command),427 context.commands.map((entry) => entry.command),
448 [NODE_PATH, "npm", NODE_PATH],428 [NODE_PATH, "npm", NODE_PATH, NODE_PATH],
449 );429 );
450 }).pipe(Effect.provide(TestClock.layer())),430 }).pipe(Effect.provide(TestClock.layer())),
451 );431 );
@@ -506,6 +486,7 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { · 54 unmodified lines
506 [486 [
507 `npm install --prefix ${context.path.join(context.baseDir, "runtime/versions/0.0.29")} --no-fund --no-audit t3@0.0.29`,487 `npm install --prefix ${context.path.join(context.baseDir, "runtime/versions/0.0.29")} --no-fund --no-audit t3@0.0.29`,
508 `${NODE_PATH} ${pinnedEntry} --version`,488 `${NODE_PATH} ${pinnedEntry} --version`,
489 `${NODE_PATH} ${pinnedEntry} service _validate-update --database ${context.path.join(context.baseDir, "userdata/state.sqlite")}`,
509 ],490 ],
510 );491 );
511 492
@@ -521,9 +502,17 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { · 9 unmodified lines
521 }).pipe(Effect.provide(TestClock.layer())),502 }).pipe(Effect.provide(TestClock.layer())),
522 );503 );
523 504
524 it.effect("rewrites the systemd unit and restarts the boot service", () =>505 it.effect("stages the systemd unit and hands activation to a transient service", () =>
525 Effect.gen(function* () {506 Effect.gen(function* () {
526 const context = yield* makeContext({ bootService: true });507 const context = yield* makeContext({ bootService: true });
508 const unitPath = context.path.join(
509 context.home,
510 ".config",
511 "systemd",
512 "user",
513 BOOT_SERVICE_UNIT_FILE,
514 );
515 const previousUnit = yield* context.fs.readFileString(unitPath);
527 const result = yield* context.service.update({ targetVersion: "0.0.29" });516 const result = yield* context.service.update({ targetVersion: "0.0.29" });
528 assert.deepEqual(result, { targetVersion: "0.0.29", method: "boot-service" });517 assert.deepEqual(result, { targetVersion: "0.0.29", method: "boot-service" });
529 518
@@ -531,46 +520,55 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { · 1 unmodified line
531 context.baseDir,520 context.baseDir,
532 "runtime/versions/0.0.29/node_modules/t3/dist/bin.mjs",521 "runtime/versions/0.0.29/node_modules/t3/dist/bin.mjs",
533 );522 );
534 const unit = yield* context.fs.readFileString(523 // The RPC process does not rewrite the unit it is currently running
535 context.path.join(context.home, ".config", "systemd", "user", "t3code.service"),524 // under. The independent transient service applies the staged unit.
525 assert.equal(yield* context.fs.readFileString(unitPath), previousUnit);
526 const planPath = context.path.join(
527 context.baseDir,
528 "userdata",
529 SYSTEMD_SELF_UPDATE_DIRECTORY,
530 SYSTEMD_SELF_UPDATE_PLAN_FILE,
531 );
532 const plan = yield* context.fs
533 .readFileString(planPath)
534 .pipe(
535 Effect.flatMap(Schema.decodeUnknownEffect(Schema.fromJsonString(SystemdSelfUpdatePlan))),
536 );
537 assert.equal(plan.previousUnit, previousUnit);
538 assert.include(plan.nextUnit, `ExecStart=${NODE_PATH} ${pinnedEntry} serve`);
539 assert.equal(
540 plan.runtimeStatePath,
541 context.path.join(context.baseDir, "userdata/server-runtime.json"),
536 );542 );
537 assert.include(unit, `ExecStart=${NODE_PATH} ${pinnedEntry} serve`);
538 assert.deepEqual(543 assert.deepEqual(
539 context.commands.map((entry) => entry.command),544 context.commands.map((entry) => entry.command),
540 ["npm", NODE_PATH, "systemctl"],545 ["npm", NODE_PATH, NODE_PATH, "systemd-run"],
541 );546 );
542 assert.deepEqual(context.commands[2]?.args, ["--user", "daemon-reload"]);547 assert.include(context.commands[3]?.args ?? [], `--unit=${SYSTEMD_SELF_UPDATE_UNIT}`);
543 548 assert.include(context.commands[3]?.args ?? [], planPath);
544 // Restart waits until after the update acknowledgement can flush.
545 yield* TestClock.adjust(Duration.seconds(10));
546 assert.deepEqual(context.commands[3], {
547 command: "systemctl",
548 args: ["--user", "restart", "--no-block", "t3code.service"],
549 });
550 assert.lengthOf(context.spawns, 0);549 assert.lengthOf(context.spawns, 0);
551 // systemd replaces the process; the server must not exit itself.
552 assert.equal(context.exitCount(), 0);550 assert.equal(context.exitCount(), 0);
553 551
554 // The queued restart returns while this process is still shutting552 // The lock stays held after a successful handoff because the transient
555 // down; the lock must stay held so a second update cannot rewrite the553 // service is now responsible for replacing this process.
556 // unit mid-teardown.
557 const concurrentError = yield* context.service554 const concurrentError = yield* context.service
558 .update({ targetVersion: "0.0.30" })555 .update({ targetVersion: "0.0.30" })
559 .pipe(Effect.flip);556 .pipe(Effect.flip);
560 assert.include(concurrentError.reason, "already in progress");557 assert.include(concurrentError.reason, "already in progress");
561 }).pipe(Effect.provide(TestClock.layer())),558 }),
562 );559 );
563 560
564 it.effect("restores the previous unit and permits a retry when systemd restart fails", () =>561 it.effect("leaves the live unit untouched and permits a retry when the handoff fails", () =>
565 Effect.gen(function* () {562 Effect.gen(function* () {
566 let failRestart = true;563 let failHandoff = true;
567 const context = yield* makeContext({564 const context = yield* makeContext({
568 bootService: true,565 bootService: true,
569 failWhen: (command, args) => {566 failWhen: (command, args) => {
570 if (command !== "systemctl" || args[1] !== "restart" || !failRestart) {567 if (command !== "systemd-run" || !failHandoff) {
571 return false;568 return false;
572 }569 }
573 failRestart = false;570 assert.include(args, `--unit=${SYSTEMD_SELF_UPDATE_UNIT}`);
571 failHandoff = false;
574 return true;572 return true;
575 },573 },
576 });574 });
@@ -583,29 +581,20 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { · 6 unmodified lines
583 );581 );
584 const previousUnit = yield* context.fs.readFileString(unitPath);582 const previousUnit = yield* context.fs.readFileString(unitPath);
585 583
586 const first = yield* context.service.update({ targetVersion: "0.0.29" });584 const first = yield* context.service.update({ targetVersion: "0.0.29" }).pipe(Effect.flip);
587 assert.deepEqual(first, { targetVersion: "0.0.29", method: "boot-service" });585 assert.include(first.reason, "Systemd rejected the update handoff");
588 yield* TestClock.adjust(Duration.seconds(10));586 assert.equal(yield* context.fs.readFileString(unitPath), previousUnit);
589 yield* eventuallyFileString(unitPath, previousUnit);
590 yield* eventuallyTrue(() => context.commands.at(-1)?.args[1] === "daemon-reload");
591 assert.deepEqual(
592 context.commands.slice(-2).map((entry) => entry.args),
593 [
594 ["--user", "restart", "--no-block", BOOT_SERVICE_UNIT_FILE],
595 ["--user", "daemon-reload"],
596 ],
597 );
598 587
599 const retry = yield* context.service.update({ targetVersion: "0.0.30" });588 const retry = yield* context.service.update({ targetVersion: "0.0.30" });
600 assert.deepEqual(retry, { targetVersion: "0.0.30", method: "boot-service" });589 assert.deepEqual(retry, { targetVersion: "0.0.30", method: "boot-service" });
601 }).pipe(Effect.provide(TestClock.layer())),590 }),
602 );591 );
603 592
604 it.effect("restores the previous systemd unit when daemon-reload fails", () =>593 it.effect("rejects a target that cannot validate the current database", () =>
605 Effect.gen(function* () {594 Effect.gen(function* () {
606 const context = yield* makeContext({595 const context = yield* makeContext({
607 bootService: true,596 bootService: true,
608 failWhen: (command) => command === "systemctl",597 failWhen: (command, args) => command === NODE_PATH && args.includes("_validate-update"),
609 });598 });
610 const unitPath = context.path.join(599 const unitPath = context.path.join(
611 context.home,600 context.home,
@@ -617,16 +606,15 @@ it.layer(NodeServices.layer)("ServerSelfUpdate.update", (it) => { · 5 unmodified lines
617 const previousUnit = yield* context.fs.readFileString(unitPath);606 const previousUnit = yield* context.fs.readFileString(unitPath);
618 607
619 const error = yield* context.service.update({ targetVersion: "0.0.29" }).pipe(Effect.flip);608 const error = yield* context.service.update({ targetVersion: "0.0.29" }).pipe(Effect.flip);
620 assert.include(error.reason, "Reloading systemd units failed");609 assert.include(error.reason, "not compatible with this server's database");
621 assert.equal(yield* context.fs.readFileString(unitPath), previousUnit);610 assert.equal(yield* context.fs.readFileString(unitPath), previousUnit);
622 assert.deepEqual(611 assert.deepEqual(
623 context.commands.map((entry) => entry.command),612 context.commands.map((entry) => entry.command),
624 ["npm", NODE_PATH, "systemctl", "systemctl"],613 ["npm", NODE_PATH, NODE_PATH],
625 );614 );
626 615
627 yield* TestClock.adjust(Duration.seconds(10));
628 assert.lengthOf(context.spawns, 0);616 assert.lengthOf(context.spawns, 0);
629 assert.equal(context.exitCount(), 0);617 assert.equal(context.exitCount(), 0);
630 }).pipe(Effect.provide(TestClock.layer())),618 }),
631 );619 );
632});620});
apps/server/src/cloud/selfUpdate.ts+83−88
@@ -24,8 +24,8 @@ import * as Layer from "effect/Layer"; · 23 unmodified lines
24import * as Path from "effect/Path";24import * as Path from "effect/Path";
25import * as Ref from "effect/Ref";25import * as Ref from "effect/Ref";
26 26
27import packageJson from "../../package.json" with { type: "json" };
27import * as ServerConfig from "../config.ts";28import * as ServerConfig from "../config.ts";
28import { writeFileStringAtomically } from "../atomicWrite.ts";
29import * as ProcessRunner from "../processRunner.ts";29import * as ProcessRunner from "../processRunner.ts";
30import {30import {
31 BOOT_SERVICE_UNIT_ENV,31 BOOT_SERVICE_UNIT_ENV,
@@ -34,6 +34,14 @@ import { · 2 unmodified lines
34 renderBootServiceUnit,34 renderBootServiceUnit,
35} from "./bootService.ts";35} from "./bootService.ts";
36import { ensurePinnedRuntimeInstalled, removePinnedRuntimeInstallation } from "./pinnedRuntime.ts";36import { ensurePinnedRuntimeInstalled, removePinnedRuntimeInstallation } from "./pinnedRuntime.ts";
37import {
38 SYSTEMD_SELF_UPDATE_DIRECTORY,
39 SYSTEMD_SELF_UPDATE_PLAN_FILE,
40 SYSTEMD_SELF_UPDATE_RECEIPT_FILE,
41 SYSTEMD_SELF_UPDATE_UNIT,
42 SystemdSelfUpdatePlan,
43 writeSystemdSelfUpdatePlan,
44} from "./systemdSelfUpdate.ts";
37 45
38/**46/**
39 * Lets a connected client replace this server with another published `t3`47 * Lets a connected client replace this server with another published `t3`
@@ -221,11 +229,6 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* (option · 181 unmodified lines
221 Effect.andThen(restart),229 Effect.andThen(restart),
222 Effect.forkDetach({ startImmediately: true }),230 Effect.forkDetach({ startImmediately: true }),
223 );231 );
224 const writeUnitAtomically = (filePath: string, contents: string) =>
225 writeFileStringAtomically({ filePath, contents }).pipe(
226 Effect.provideService(FileSystem.FileSystem, fs),
227 Effect.provideService(Path.Path, path),
228 );
229 232
230 const update: ServerSelfUpdate["Service"]["update"] = Effect.fn(233 const update: ServerSelfUpdate["Service"]["update"] = Effect.fn(
231 "cloud.server_self_update.update",234 "cloud.server_self_update.update",
@@ -301,6 +304,33 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* (option · 69 unmodified lines
301 );304 );
302 }305 }
303 306
307 // Run this command from the target artifact while the known-good server
308 // is still alive. Older artifacts without the compatibility check, or a
309 // build whose migration identities disagree with this database, cannot
310 // be activated remotely.
311 const databasePreflight = yield* runner
312 .run({
313 command: host.execPath,
314 args: [
315 runtimePaths.entryPath,
316 "service",
317 "_validate-update",
318 "--database",
319 serverConfig.dbPath,
320 ],
321 timeout: PREFLIGHT_TIMEOUT,
322 })
323 .pipe(
324 Effect.mapError((cause) =>
325 failWith(`Could not verify t3@${targetVersion} against this database.`, cause),
326 ),
327 );
328 if (databasePreflight.code !== 0) {
329 return yield* failWith(
330 `The installed t3@${targetVersion} is not compatible with this server's database (exit code ${String(databasePreflight.code)}).`,
331 );
332 }
333
304 if (activeMethod === "boot-service") {334 if (activeMethod === "boot-service") {
305 const homeDir = env.HOME ?? "";335 const homeDir = env.HOME ?? "";
306 const unitPath = path.join(homeDir, ".config", "systemd", "user", BOOT_SERVICE_UNIT_FILE);336 const unitPath = path.join(homeDir, ".config", "systemd", "user", BOOT_SERVICE_UNIT_FILE);
@@ -311,98 +341,63 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* (option · 4 unmodified lines
311 );341 );
312 // Same shape bootService.install writes, so host lifecycle commands342 // Same shape bootService.install writes, so host lifecycle commands
313 // still recognize the unit as current.343 // still recognize the unit as current.
314 const unit = renderBootServiceUnit({344 const nextUnit = renderBootServiceUnit({
315 nodePath: host.execPath,345 nodePath: host.execPath,
316 t3EntryPath: runtimePaths.entryPath,346 t3EntryPath: runtimePaths.entryPath,
317 baseDir: serverConfig.baseDir,347 baseDir: serverConfig.baseDir,
318 logPath: path.join(serverConfig.logsDir, "boot-service.log"),348 logPath: path.join(serverConfig.logsDir, "boot-service.log"),
319 unitPath,349 unitPath,
320 });350 });
321 yield* writeUnitAtomically(unitPath, unit).pipe(351 const updateDirectory = path.join(serverConfig.stateDir, SYSTEMD_SELF_UPDATE_DIRECTORY);
322 Effect.mapError((cause) => failWith("Could not update the systemd unit.", cause)),352 const planPath = path.join(updateDirectory, SYSTEMD_SELF_UPDATE_PLAN_FILE);
323 );353 const plan = new SystemdSelfUpdatePlan({
324 354 version: 1,
325 const reloadSystemd = Effect.fn("cloud.server_self_update.reload_systemd")(function* () {355 fromVersion: packageJson.version,
326 const reload = yield* runner356 targetVersion,
327 .run({ command: "systemctl", args: ["--user", "daemon-reload"] })357 currentPid: process.pid,
328 .pipe(Effect.mapError((cause) => failWith("Could not reload systemd units.", cause)));358 unitPath,
329 if (reload.code !== 0) {359 previousUnit,
330 return yield* failWith(360 nextUnit,
331 `Reloading systemd units failed (exit code ${String(reload.code)}).`,361 runtimeStatePath: serverConfig.serverRuntimeStatePath,
332 );362 receiptPath: path.join(updateDirectory, SYSTEMD_SELF_UPDATE_RECEIPT_FILE),
333 }
334 });363 });
335 364 yield* writeSystemdSelfUpdatePlan(planPath, plan).pipe(
336 yield* reloadSystemd().pipe(365 Effect.provideService(FileSystem.FileSystem, fs),
337 Effect.catch((reloadError) =>366 Effect.provideService(Path.Path, path),
338 writeUnitAtomically(unitPath, previousUnit).pipe(367 Effect.mapError((cause) => failWith("Could not stage the systemd update plan.", cause)),
339 Effect.mapError((rollbackCause) =>
340 failWith("Could not restore the previous systemd unit.", {
341 reloadError,
342 rollbackCause,
343 }),
344 ),
345 // Systemd should still have the old unit in memory after the
346 // failed reload, but retry after restoring in case it applied a
347 // partial update before returning an error.
348 Effect.andThen(reloadSystemd().pipe(Effect.ignore)),
349 Effect.andThen(Effect.fail(reloadError)),
350 ),
351 ),
352 );368 );
353 yield* Effect.logInfo("Server self-update installed; restarting boot service.", {369
370 // The transient unit has its own cgroup, so it survives the restart of
371 // t3code.service and remains responsible for readiness and rollback.
372 const handoff = yield* runner
373 .run({
374 command: "systemd-run",
375 args: [
376 "--user",
377 "--collect",
378 "--service-type=exec",
379 `--unit=${SYSTEMD_SELF_UPDATE_UNIT}`,
380 host.execPath,
381 host.cliEntryPath,
382 "service",
383 "_apply-update",
384 "--plan",
385 planPath,
386 ],
387 timeout: PREFLIGHT_TIMEOUT,
388 })
389 .pipe(
390 Effect.mapError((cause) => failWith("Could not hand the update to systemd.", cause)),
391 );
392 if (handoff.code !== 0) {
393 return yield* failWith(
394 `Systemd rejected the update handoff (exit code ${String(handoff.code)}).`,
395 );
396 }
397 yield* Effect.logInfo("Server self-update staged; systemd will activate it.", {
354 targetVersion,398 targetVersion,
399 planPath,
355 });400 });
356 // Restart after the acknowledgement has had time to cross any relay
357 // hop. --no-block queues the restart job and exits before systemd
358 // stops this unit: a blocking restart's SIGTERM reaches the systemctl
359 // child (it shares this service's cgroup), which read as a restart
360 // failure and rolled the new unit back while the old server finished
361 // shutting down. With the handoff race gone, a non-zero exit or spawn
362 // error means systemd genuinely rejected the job while this process is
363 // still alive, so restoring the previous unit below stays correct.
364 yield* scheduleRestart(
365 Effect.gen(function* () {
366 const restart = yield* runner
367 .run({
368 command: "systemctl",
369 args: ["--user", "restart", "--no-block", BOOT_SERVICE_UNIT_FILE],
370 })
371 .pipe(
372 Effect.mapError((cause) =>
373 failWith("Could not restart the systemd boot service.", cause),
374 ),
375 );
376 if (restart.code !== 0) {
377 return yield* failWith(
378 `Restarting the systemd boot service failed (exit code ${String(restart.code)}).`,
379 );
380 }
381 }).pipe(
382 Effect.catch((restartError) =>
383 writeUnitAtomically(unitPath, previousUnit).pipe(
384 Effect.andThen(reloadSystemd()),
385 Effect.mapError((rollbackError) =>
386 failWith("Could not restore the previous systemd unit.", {
387 restartError,
388 rollbackError,
389 }),
390 ),
391 Effect.andThen(Effect.fail(restartError)),
392 ),
393 ),
394 Effect.catch((error) =>
395 Effect.logError("Server self-update could not restart the boot service.").pipe(
396 Effect.annotateLogs({ targetVersion, error: error.reason }),
397 // Permit a retry only after the failed handoff was rolled
398 // back. A queued restart returns while this process is still
399 // shutting down; releasing the lock then would let a second
400 // update rewrite the unit mid-teardown.
401 Effect.andThen(Ref.set(inFlight, false)),
402 ),
403 ),
404 ),
405 );
406 } else {401 } else {
407 // Spawn the shim before acknowledging the RPC so ENOENT/EACCES and402 // Spawn the shim before acknowledging the RPC so ENOENT/EACCES and
408 // other launch failures leave this server alive and return a useful403 // other launch failures leave this server alive and return a useful
apps/server/src/cloud/systemdSelfUpdate.test.ts+165−0
@@ -0,0 +1,165 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it } from "@effect/vitest";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as FileSystem from "effect/FileSystem";
6import * as Layer from "effect/Layer";
7import * as Path from "effect/Path";
8import * as Schema from "effect/Schema";
9import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
10
11import * as ProcessRunner from "../processRunner.ts";
12import {
13 SystemdSelfUpdateActivationError,
14 SystemdSelfUpdatePlan,
15 SystemdSelfUpdateReceipt,
16 activateSystemdSelfUpdatePlan,
17} from "./systemdSelfUpdate.ts";
18
19interface RecordedCommand {
20 readonly command: string;
21 readonly args: ReadonlyArray<string>;
22}
23
24const makeRunnerLayer = (commands: Array<RecordedCommand>) =>
25 Layer.succeed(
26 ProcessRunner.ProcessRunner,
27 ProcessRunner.ProcessRunner.of({
28 run: (input) =>
29 Effect.sync(() => {
30 commands.push({ command: input.command, args: input.args });
31 return {
32 stdout: "",
33 stderr: "",
34 code: ChildProcessSpawner.ExitCode(0),
35 timedOut: false,
36 stdoutTruncated: false,
37 stderrTruncated: false,
38 };
39 }),
40 }),
41 );
42
43it.layer(NodeServices.layer)("systemd self-update activation", (it) => {
44 const makeFixture = Effect.fn("test.makeSystemdUpdateFixture")(function* () {
45 const fs = yield* FileSystem.FileSystem;
46 const path = yield* Path.Path;
47 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-systemd-update-test-" });
48 const unitPath = path.join(directory, "t3code.service");
49 const receiptPath = path.join(directory, "receipt.json");
50 const previousUnit = "old unit\n";
51 yield* fs.writeFileString(unitPath, previousUnit);
52 return {
53 fs,
54 unitPath,
55 receiptPath,
56 previousUnit,
57 plan: new SystemdSelfUpdatePlan({
58 version: 1,
59 fromVersion: "0.0.28",
60 targetVersion: "0.0.29",
61 currentPid: 123,
62 unitPath,
63 previousUnit,
64 nextUnit: "new unit\n",
65 runtimeStatePath: path.join(directory, "server-runtime.json"),
66 receiptPath,
67 }),
68 };
69 });
70
71 const readReceipt = Effect.fn("test.readSystemdUpdateReceipt")(function* (receiptPath: string) {
72 const fs = yield* FileSystem.FileSystem;
73 return yield* fs
74 .readFileString(receiptPath)
75 .pipe(
76 Effect.flatMap(Schema.decodeUnknownEffect(Schema.fromJsonString(SystemdSelfUpdateReceipt))),
77 );
78 });
79
80 it.effect("activates and verifies the new service from the helper", () =>
81 Effect.gen(function* () {
82 const fixture = yield* makeFixture();
83 const commands: Array<RecordedCommand> = [];
84 const readinessVersions: Array<string> = [];
85
86 yield* activateSystemdSelfUpdatePlan(
87 fixture.plan,
88 (input) => Effect.sync(() => readinessVersions.push(input.expectedVersion)),
89 { restartDelay: Duration.zero },
90 ).pipe(Effect.provide(makeRunnerLayer(commands)));
91
92 assert.equal(yield* fixture.fs.readFileString(fixture.unitPath), "new unit\n");
93 assert.deepEqual(readinessVersions, ["0.0.29"]);
94 assert.deepEqual(
95 commands.map(({ args }) => args),
96 [
97 ["--user", "daemon-reload"],
98 ["--user", "restart", "t3code.service"],
99 ],
100 );
101 assert.equal((yield* readReceipt(fixture.receiptPath)).phase, "healthy");
102 }),
103 );
104
105 it.effect("restores and verifies the previous service when the target never becomes ready", () =>
106 Effect.gen(function* () {
107 const fixture = yield* makeFixture();
108 const commands: Array<RecordedCommand> = [];
109 const readinessVersions: Array<string> = [];
110
111 yield* activateSystemdSelfUpdatePlan(
112 fixture.plan,
113 (input) =>
114 Effect.sync(() => readinessVersions.push(input.expectedVersion)).pipe(
115 Effect.andThen(
116 input.expectedVersion === fixture.plan.targetVersion
117 ? Effect.fail(
118 new SystemdSelfUpdateActivationError({
119 reason: "target did not become ready",
120 }),
121 )
122 : Effect.void,
123 ),
124 ),
125 { restartDelay: Duration.zero },
126 ).pipe(Effect.provide(makeRunnerLayer(commands)));
127
128 assert.equal(yield* fixture.fs.readFileString(fixture.unitPath), fixture.previousUnit);
129 assert.deepEqual(readinessVersions, ["0.0.29", "0.0.28"]);
130 assert.deepEqual(
131 commands.map(({ args }) => args),
132 [
133 ["--user", "daemon-reload"],
134 ["--user", "restart", "t3code.service"],
135 ["--user", "daemon-reload"],
136 ["--user", "reset-failed", "t3code.service"],
137 ["--user", "restart", "t3code.service"],
138 ],
139 );
140 const receipt = yield* readReceipt(fixture.receiptPath);
141 assert.equal(receipt.phase, "rolled-back");
142 assert.equal(receipt.detail, "target did not become ready");
143 }),
144 );
145
146 it.effect("records a recovery failure instead of claiming the update completed", () =>
147 Effect.gen(function* () {
148 const fixture = yield* makeFixture();
149 const commands: Array<RecordedCommand> = [];
150 const error = yield* activateSystemdSelfUpdatePlan(
151 fixture.plan,
152 (input) =>
153 Effect.fail(
154 new SystemdSelfUpdateActivationError({
155 reason: `${input.expectedVersion} did not become ready`,
156 }),
157 ),
158 { restartDelay: Duration.zero },
159 ).pipe(Effect.provide(makeRunnerLayer(commands)), Effect.flip);
160
161 assert.include(error.reason, "previous server could not be restored");
162 assert.equal((yield* readReceipt(fixture.receiptPath)).phase, "recovery-failed");
163 }),
164 );
165});
apps/server/src/cloud/systemdSelfUpdate.ts+310−0
@@ -0,0 +1,310 @@
1import { ExecutionEnvironmentDescriptor } from "@t3tools/contracts";
2import * as DateTime from "effect/DateTime";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as FileSystem from "effect/FileSystem";
6import * as Option from "effect/Option";
7import * as Path from "effect/Path";
8import * as Schedule from "effect/Schedule";
9import * as Schema from "effect/Schema";
10import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http";
11
12import { writeFileStringAtomically } from "../atomicWrite.ts";
13import * as ProcessRunner from "../processRunner.ts";
14import { readPersistedServerRuntimeState } from "../serverRuntimeState.ts";
15import { BOOT_SERVICE_UNIT_FILE } from "./bootService.ts";
16
17export const SYSTEMD_SELF_UPDATE_UNIT = "t3code-self-update.service";
18export const SYSTEMD_SELF_UPDATE_DIRECTORY = "self-update";
19export const SYSTEMD_SELF_UPDATE_PLAN_FILE = "plan.json";
20export const SYSTEMD_SELF_UPDATE_RECEIPT_FILE = "receipt.json";
21
22const DEFAULT_RESTART_DELAY = Duration.seconds(2);
23const DEFAULT_READINESS_TIMEOUT = Duration.seconds(60);
24const DEFAULT_READINESS_INTERVAL = Duration.millis(500);
25const READINESS_REQUEST_TIMEOUT = Duration.seconds(2);
26const WELL_KNOWN_ENVIRONMENT_PATH = "/.well-known/t3/environment";
27
28export class SystemdSelfUpdatePlan extends Schema.Class<SystemdSelfUpdatePlan>(
29 "SystemdSelfUpdatePlan",
30)({
31 version: Schema.Literal(1),
32 fromVersion: Schema.String,
33 targetVersion: Schema.String,
34 currentPid: Schema.Int,
35 unitPath: Schema.String,
36 previousUnit: Schema.String,
37 nextUnit: Schema.String,
38 runtimeStatePath: Schema.String,
39 receiptPath: Schema.String,
40}) {}
41
42export const SystemdSelfUpdateReceipt = Schema.Struct({
43 version: Schema.Literal(1),
44 targetVersion: Schema.String,
45 phase: Schema.Literals(["prepared", "activating", "healthy", "rolled-back", "recovery-failed"]),
46 updatedAt: Schema.String,
47 detail: Schema.optional(Schema.String),
48});
49export type SystemdSelfUpdateReceipt = typeof SystemdSelfUpdateReceipt.Type;
50
51export class SystemdSelfUpdateActivationError extends Schema.TaggedErrorClass<SystemdSelfUpdateActivationError>()(
52 "SystemdSelfUpdateActivationError",
53 {
54 reason: Schema.String,
55 cause: Schema.optional(Schema.Defect()),
56 },
57) {
58 override get message(): string {
59 return this.reason;
60 }
61}
62
63const decodePlan = Schema.decodeUnknownEffect(Schema.fromJsonString(SystemdSelfUpdatePlan));
64const encodePlan = Schema.encodeEffect(Schema.fromJsonString(SystemdSelfUpdatePlan));
65const encodeReceipt = Schema.encodeEffect(Schema.fromJsonString(SystemdSelfUpdateReceipt));
66
67const provideFileServices = <A, E>(
68 effect: Effect.Effect<A, E, FileSystem.FileSystem | Path.Path>,
69 fs: FileSystem.FileSystem,
70 path: Path.Path,
71) =>
72 effect.pipe(
73 Effect.provideService(FileSystem.FileSystem, fs),
74 Effect.provideService(Path.Path, path),
75 );
76
77const writeAtomically = (
78 filePath: string,
79 contents: string,
80 fs: FileSystem.FileSystem,
81 path: Path.Path,
82) => provideFileServices(writeFileStringAtomically({ filePath, contents }), fs, path);
83
84export const writeSystemdSelfUpdatePlan = Effect.fn("systemdSelfUpdate.writePlan")(function* (
85 planPath: string,
86 plan: SystemdSelfUpdatePlan,
87) {
88 const fs = yield* FileSystem.FileSystem;
89 const path = yield* Path.Path;
90 const encoded = yield* encodePlan(plan);
91 yield* writeAtomically(planPath, `${encoded}\n`, fs, path);
92 yield* writeReceipt(plan, "prepared");
93});
94
95const writeReceipt = Effect.fn("systemdSelfUpdate.writeReceipt")(function* (
96 plan: SystemdSelfUpdatePlan,
97 phase: SystemdSelfUpdateReceipt["phase"],
98 detail?: string,
99) {
100 const fs = yield* FileSystem.FileSystem;
101 const path = yield* Path.Path;
102 const now = yield* DateTime.now;
103 const receipt = {
104 version: 1,
105 targetVersion: plan.targetVersion,
106 phase,
107 updatedAt: DateTime.formatIso(now),
108 ...(detail === undefined ? {} : { detail }),
109 } satisfies SystemdSelfUpdateReceipt;
110 const encoded = yield* encodeReceipt(receipt);
111 yield* writeAtomically(plan.receiptPath, `${encoded}\n`, fs, path);
112});
113
114const runSystemctl = Effect.fn("systemdSelfUpdate.runSystemctl")(function* (
115 args: ReadonlyArray<string>,
116 failureDescription: string,
117) {
118 const runner = yield* ProcessRunner.ProcessRunner;
119 const result = yield* runner.run({ command: "systemctl", args }).pipe(
120 Effect.mapError(
121 (cause) =>
122 new SystemdSelfUpdateActivationError({
123 reason: failureDescription,
124 cause,
125 }),
126 ),
127 );
128 if (result.code !== 0) {
129 return yield* new SystemdSelfUpdateActivationError({
130 reason: `${failureDescription} (exit code ${String(result.code)}).`,
131 });
132 }
133});
134
135const installUnit = Effect.fn("systemdSelfUpdate.installUnit")(function* (
136 plan: SystemdSelfUpdatePlan,
137 contents: string,
138) {
139 const fs = yield* FileSystem.FileSystem;
140 const path = yield* Path.Path;
141 yield* writeAtomically(plan.unitPath, contents, fs, path).pipe(
142 Effect.mapError(
143 (cause) =>
144 new SystemdSelfUpdateActivationError({
145 reason: "Could not write the systemd boot service unit.",
146 cause,
147 }),
148 ),
149 );
150 yield* runSystemctl(["--user", "daemon-reload"], "Could not reload systemd units");
151});
152
153const restartBootService = runSystemctl(
154 ["--user", "restart", BOOT_SERVICE_UNIT_FILE],
155 "Could not restart the systemd boot service",
156);
157
158export interface SystemdSelfUpdateReadinessInput {
159 readonly expectedVersion: string;
160 readonly previousPid: number;
161 readonly runtimeStatePath: string;
162}
163
164export const waitForSystemdSelfUpdateReadiness = Effect.fn("systemdSelfUpdate.waitForReadiness")(
165 function* (
166 input: SystemdSelfUpdateReadinessInput,
167 options: {
168 readonly timeout?: Duration.Input;
169 readonly interval?: Duration.Input;
170 } = {},
171 ) {
172 const fs = yield* FileSystem.FileSystem;
173 const path = yield* Path.Path;
174 const client = yield* HttpClient.HttpClient;
175 const attempt = Effect.gen(function* () {
176 const state = yield* provideFileServices(
177 readPersistedServerRuntimeState(input.runtimeStatePath),
178 fs,
179 path,
180 ).pipe(
181 Effect.mapError(
182 (cause) =>
183 new SystemdSelfUpdateActivationError({
184 reason: "Could not read the restarted server's runtime state.",
185 cause,
186 }),
187 ),
188 );
189 if (Option.isNone(state) || state.value.pid === input.previousPid) {
190 return yield* new SystemdSelfUpdateActivationError({
191 reason: "The restarted server has not published fresh runtime state yet.",
192 });
193 }
194
195 const request = HttpClientRequest.get(
196 new URL(WELL_KNOWN_ENVIRONMENT_PATH, state.value.origin).toString(),
197 );
198 const descriptor = yield* client.execute(request).pipe(
199 Effect.timeout(READINESS_REQUEST_TIMEOUT),
200 Effect.flatMap(HttpClientResponse.filterStatusOk),
201 Effect.flatMap(HttpClientResponse.schemaBodyJson(ExecutionEnvironmentDescriptor)),
202 Effect.mapError(
203 (cause) =>
204 new SystemdSelfUpdateActivationError({
205 reason: "The restarted server is not ready yet.",
206 cause,
207 }),
208 ),
209 );
210 if (descriptor.serverVersion !== input.expectedVersion) {
211 return yield* new SystemdSelfUpdateActivationError({
212 reason: `The restarted server reported t3@${descriptor.serverVersion}, expected t3@${input.expectedVersion}.`,
213 });
214 }
215 });
216
217 const ready = yield* attempt.pipe(
218 Effect.retry(Schedule.spaced(options.interval ?? DEFAULT_READINESS_INTERVAL)),
219 Effect.timeoutOption(options.timeout ?? DEFAULT_READINESS_TIMEOUT),
220 );
221 if (Option.isNone(ready)) {
222 return yield* new SystemdSelfUpdateActivationError({
223 reason: `The restarted server did not become ready as t3@${input.expectedVersion}.`,
224 });
225 }
226 },
227);
228
229export const activateSystemdSelfUpdatePlan = Effect.fn("systemdSelfUpdate.activatePlan")(function* <
230 R,
231>(
232 plan: SystemdSelfUpdatePlan,
233 waitForReadiness: (
234 input: SystemdSelfUpdateReadinessInput,
235 ) => Effect.Effect<void, SystemdSelfUpdateActivationError, R>,
236 options: { readonly restartDelay?: Duration.Input } = {},
237) {
238 const activate = Effect.gen(function* () {
239 yield* writeReceipt(plan, "activating").pipe(Effect.ignore);
240 yield* Effect.sleep(options.restartDelay ?? DEFAULT_RESTART_DELAY);
241 yield* installUnit(plan, plan.nextUnit);
242 yield* restartBootService;
243 yield* waitForReadiness({
244 expectedVersion: plan.targetVersion,
245 previousPid: plan.currentPid,
246 runtimeStatePath: plan.runtimeStatePath,
247 });
248 yield* writeReceipt(plan, "healthy").pipe(Effect.ignore);
249 });
250
251 yield* activate.pipe(
252 Effect.catch((activationError) =>
253 Effect.gen(function* () {
254 yield* Effect.logError("Systemd self-update failed; restoring the previous version.").pipe(
255 Effect.annotateLogs({
256 targetVersion: plan.targetVersion,
257 error: activationError.reason,
258 }),
259 );
260 const recovery = Effect.gen(function* () {
261 yield* installUnit(plan, plan.previousUnit);
262 yield* runSystemctl(
263 ["--user", "reset-failed", BOOT_SERVICE_UNIT_FILE],
264 "Could not reset the failed systemd boot service",
265 );
266 yield* restartBootService;
267 yield* waitForReadiness({
268 expectedVersion: plan.fromVersion,
269 previousPid: plan.currentPid,
270 runtimeStatePath: plan.runtimeStatePath,
271 });
272 yield* writeReceipt(plan, "rolled-back", activationError.reason).pipe(Effect.ignore);
273 });
274
275 yield* recovery.pipe(
276 Effect.catch((recoveryError) =>
277 writeReceipt(plan, "recovery-failed", recoveryError.reason).pipe(
278 Effect.ignore,
279 Effect.andThen(
280 Effect.fail(
281 new SystemdSelfUpdateActivationError({
282 reason: "The update failed and the previous server could not be restored.",
283 cause: { activationError, recoveryError },
284 }),
285 ),
286 ),
287 ),
288 ),
289 );
290 }),
291 ),
292 );
293});
294
295export const applySystemdSelfUpdatePlan = Effect.fn("systemdSelfUpdate.applyPlan")(function* (
296 planPath: string,
297) {
298 const fs = yield* FileSystem.FileSystem;
299 const plan = yield* fs.readFileString(planPath).pipe(
300 Effect.flatMap(decodePlan),
301 Effect.mapError(
302 (cause) =>
303 new SystemdSelfUpdateActivationError({
304 reason: `Could not read the staged update plan at ${planPath}.`,
305 cause,
306 }),
307 ),
308 );
309 yield* activateSystemdSelfUpdatePlan(plan, waitForSystemdSelfUpdateReadiness);
310});
apps/server/src/persistence/Migrations.test.ts+53−0
@@ -0,0 +1,53 @@
1import { assert, it } from "@effect/vitest";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as SqlClient from "effect/unstable/sql/SqlClient";
5
6import { MigrationIdentityMismatchError, validateMigrationIdentities } from "./Migrations.ts";
7import * as NodeSqliteClient from "./NodeSqliteClient.ts";
8
9const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory()));
10
11layer("migration identity validation", (it) => {
12 const createTrackingTable = Effect.fn("test.createMigrationTrackingTable")(function* () {
13 const sql = yield* SqlClient.SqlClient;
14 yield* sql`
15 CREATE TABLE IF NOT EXISTS effect_sql_migrations (
16 migration_id integer PRIMARY KEY NOT NULL,
17 name varchar(255) NOT NULL,
18 created_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP
19 )
20 `;
21 yield* sql`DELETE FROM effect_sql_migrations`;
22 });
23
24 it.effect("rejects a reused migration ID before migrations run", () =>
25 Effect.gen(function* () {
26 const sql = yield* SqlClient.SqlClient;
27 yield* createTrackingTable();
28 yield* sql`
29 INSERT INTO effect_sql_migrations (migration_id, name)
30 VALUES (35, 'DifferentMigration')
31 `;
32
33 const error = yield* validateMigrationIdentities().pipe(Effect.flip);
34 assert.instanceOf(error, MigrationIdentityMismatchError);
35 assert.equal(error.migrationId, 35);
36 assert.equal(error.expectedName, "ProjectionThreadTitleRegeneration");
37 assert.equal(error.actualName, "DifferentMigration");
38 }),
39 );
40
41 it.effect("allows migration IDs this build does not know about", () =>
42 Effect.gen(function* () {
43 const sql = yield* SqlClient.SqlClient;
44 yield* createTrackingTable();
45 yield* sql`
46 INSERT INTO effect_sql_migrations (migration_id, name)
47 VALUES (999, 'FutureMigration')
48 `;
49
50 yield* validateMigrationIdentities();
51 }),
52 );
53});
apps/server/src/persistence/Migrations.ts+56−0
@@ -11,6 +11,8 @@ · 10 unmodified lines
11import * as Migrator from "effect/unstable/sql/Migrator";11import * as Migrator from "effect/unstable/sql/Migrator";
12import * as Layer from "effect/Layer";12import * as Layer from "effect/Layer";
13import * as Effect from "effect/Effect";13import * as Effect from "effect/Effect";
14import * as Schema from "effect/Schema";
15import * as SqlClient from "effect/unstable/sql/SqlClient";
14 16
15// Import all migrations statically17// Import all migrations statically
16import Migration0001 from "./Migrations/001_OrchestrationEvents.ts";18import Migration0001 from "./Migrations/001_OrchestrationEvents.ts";
@@ -112,6 +114,59 @@ export const makeMigrationLoader = (throughId?: number) => · 95 unmodified lines
112 */114 */
113const run = Migrator.make({});115const run = Migrator.make({});
114 116
117export class MigrationIdentityMismatchError extends Schema.TaggedErrorClass<MigrationIdentityMismatchError>()(
118 "MigrationIdentityMismatchError",
119 {
120 migrationId: Schema.Int,
121 expectedName: Schema.String,
122 actualName: Schema.String,
123 },
124) {
125 override get message(): string {
126 return `Migration ${String(this.migrationId)} is recorded as '${this.actualName}', but this t3 build expects '${this.expectedName}'.`;
127 }
128}
129
130/**
131 * Refuse to run a build whose migration history disagrees with the database.
132 * Unknown newer IDs are intentionally allowed so a compatible older build can
133 * inspect a database created by a newer release without rewriting its history.
134 */
135export const validateMigrationIdentities = Effect.fn("validateMigrationIdentities")(function* () {
136 const sql = yield* SqlClient.SqlClient;
137 const tables = yield* sql<{ readonly name: string }>`
138 SELECT name
139 FROM sqlite_master
140 WHERE type = 'table' AND name = 'effect_sql_migrations'
141 `;
142 if (tables.length === 0) {
143 return;
144 }
145
146 const expectedNames = new Map<number, string>(
147 migrationEntries.map(([migrationId, name]) => [migrationId, name]),
148 );
149 const persisted = yield* sql<{
150 readonly migration_id: number;
151 readonly name: string;
152 }>`
153 SELECT migration_id, name
154 FROM effect_sql_migrations
155 ORDER BY migration_id
156 `;
157
158 for (const migration of persisted) {
159 const expectedName = expectedNames.get(migration.migration_id);
160 if (expectedName !== undefined && expectedName !== migration.name) {
161 return yield* new MigrationIdentityMismatchError({
162 migrationId: migration.migration_id,
163 expectedName,
164 actualName: migration.name,
165 });
166 }
167 }
168});
169
115export interface RunMigrationsOptions {170export interface RunMigrationsOptions {
116 readonly toMigrationInclusive?: number | undefined;171 readonly toMigrationInclusive?: number | undefined;
117}172}
@@ -129,6 +184,7 @@ export interface RunMigrationsOptions { · 11 unmodified lines
129export const runMigrations = Effect.fn("runMigrations")(function* ({184export const runMigrations = Effect.fn("runMigrations")(function* ({
130 toMigrationInclusive,185 toMigrationInclusive,
131}: RunMigrationsOptions = {}) {186}: RunMigrationsOptions = {}) {
187 yield* validateMigrationIdentities();
132 const executedMigrations = yield* run({ loader: makeMigrationLoader(toMigrationInclusive) });188 const executedMigrations = yield* run({ loader: makeMigrationLoader(toMigrationInclusive) });
133 const migrations = executedMigrations.map(([id, name]) => `${id}_${name}`);189 const migrations = executedMigrations.map(([id, name]) => `${id}_${name}`);
134 yield* migrations.length === 0190 yield* migrations.length === 0
docs/internals/server-updates.md+26−13
@@ -66,10 +66,15 @@ flowchart TD · 65 unmodified lines
66 H --> I[Run version preflight]66 H --> I[Run version preflight]
67 I -->|bad code or version| J[Remove candidate runtime and keep current server]67 I -->|bad code or version| J[Remove candidate runtime and keep current server]
68 I -->|cannot run preflight| J2[Keep candidate and current server]68 I -->|cannot run preflight| J2[Keep candidate and current server]
69 I -->|passes| K{Handoff method}69 I -->|passes| I2[Run target database compatibility check]
70 K -->|boot-service| L[Rewrite and restart T3 systemd unit]70 I2 -->|fails| J2
71 I2 -->|passes| K{Handoff method}
72 K -->|boot-service| L[Stage plan and start transient systemd helper]
71 K -->|respawn| M[Start delayed replacement and exit current process]73 K -->|respawn| M[Start delayed replacement and exit current process]
72 L --> N[Reconnect with fresh backoff]74 L --> L2[Helper activates unit and verifies target]
75 L2 -->|target fails| L3[Restore and verify previous unit]
76 L2 -->|target ready| N[Reconnect with fresh backoff]
77 L3 --> N
73 M --> N78 M --> N
74 N --> O[Replacement publishes ready at target version]79 N --> O[Replacement publishes ready at target version]
75```80```
@@ -84,8 +89,10 @@ The update service permits one update at a time. It installs `t3@<version>` unde · 8 unmodified lines
84successfully. Boot-service setup and self-update share the same process-wide installation lock, so89successfully. Boot-service setup and self-update share the same process-wide installation lock, so
85they cannot mutate a pinned runtime concurrently.90they cannot mutate a pinned runtime concurrently.
86 91
87Before any restart, the current Node executable runs the replacement with `--version`. A failed92Before any restart, the current Node executable runs the replacement with `--version`, then asks
88install, failed preflight, or wrong reported version leaves the current server running.93the target artifact to validate its migration identities against the live database in read-only
94mode. A failed install, version preflight, or compatibility check leaves the current server running.
95Targets old enough to lack the compatibility command are not eligible for remote activation.
89 96
90Candidate cleanup is narrower than "any failed preflight". The candidate runtime is removed only when97Candidate cleanup is narrower than "any failed preflight". The candidate runtime is removed only when
91the preflight process actually completes and reports a bad exit code or the wrong version: that is98the preflight process actually completes and reports a bad exit code or the wrong version: that is
@@ -107,10 +114,14 @@ authorization; it does not uninstall the host service. · 15 unmodified lines
107 114
108## Process Handoff115## Process Handoff
109 116
110For `boot-service`, the server atomically rewrites the T3-managed user unit to point at the verified117For `boot-service`, the request process writes a typed plan but leaves its live unit untouched. It
111runtime and reloads systemd. It acknowledges the handoff, then restarts the unit after the same118starts a fixed-name `t3code-self-update.service` transient unit with `systemd-run --user`; the
112short grace period used by foreground respawn. A rejected deferred restart restores the previous119transient unit has a separate cgroup and acts as the cross-process update lock. After a short grace
113unit and is logged by the still-running process.120period it atomically installs the candidate unit, reloads systemd, performs a blocking restart, and
121probes the public environment descriptor for the exact target version. If activation or readiness
122fails, the helper restores the previous unit, clears systemd's failed state, restarts it, and proves
123the previous version is ready. The plan and latest receipt live under the environment state
124directory's `self-update` folder for diagnosis.
114 125
115For `respawn`, the server starts a detached, delayed replacement that replays the original CLI126For `respawn`, the server starts a detached, delayed replacement that replays the original CLI
116arguments. It then acknowledges the request and schedules the current process to exit. The delays127arguments. It then acknowledges the request and schedules the current process to exit. The delays
@@ -118,10 +129,10 @@ give the acknowledgement time to cross direct or relayed connections before the · 1 unmodified line
118 129
119Progress-capable servers emit `downloading` before installing the pinned runtime and `installing`130Progress-capable servers emit `downloading` before installing the pinned runtime and `installing`
120before preflight and handoff. A terminal stream event acknowledges that restart is scheduled. The131before preflight and handoff. A terminal stream event acknowledges that restart is scheduled. The
121client then enters `resuming`, waits for the replacement lifecycle stream to publish `ready` with132client then enters `resuming`, waits for the intentional disconnect before accepting a new lifecycle
122the target version, and only then completes the operation. It watches for the intentional133event, and requests one fresh retry to clear historical backoff debt. The first new `ready` event
123disconnect's first backoff state and requests one fresh retry, which clears historical backoff debt134must report the target version. A verified rollback therefore becomes an immediate wrong-version
124without adding a separate reconnect loop.135error instead of looking like a two-minute timeout.
125 136
126## Release Invariant137## Release Invariant
127 138
@@ -134,7 +145,9 @@ the hosted web deployment depends on that release. See [Release Checklist](../op · 6 unmodified lines
134- Capability contract: `packages/contracts/src/environment.ts`145- Capability contract: `packages/contracts/src/environment.ts`
135- Update RPC contract: `packages/contracts/src/server.ts` and `packages/contracts/src/rpc.ts`146- Update RPC contract: `packages/contracts/src/server.ts` and `packages/contracts/src/rpc.ts`
136- Capability detection and handoff: `apps/server/src/cloud/selfUpdate.ts`147- Capability detection and handoff: `apps/server/src/cloud/selfUpdate.ts`
148- Systemd activation, health check, and rollback: `apps/server/src/cloud/systemdSelfUpdate.ts`
137- Host service commands: `apps/server/src/cli/service.ts`149- Host service commands: `apps/server/src/cli/service.ts`
138- Pinned runtime installation: `apps/server/src/cloud/pinnedRuntime.ts`150- Pinned runtime installation: `apps/server/src/cloud/pinnedRuntime.ts`
151- Migration identity validation: `apps/server/src/persistence/Migrations.ts`
139- Client version comparison: `apps/web/src/versionSkew.ts`152- Client version comparison: `apps/web/src/versionSkew.ts`
140- Shared update action: `apps/web/src/components/ServerUpdateAction.tsx`153- Shared update action: `apps/web/src/components/ServerUpdateAction.tsx`
docs/user/updating.md+4−0
@@ -59,6 +59,10 @@ Keep the web or desktop app open while the server restarts. The update completes · 58 unmodified lines
59replacement server reports the requested version and is ready to accept commands. The warning and59replacement server reports the requested version and is ready to accept commands. The warning and
60progress rail then disappear.60progress rail then disappear.
61 61
62When a background-service replacement cannot become ready, T3 Code restores and verifies the
63previous server version automatically. The update remains failed so you can retry after reading the
64reported reason, but the remote server should remain reachable.
65
62If a step fails:66If a step fails:
63 67
641. Retry the offered action once.681. Retry the offered action once.
packages/client-runtime/src/state/server.test.ts+23−0
@@ -34,6 +34,8 @@ import { · 33 unmodified lines
34 resolveServerUpdateProgressResult,34 resolveServerUpdateProgressResult,
35 serverUpdateStateForProgressEvent,35 serverUpdateStateForProgressEvent,
36 serverUpdateStateForServerVersion,36 serverUpdateStateForServerVersion,
37 ServerUpdateWrongVersionError,
38 validateResumedServerVersion,
37} from "./server.ts";39} from "./server.ts";
38 40
39const CONFIG = {41const CONFIG = {
@@ -152,6 +154,27 @@ describe("server state projection", () => { · 112 unmodified lines
152 });154 });
153 });155 });
154 156
157 it.effect("reports the rollback version as soon as the server resumes", () =>
158 Effect.gen(function* () {
159 const error = yield* validateResumedServerVersion({
160 environmentId: TARGET.environmentId,
161 targetVersion: "0.0.31",
162 actualVersion: "0.0.30",
163 }).pipe(Effect.flip);
164
165 expect(error).toBeInstanceOf(ServerUpdateWrongVersionError);
166 expect(error.message).toBe("The server resumed on t3@0.0.30 instead of t3@0.0.31.");
167 }),
168 );
169
170 it.effect("accepts the requested version after reconnect", () =>
171 validateResumedServerVersion({
172 environmentId: TARGET.environmentId,
173 targetVersion: "0.0.31",
174 actualVersion: "0.0.31",
175 }),
176 );
177
155 it("keeps active update state and hides stale failures after a version change", () => {178 it("keeps active update state and hides stale failures after a version change", () => {
156 const running = {179 const running = {
157 status: "running" as const,180 status: "running" as const,
packages/client-runtime/src/state/server.ts+80−17
@@ -86,6 +86,43 @@ export class ServerUpdateResumeTimeoutError extends Schema.TaggedErrorClass<Serv · 85 unmodified lines
86 }86 }
87}87}
88 88
89export class ServerUpdateRestartTimeoutError extends Schema.TaggedErrorClass<ServerUpdateRestartTimeoutError>()(
90 "ServerUpdateRestartTimeoutError",
91 {
92 environmentId: Schema.String,
93 targetVersion: Schema.String,
94 },
95) {
96 override get message(): string {
97 return `The server accepted t3@${this.targetVersion} but did not restart.`;
98 }
99}
100
101export class ServerUpdateWrongVersionError extends Schema.TaggedErrorClass<ServerUpdateWrongVersionError>()(
102 "ServerUpdateWrongVersionError",
103 {
104 environmentId: Schema.String,
105 targetVersion: Schema.String,
106 actualVersion: Schema.String,
107 },
108) {
109 override get message(): string {
110 return `The server resumed on t3@${this.actualVersion} instead of t3@${this.targetVersion}.`;
111 }
112}
113
114export const validateResumedServerVersion = Effect.fn("server.validateResumedUpdateVersion")(
115 function* (input: {
116 readonly environmentId: EnvironmentId;
117 readonly targetVersion: string;
118 readonly actualVersion: string;
119 }) {
120 if (input.actualVersion !== input.targetVersion) {
121 return yield* new ServerUpdateWrongVersionError(input);
122 }
123 },
124);
125
89export class ServerUpdateProgressIncompleteError extends Schema.TaggedErrorClass<ServerUpdateProgressIncompleteError>()(126export class ServerUpdateProgressIncompleteError extends Schema.TaggedErrorClass<ServerUpdateProgressIncompleteError>()(
90 "ServerUpdateProgressIncompleteError",127 "ServerUpdateProgressIncompleteError",
91 {128 {
@@ -446,6 +483,7 @@ export function createServerEnvironmentAtoms<R, E>( · 354 unmodified lines
446 atomRegistry.get(configValueAtom(target.environmentId))?.environment.serverVersion ??483 atomRegistry.get(configValueAtom(target.environmentId))?.environment.serverVersion ??
447 targetVersion;484 targetVersion;
448 let currentStage: ServerUpdateStage = "downloading";485 let currentStage: ServerUpdateStage = "downloading";
486 let transportAlreadyDisconnected = false;
449 atomRegistry.set(stateAtom, {487 atomRegistry.set(stateAtom, {
450 status: "running",488 status: "running",
451 stage: currentStage,489 stage: currentStage,
@@ -500,9 +538,14 @@ export function createServerEnvironmentAtoms<R, E>( · 48 unmodified lines
500 ),538 ),
501 Effect.exit,539 Effect.exit,
502 );540 );
541 const terminalResult = yield* Ref.get(terminal);
542 transportAlreadyDisconnected =
543 Option.isSome(terminalResult) &&
544 Exit.isFailure(streamExit) &&
545 isLegacyUpdateHandoffLoss(streamExit.cause);
503 return yield* resolveServerUpdateProgressResult(546 return yield* resolveServerUpdateProgressResult(
504 targetVersion,547 targetVersion,
505 yield* Ref.get(terminal),548 terminalResult,
506 streamExit,549 streamExit,
507 );550 );
508 })551 })
@@ -521,6 +564,7 @@ export function createServerEnvironmentAtoms<R, E>( · 12 unmodified lines
521 // Older servers can tear down the transport before their564 // Older servers can tear down the transport before their
522 // unary acknowledgement arrives. Treat only that transport565 // unary acknowledgement arrives. Treat only that transport
523 // loss as a handoff, then prove it by waiting for target ready.566 // loss as a handoff, then prove it by waiting for target ready.
567 transportAlreadyDisconnected = true;
524 return { targetVersion, method: selfUpdateMethod };568 return { targetVersion, method: selfUpdateMethod };
525 }569 }
526 return yield* Effect.failCause(exit.cause);570 return yield* Effect.failCause(exit.cause);
@@ -537,26 +581,40 @@ export function createServerEnvironmentAtoms<R, E>( · 10 unmodified lines
537 }),581 }),
538 );582 );
539 583
540 // The update restart is intentional. As soon as the supervisor sees584 // Subscribe before following lifecycle again so a cached ready event
541 // that first failed connection, discard any prior backoff debt and585 // from the still-running old process cannot satisfy the update. The
542 // retry immediately instead of carrying an old 16-second delay.586 // helper waits before restarting, giving this listener time to attach.
543 yield* environmentRegistry.stateChanges(target.environmentId).pipe(587 if (!transportAlreadyDisconnected) {
544 Stream.filter((state) => state.phase === "backoff"),588 const disconnected = yield* environmentRegistry.stateChanges(target.environmentId).pipe(
545 Stream.take(1),589 Stream.filter((state) => state.phase === "backoff"),
546 Stream.runDrain,590 Stream.take(1),
547 Effect.andThen(environmentRegistry.retryNow(target.environmentId)),591 Stream.runHead,
548 Effect.timeoutOption(Duration.seconds(30)),592 Effect.timeoutOption(Duration.seconds(30)),
549 Effect.ignore,593 Effect.map(Option.flatten),
550 Effect.forkChild,594 );
551 );595 if (Option.isNone(disconnected)) {
596 return yield* new ServerUpdateRestartTimeoutError({
597 environmentId: target.environmentId,
598 targetVersion,
599 });
600 }
601 // The update restart is intentional. Discard any prior backoff debt
602 // instead of carrying an old 16-second delay into recovery.
603 yield* environmentRegistry.retryNow(target.environmentId);
604 } else {
605 // Legacy servers may close the RPC before this listener can attach.
606 // If the supervisor is still waiting, clear that delay; if it has
607 // already reconnected, do not interrupt the healthy replacement.
608 const state = yield* environmentRegistry.state(target.environmentId);
609 if (state.phase === "backoff") {
610 yield* environmentRegistry.retryNow(target.environmentId);
611 }
612 }
552 613
553 const resumed = yield* environmentRegistry614 const resumed = yield* environmentRegistry
554 .followStream(target.environmentId, subscribe(WS_METHODS.subscribeServerLifecycle, {}))615 .followStream(target.environmentId, subscribe(WS_METHODS.subscribeServerLifecycle, {}))
555 .pipe(616 .pipe(
556 Stream.filter(617 Stream.filter((event) => event.type === "ready"),
557 (event) =>
558 event.type === "ready" && event.payload.environment.serverVersion === targetVersion,
559 ),
560 Stream.runHead,618 Stream.runHead,
561 Effect.timeoutOption(Duration.seconds(120)),619 Effect.timeoutOption(Duration.seconds(120)),
562 Effect.map(Option.flatten),620 Effect.map(Option.flatten),
@@ -567,6 +625,11 @@ export function createServerEnvironmentAtoms<R, E>( · 4 unmodified lines
567 targetVersion,625 targetVersion,
568 });626 });
569 }627 }
628 yield* validateResumedServerVersion({
629 environmentId: target.environmentId,
630 targetVersion,
631 actualVersion: resumed.value.payload.environment.serverVersion,
632 });
570 633
571 atomRegistry.set(stateAtom, IDLE_SERVER_UPDATE_STATE);634 atomRegistry.set(stateAtom, IDLE_SERVER_UPDATE_STATE);
572 return result;635 return result;