@@ -1,5 +1,10 @@
11import { randomUUID } from "node:crypto";
22import net from "node:net";
3+import {
4+measureGatewayRestartTrace,
5+markGatewayRestartTrace,
6+startGatewayRestartTrace,
7+} from "../../gateway/restart-trace.js";
38import type { startGatewayServer } from "../../gateway/server.js";
49import { formatErrorMessage } from "../../infra/errors.js";
510import { acquireGatewayLock } from "../../infra/gateway-lock.js";
@@ -187,6 +192,11 @@ export async function runGatewayLoop(params: {
187192 processInstanceId,
188193supervisorMode: supervisorMode ?? "external",
189194});
195+markGatewayRestartTrace("restart.full-process-handoff", [
196+["kind", "update-process"],
197+["mode", respawn.mode],
198+["supervisorMode", supervisorMode ?? "external"],
199+]);
190200gatewayLog.info("restart mode: update process respawn (supervisor restart)");
191201if (supervisorMode === "launchd") {
192202await new Promise((resolve) => {
@@ -235,6 +245,12 @@ export async function runGatewayLoop(params: {
235245supervisorMode: supervisorMode ?? "external",
236246});
237247}
248+markGatewayRestartTrace("restart.full-process-handoff", [
249+["kind", "full-process"],
250+["mode", respawn.mode],
251+["pid", respawn.mode === "spawned" ? (respawn.pid ?? "unknown") : "none"],
252+["supervisorMode", supervisorMode ?? "none"],
253+]);
238254gatewayLog.info(`restart mode: full process restart (${modeLabel})`);
239255if (supervisorMode === "launchd") {
240256// A short clean-exit pause keeps rapid SIGUSR1/config restarts from
@@ -301,6 +317,14 @@ export async function runGatewayLoop(params: {
301317shuttingDown = true;
302318const isRestart = action === "restart";
303319gatewayLog.info(`received ${signal}; ${isRestart ? "restarting" : "shutting down"}`);
320+if (isRestart) {
321+startGatewayRestartTrace("restart.signal.received", [
322+["signal", signal],
323+["reason", restartReason ?? signal],
324+["force", restartIntent?.force === true],
325+["waitMs", restartIntent?.waitMs ?? "default"],
326+]);
327+}
304328305329let forceExitTimer: ReturnType<typeof setTimeout> | null = null;
306330const armForceExitTimer = (forceExitMs: number) => {
@@ -365,105 +389,124 @@ export async function runGatewayLoop(params: {
365389// On restart, wait for in-flight agent turns to finish before
366390// tearing down the server so buffered messages are delivered.
367391if (isRestart) {
368-const {
369- abortEmbeddedPiRun,
370- getInspectableActiveTaskRestartBlockers,
371- getActiveEmbeddedRunCount,
372- getActiveTaskCount,
373- markGatewayDraining,
374- waitForActiveEmbeddedRuns,
375- waitForActiveTasks,
376-} = await loadGatewayLifecycleRuntimeModule();
377-const formatTaskBlockers = () => {
378-const blockers = getInspectableActiveTaskRestartBlockers();
379-if (blockers.length === 0) {
380-return null;
381-}
382-const shown = blockers
383-.slice(0, 8)
384-.map((task) =>
385-[
386-`taskId=${task.taskId}`,
387-task.runId ? `runId=${task.runId}` : null,
388-`status=${task.status}`,
389-`runtime=${task.runtime}`,
390-task.label ? `label=${task.label}` : null,
391-task.title ? `title=${task.title.slice(0, 80)}` : null,
392-]
393-.filter((value): value is string => Boolean(value))
394-.join(" "),
395-);
396-const omitted = blockers.length - shown.length;
397-return omitted > 0 ? `${shown.join("; ")}; +${omitted} more` : shown.join("; ");
398-};
399-const createStillPendingDrainLogger = () =>
400-setInterval(() => {
401-gatewayLog.warn(
402-`still draining ${getActiveTaskCount()} active task(s) and ${getActiveEmbeddedRunCount()} active embedded run(s) before restart`,
403-);
404-}, RESTART_DRAIN_STILL_PENDING_WARN_MS);
392+let activeTasksAtDrainStart = 0;
393+let activeRunsAtDrainStart = 0;
394+let drainTimedOut = false;
395+await measureGatewayRestartTrace(
396+"restart.drain",
397+async () => {
398+const {
399+ abortEmbeddedPiRun,
400+ getInspectableActiveTaskRestartBlockers,
401+ getActiveEmbeddedRunCount,
402+ getActiveTaskCount,
403+ markGatewayDraining,
404+ waitForActiveEmbeddedRuns,
405+ waitForActiveTasks,
406+} = await loadGatewayLifecycleRuntimeModule();
407+const formatTaskBlockers = () => {
408+const blockers = getInspectableActiveTaskRestartBlockers();
409+if (blockers.length === 0) {
410+return null;
411+}
412+const shown = blockers
413+.slice(0, 8)
414+.map((task) =>
415+[
416+`taskId=${task.taskId}`,
417+task.runId ? `runId=${task.runId}` : null,
418+`status=${task.status}`,
419+`runtime=${task.runtime}`,
420+task.label ? `label=${task.label}` : null,
421+task.title ? `title=${task.title.slice(0, 80)}` : null,
422+]
423+.filter((value): value is string => Boolean(value))
424+.join(" "),
425+);
426+const omitted = blockers.length - shown.length;
427+return omitted > 0 ? `${shown.join("; ")}; +${omitted} more` : shown.join("; ");
428+};
429+const createStillPendingDrainLogger = () =>
430+setInterval(() => {
431+gatewayLog.warn(
432+`still draining ${getActiveTaskCount()} active task(s) and ${getActiveEmbeddedRunCount()} active embedded run(s) before restart`,
433+);
434+}, RESTART_DRAIN_STILL_PENDING_WARN_MS);
405435406-// Reject new enqueues immediately during the drain window so
407-// sessions get an explicit restart error instead of silent task loss.
408-markGatewayDraining();
409-const activeTasks = getActiveTaskCount();
410-const activeRuns = getActiveEmbeddedRunCount();
436+// Reject new enqueues immediately during the drain window so
437+// sessions get an explicit restart error instead of silent task loss.
438+markGatewayDraining();
439+const activeTasks = getActiveTaskCount();
440+const activeRuns = getActiveEmbeddedRunCount();
441+activeTasksAtDrainStart = activeTasks;
442+activeRunsAtDrainStart = activeRuns;
411443412-// Best-effort abort for compacting runs so long compaction operations
413-// don't hold session write locks across restart boundaries.
414-if (activeRuns > 0) {
415-abortEmbeddedPiRun(undefined, { mode: "compacting" });
416-}
444+ // Best-effort abort for compacting runs so long compaction operations
445+ // don't hold session write locks across restart boundaries.
446+ if (activeRuns > 0) {
447+ abortEmbeddedPiRun(undefined, { mode: "compacting" });
448+ }
417449418-if (activeTasks > 0 || activeRuns > 0) {
419-const taskBlockers = formatTaskBlockers();
420-gatewayLog.info(
421-`draining ${activeTasks} active task(s) and ${activeRuns} active embedded run(s) before restart ${formatRestartDrainBudget()}`,
422-);
423-if (taskBlockers) {
424-gatewayLog.warn(`restart blocked by active background task run(s): ${taskBlockers}`);
425-}
426-if (restartIntent?.force) {
427-gatewayLog.warn("forced restart requested; skipping active work drain");
428-abortEmbeddedPiRun(undefined, { mode: "all" });
429-} else {
430-const activeRunDrainWaitMs = resolveActiveRunDrainWaitMs(activeRuns);
431-const stillPendingDrainLogger = createStillPendingDrainLogger();
432-let abortedAfterRunGrace = false;
433-let tasksDrain: { drained: boolean } = { drained: true };
434-let runsDrain: { drained: boolean } = { drained: true };
435-try {
436-const tasksDrainPromise =
437-activeTasks > 0
438- ? waitForActiveTasks(restartDrainTimeoutMs)
439- : Promise.resolve({ drained: true });
440-runsDrain =
441-activeRuns > 0
442- ? await waitForActiveEmbeddedRuns(activeRunDrainWaitMs)
443- : { drained: true };
444-if (!runsDrain.drained && activeRuns > 0) {
450+if (activeTasks > 0 || activeRuns > 0) {
451+const taskBlockers = formatTaskBlockers();
452+gatewayLog.info(
453+`draining ${activeTasks} active task(s) and ${activeRuns} active embedded run(s) before restart ${formatRestartDrainBudget()}`,
454+);
455+if (taskBlockers) {
445456gatewayLog.warn(
446-"active embedded run drain grace reached; aborting active run(s) before restart",
457+`restart blocked by active background task run(s): ${taskBlockers}`,
447458);
448-abortEmbeddedPiRun(undefined, { mode: "all" });
449-abortedAfterRunGrace = true;
450459}
451-tasksDrain = await tasksDrainPromise;
452-} finally {
453-clearInterval(stillPendingDrainLogger);
454-}
455-if (tasksDrain.drained && runsDrain.drained) {
456-gatewayLog.info("all active work drained");
457-} else {
458-gatewayLog.warn("drain timeout reached; proceeding with restart");
459-// Final best-effort abort to avoid carrying active runs into the
460-// next lifecycle when drain time budget is exhausted.
461-if (!abortedAfterRunGrace) {
460+if (restartIntent?.force) {
461+gatewayLog.warn("forced restart requested; skipping active work drain");
462462abortEmbeddedPiRun(undefined, { mode: "all" });
463+} else {
464+const activeRunDrainWaitMs = resolveActiveRunDrainWaitMs(activeRuns);
465+const stillPendingDrainLogger = createStillPendingDrainLogger();
466+let abortedAfterRunGrace = false;
467+let tasksDrain: { drained: boolean } = { drained: true };
468+let runsDrain: { drained: boolean } = { drained: true };
469+try {
470+const tasksDrainPromise =
471+activeTasks > 0
472+ ? waitForActiveTasks(restartDrainTimeoutMs)
473+ : Promise.resolve({ drained: true });
474+runsDrain =
475+activeRuns > 0
476+ ? await waitForActiveEmbeddedRuns(activeRunDrainWaitMs)
477+ : { drained: true };
478+if (!runsDrain.drained && activeRuns > 0) {
479+gatewayLog.warn(
480+"active embedded run drain grace reached; aborting active run(s) before restart",
481+);
482+abortEmbeddedPiRun(undefined, { mode: "all" });
483+abortedAfterRunGrace = true;
484+}
485+tasksDrain = await tasksDrainPromise;
486+} finally {
487+clearInterval(stillPendingDrainLogger);
488+}
489+if (tasksDrain.drained && runsDrain.drained) {
490+gatewayLog.info("all active work drained");
491+} else {
492+drainTimedOut = true;
493+gatewayLog.warn("drain timeout reached; proceeding with restart");
494+// Final best-effort abort to avoid carrying active runs into the
495+// next lifecycle when drain time budget is exhausted.
496+if (!abortedAfterRunGrace) {
497+abortEmbeddedPiRun(undefined, { mode: "all" });
498+}
499+}
463500}
464501}
465-}
466-}
502+},
503+() => [
504+["activeTasks", activeTasksAtDrainStart],
505+["activeRuns", activeRunsAtDrainStart],
506+["timedOut", drainTimedOut],
507+["force", restartIntent?.force === true],
508+],
509+);
467510}
468511469512armCloseForceExitTimerForIndefiniteRestart();
@@ -557,6 +600,7 @@ export async function runGatewayLoop(params: {
557600resetAllLanes();
558601resetGatewayRestartStateForInProcessRestart();
559602reloadTaskRegistryFromStore();
603+markGatewayRestartTrace("restart.next-start");
560604});
561605562606// Keep process alive; SIGUSR1 triggers an in-process restart (no supervisor required).
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。