fix(supervisor): wide-event review fixes + noisy-routes flag + socket lifecycle

Author: nicktrn

apps/supervisor/src/env.ts

@@ -261,6 +261,10 @@ const Env = z
     // line per natural unit of work (dequeue iteration, HTTP request, socket
     // lifecycle). High-QPS hotpath, so the kill switch must be honoured.
     TRIGGER_WIDE_EVENTS_ENABLED: BoolEnv.default(false),
+    // When true, also emit wide events for high-frequency HTTP routes
+    // (heartbeat, snapshots-since, logs/debug). Off in prod to keep event
+    // volume manageable; on in test environments for full-fidelity debugging.
+    TRIGGER_WIDE_EVENTS_NOISY_ROUTES: BoolEnv.default(false),
   })
   .superRefine((data, ctx) => {
     if (data.COMPUTE_SNAPSHOTS_ENABLED && !data.TRIGGER_METADATA_URL) {

apps/supervisor/src/index.ts

@@ -63,6 +63,7 @@ class ManagedSupervisor {
     env: { nodeId: env.TRIGGER_WORKER_INSTANCE_NAME },
     enabled: env.TRIGGER_WIDE_EVENTS_ENABLED,
   };
+  private readonly wideEventsNoisyRoutes = env.TRIGGER_WIDE_EVENTS_NOISY_ROUTES;
 
   constructor() {
     const {
@@ -260,7 +261,7 @@ class ManagedSupervisor {
             ...this.wideEventOpts,
             traceparent,
             setup: (state) => {
-              setMeta(state, "run_id", message.run.id);
+              setMeta(state, "run_id", message.run.friendlyId);
               setMeta(state, "env_id", message.environment.id);
               setMeta(state, "org_id", message.organization.id);
               setMeta(state, "project_id", message.project.id);
@@ -472,6 +473,7 @@ class ManagedSupervisor {
       computeManager: this.computeManager,
       tracing: this.tracing,
       wideEventOpts: this.wideEventOpts,
+      wideEventsNoisyRoutes: this.wideEventsNoisyRoutes,
     });
 
     this.workloadServer.on("runConnected", this.onRunConnected.bind(this));

apps/supervisor/src/wideEvents/emit.test.ts

@@ -43,7 +43,6 @@ describe("emit", () => {
     expect(out).not.toHaveProperty("trace_id");
     expect(out).not.toHaveProperty("version");
     expect(out).not.toHaveProperty("commit_sha");
-    expect(out).not.toHaveProperty("instance_id");
     expect(out).not.toHaveProperty("error.code");
   });
 

apps/supervisor/src/wideEvents/emit.ts

@@ -26,7 +26,6 @@ export function emit(state: State): void {
   appendIfSet(out, "commit_sha", state.commitSha);
   appendIfSet(out, "region", state.region);
   appendIfSet(out, "node_id", state.nodeId);
-  appendIfSet(out, "instance_id", state.instanceId);
 
   out.ok = state.ok;
   if (state.statusCode !== 0) out.status = state.statusCode;

apps/supervisor/src/wideEvents/middleware.ts

@@ -57,10 +57,10 @@ export async function runWideEvent<T>(
   });
   if (opts.route) state.extras.route = opts.route;
   if (opts.method) state.extras.method = opts.method;
-  if (opts.setup) opts.setup(state);
 
   const start = performance.now();
   try {
+    if (opts.setup) opts.setup(state);
     const result = await wideEventStorage.run(state, () => Promise.resolve(fn()));
     state.durationMs = Math.round(performance.now() - start);
     if (finalize) finalize(state);

apps/supervisor/src/wideEvents/state.ts

@@ -21,7 +21,6 @@ export type State = {
   commitSha?: string;
   region?: string;
   nodeId?: string;
-  instanceId?: string;
 
   // Caller-attached opaque metadata, flattened to `meta.<key>` on emit.
   meta: Record<string, string>;

apps/supervisor/src/workloadServer/index.ts

@@ -88,6 +88,8 @@ type WorkloadServerOptions = {
   computeManager?: ComputeWorkloadManager;
   tracing?: OtlpTraceService;
   wideEventOpts: WideEventOptions;
+  /** When true, high-frequency HTTP routes also emit wide events. */
+  wideEventsNoisyRoutes: boolean;
 };
 
 export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
@@ -96,6 +98,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
 
   private readonly logger = new SimpleStructuredLogger("workload-server");
   private readonly wideEventOpts: WideEventOptions;
+  private readonly wideEventsNoisyRoutes: boolean;
 
   private readonly httpServer: HttpServer;
   private readonly websocketServer: Namespace<
@@ -126,6 +129,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
     this.workerClient = opts.workerClient;
     this.checkpointClient = opts.checkpointClient;
     this.wideEventOpts = opts.wideEventOpts;
+    this.wideEventsNoisyRoutes = opts.wideEventsNoisyRoutes;
 
     if (opts.computeManager?.snapshotsEnabled) {
       this.snapshotService = new ComputeSnapshotService({
@@ -183,16 +187,25 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
    * `traceparent` and `x-request-id` from `req.headers`, attaches `run_id` /
    * `snapshot_id` / `deployment_id` meta from `params` when present, and
    * captures the response status from `res.statusCode` after `fn` returns.
+   *
+   * Pass `highFrequency: true` for noisy routes (heartbeat, polling). Those
+   * still go through the wrapper but only emit when
+   * `TRIGGER_WIDE_EVENTS_NOISY_ROUTES` is on, so prod can keep them dark
+   * while test envs capture full-fidelity traffic for debugging.
    */
   private wideRoute<T>(
     ctx: { req: IncomingMessage; res: ServerResponse; params?: unknown },
     route: string,
     method: string,
-    fn: () => Promise<T> | T
+    fn: () => Promise<T> | T,
+    routeOpts: { highFrequency?: boolean } = {}
   ): Promise<T> {
+    const enabled =
+      this.wideEventOpts.enabled && (!routeOpts.highFrequency || this.wideEventsNoisyRoutes);
     return runWideEvent(
       {
         ...this.wideEventOpts,
+        enabled,
         route,
         method,
         traceparent: this.headerValueFromRequest(ctx.req, "traceparent"),
@@ -324,7 +337,8 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
                 reply.json({
                   ok: true,
                 } satisfies WorkloadHeartbeatResponseBody);
-              }
+              },
+              { highFrequency: true }
             ),
         }
       )
@@ -494,7 +508,8 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
                 reply.json({
                   snapshots: sinceSnapshotResponse.data.snapshots.map(legacifyCheckpointType),
                 } satisfies WorkloadRunSnapshotsSinceResponseBody);
-              }
+              },
+              { highFrequency: true }
             ),
         }
       )
@@ -550,7 +565,8 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
                 body,
                 this.runnerIdFromRequest(req)
               );
-            }
+            },
+            { highFrequency: true }
           ),
       });
     } else {
@@ -563,7 +579,8 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
             "POST",
             async () => {
               ctx.reply.empty(204);
-            }
+            },
+            { highFrequency: true }
           ),
       });
     }
