// @effect-diagnostics nodeBuiltinImport:off// @effect-diagnostics globalTimers:off// This file is shipped as a standalone bundle and copied to a stable path by// `t3 service update`. Keep runtime imports limited to Node built-ins.import * as NodeChildProcess from "node:child_process";import * as NodeCrypto from "node:crypto";import * as NodeFS from "node:fs";import * as NodeFSP from "node:fs/promises";import * as NodePath from "node:path"; import type {  PendingServiceUpdate,  ServiceLauncherChildMessage,  ServiceLauncherContext,  ServiceLauncherParentMessage,  ServiceState,  ServiceUpdateRecord,} from "./cloud/serviceProtocol.ts";import {  compareExactServiceVersions,  decodeServiceLauncherChildMessage,  isExactServiceVersion,  parseServiceState,  SERVICE_LAUNCHER_CONTEXT_ENV,  SERVICE_LAUNCHER_PROTOCOL,  SERVICE_STATE_FILE,  SERVICE_STOP_MARKER_FILE,} from "./cloud/serviceProtocol.ts";import { isEntrypoint } from "./entrypoint.ts"; const HANDOFF_DELAY_MS = 2_000;const PREPARED_TIMEOUT_MS = 120_000;const TERMINATE_GRACE_MS = 5_000; type TerminalStatus = "committed" | "rolled-back" | "failed";type ChildRole = "active" | "trial"; interface ManagedChild {  readonly version: string;  role: ChildRole;  readonly process: NodeChildProcess.ChildProcess;} const runtimePaths = (baseDir: string, version: string) => {  const versionDir = NodePath.join(baseDir, "runtime", "versions", version);  return {    versionDir,    entryPath: NodePath.join(versionDir, "node_modules", "t3", "dist", "bin.mjs"),    sentinelPath: NodePath.join(versionDir, ".install-complete"),  };}; /** SQLite persists across the main file plus its WAL and shared-memory sidecars. */const DB_FILE_SUFFIXES = ["", "-wal", "-shm"] as const;const RESTORE_MARKER = ".restore-pending"; const databaseBackupDir = (baseDir: string, updateId: string) =>  NodePath.join(baseDir, "runtime", "db-backup", updateId); const databaseBackupFile = (backupDir: string, suffix: (typeof DB_FILE_SUFFIXES)[number]) =>  NodePath.join(backupDir, suffix === "" ? "database" : `database${suffix}`); async function pathExists(target: string): Promise<boolean> {  try {    await NodeFSP.access(target);    return true;  } catch (cause) {    if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") return false;    throw cause;  }} // Opened read-write: Windows refuses to flush a handle without write access.async function syncFile(filePath: string): Promise<void> {  const handle = await NodeFSP.open(filePath, "r+");  try {    await handle.sync();  } finally {    await handle.close();  }} // Flushes a directory entry so a rename into it survives power loss. Windows// has no directory fsync: the handle opens but sync fails with EPERM, and// NTFS journals the rename on its own.async function syncDirectory(directory: string): Promise<void> {  const handle = await NodeFSP.open(directory, "r");  try {    await handle.sync();  } catch (error) {    if ((error as NodeJS.ErrnoException).code !== "EPERM") throw error;  } finally {    await handle.close();  }} /** * Snapshots the database once per update before the first trial. A completed * backup is never overwritten because a restarted launcher may be looking at * database writes from an earlier attempt by the same trial. */async function backupDatabaseOnce(baseDir: string, pending: PendingServiceUpdate): Promise<void> {  const backupDir = databaseBackupDir(baseDir, pending.id);  if (await pathExists(backupDir)) return;   const stagingDir = `${backupDir}.staging`;  await NodeFSP.rm(stagingDir, { recursive: true, force: true });  await NodeFSP.mkdir(stagingDir, { recursive: true, mode: 0o700 });  try {    for (const suffix of DB_FILE_SUFFIXES) {      const source = `${pending.dbPath}${suffix}`;      if (suffix !== "" && !(await pathExists(source))) continue;      const destination = databaseBackupFile(stagingDir, suffix);      await NodeFSP.copyFile(source, destination);      await syncFile(destination);    }    await NodeFSP.rename(stagingDir, backupDir);    await syncDirectory(NodePath.dirname(backupDir));  } catch (cause) {    await NodeFSP.rm(stagingDir, { recursive: true, force: true }).catch(() => undefined);    throw cause;  }} const restoreMarkerPath = (baseDir: string, updateId: string) =>  NodePath.join(databaseBackupDir(baseDir, updateId), RESTORE_MARKER); const databaseRestorePending = (baseDir: string, pending: PendingServiceUpdate) =>  pathExists(restoreMarkerPath(baseDir, pending.id)); /** Mark rollback before changing live files so launcher recovery cannot boot a partial restore. */async function markDatabaseRestorePending(backupDir: string): Promise<void> {  const markerPath = NodePath.join(backupDir, RESTORE_MARKER);  if (!(await pathExists(markerPath))) {    const handle = await NodeFSP.open(markerPath, "wx", 0o600);    try {      await handle.sync();    } finally {      await handle.close();    }    await syncDirectory(backupDir);  }} /** Restore is retryable after any process crash while the backup directory remains. */async function restoreDatabaseBackup(  baseDir: string,  pending: PendingServiceUpdate,): Promise<void> {  const backupDir = databaseBackupDir(baseDir, pending.id);  if (!(await pathExists(backupDir))) return;   await markDatabaseRestorePending(backupDir);  for (const suffix of DB_FILE_SUFFIXES) {    const target = `${pending.dbPath}${suffix}`;    const source = databaseBackupFile(backupDir, suffix);    if (await pathExists(source)) {      await NodeFSP.copyFile(source, target);      await syncFile(target);    } else {      await NodeFSP.rm(target, { force: true });    }  }  await syncDirectory(NodePath.dirname(pending.dbPath));} async function discardDatabaseBackup(baseDir: string, updateId: string): Promise<void> {  const backupDir = databaseBackupDir(baseDir, updateId);  if (!(await pathExists(backupDir))) return;  await NodeFSP.rm(backupDir, { recursive: true, force: true });  await syncDirectory(NodePath.dirname(backupDir));} export async function readServiceState(filePath: string): Promise<ServiceState> {  const contents = await NodeFSP.readFile(filePath, "utf8");  const state = parseServiceState(contents);  if (state === undefined) throw new Error("Service state is invalid or unsupported.");  return state;} /** Durable same-directory replacement used for every runtime state transition. */export async function writeServiceState(filePath: string, state: ServiceState): Promise<void> {  const directory = NodePath.dirname(filePath);  await NodeFSP.mkdir(directory, { recursive: true, mode: 0o700 });  const tempPath = NodePath.join(    directory,    `.${NodePath.basename(filePath)}.${process.pid}.${NodeCrypto.randomUUID()}`,  );  let handle: NodeFSP.FileHandle | undefined;  try {    handle = await NodeFSP.open(tempPath, "wx", 0o600);    await handle.writeFile(`${JSON.stringify(state, null, 2)}\n`, "utf8");    await handle.sync();    await handle.close();    handle = undefined;    await NodeFSP.rename(tempPath, filePath);    await syncDirectory(directory);  } finally {    await handle?.close().catch(() => undefined);    await NodeFSP.rm(tempPath, { force: true }).catch(() => undefined);  }} async function runtimeExists(baseDir: string, version: string): Promise<boolean> {  const paths = runtimePaths(baseDir, version);  try {    const [entry, sentinel] = await Promise.all([      NodeFSP.stat(paths.entryPath),      NodeFSP.readFile(paths.sentinelPath, "utf8"),    ]);    return entry.isFile() && sentinel.trim() === version;  } catch {    return false;  }} function terminalUpdate<S extends TerminalStatus>(input: {  readonly pending: PendingServiceUpdate;  readonly status: S;  readonly reason?: string;}): Exclude<ServiceUpdateRecord, PendingServiceUpdate> & { readonly status: S } {  return {    id: input.pending.id,    fromVersion: input.pending.fromVersion,    targetVersion: input.pending.targetVersion,    status: input.status,    ...(input.reason === undefined ? {} : { reason: input.reason }),  };} function sendMessage(  child: NodeChildProcess.ChildProcess,  message: ServiceLauncherParentMessage,): Promise<void> {  return new Promise((resolve, reject) => {    if (!child.connected || child.send === undefined) {      reject(new Error("service child IPC is disconnected."));      return;    }    child.send(message, (error) => (error === null ? resolve() : reject(error)));  });} function waitForExit(child: NodeChildProcess.ChildProcess): Promise<void> {  if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve();  return new Promise((resolve) => child.once("exit", () => resolve()));} async function terminateChild(  child: NodeChildProcess.ChildProcess,  signal: NodeJS.Signals = "SIGTERM",): Promise<void> {  if (child.exitCode !== null || child.signalCode !== null) return;  child.kill(signal);  const force = setTimeout(() => child.kill("SIGKILL"), TERMINATE_GRACE_MS);  try {    await waitForExit(child);  } finally {    clearTimeout(force);  }} const stopMarkerPath = (baseDir: string) =>  NodePath.join(baseDir, "runtime", SERVICE_STOP_MARKER_FILE); export class Launcher {  readonly #baseDir: string;  readonly #statePath: string;  #state: ServiceState;  #child: ManagedChild | null = null;  #timer: NodeJS.Timeout | undefined;  #transitions: Promise<void> = Promise.resolve();  #stopRequested = false;  #stopping = false;  #done = false;  readonly #completion = Promise.withResolvers<void>();   constructor(baseDir: string, state: ServiceState) {    this.#baseDir = baseDir;    this.#statePath = NodePath.join(baseDir, "runtime", SERVICE_STATE_FILE);    this.#state = state;  }   async run(): Promise<void> {    const onSigterm = () => void this.stop("SIGTERM");    const onSigint = () => void this.stop("SIGINT");    process.once("SIGTERM", onSigterm);    process.once("SIGINT", onSigint);    try {      this.#enqueue(() => this.#recover());      await this.#completion.promise;    } finally {      process.off("SIGTERM", onSigterm);      process.off("SIGINT", onSigint);    }  }   #enqueue(transition: () => Promise<void>): void {    this.#transitions = this.#transitions      .then(transition, transition)      .catch((cause: unknown) =>        this.#fatal(cause instanceof Error ? cause : new Error(String(cause))),      );  }   async #fatal(error: Error): Promise<void> {    if (this.#done) return;    this.#done = true;    this.#stopping = true;    this.#clearTimer();    const child = this.#child?.process;    this.#child = null;    if (child !== undefined) await terminateChild(child);    this.#completion.reject(error);  }   async stop(signal: NodeJS.Signals): Promise<void> {    // This must happen synchronously at signal receipt. A queued update    // transition may already be terminating the active child, and that child    // needs to see the marker in its shutdown finalizer. KillMode=mixed also    // ensures systemd signals the launcher before the rest of the cgroup, and    // launchd signals only the job's main process (this launcher), so the    // marker lands before the child sees any signal on both platforms.    try {      NodeFS.writeFileSync(stopMarkerPath(this.#baseDir), "", { mode: 0o600 });    } catch {      // Err toward keeping the tunnel; the next link or unlink reconciles it.    }    if (this.#stopRequested || this.#stopping) {      await this.#completion.promise.catch(() => undefined);      return;    }    this.#stopRequested = true;    this.#clearTimer();    this.#enqueue(async () => {      // Let an update transition already in progress start its replacement      // before this queued stop tears it down. That replacement owns the      // pre-activation tunnel cleanup path and observes the marker above.      this.#stopping = true;      const child = this.#child?.process;      this.#child = null;      if (child !== undefined) await terminateChild(child, signal);      this.#done = true;      this.#completion.resolve();    });    await this.#completion.promise.catch(() => undefined);  }   #clearTimer(): void {    clearTimeout(this.#timer);    this.#timer = undefined;  }   async #recover(): Promise<void> {    // A fresh launcher means servers are running again: any stop marker from    // a previous explicit stop is stale and must not make a future update    // handoff release its tunnel.    await NodeFSP.rm(stopMarkerPath(this.#baseDir), { force: true }).catch(() => undefined);    const update = this.#state.update;    if (update?.status !== "pending") {      if (update !== undefined) {        await discardDatabaseBackup(this.#baseDir, update.id).catch(() => undefined);      }      await this.#startChild(this.#state.activeVersion, "active", update);      return;    }    if (await databaseRestorePending(this.#baseDir, update)) {      await this.#returnToPrevious(update, "failed", "rollback-interrupted");      return;    }    if (!(await runtimeExists(this.#baseDir, update.targetVersion))) {      await this.#returnToPrevious(update, "failed", "target-runtime-missing");      return;    }    await this.#startTrial(update);  }   async #startTrial(pending: PendingServiceUpdate): Promise<void> {    // The previous child is dead here, so all three SQLite files are quiescent.    try {      await backupDatabaseOnce(this.#baseDir, pending);    } catch {      await this.#returnToPrevious(pending, "failed", "db-backup-failed");      return;    }    try {      await this.#startChild(pending.targetVersion, "trial", pending);    } catch {      await this.#returnToPrevious(pending, "failed", "candidate-start-failed");    }  }   async #startChild(version: string, role: ChildRole, update?: ServiceUpdateRecord): Promise<void> {    if (this.#stopping) return;    if (!(await runtimeExists(this.#baseDir, version))) {      throw new Error(`Selected t3@${version} runtime is missing or incomplete.`);    }    if (this.#stopping) return;    const paths = runtimePaths(this.#baseDir, version);    const context: ServiceLauncherContext = {      protocol: SERVICE_LAUNCHER_PROTOCOL,      childVersion: version,      ...(update === undefined ? {} : { update }),    };    const child = NodeChildProcess.spawn(process.execPath, [paths.entryPath, "serve"], {      env: { ...process.env, [SERVICE_LAUNCHER_CONTEXT_ENV]: JSON.stringify(context) },      stdio: ["inherit", "inherit", "inherit", "ipc"],    });    await new Promise<void>((resolve, reject) => {      const onError = (error: Error) => reject(error);      child.once("error", onError);      child.once("spawn", () => {        child.removeListener("error", onError);        child.on("error", (error) => this.#enqueue(() => Promise.reject(error)));        resolve();      });    });    if (this.#stopping) {      await terminateChild(child);      return;    }     const managed: ManagedChild = {      version,      role,      process: child,    };    this.#child = managed;    child.on("message", (value) => {      const message = decodeServiceLauncherChildMessage(value);      if (message !== undefined) this.#enqueue(() => this.#handleMessage(managed, message));    });    child.once("exit", (code, signal) =>      this.#enqueue(() => this.#handleExit(managed, code, signal)),    );     if (role === "trial") {      this.#timer = setTimeout(        () => this.#enqueue(() => this.#handlePreparedTimeout(managed)),        PREPARED_TIMEOUT_MS,      );    }  }   async #handleMessage(child: ManagedChild, message: ServiceLauncherChildMessage): Promise<void> {    if (this.#child !== child || this.#stopping) return;    if (message.type === "request-update") {      await this.#handleUpdateRequest(child, message);      return;    }    await this.#handlePrepared(child, message.updateId);  }   async #handleUpdateRequest(    child: ManagedChild,    message: Extract<ServiceLauncherChildMessage, { readonly type: "request-update" }>,  ): Promise<void> {    const reject = (reason: string) =>      sendMessage(child.process, { type: "update-rejected", reason });    if (child.role !== "active") {      await reject("Only the active server can request an update.");      return;    }    if (child.version !== this.#state.activeVersion) {      await reject("The requesting server is not the selected active version.");      return;    }    if (this.#state.update?.status === "pending") {      await reject("Another server update is already pending.");      return;    }    if (!isExactServiceVersion(message.targetVersion)) {      await reject("The requested target is not an exact version.");      return;    }    if (compareExactServiceVersions(message.targetVersion, child.version) <= 0) {      await reject("Remote updates must select a newer server version.");      return;    }    if (!NodePath.isAbsolute(message.dbPath)) {      await reject("The requested database path is not absolute.");      return;    }    if (!(await runtimeExists(this.#baseDir, message.targetVersion))) {      await reject("The requested target runtime is missing or incomplete.");      return;    }     const pending: PendingServiceUpdate = {      id: NodeCrypto.randomUUID(),      fromVersion: child.version,      targetVersion: message.targetVersion,      dbPath: message.dbPath,      status: "pending",    };    const next: ServiceState = { ...this.#state, update: pending };    await writeServiceState(this.#statePath, next);    this.#state = next;    await sendMessage(child.process, { type: "update-accepted", updateId: pending.id });    this.#timer = setTimeout(() => this.#enqueue(() => this.#beginTrial(child)), HANDOFF_DELAY_MS);

Lines 1–500 of 627. Load more