fix: make snapshot dispatch limit configurable via COMPUTE_SNAPSHOT_DISPATCH_LIMIT

Author: nicktrn

apps/supervisor/src/env.ts

@@ -86,6 +86,7 @@ const Env = z
     COMPUTE_TRACE_SPANS_ENABLED: BoolEnv.default(true),
     COMPUTE_TRACE_OTLP_ENDPOINT: z.string().url().optional(), // Override for span export (derived from TRIGGER_API_URL if unset)
     COMPUTE_SNAPSHOT_DELAY_MS: z.coerce.number().int().min(0).max(60_000).default(5_000),
+    COMPUTE_SNAPSHOT_DISPATCH_LIMIT: z.coerce.number().int().min(1).max(100).default(10),
 
     // Kubernetes settings
     KUBERNETES_FORCE_ENABLED: BoolEnv.default(false),

apps/supervisor/src/index.ts

@@ -110,6 +110,7 @@ class ManagedSupervisor {
         snapshots: {
           enabled: env.COMPUTE_SNAPSHOTS_ENABLED,
           delayMs: env.COMPUTE_SNAPSHOT_DELAY_MS,
+          dispatchLimit: env.COMPUTE_SNAPSHOT_DISPATCH_LIMIT,
           callbackUrl,
         },
         tracing: this.tracing,

apps/supervisor/src/workloadManager/compute.ts

@@ -20,6 +20,7 @@ type ComputeWorkloadManagerOptions = WorkloadManagerOptions & {
   snapshots: {
     enabled: boolean;
     delayMs: number;
+    dispatchLimit: number;
     callbackUrl: string;
   };
   tracing?: OtlpTraceService;
@@ -56,6 +57,10 @@ export class ComputeWorkloadManager implements WorkloadManager {
     return this.opts.snapshots.delayMs;
   }
 
+  get snapshotDispatchLimit(): number {
+    return this.opts.snapshots.dispatchLimit;
+  }
+
   get traceSpansEnabled(): boolean {
     return !!this.opts.tracing;
   }

apps/supervisor/src/workloadServer/index.ts

@@ -120,10 +120,9 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
   // hours later after a checkpoint/restore cycle. Using a capped map avoids unbounded
   // growth while keeping recent contexts available. Oldest entries are evicted first.
   private static readonly MAX_TRACE_CONTEXTS = 10_000;
-  private static readonly SNAPSHOT_CONCURRENCY = 10;
   private readonly runTraceContexts = new Map<string, RunTraceContext>();
   private readonly snapshotDelayWheel?: TimerWheel<DelayedSnapshot>;
-  private readonly snapshotLimit = pLimit(WorkloadServer.SNAPSHOT_CONCURRENCY);
+  private readonly snapshotLimit?: ReturnType<typeof pLimit>;
 
   constructor(opts: WorkloadServerOptions) {
     super();
@@ -137,10 +136,12 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
     this.tracing = opts.tracing;
 
     if (this.computeManager?.snapshotsEnabled) {
+      const snapshotLimit = pLimit(this.computeManager.snapshotDispatchLimit);
+      this.snapshotLimit = snapshotLimit;
       this.snapshotDelayWheel = new TimerWheel<DelayedSnapshot>({
         delayMs: this.computeManager.snapshotDelayMs,
         onExpire: (item) => {
-          this.snapshotLimit(() => this.dispatchComputeSnapshot(item.data)).catch((error) => {
+          snapshotLimit(() => this.dispatchComputeSnapshot(item.data)).catch((error) => {
             this.logger.error("Compute snapshot dispatch failed", {
               runId: item.data.runFriendlyId,
               runnerId: item.data.runnerId,