@@ -654,6 +671,26 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         };
       };
 
+      const emitSocketLifecycle = (
+        event: "run_connected" | "run_disconnected",
+        friendlyId: string,
+        disconnectReason?: string
+      ) => {
+        emitOneShot({
+          ...this.wideEventOpts,
+          populate: (state) => {
+            state.extras.event = event;
+            setMeta(state, "run_id", friendlyId);
+            if (socket.data.deploymentId) {
+              setMeta(state, "deployment_id", socket.data.deploymentId);
+            }
+            if (socket.data.runnerId) setMeta(state, "runner_id", socket.data.runnerId);
+            state.extras.socket_id = socket.id;
+            if (disconnectReason) state.extras.disconnect_reason = disconnectReason;
+          },
+        });
+      };
+
       const runConnected = (friendlyId: string) => {
         socketLogger.debug("runConnected", { ...getSocketMetadata() });
 
@@ -664,20 +701,22 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
             newRunId: friendlyId,
             oldRunId: socket.data.runFriendlyId,
           });
-          runDisconnected(socket.data.runFriendlyId);
+          runDisconnected(socket.data.runFriendlyId, "socket_run_replaced");
         }
 
         this.runSockets.set(friendlyId, socket);
         this.emit("runConnected", { run: { friendlyId } });
         socket.data.runFriendlyId = friendlyId;
+        emitSocketLifecycle("run_connected", friendlyId);
       };
 
-      const runDisconnected = (friendlyId: string) => {
+      const runDisconnected = (friendlyId: string, reason: string) => {
         socketLogger.debug("runDisconnected", { ...getSocketMetadata() });
 
         this.runSockets.delete(friendlyId);
         this.emit("runDisconnected", { run: { friendlyId } });
         socket.data.runFriendlyId = undefined;
+        emitSocketLifecycle("run_disconnected", friendlyId, reason);
       };
 
       socketLogger.debug("wsServer socket connected", { ...getSocketMetadata() });
@@ -695,7 +734,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         });
 
         if (socket.data.runFriendlyId) {
-          runDisconnected(socket.data.runFriendlyId);
+          runDisconnected(socket.data.runFriendlyId, `socket_disconnecting:${reason}`);
         }
       });
 
@@ -725,18 +764,6 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
 
         try {
           runConnected(message.run.friendlyId);
-          emitOneShot({
-            ...this.wideEventOpts,
-            populate: (state) => {
-              state.extras.event = "run:start";
-              setMeta(state, "run_id", message.run.friendlyId);
-              if (socket.data.deploymentId) {
-                setMeta(state, "deployment_id", socket.data.deploymentId);
-              }
-              if (socket.data.runnerId) setMeta(state, "runner_id", socket.data.runnerId);
-              state.extras.socket_id = socket.id;
-            },
-          });
         } catch (error) {
           log.error("run:start error", { error });
         }
@@ -752,22 +779,10 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         log.debug("Handling run:stop");
 
         try {
-          runDisconnected(message.run.friendlyId);
+          runDisconnected(message.run.friendlyId, "run_stop_message");
           // Don't delete trace context here - run:stop fires after each snapshot/shutdown
           // but the run may be restored on a new VM and snapshot again. Trace context is
           // re-populated on dequeue, and entries are small (4 strings per run).
-          emitOneShot({
-            ...this.wideEventOpts,
-            populate: (state) => {
-              state.extras.event = "run:stop";
-              setMeta(state, "run_id", message.run.friendlyId);
-              if (socket.data.deploymentId) {
-                setMeta(state, "deployment_id", socket.data.deploymentId);
-              }
-              if (socket.data.runnerId) setMeta(state, "runner_id", socket.data.runnerId);
-              state.extras.socket_id = socket.id;
-            },
-          });
         } catch (error) {
           log.error("run:stop error", { error });
         }