Skip to content

Commit 7eae09b

Browse files
committed
feat: stamp the worker cell and lane on runs at attempt start
The supervisor adds its cell name and isolation lane to every attempt start it proxies for a runner, overriding anything the runner sends. The engine merges both into the run's existing annotations, inside the update that already starts the attempt, and replication projects them into two new task_runs_v2 columns (worker_cell, worker_lane). The cell name comes from the new optional TRIGGER_WORKER_CELL_NAME env var (helm: supervisor.config.cellName). The lane is derived from the selected backend: compute and microvm Runners stamp microvm, pods and containers stamp container. Both also go into the worker instance metadata sent on connect. Admins see both on the run inspector.
1 parent 8e97e44 commit 7eae09b

30 files changed

Lines changed: 734 additions & 8 deletions

File tree

‎apps/supervisor/src/env.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,11 @@ export const Env = z
1414
.object({
1515
// This will come from `spec.nodeName` in k8s
1616
TRIGGER_WORKER_INSTANCE_NAME: z.string().default(randomUUID()),
17+
// Name of the worker cluster (cell) this supervisor serves. Stamped on every run it starts.
18+
TRIGGER_WORKER_CELL_NAME: z
19+
.string()
20+
.regex(/^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/)
21+
.optional(),
1722
TRIGGER_WORKER_HEARTBEAT_INTERVAL_SECONDS: z.coerce.number().default(30),
1823

1924
// Opt-in, dev-only: stream this process's logs over a local telnet/TCP socket on this port.

‎apps/supervisor/src/index.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import { KubernetesWorkloadManager } from "./workloadManager/kubernetes.js";
2121
import {
2222
RunCrdWorkloadManager,
2323
RunnerRestoreInformer,
24+
type RunnerRuntime,
2425
settleRestoreFailure,
2526
} from "./workloadManager/runCrd.js";
2627
import { RestoreWatcher, type RestoreFailure } from "./services/restoreWatcher.js";
@@ -385,7 +386,13 @@ class ManagedSupervisor {
385386

386387
const workerToken = getWorkerToken();
387388

389+
const workerStamp = {
390+
workerCell: env.TRIGGER_WORKER_CELL_NAME,
391+
workerLane: this.workerLane(),
392+
};
393+
388394
this.workerSession = new SupervisorSession({
395+
metadata: workerStamp,
389396
resolveResponseSchema,
390397
workerToken,
391398
apiUrl: env.TRIGGER_API_URL,
@@ -669,6 +676,7 @@ class ManagedSupervisor {
669676

670677
// Responds to workload requests only
671678
this.workloadServer = new WorkloadServer({
679+
workerStamp,
672680
port: env.TRIGGER_WORKLOAD_API_PORT_INTERNAL,
673681
host: env.TRIGGER_WORKLOAD_API_HOST_INTERNAL,
674682
workerClient: this.workerSession.httpClient,
@@ -685,6 +693,19 @@ class ManagedSupervisor {
685693
this.workloadServer.on("runDisconnected", this.onRunDisconnected.bind(this));
686694
}
687695

696+
/** The isolation lane every run started by this supervisor executes in. */
697+
private workerLane(): RunnerRuntime {
698+
switch (this.workloadManagerBackend) {
699+
case "compute":
700+
return "microvm";
701+
case "run-crd":
702+
return env.KUBERNETES_RUNNER_RUNTIME;
703+
case "kubernetes":
704+
case "docker":
705+
return "container";
706+
}
707+
}
708+
688709
async onRunConnected({ run }: { run: { friendlyId: string } }) {
689710
this.logger.debug("Run connected", { run });
690711
// The dispatched run reached a runner on this node - no fallback needed.

‎apps/supervisor/src/workloadServer/index.ts‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,8 @@ type WorkloadServerOptions = {
120120
wideEventOpts: WideEventOptions;
121121
/** When true, high-frequency HTTP routes also emit wide events. */
122122
wideEventsNoisyRoutes: boolean;
123+
/** Where runs started through this server execute. Overrides anything the runner sends. */
124+
workerStamp?: { workerCell?: string; workerLane?: string };
123125
};
124126

125127
export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
@@ -149,6 +151,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
149151
>();
150152

151153
private readonly workerClient: SupervisorHttpClient;
154+
private readonly workerStamp: { workerCell?: string; workerLane?: string };
152155

153156
constructor(opts: WorkloadServerOptions) {
154157
super();
@@ -157,6 +160,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
157160
const port = opts.port;
158161

159162
this.workerClient = opts.workerClient;
163+
this.workerStamp = opts.workerStamp ?? {};
160164
this.checkpointClient = opts.checkpointClient;
161165
this.wideEventOpts = opts.wideEventOpts;
162166
this.wideEventsNoisyRoutes = opts.wideEventsNoisyRoutes;
@@ -416,7 +420,11 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
416420
const startResponse = await this.workerClient.startRunAttempt(
417421
params.runFriendlyId,
418422
params.snapshotFriendlyId,
419-
body,
423+
{
424+
...body,
425+
workerCell: this.workerStamp.workerCell,
426+
workerLane: this.workerStamp.workerLane,
427+
},
420428
this.runnerIdFromRequest(req),
421429
auth.environmentId
422430
);
Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
import { WORKLOAD_HEADERS } from "@trigger.dev/core/v3/workers";
2+
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
3+
4+
vi.mock("std-env", () => ({
5+
env: {
6+
TRIGGER_API_URL: "http://localhost:3030",
7+
TRIGGER_WORKER_TOKEN: "test-token",
8+
MANAGED_WORKER_SECRET: "test-secret",
9+
OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318",
10+
},
11+
}));
12+
13+
const { WorkloadServer } = await import("./index.js");
14+
15+
const PORT = 18733;
16+
const BASE = `http://127.0.0.1:${PORT}`;
17+
18+
const startRunAttempt = vi.fn(async () => ({
19+
success: true as const,
20+
data: { envVars: {} },
21+
}));
22+
23+
const workerClient = { startRunAttempt } as any;
24+
25+
let server: InstanceType<typeof WorkloadServer>;
26+
27+
beforeAll(async () => {
28+
server = new WorkloadServer({
29+
port: PORT,
30+
workerClient,
31+
snapshotCallbackSecret: "snapshot-callback-secret",
32+
wideEventOpts: { service: "supervisor", env: { nodeId: "test" }, enabled: false },
33+
wideEventsNoisyRoutes: false,
34+
workerStamp: { workerCell: "cell-a", workerLane: "microvm" },
35+
});
36+
await server.start();
37+
});
38+
39+
afterAll(async () => {
40+
await server.stop();
41+
});
42+
43+
function startAttempt(body: Record<string, unknown>) {
44+
return fetch(`${BASE}/api/v1/workload-actions/runs/run_1/snapshots/snap_1/attempts/start`, {
45+
method: "POST",
46+
headers: {
47+
"content-type": "application/json",
48+
[WORKLOAD_HEADERS.RUNNER_ID]: "runner_1",
49+
},
50+
body: JSON.stringify(body),
51+
});
52+
}
53+
54+
describe("WorkloadServer attempt start", () => {
55+
it("forwards the runner body with the supervisor's cell and lane stamped on", async () => {
56+
const res = await startAttempt({ isWarmStart: true });
57+
58+
expect(res.status).toBe(200);
59+
expect(startRunAttempt).toHaveBeenLastCalledWith(
60+
"run_1",
61+
"snap_1",
62+
{ isWarmStart: true, workerCell: "cell-a", workerLane: "microvm" },
63+
"runner_1",
64+
undefined
65+
);
66+
});
67+
68+
it("does not let the runner choose its own cell or lane", async () => {
69+
const res = await startAttempt({ workerCell: "spoofed", workerLane: "container" });
70+
71+
expect(res.status).toBe(200);
72+
expect(startRunAttempt).toHaveBeenLastCalledWith(
73+
"run_1",
74+
"snap_1",
75+
{ workerCell: "cell-a", workerLane: "microvm" },
76+
"runner_1",
77+
undefined
78+
);
79+
});
80+
});

‎apps/webapp/app/presenters/v3/SpanPresenter.server.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import {
44
prettyPrintPacket,
55
RunAnnotations,
66
SemanticInternalAttributes,
7+
WorkerStampAnnotations,
78
type TaskRunContext,
89
TaskRunError,
910
TriggerTraceContext,
@@ -325,6 +326,7 @@ export class SpanPresenter extends BasePresenter {
325326
const externalTraceId = this.#getExternalTraceId(run.traceContext);
326327

327328
const taskKind = RunAnnotations.safeParse(run.annotations).data?.taskKind;
329+
const workerStamp = WorkerStampAnnotations.safeParse(run.annotations).data;
328330
const isAgentRun = taskKind === "AGENT";
329331
const isScheduled = taskKind === "SCHEDULED";
330332
const isWebhook = taskKind === "WEBHOOK";
@@ -467,6 +469,8 @@ export class SpanPresenter extends BasePresenter {
467469
engine: run.engine,
468470
region,
469471
workerQueue: run.workerQueue,
472+
workerCell: workerStamp?.workerCell,
473+
workerLane: workerStamp?.workerLane,
470474
traceId: run.traceId,
471475
spanId: run.spanId,
472476
isCached: !!linkedRunId,

‎apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@ export const action = createActionWorkerApiRoute(
2626
runFriendlyId,
2727
snapshotFriendlyId,
2828
isWarmStart: body.isWarmStart,
29+
workerCell: body.workerCell,
30+
workerLane: body.workerLane,
2931
runnerId,
3032
environmentId,
3133
});

‎apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1192,6 +1192,14 @@ function RunBody({
11921192
<Property.Label>Worker queue</Property.Label>
11931193
<Property.Value>{run.workerQueue}</Property.Value>
11941194
</Property.Item>
1195+
<Property.Item>
1196+
<Property.Label>Worker cell</Property.Label>
1197+
<Property.Value>{run.workerCell ?? "–"}</Property.Value>
1198+
</Property.Item>
1199+
<Property.Item>
1200+
<Property.Label>Worker lane</Property.Label>
1201+
<Property.Value>{run.workerLane ?? "–"}</Property.Value>
1202+
</Property.Item>
11951203
<Property.Item>
11961204
<Property.Label>Trace ID</Property.Label>
11971205
<Property.Value>{run.traceId}</Property.Value>

‎apps/webapp/app/services/runsReplicationService.server.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ import {
3434
unsafeExtractIdempotencyKeyScope,
3535
unsafeExtractIdempotencyKeyUser,
3636
} from "@trigger.dev/core/v3/serverOnly";
37-
import { RunAnnotations } from "@trigger.dev/core/v3";
37+
import { RunAnnotations, WorkerStampAnnotations } from "@trigger.dev/core/v3";
3838
import { type TaskRun } from "@trigger.dev/database";
3939
import { PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON } from "@internal/run-engine";
4040
import { nanoid } from "nanoid";
@@ -1323,6 +1323,7 @@ export class RunsReplicationService {
13231323
: "";
13241324

13251325
const annotations = this.#parseAnnotations(run.annotations);
1326+
const workerStamp = WorkerStampAnnotations.safeParse(run.annotations).data;
13261327

13271328
// Return array matching TASK_RUN_COLUMNS order
13281329
return [
@@ -1382,6 +1383,8 @@ export class RunsReplicationService {
13821383
annotations?.taskKind ?? "", // task_kind
13831384
run.isWarmStart ?? null, // is_warm_start
13841385
this.#readExternalDeploymentId(run),
1386+
workerStamp?.workerCell ?? "", // worker_cell
1387+
workerStamp?.workerLane ?? "", // worker_lane
13851388
];
13861389
}
13871390

‎apps/webapp/app/v3/mollifier/syntheticSpanRun.server.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,8 @@ export async function buildSyntheticSpanRun(args: {
197197
engine: "V2",
198198
region: null,
199199
workerQueue: run.workerQueue ?? "",
200+
workerCell: undefined,
201+
workerLane: undefined,
200202
traceId: run.traceId ?? "",
201203
spanId: run.spanId ?? "",
202204
isCached: false,

‎apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -515,12 +515,16 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
515515
runFriendlyId,
516516
snapshotFriendlyId,
517517
isWarmStart,
518+
workerCell,
519+
workerLane,
518520
runnerId,
519521
environmentId,
520522
}: {
521523
runFriendlyId: string;
522524
snapshotFriendlyId: string;
523525
isWarmStart?: boolean;
526+
workerCell?: string;
527+
workerLane?: string;
524528
runnerId?: string;
525529
environmentId?: string;
526530
}): Promise<
@@ -538,6 +542,8 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
538542
runId: fromFriendlyId(runFriendlyId),
539543
snapshotId: fromFriendlyId(snapshotFriendlyId),
540544
isWarmStart,
545+
workerCell,
546+
workerLane,
541547
workerId: this.workerInstanceId,
542548
runnerId,
543549
environmentId,

0 commit comments

Comments
 (0)