diff --git a/scripts/bench-gateway-restart.ts b/scripts/bench-gateway-restart.ts index 062e89d0b5c9..8142d975d875 100644 --- a/scripts/bench-gateway-restart.ts +++ b/scripts/bench-gateway-restart.ts @@ -7,7 +7,7 @@ import path from "node:path"; import { performance } from "node:perf_hooks"; import { pathToFileURL } from "node:url"; import { expectDefined } from "../packages/normalization-core/src/expect.js"; -import { writeGatewayRestartIntentSync } from "../src/infra/restart.js"; +import { writeGatewayRestartIntentSync } from "../src/infra/restart-intent.js"; import { parseStrictIntegerOption } from "./lib/dev-tooling-safety.ts"; import { delay, stopChild, type StopChildResult } from "./lib/gateway-bench-child.ts"; import { diff --git a/src/cli/daemon-cli/lifecycle-core.test.ts b/src/cli/daemon-cli/lifecycle-core.test.ts index df58361c6f0c..44fd87f4c410 100644 --- a/src/cli/daemon-cli/lifecycle-core.test.ts +++ b/src/cli/daemon-cli/lifecycle-core.test.ts @@ -32,7 +32,7 @@ vi.mock("../../runtime.js", () => ({ defaultRuntime, })); -vi.mock("../../infra/restart.js", () => ({ +vi.mock("../../infra/restart-intent.js", () => ({ clearGatewayRestartIntentSync: () => clearGatewayRestartIntentSync(), writeGatewayRestartIntentSync: (opts: unknown) => writeGatewayRestartIntentSync(opts), })); diff --git a/src/cli/daemon-cli/lifecycle-core.ts b/src/cli/daemon-cli/lifecycle-core.ts index 4277a9555875..bae4aee26be0 100644 --- a/src/cli/daemon-cli/lifecycle-core.ts +++ b/src/cli/daemon-cli/lifecycle-core.ts @@ -21,7 +21,7 @@ import { clearGatewayRestartIntentSync, type GatewayRestartIntent, writeGatewayRestartIntentSync, -} from "../../infra/restart.js"; +} from "../../infra/restart-intent.js"; import { isWSL } from "../../infra/wsl.js"; import { defaultRuntime } from "../../runtime.js"; import { formatCliCommand } from "../command-format.js"; diff --git a/src/cli/daemon-cli/lifecycle.ts b/src/cli/daemon-cli/lifecycle.ts index 07d800954898..dc0906467f91 100644 --- a/src/cli/daemon-cli/lifecycle.ts +++ b/src/cli/daemon-cli/lifecycle.ts @@ -19,7 +19,10 @@ import { signalVerifiedGatewayPidSync, } from "../../infra/gateway-processes.js"; import type { SafeGatewayRestartRequestResult } from "../../infra/restart-coordinator.js"; -import { type GatewayRestartIntent, writeGatewayRestartIntentSync } from "../../infra/restart.js"; +import { + type GatewayRestartIntent, + writeGatewayRestartIntentSync, +} from "../../infra/restart-intent.js"; import { defaultRuntime } from "../../runtime.js"; import { formatCliCommand } from "../command-format.js"; import { parseDurationMs } from "../parse-duration.js"; diff --git a/src/cli/gateway-cli/lifecycle.runtime.ts b/src/cli/gateway-cli/lifecycle.runtime.ts index 2bd7f17c2d7e..bd78796ab257 100644 --- a/src/cli/gateway-cli/lifecycle.runtime.ts +++ b/src/cli/gateway-cli/lifecycle.runtime.ts @@ -14,9 +14,7 @@ export { } from "../../infra/process-respawn.js"; export { resolveGatewayRestartDeferralTimeoutMs, - consumeGatewayRestartIntentPayloadSync, consumeGatewaySigusr1RestartIntent, - consumeGatewayRestartIntentSync, consumeGatewaySigusr1RestartAuthorization, isGatewaySigusr1RestartExternallyAllowed, markGatewaySigusr1RestartHandled, @@ -26,6 +24,10 @@ export { rollbackGatewayRestartSignalAdmission, scheduleGatewaySigusr1Restart, } from "../../infra/restart.js"; +export { + consumeGatewayRestartIntentPayloadSync, + consumeGatewayRestartIntentSync, +} from "../../infra/restart-intent.js"; export { writeGatewayRestartHandoffSync } from "../../infra/restart-handoff.js"; export { resetGatewaySuspendCoordinatorForLifecycleRestart } from "../../infra/gateway-suspend-coordinator.js"; export { rotateAgentEventLifecycleGeneration } from "../../infra/agent-events.js"; diff --git a/src/cli/gateway-cli/run-loop.test.ts b/src/cli/gateway-cli/run-loop.test.ts index c978ea21d54e..a0fb353e4a89 100644 --- a/src/cli/gateway-cli/run-loop.test.ts +++ b/src/cli/gateway-cli/run-loop.test.ts @@ -129,10 +129,8 @@ vi.mock("../../infra/gateway-lock.js", () => ({ })); vi.mock("../../infra/restart.js", () => ({ - consumeGatewayRestartIntentPayloadSync: () => consumeGatewayRestartIntentPayloadSync(), consumeGatewaySigusr1RestartIntent: () => consumeGatewaySigusr1RestartIntent(), consumeGatewaySigusr1RestartAuthorization: () => consumeGatewaySigusr1RestartAuthorization(), - consumeGatewayRestartIntentSync: () => consumeGatewayRestartIntentSync(), isGatewaySigusr1RestartExternallyAllowed: () => isGatewaySigusr1RestartExternallyAllowed(), markGatewaySigusr1RestartHandled: () => markGatewaySigusr1RestartHandled(), peekGatewaySigusr1RestartReason: () => peekGatewaySigusr1RestartReason(), @@ -152,6 +150,11 @@ vi.mock("../../infra/restart.js", () => ({ scheduleGatewaySigusr1Restart(opts), })); +vi.mock("../../infra/restart-intent.js", () => ({ + consumeGatewayRestartIntentPayloadSync: () => consumeGatewayRestartIntentPayloadSync(), + consumeGatewayRestartIntentSync: () => consumeGatewayRestartIntentSync(), +})); + vi.mock("../../infra/gateway-suspend-coordinator.js", () => ({ resetGatewaySuspendCoordinatorForLifecycleRestart: () => resetGatewaySuspendCoordinatorForLifecycleRestart(), diff --git a/src/cron/service.every-jobs-fire.test.ts b/src/cron/service.every-jobs-fire.test.ts index 8c3c33fb806d..769a883889fc 100644 --- a/src/cron/service.every-jobs-fire.test.ts +++ b/src/cron/service.every-jobs-fire.test.ts @@ -244,11 +244,12 @@ describe("CronService interval/cron jobs fire on time", () => { }); const pendingSignal = beginGatewayRestartSignalAdmission(); + expect(pendingSignal).not.toBeNull(); const finishedRun = finished.waitForOk(job.id); await vi.advanceTimersByTimeAsync(10_005); expect(enqueueSystemEvent).not.toHaveBeenCalled(); - expect(pendingSignal.rollback()).toBe(true); + expect(pendingSignal?.rollback()).toBe(true); await finishedRun; expectMainSystemEvent(enqueueSystemEvent, "rollback-tick", job.id); } finally { diff --git a/src/gateway/server-reload-handlers.ts b/src/gateway/server-reload-handlers.ts index 3f317bd2a2c8..a40c3f1d0a44 100644 --- a/src/gateway/server-reload-handlers.ts +++ b/src/gateway/server-reload-handlers.ts @@ -24,10 +24,10 @@ import { isTruthyEnvValue } from "../infra/env.js"; import { formatErrorMessage } from "../infra/errors.js"; import type { HeartbeatRunner } from "../infra/heartbeat-runner.js"; import { resetDirectoryCache } from "../infra/outbound/target-resolver.js"; +import type { GatewayRestartIntent } from "../infra/restart-intent.js"; import { deferGatewayRestartUntilIdle, type GatewayRestartEmitter, - type GatewayRestartIntent, type RestartDeferralHandle, resolveGatewayRestartDeferralTimeoutMs, setGatewaySigusr1RestartPolicy, diff --git a/src/infra/infra-runtime.test.ts b/src/infra/infra-runtime.test.ts index 4d90ca7e8bbc..8816ed02349c 100644 --- a/src/infra/infra-runtime.test.ts +++ b/src/infra/infra-runtime.test.ts @@ -3,6 +3,7 @@ import { importFreshModule } from "openclaw/plugin-sdk/test-fixtures"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { clearRuntimeConfigSnapshot, setRuntimeConfigSnapshot } from "../config/config.js"; import { + beginGatewayRestartSignalAdmission, isGatewayWorkAdmissionClosed, resetGatewayWorkAdmission, tryBeginGatewayRootWorkAdmission, @@ -16,6 +17,7 @@ let isGatewaySigusr1RestartExternallyAllowed: RestartModule["isGatewaySigusr1Res let markGatewaySigusr1RestartHandled: RestartModule["markGatewaySigusr1RestartHandled"]; let peekGatewaySigusr1RestartReason: RestartModule["peekGatewaySigusr1RestartReason"]; let requestGatewayRestartWithSignalAdmission: RestartModule["requestGatewayRestartWithSignalAdmission"]; +let rollbackGatewayRestartSignalAdmission: RestartModule["rollbackGatewayRestartSignalAdmission"]; let scheduleGatewaySigusr1Restart: RestartModule["scheduleGatewaySigusr1Restart"]; let setGatewaySigusr1RestartPolicy: RestartModule["setGatewaySigusr1RestartPolicy"]; let setPreRestartDeferralCheck: RestartModule["setPreRestartDeferralCheck"]; @@ -106,6 +108,7 @@ describe("infra runtime", () => { markGatewaySigusr1RestartHandled, peekGatewaySigusr1RestartReason, requestGatewayRestartWithSignalAdmission, + rollbackGatewayRestartSignalAdmission, scheduleGatewaySigusr1Restart, setGatewaySigusr1RestartPolicy, setPreRestartDeferralCheck, @@ -172,6 +175,128 @@ describe("infra runtime", () => { } }); + it("reopens admission when refused-handler rollback finds no live emission lease", () => { + // Fence closed outside restart.ts ownership (lost/overwritten lease). + const orphanLease = beginGatewayRestartSignalAdmission(); + expect(orphanLease).not.toBeNull(); + expect(isGatewayWorkAdmissionClosed()).toBe(true); + + // Run-loop refused path: mark handled / explicit rollback with no stored lease. + expect(rollbackGatewayRestartSignalAdmission()).toBe(true); + expect(isGatewayWorkAdmissionClosed()).toBe(false); + expect(orphanLease?.rollback()).toBe(false); + + const root = tryBeginGatewayRootWorkAdmission(); + expect(root).not.toBeNull(); + root?.release(); + }); + + it("does not leave admission closed when a deferred emission is cancelled mid-prepare", async () => { + let releasePrepare: (() => void) | undefined; + const prepareGate = new Promise((resolve) => { + releasePrepare = resolve; + }); + const handle = deferGatewayRestartUntilIdle({ + getPendingCount: () => 0, + reason: "config.reload.cancelled", + emitHooks: { + beforeEmit: async () => { + await prepareGate; + }, + }, + }); + await Promise.resolve(); + expect(isGatewayWorkAdmissionClosed()).toBe(true); + + handle.cancel(); + releasePrepare?.(); + await vi.advanceTimersByTimeAsync(0); + await Promise.resolve(); + await Promise.resolve(); + + expect(isGatewayWorkAdmissionClosed()).toBe(false); + const root = tryBeginGatewayRootWorkAdmission(); + expect(root).not.toBeNull(); + root?.release(); + }); + + it("keeps admission open when a deferred restart emission races config supersession", async () => { + let pending = 1; + let releasePrepare: (() => void) | undefined; + const prepareGate = new Promise((resolve) => { + releasePrepare = resolve; + }); + const handle = deferGatewayRestartUntilIdle({ + getPendingCount: () => pending, + reason: "config.reload.superseded", + emitHooks: { + beforeEmit: async () => { + await prepareGate; + }, + emitRestart: () => ({ status: "coalesced" as const }), + }, + }); + expect(isGatewayWorkAdmissionClosed()).toBe(false); + + pending = 0; + await vi.advanceTimersByTimeAsync(500); + await Promise.resolve(); + expect(isGatewayWorkAdmissionClosed()).toBe(true); + + // Superseding reload cancels the in-flight emission before signal delivery. + handle.cancel(); + releasePrepare?.(); + await Promise.resolve(); + await Promise.resolve(); + + expect(isGatewayWorkAdmissionClosed()).toBe(false); + }); + + it("keeps the signal fence closed when cancel races a concurrent emitted SIGUSR1", async () => { + let releasePrepare: (() => void) | undefined; + const prepareGate = new Promise((resolve) => { + releasePrepare = resolve; + }); + const handler = () => {}; + process.on("SIGUSR1", handler); + try { + const handle = deferGatewayRestartUntilIdle({ + getPendingCount: () => 0, + reason: "config.reload.shared-fence", + emitHooks: { + beforeEmit: async () => { + await prepareGate; + }, + }, + }); + await Promise.resolve(); + expect(isGatewayWorkAdmissionClosed()).toBe(true); + + // Concurrent path reuses the deferred prepare lease and queues SIGUSR1. + expect(requestGatewayRestartWithSignalAdmission("concurrent.emit")).toEqual({ + status: "emitted", + }); + expect(isGatewayWorkAdmissionClosed()).toBe(true); + + handle.cancel(); + expect(isGatewayWorkAdmissionClosed()).toBe(true); + expect(tryBeginGatewayRootWorkAdmission()).toBeNull(); + + releasePrepare?.(); + await Promise.resolve(); + await Promise.resolve(); + + // In-flight signal still owns the fence until the handled path reopens it. + expect(isGatewayWorkAdmissionClosed()).toBe(true); + expect(tryBeginGatewayRootWorkAdmission()).toBeNull(); + + markGatewaySigusr1RestartHandled(); + expect(isGatewayWorkAdmissionClosed()).toBe(false); + } finally { + process.removeListener("SIGUSR1", handler); + } + }); + it("backs off before an emoji that crosses the restart reason limit", () => { const restart = scheduleGatewaySigusr1Restart({ delayMs: 0, diff --git a/src/infra/restart-intent.test.ts b/src/infra/restart-intent.test.ts index 4a084a2a9204..777b0d1b47c3 100644 --- a/src/infra/restart-intent.test.ts +++ b/src/infra/restart-intent.test.ts @@ -17,7 +17,7 @@ import { consumeGatewayRestartIntentPayloadSync, consumeGatewayRestartIntentSync, writeGatewayRestartIntentSync, -} from "./restart.js"; +} from "./restart-intent.js"; const tempDirs: string[] = []; type GatewayRestartIntentDatabase = Pick; diff --git a/src/infra/restart-intent.ts b/src/infra/restart-intent.ts new file mode 100644 index 000000000000..e2123224d76e --- /dev/null +++ b/src/infra/restart-intent.ts @@ -0,0 +1,192 @@ +// Persists short-lived gateway restart intent for supervisor SIGTERM handoff. +import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; +import { createSubsystemLogger } from "../logging/subsystem.js"; +import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; +import { + openOpenClawStateDatabase, + runOpenClawStateWriteTransaction, +} from "../state/openclaw-state-db.js"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, + getNodeSqliteKysely, +} from "./kysely-sync.js"; + +const GATEWAY_RESTART_INTENT_KEY = "gateway-restart"; +const GATEWAY_RESTART_INTENT_TTL_MS = 60_000; + +const restartLog = createSubsystemLogger("restart"); +type GatewayRestartIntentDatabase = Pick; + +type GatewayRestartIntentPayload = { + kind: "gateway-restart"; + pid: number; + createdAt: number; + reason?: string; + force?: boolean; + waitMs?: number; +}; + +export type GatewayRestartIntent = { + reason?: string; + force?: boolean; + waitMs?: number; +}; + +function normalizeRestartIntentPid(pid: number | undefined): number | null { + return typeof pid === "number" && Number.isSafeInteger(pid) && pid > 0 ? pid : null; +} + +export function normalizeRestartIntentReason(reason: string | undefined): string | undefined { + const normalized = reason?.trim(); + return normalized ? truncateUtf16Safe(normalized, 200) : undefined; +} + +export function writeGatewayRestartIntentSync(opts: { + env?: NodeJS.ProcessEnv; + targetPid?: number; + intent?: GatewayRestartIntent; + reason?: string; +}): boolean { + const targetPid = normalizeRestartIntentPid(opts.targetPid); + if (targetPid === null) { + return false; + } + const env = opts.env ?? process.env; + try { + const reason = normalizeRestartIntentReason(opts.reason ?? opts.intent?.reason); + const waitMs = + typeof opts.intent?.waitMs === "number" && + Number.isFinite(opts.intent.waitMs) && + opts.intent.waitMs >= 0 + ? Math.floor(opts.intent.waitMs) + : null; + const createdAt = Date.now(); + runOpenClawStateWriteTransaction( + ({ db }) => { + const stateDb = getNodeSqliteKysely(db); + executeSqliteQuerySync( + db, + stateDb + .insertInto("gateway_restart_intent") + .values({ + intent_key: GATEWAY_RESTART_INTENT_KEY, + kind: "gateway-restart", + pid: targetPid, + created_at: createdAt, + reason: reason ?? null, + force: opts.intent?.force ? 1 : null, + wait_ms: waitMs, + updated_at_ms: createdAt, + }) + .onConflict((conflict) => + conflict.column("intent_key").doUpdateSet({ + kind: (eb) => eb.ref("excluded.kind"), + pid: (eb) => eb.ref("excluded.pid"), + created_at: (eb) => eb.ref("excluded.created_at"), + reason: (eb) => eb.ref("excluded.reason"), + force: (eb) => eb.ref("excluded.force"), + wait_ms: (eb) => eb.ref("excluded.wait_ms"), + updated_at_ms: (eb) => eb.ref("excluded.updated_at_ms"), + }), + ), + ); + }, + { env }, + ); + return true; + } catch (err) { + restartLog.warn(`failed to write gateway restart intent: ${String(err)}`); + return false; + } +} + +export function clearGatewayRestartIntentSync(env: NodeJS.ProcessEnv = process.env): void { + try { + runOpenClawStateWriteTransaction( + ({ db }) => { + const stateDb = getNodeSqliteKysely(db); + executeSqliteQuerySync( + db, + stateDb + .deleteFrom("gateway_restart_intent") + .where("intent_key", "=", GATEWAY_RESTART_INTENT_KEY), + ); + }, + { env }, + ); + } catch {} +} + +function readGatewayRestartIntentPayloadSync( + env: NodeJS.ProcessEnv, +): GatewayRestartIntentPayload | null { + try { + const { db } = openOpenClawStateDatabase({ env }); + const stateDb = getNodeSqliteKysely(db); + const parsed = executeSqliteQueryTakeFirstSync( + db, + stateDb + .selectFrom("gateway_restart_intent") + .select(["kind", "pid", "created_at", "reason", "force", "wait_ms"]) + .where("intent_key", "=", GATEWAY_RESTART_INTENT_KEY), + ); + if ( + parsed?.kind === "gateway-restart" && + typeof parsed.pid === "number" && + Number.isFinite(parsed.pid) && + typeof parsed.created_at === "number" && + Number.isFinite(parsed.created_at) && + (parsed.reason === null || typeof parsed.reason === "string") && + (parsed.force === null || + (typeof parsed.force === "number" && Number.isFinite(parsed.force))) && + (parsed.wait_ms === null || + (typeof parsed.wait_ms === "number" && + Number.isFinite(parsed.wait_ms) && + parsed.wait_ms >= 0)) + ) { + const reason = normalizeRestartIntentReason(parsed.reason ?? undefined); + return { + kind: "gateway-restart", + pid: parsed.pid, + createdAt: parsed.created_at, + ...(reason ? { reason } : {}), + ...(parsed.force ? { force: true } : {}), + ...(typeof parsed.wait_ms === "number" ? { waitMs: Math.floor(parsed.wait_ms) } : {}), + }; + } + } catch { + return null; + } + return null; +} + +export function consumeGatewayRestartIntentPayloadSync( + env: NodeJS.ProcessEnv = process.env, + now = Date.now(), +): GatewayRestartIntent | null { + const payload = readGatewayRestartIntentPayloadSync(env); + clearGatewayRestartIntentSync(env); + if (!payload) { + return null; + } + if (payload.pid !== process.pid) { + return null; + } + const ageMs = now - payload.createdAt; + if (ageMs < 0 || ageMs > GATEWAY_RESTART_INTENT_TTL_MS) { + return null; + } + return { + ...(payload.reason ? { reason: payload.reason } : {}), + ...(payload.force ? { force: true } : {}), + ...(typeof payload.waitMs === "number" ? { waitMs: payload.waitMs } : {}), + }; +} + +export function consumeGatewayRestartIntentSync( + env: NodeJS.ProcessEnv = process.env, + now = Date.now(), +): boolean { + return consumeGatewayRestartIntentPayloadSync(env, now) !== null; +} diff --git a/src/infra/restart.ts b/src/infra/restart.ts index f03a0badd5b5..4bd9e1ae4a6b 100644 --- a/src/infra/restart.ts +++ b/src/infra/restart.ts @@ -2,7 +2,6 @@ import { spawnSync } from "node:child_process"; import os from "node:os"; import path from "node:path"; -import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; import { getRuntimeConfig } from "../config/config.js"; import { resolveGatewayLaunchAgentLabel, @@ -13,20 +12,12 @@ import { beginGatewayRestartSignalAdmission, getActiveGatewayRootWorkCount, isGatewayRestartDraining, + rollbackGatewayRestartSignalFence, runWithGatewayIndependentRootWorkAdmission, type GatewayRestartSignalAdmissionLease, } from "../process/gateway-work-admission.js"; import { resolveTimerTimeoutMs } from "../shared/number-coercion.js"; -import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; -import { - openOpenClawStateDatabase, - runOpenClawStateWriteTransaction, -} from "../state/openclaw-state-db.js"; -import { - executeSqliteQuerySync, - executeSqliteQueryTakeFirstSync, - getNodeSqliteKysely, -} from "./kysely-sync.js"; +import { type GatewayRestartIntent, normalizeRestartIntentReason } from "./restart-intent.js"; import { cleanStaleGatewayProcessesSync } from "./restart-stale-pids.js"; import type { RestartAttempt } from "./restart.types.js"; import { relaunchGatewayScheduledTask } from "./windows-task-restart.js"; @@ -38,11 +29,8 @@ const DEFAULT_DEFERRAL_STILL_PENDING_WARN_MS = 30_000; const DEFAULT_RESTART_DEFERRAL_TIMEOUT_MS = 300_000; const RESTART_COOLDOWN_MS = 30_000; const LAUNCHCTL_ALREADY_LOADED_EXIT_CODE = 37; -const GATEWAY_RESTART_INTENT_KEY = "gateway-restart"; -const GATEWAY_RESTART_INTENT_TTL_MS = 60_000; const restartLog = createSubsystemLogger("restart"); -type GatewayRestartIntentDatabase = Pick; let sigusr1AuthorizedCount = 0; let sigusr1AuthorizedUntil = 0; @@ -88,9 +76,15 @@ function clearPendingScheduledRestart(): void { } function clearPendingRestartSignalAdmission(): boolean { - const rolledBack = pendingRestartSignalAdmission?.rollback() ?? false; + const lease = pendingRestartSignalAdmission; pendingRestartSignalAdmission = null; - return rolledBack; + if (lease?.rollback()) { + return true; + } + // A concurrent emission must never replace a live lease with a dead handle. + // If that still happens, reopen the reversible fence directly so refused or + // abandoned signals cannot wedge process admission forever. + return rollbackGatewayRestartSignalFence(); } /** Releases a signal fence when the run loop rejects or fails to handle the signal. */ @@ -170,179 +164,6 @@ type RestartAuditInfo = { changedPaths?: string[]; }; -type GatewayRestartIntentPayload = { - kind: "gateway-restart"; - pid: number; - createdAt: number; - reason?: string; - force?: boolean; - waitMs?: number; -}; - -export type GatewayRestartIntent = { - reason?: string; - force?: boolean; - waitMs?: number; -}; - -function normalizeRestartIntentPid(pid: number | undefined): number | null { - return typeof pid === "number" && Number.isSafeInteger(pid) && pid > 0 ? pid : null; -} - -export function writeGatewayRestartIntentSync(opts: { - env?: NodeJS.ProcessEnv; - targetPid?: number; - intent?: GatewayRestartIntent; - reason?: string; -}): boolean { - const targetPid = normalizeRestartIntentPid(opts.targetPid); - if (targetPid === null) { - return false; - } - const env = opts.env ?? process.env; - try { - const reason = normalizeRestartIntentReason(opts.reason ?? opts.intent?.reason); - const waitMs = - typeof opts.intent?.waitMs === "number" && - Number.isFinite(opts.intent.waitMs) && - opts.intent.waitMs >= 0 - ? Math.floor(opts.intent.waitMs) - : null; - const createdAt = Date.now(); - runOpenClawStateWriteTransaction( - ({ db }) => { - const stateDb = getNodeSqliteKysely(db); - executeSqliteQuerySync( - db, - stateDb - .insertInto("gateway_restart_intent") - .values({ - intent_key: GATEWAY_RESTART_INTENT_KEY, - kind: "gateway-restart", - pid: targetPid, - created_at: createdAt, - reason: reason ?? null, - force: opts.intent?.force ? 1 : null, - wait_ms: waitMs, - updated_at_ms: createdAt, - }) - .onConflict((conflict) => - conflict.column("intent_key").doUpdateSet({ - kind: (eb) => eb.ref("excluded.kind"), - pid: (eb) => eb.ref("excluded.pid"), - created_at: (eb) => eb.ref("excluded.created_at"), - reason: (eb) => eb.ref("excluded.reason"), - force: (eb) => eb.ref("excluded.force"), - wait_ms: (eb) => eb.ref("excluded.wait_ms"), - updated_at_ms: (eb) => eb.ref("excluded.updated_at_ms"), - }), - ), - ); - }, - { env }, - ); - return true; - } catch (err) { - restartLog.warn(`failed to write gateway restart intent: ${String(err)}`); - return false; - } -} - -export function clearGatewayRestartIntentSync(env: NodeJS.ProcessEnv = process.env): void { - try { - runOpenClawStateWriteTransaction( - ({ db }) => { - const stateDb = getNodeSqliteKysely(db); - executeSqliteQuerySync( - db, - stateDb - .deleteFrom("gateway_restart_intent") - .where("intent_key", "=", GATEWAY_RESTART_INTENT_KEY), - ); - }, - { env }, - ); - } catch {} -} - -function readGatewayRestartIntentPayloadSync( - env: NodeJS.ProcessEnv, -): GatewayRestartIntentPayload | null { - try { - const { db } = openOpenClawStateDatabase({ env }); - const stateDb = getNodeSqliteKysely(db); - const parsed = executeSqliteQueryTakeFirstSync( - db, - stateDb - .selectFrom("gateway_restart_intent") - .select(["kind", "pid", "created_at", "reason", "force", "wait_ms"]) - .where("intent_key", "=", GATEWAY_RESTART_INTENT_KEY), - ); - if ( - parsed?.kind === "gateway-restart" && - typeof parsed.pid === "number" && - Number.isFinite(parsed.pid) && - typeof parsed.created_at === "number" && - Number.isFinite(parsed.created_at) && - (parsed.reason === null || typeof parsed.reason === "string") && - (parsed.force === null || - (typeof parsed.force === "number" && Number.isFinite(parsed.force))) && - (parsed.wait_ms === null || - (typeof parsed.wait_ms === "number" && - Number.isFinite(parsed.wait_ms) && - parsed.wait_ms >= 0)) - ) { - const reason = normalizeRestartIntentReason(parsed.reason ?? undefined); - return { - kind: "gateway-restart", - pid: parsed.pid, - createdAt: parsed.created_at, - ...(reason ? { reason } : {}), - ...(parsed.force ? { force: true } : {}), - ...(typeof parsed.wait_ms === "number" ? { waitMs: Math.floor(parsed.wait_ms) } : {}), - }; - } - } catch { - return null; - } - return null; -} - -function normalizeRestartIntentReason(reason: string | undefined): string | undefined { - const normalized = reason?.trim(); - return normalized ? truncateUtf16Safe(normalized, 200) : undefined; -} - -export function consumeGatewayRestartIntentPayloadSync( - env: NodeJS.ProcessEnv = process.env, - now = Date.now(), -): GatewayRestartIntent | null { - const payload = readGatewayRestartIntentPayloadSync(env); - clearGatewayRestartIntentSync(env); - if (!payload) { - return null; - } - if (payload.pid !== process.pid) { - return null; - } - const ageMs = now - payload.createdAt; - if (ageMs < 0 || ageMs > GATEWAY_RESTART_INTENT_TTL_MS) { - return null; - } - return { - ...(payload.reason ? { reason: payload.reason } : {}), - ...(payload.force ? { force: true } : {}), - ...(typeof payload.waitMs === "number" ? { waitMs: payload.waitMs } : {}), - }; -} - -export function consumeGatewayRestartIntentSync( - env: NodeJS.ProcessEnv = process.env, - now = Date.now(), -): boolean { - return consumeGatewayRestartIntentPayloadSync(env, now) !== null; -} - function summarizeChangedPaths(paths: string[] | undefined, maxPaths = 6): string | null { if (!Array.isArray(paths) || paths.length === 0) { return null; @@ -444,8 +265,20 @@ function emitGatewayRestartWithSignalAdmission( reasonOverride?: string, intent?: GatewayRestartIntent, ): boolean { - const signalAdmission = pendingRestartSignalAdmission ?? beginGatewayRestartSignalAdmission(); - pendingRestartSignalAdmission = signalAdmission; + let signalAdmission = pendingRestartSignalAdmission; + if (!signalAdmission) { + // Orphan fence: pending without a lease and without a delivered signal. + // Reopen before acquiring so a lost lease cannot block all future emissions. + if (!hasUnconsumedRestartSignal()) { + rollbackGatewayRestartSignalFence(); + } + signalAdmission = beginGatewayRestartSignalAdmission(); + if (!signalAdmission) { + // Another emission owns the fence, or one-way drain already closed admission. + return false; + } + pendingRestartSignalAdmission = signalAdmission; + } const hadUnconsumedRestartSignal = hasUnconsumedRestartSignal(); const emitted = emitGatewayRestart(reasonOverride, intent); if (!emitted && !hadUnconsumedRestartSignal) { @@ -773,12 +606,39 @@ async function emitPreparedGatewayRestart( if (transientGeneration !== restartTransientGeneration) { return false; } - // Close new roots before the final synchronous idle check. The independent - // emission owner is excluded; any other admitted root makes this attempt retry. - const signalAdmission = beginGatewayRestartSignalAdmission(); - pendingRestartSignalAdmission = signalAdmission; + // SIGUSR1 already queued: coalesce. Run loop owns reopen-or-drain. + if (hasUnconsumedRestartSignal()) { + return false; + } + // Single live lease, multiple attempts may share it (deferred prepare → + // concurrent emit / retry). Never invent a dead stand-in lease. + let signalAdmission = pendingRestartSignalAdmission; + let ownsFenceLease = false; + if (!signalAdmission) { + // Orphan fence: pending without a lease and without a delivered signal. + rollbackGatewayRestartSignalFence(); + signalAdmission = beginGatewayRestartSignalAdmission(); + if (!signalAdmission) { + return false; + } + pendingRestartSignalAdmission = signalAdmission; + ownsFenceLease = true; + } let fenceActive = true; + let keepFenceForRunLoop = false; const rollbackFence = () => { + // A concurrent emitter may queue SIGUSR1 on this shared lease while we + // await beforeEmit. Cancel/finally must not reopen over an in-flight + // signal — the run loop owns reopen-or-drain from here. + if (keepFenceForRunLoop || hasUnconsumedRestartSignal()) { + return; + } + // Adopters share the lease with a still-active prepare/deferral owner. + // Only the creator may reopen on abandon; stop this attempt's canEmit. + if (!ownsFenceLease) { + fenceActive = false; + return; + } fenceActive = false; signalAdmission.rollback(); if (pendingRestartSignalAdmission === signalAdmission) { @@ -786,37 +646,37 @@ async function emitPreparedGatewayRestart( } }; setFenceRollback?.(rollbackFence); - let isIdle: boolean; try { - isIdle = finalIdleCheck + const isIdle = finalIdleCheck ? finalIdleCheck() && getActiveGatewayRootWorkCount({ excludeCurrent: true }) === 0 : true; - } catch (err) { - rollbackFence(); + if (!isIdle) { + return false; + } + const emitResult = await emitPreparedGatewayRestartUnderAdmission( + hooks, + reasonOverride, + intent, + transientGeneration, + () => fenceActive, + ); + if ( + emitResult && + (emitResult.status === "emitted" || + (emitResult.status === "coalesced" && hasUnconsumedRestartSignal())) + ) { + // Delivered or already-in-flight signal: run loop owns reopen-or-drain. + keepFenceForRunLoop = true; + return true; + } + return emitResult !== null; + } finally { + // Creator non-delivery reopens; adopters leave the live prepare lease. + if (!keepFenceForRunLoop) { + rollbackFence(); + } setFenceRollback?.(null); - throw err; } - if (!isIdle) { - rollbackFence(); - setFenceRollback?.(null); - return false; - } - const emitResult = await emitPreparedGatewayRestartUnderAdmission( - hooks, - reasonOverride, - intent, - transientGeneration, - () => fenceActive, - ); - if ( - !emitResult || - emitResult.status === "failed" || - (emitResult.status === "coalesced" && !hasUnconsumedRestartSignal()) - ) { - rollbackFence(); - } - setFenceRollback?.(null); - return emitResult !== null; }); } catch (err) { if (!isGatewayRestartDraining()) { @@ -887,6 +747,8 @@ export function deferGatewayRestartUntilIdle(opts: { ) .then((attempted) => { attemptingEmission = false; + // Successful delivery clears the cancel hook after the fence is owned by + // the run loop. Failed attempts already reopened via emitPrepared finally. cancelEmissionFence = null; if (cancelled || !attempted) { return; @@ -898,6 +760,10 @@ export function deferGatewayRestartUntilIdle(opts: { }) .catch((err: unknown) => { attemptingEmission = false; + // Invoke before clearing: a thrown emission must reopen the fence even + // when emitPreparedGatewayRestart's finally did not run (for example a + // rejection from the independent-root wrapper after cancel raced). + cancelEmissionFence?.(); cancelEmissionFence = null; stopPoll(); opts.hooks?.onCheckError?.(err); diff --git a/src/process/gateway-work-admission.test.ts b/src/process/gateway-work-admission.test.ts index c808112f7256..ac626d0d3877 100644 --- a/src/process/gateway-work-admission.test.ts +++ b/src/process/gateway-work-admission.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, expect, it, vi } from "vitest"; import { beginGatewayRestartSignalAdmission, + beginGatewayRootWorkAdmissionWhenOpen, GatewayDrainingError, getActiveGatewayRootWorkCount, isGatewaySubordinateWorkAdmissionClosed, @@ -9,6 +10,7 @@ import { markGatewayRestartDraining, retainGatewayRootWorkAdmissionContinuation, resetGatewayWorkAdmission, + rollbackGatewayRestartSignalFence, runWithGatewayIndependentRootWorkContinuation, tryBeginGatewayRootWorkAdmission, tryBeginGatewaySuspendAdmission, @@ -176,25 +178,62 @@ it("does not let a stale suspension release clear restart drain", () => { it("blocks suspension while restart signal handling is pending", () => { const pendingSignal = beginGatewayRestartSignalAdmission(); + expect(pendingSignal).not.toBeNull(); expect(isGatewayWorkAdmissionClosed()).toBe(true); expect(tryBeginGatewayRootWorkAdmission()).toBeNull(); expect(tryBeginGatewaySuspendAdmission(() => {})).toBeNull(); - expect(pendingSignal.rollback()).toBe(true); + expect(beginGatewayRestartSignalAdmission()).toBeNull(); + expect(pendingSignal?.rollback()).toBe(true); expect(isGatewayWorkAdmissionClosed()).toBe(false); expect(tryBeginGatewaySuspendAdmission(() => {})?.rollback()).toBe(true); }); it("promotes a pending restart signal to one-way drain", () => { const pendingSignal = beginGatewayRestartSignalAdmission(); + expect(pendingSignal).not.toBeNull(); markGatewayRestartDraining(); - expect(pendingSignal.rollback()).toBe(false); + expect(pendingSignal?.rollback()).toBe(false); expect(isGatewayWorkAdmissionClosed()).toBe(true); expect(tryBeginGatewayRootWorkAdmission()).toBeNull(); }); +it("force-rolls back an orphan restart-signal fence without a live lease", () => { + const pendingSignal = beginGatewayRestartSignalAdmission(); + expect(pendingSignal).not.toBeNull(); + expect(isGatewayWorkAdmissionClosed()).toBe(true); + + // Drop the lease the way a concurrent emission overwrite used to: the fence + // stays closed with no handle that can reopen it. + expect(rollbackGatewayRestartSignalFence()).toBe(true); + expect(pendingSignal?.rollback()).toBe(false); + expect(isGatewayWorkAdmissionClosed()).toBe(false); + const root = tryBeginGatewayRootWorkAdmission(); + expect(root).not.toBeNull(); + root?.release(); +}); + +it("wakes beginGatewayRootWorkAdmissionWhenOpen waiters when the signal fence rolls back", async () => { + const pendingSignal = beginGatewayRestartSignalAdmission(); + expect(pendingSignal).not.toBeNull(); + + const waiting = beginGatewayRootWorkAdmissionWhenOpen(); + let resolved = false; + void waiting.then(() => { + resolved = true; + }); + await Promise.resolve(); + expect(resolved).toBe(false); + + expect(pendingSignal?.rollback()).toBe(true); + const admission = await waiting; + expect(resolved).toBe(true); + expect(admission.ownsRoot).toBe(true); + admission.release(); +}); + it("defers required internal root work until suspension reopens", async () => { const suspension = tryBeginGatewaySuspendAdmission(() => {}); expect(suspension?.commit()).toBe(true); diff --git a/src/process/gateway-work-admission.ts b/src/process/gateway-work-admission.ts index 0cbba648352b..6e50414950bc 100644 --- a/src/process/gateway-work-admission.ts +++ b/src/process/gateway-work-admission.ts @@ -1,9 +1,13 @@ // Coordinates process-wide root work admission with reversible host suspension. import { AsyncLocalStorage } from "node:async_hooks"; +import { createSubsystemLogger } from "../logging/subsystem.js"; import { resolveGlobalSingleton } from "../shared/global-singleton.js"; type GatewaySuspendAdmissionPhase = "accepting" | "preparing" | "prepared"; +type AdmissionCloseReason = "restart-signal fence" | "restart drain" | "suspend phase"; +type AdmissionReopenReason = "restart-signal fence" | "suspend phase"; + export class GatewayDrainingError extends Error { constructor() { super("Gateway is draining; new tasks are not accepted"); @@ -29,6 +33,8 @@ type GatewayWorkAdmissionState = { suspendOpenWaiters: Set<() => void>; }; +const admissionLog = createSubsystemLogger("gateway/admission"); + const GATEWAY_WORK_ADMISSION_STATE = resolveGlobalSingleton( Symbol.for("openclaw.gatewayWorkAdmissionState"), (): GatewayWorkAdmissionState => ({ @@ -44,6 +50,14 @@ const GATEWAY_WORK_ADMISSION_STATE = resolveGlobalSingleton( }), ); +function logAdmissionClosed(reason: AdmissionCloseReason): void { + admissionLog.info(`admission closed: ${reason}`); +} + +function logAdmissionReopened(reason: AdmissionReopenReason): void { + admissionLog.info(`admission reopened: ${reason}`); +} + type GatewayRootWorkAdmissionLease = { ownsRoot: boolean; release: () => void; @@ -105,13 +119,32 @@ function resolveRootDrainWaiters(): void { function invalidateSuspendAdmission(): void { const callback = GATEWAY_WORK_ADMISSION_STATE.suspendInvalidated; + const wasClosed = GATEWAY_WORK_ADMISSION_STATE.suspendPhase !== "accepting"; GATEWAY_WORK_ADMISSION_STATE.suspendInvalidated = undefined; GATEWAY_WORK_ADMISSION_STATE.suspendPhase = "accepting"; GATEWAY_WORK_ADMISSION_STATE.suspendGeneration += 1; resolveSuspendOpenWaiters(); + // Restart drain supersedes suspension without reopening process admission. + if (wasClosed && !GATEWAY_WORK_ADMISSION_STATE.restartDraining) { + logAdmissionReopened("suspend phase"); + } callback?.(); } +function clearRestartSignalFence(): boolean { + if ( + GATEWAY_WORK_ADMISSION_STATE.restartDraining || + !GATEWAY_WORK_ADMISSION_STATE.restartSignalPending + ) { + return false; + } + GATEWAY_WORK_ADMISSION_STATE.restartSignalPending = false; + GATEWAY_WORK_ADMISSION_STATE.restartSignalGeneration += 1; + resolveSuspendOpenWaiters(); + logAdmissionReopened("restart-signal fence"); + return true; +} + function resolveSuspendOpenWaiters(): void { const waiters = Array.from(GATEWAY_WORK_ADMISSION_STATE.suspendOpenWaiters); GATEWAY_WORK_ADMISSION_STATE.suspendOpenWaiters.clear(); @@ -160,10 +193,16 @@ export function isGatewayRestartDraining(): boolean { /** Restart drain is one-way until the in-process restart resets runtime state. */ export function markGatewayRestartDraining(): void { + if (GATEWAY_WORK_ADMISSION_STATE.restartDraining) { + return; + } + // Drain supersedes the reversible signal fence; do not reopen before the + // one-way close, or waiters could briefly admit work into a dying process. GATEWAY_WORK_ADMISSION_STATE.restartSignalPending = false; GATEWAY_WORK_ADMISSION_STATE.restartSignalGeneration += 1; GATEWAY_WORK_ADMISSION_STATE.restartDraining = true; resolveSuspendOpenWaiters(); + logAdmissionClosed("restart drain"); if (GATEWAY_WORK_ADMISSION_STATE.suspendPhase !== "accepting") { // A restart supersedes a reversible suspension. The coordinator callback // drops its timer/token without reopening the scheduler being shut down. @@ -171,13 +210,22 @@ export function markGatewayRestartDraining(): void { } } -/** Blocks suspension across signal emission until the run loop starts restart drain. */ -export function beginGatewayRestartSignalAdmission(): GatewayRestartSignalAdmissionLease { - if (GATEWAY_WORK_ADMISSION_STATE.restartSignalPending) { - return { rollback: () => false }; +/** + * Blocks suspension across signal emission until the run loop starts restart drain. + * Returns null when another owner already holds the fence or one-way drain is active. + * Callers must not invent a stand-in lease: a dead rollback handle is how the fence + * can stay closed after the real owner is lost. + */ +export function beginGatewayRestartSignalAdmission(): GatewayRestartSignalAdmissionLease | null { + if ( + GATEWAY_WORK_ADMISSION_STATE.restartDraining || + GATEWAY_WORK_ADMISSION_STATE.restartSignalPending + ) { + return null; } GATEWAY_WORK_ADMISSION_STATE.restartSignalPending = true; const generation = ++GATEWAY_WORK_ADMISSION_STATE.restartSignalGeneration; + logAdmissionClosed("restart-signal fence"); return { rollback: () => { if ( @@ -186,14 +234,19 @@ export function beginGatewayRestartSignalAdmission(): GatewayRestartSignalAdmiss ) { return false; } - GATEWAY_WORK_ADMISSION_STATE.restartSignalPending = false; - GATEWAY_WORK_ADMISSION_STATE.restartSignalGeneration += 1; - resolveSuspendOpenWaiters(); - return true; + return clearRestartSignalFence(); }, }; } +/** + * Reopens a reversible restart-signal fence that no longer has a live lease. + * No-op while one-way restart drain owns admission. + */ +export function rollbackGatewayRestartSignalFence(): boolean { + return clearRestartSignalFence(); +} + /** Root RPC/timer admission. Nested work in the same async chain counts once. */ export function tryBeginGatewayRootWorkAdmission(): GatewayRootWorkAdmissionLease | null { const current = GATEWAY_WORK_ADMISSION_STATE.currentRootWork.getStore(); @@ -355,6 +408,7 @@ export function tryBeginGatewaySuspendAdmission( GATEWAY_WORK_ADMISSION_STATE.suspendPhase = "preparing"; const generation = ++GATEWAY_WORK_ADMISSION_STATE.suspendGeneration; GATEWAY_WORK_ADMISSION_STATE.suspendInvalidated = onInvalidated; + logAdmissionClosed("suspend phase"); const transition = ( expected: GatewaySuspendAdmissionPhase, @@ -370,6 +424,7 @@ export function tryBeginGatewaySuspendAdmission( if (next === "accepting") { GATEWAY_WORK_ADMISSION_STATE.suspendInvalidated = undefined; resolveSuspendOpenWaiters(); + logAdmissionReopened("suspend phase"); } return true; };