feat(webapp): accept previous worker auth secrets during rotation

Mono-RevId: 5147714aeb571087890f5bb0b5c4a7a3a592bb39

Author: nicktrn

apps/webapp/app/env.server.ts

@@ -1034,6 +1034,10 @@ const EnvironmentSchema = z
     EVENTS_LOAD_SHEDDING_ENABLED: z.string().default("1"),
 
     MANAGED_WORKER_SECRET: z.string().min(1).refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),
+    MANAGED_WORKER_SECRET_PREVIOUS: z.preprocess(
+      (v) => (v === "" ? undefined : v),
+      z.string().min(1).refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE).optional()
+    ),
 
     // Allow booting with a known-insecure published default secret. Temporary
     // bridge for deployments that can't rotate yet; rotate as soon as possible.

apps/webapp/app/routes/admin.api.v1.workers.$workerGroupId.token.ts

@@ -0,0 +1,88 @@
+import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { tryCatch } from "@trigger.dev/core/utils";
+import { z } from "zod";
+import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
+import { WorkerGroupTokenService } from "~/v3/services/worker/workerGroupTokenService.server";
+
+const DEFAULT_GRACE_SECONDS = 48 * 60 * 60;
+
+const ParamsSchema = z.object({
+  workerGroupId: z.string(),
+});
+
+const RequestBodySchema = z.discriminatedUnion("action", [
+  z.object({
+    action: z.literal("rotate"),
+    graceSeconds: z.number().int().positive().default(DEFAULT_GRACE_SECONDS),
+  }),
+  z.object({
+    action: z.literal("revoke-previous"),
+  }),
+]);
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  await requireAdminApiRequest(request);
+
+  if (request.method !== "POST") {
+    return json({ error: "method not allowed" }, { status: 405 });
+  }
+
+  const { workerGroupId } = ParamsSchema.parse(params);
+
+  const [jsonError, rawBody] = await tryCatch(request.json());
+
+  if (jsonError) {
+    return json({ error: "invalid request body" }, { status: 400 });
+  }
+
+  const parsedBody = RequestBodySchema.safeParse(rawBody ?? {});
+
+  if (!parsedBody.success) {
+    return json(
+      { error: "invalid request body", issues: parsedBody.error.issues },
+      { status: 400 }
+    );
+  }
+
+  const service = new WorkerGroupTokenService();
+
+  switch (parsedBody.data.action) {
+    case "rotate": {
+      const result = await service.rotateToken({
+        workerGroupId,
+        graceMs: parsedBody.data.graceSeconds * 1000,
+      });
+
+      switch (result.outcome) {
+        case "not_found":
+          return json({ error: "worker group not found" }, { status: 404 });
+        case "conflict":
+          return json({ error: "worker group token changed concurrently" }, { status: 409 });
+        case "grace_active":
+          return json(
+            {
+              error:
+                "previous worker group token is still within its grace period; revoke it first",
+              previousTokenValidUntil: result.previousTokenValidUntil,
+            },
+            { status: 409 }
+          );
+        case "rotated":
+          return json({
+            outcome: "rotated worker group token",
+            token: result.plaintext,
+            previousTokenValidUntil: result.previousTokenValidUntil,
+          });
+      }
+    }
+    case "revoke-previous": {
+      const result = await service.revokePreviousToken({ workerGroupId });
+
+      if (!result) {
+        return json({ error: "worker group not found" }, { status: 404 });
+      }
+
+      return json({ outcome: "revoked previous worker group token" });
+    }
+  }
+}

apps/webapp/app/v3/services/worker/managedWorkerSecret.server.ts

@@ -0,0 +1,25 @@
+import { timingSafeEqual } from "node:crypto";
+
+export function matchesAnyManagedWorkerSecret(
+  presented: string,
+  accepted: ReadonlyArray<string | undefined>
+): boolean {
+  const encoder = new TextEncoder();
+  const a = encoder.encode(presented);
+
+  let matched = false;
+
+  for (const candidate of accepted) {
+    if (!candidate) {
+      continue;
+    }
+
+    const b = encoder.encode(candidate);
+
+    if (a.byteLength === b.byteLength && timingSafeEqual(a, b)) {
+      matched = true;
+    }
+  }
+
+  return matched;
+}

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

@@ -28,8 +28,10 @@ import {
   type WorkerInstanceGroupType,
 } from "@trigger.dev/database";
 import { acceptsWorkerStamp } from "./workerGroupAccess";
+import { matchesAnyManagedWorkerSecret } from "./managedWorkerSecret.server";
+import { sanitizeWorkerHeaders } from "./sanitizeWorkerHeaders";
 import { json } from "@remix-run/server-runtime";
-import { createHash, timingSafeEqual } from "crypto";
+import { createHash } from "crypto";
 import { customAlphabet } from "nanoid";
 import { z } from "zod";
 import { env } from "~/env.server";
@@ -129,51 +131,108 @@ export class WorkerGroupTokenService extends WithRunEngine {
     const workerGroup = await this._prisma.workerInstanceGroup.findFirst({
       where: {
         token: {
-          tokenHash,
+          OR: [
+            { tokenHash },
+            { previousTokenHash: tokenHash, previousTokenValidUntil: { gt: new Date() } },
+          ],
         },
       },
+      include: {
+        token: { select: { tokenHash: true, previousTokenValidUntil: true } },
+      },
     });
 
     if (!workerGroup) {
-      logger.warn("[WorkerGroupTokenService] No matching worker group found", { token });
+      logger.warn("[WorkerGroupTokenService] No matching worker group found");
       return null;
     }
 
-    return workerGroup;
+    const { token: matchedToken, ...group } = workerGroup;
+    const tokenValidUntil =
+      matchedToken.tokenHash === tokenHash
+        ? undefined
+        : (matchedToken.previousTokenValidUntil ?? undefined);
+
+    return { ...group, tokenValidUntil };
   }
 
-  async rotateToken({ workerGroupId }: { workerGroupId: string }) {
+  async rotateToken({ workerGroupId, graceMs }: { workerGroupId: string; graceMs: number }) {
     const workerGroup = await this._prisma.workerInstanceGroup.findFirst({
       where: {
         id: workerGroupId,
       },
+      include: {
+        token: true,
+      },
     });
 
     if (!workerGroup) {
       logger.error("[WorkerGroupTokenService] WorkerGroup not found", { workerGroupId });
-      return;
+      return { outcome: "not_found" as const };
+    }
+
+    const activeGraceUntil = workerGroup.token.previousTokenValidUntil;
+    if (activeGraceUntil && activeGraceUntil.getTime() > Date.now()) {
+      logger.error("[WorkerGroupTokenService] Previous token still within grace", {
+        workerGroupId,
+        previousTokenValidUntil: activeGraceUntil,
+      });
+      return { outcome: "grace_active" as const, previousTokenValidUntil: activeGraceUntil };
     }
 
     const rawToken = await this.generateToken();
+    const previousTokenValidUntil = new Date(Date.now() + graceMs);
 
-    const workerGroupToken = await this._prisma.workerGroupToken.update({
+    const updated = await this._prisma.workerGroupToken.updateMany({
       where: {
         id: workerGroup.tokenId,
+        tokenHash: workerGroup.token.tokenHash,
       },
       data: {
         tokenHash: rawToken.hash,
+        previousTokenHash: workerGroup.token.tokenHash,
+        previousTokenValidUntil,
+      },
+    });
+
+    if (updated.count === 0) {
+      logger.error("[WorkerGroupTokenService] Concurrent token rotation detected", {
+        workerGroupId,
+      });
+      return { outcome: "conflict" as const };
+    }
+
+    return {
+      outcome: "rotated" as const,
+      plaintext: rawToken.plaintext,
+      previousTokenValidUntil,
+    };
+  }
+
+  async revokePreviousToken({ workerGroupId }: { workerGroupId: string }) {
+    const workerGroup = await this._prisma.workerInstanceGroup.findFirst({
+      where: {
+        id: workerGroupId,
       },
     });
 
-    if (!workerGroupToken) {
-      logger.error("[WorkerGroupTokenService] WorkerGroupToken not found", { workerGroupId });
+    if (!workerGroup) {
+      logger.error("[WorkerGroupTokenService] WorkerGroup not found", { workerGroupId });
       return;
     }
 
+    const workerGroupToken = await this._prisma.workerGroupToken.update({
+      where: {
+        id: workerGroup.tokenId,
+      },
+      data: {
+        previousTokenHash: null,
+        previousTokenValidUntil: null,
+      },
+    });
+
     return {
       id: workerGroupToken.id,
-      tokenHash: workerGroupToken.tokenHash,
-      plaintext: rawToken.plaintext,
     };
   }
 
@@ -203,7 +262,6 @@ export class WorkerGroupTokenService extends WithRunEngine {
 
     if (!token.startsWith(this.tokenPrefix)) {
       logger.error("[WorkerGroupTokenService] Token does not start with expected prefix", {
-        token,
         prefix: this.tokenPrefix,
       });
       return;
@@ -227,22 +285,13 @@ export class WorkerGroupTokenService extends WithRunEngine {
       return;
     }
 
-    const encoder = new TextEncoder();
-
-    const a = encoder.encode(managedWorkerSecret);
-    const b = encoder.encode(env.MANAGED_WORKER_SECRET);
-
-    if (a.byteLength !== b.byteLength) {
-      logger.error("[WorkerGroupTokenService] Managed secret length mismatch", {
-        managedWorkerSecret,
-        headers: this.sanitizeHeaders(request),
-      });
-      return;
-    }
-
-    if (!timingSafeEqual(a, b)) {
+    if (
+      !matchesAnyManagedWorkerSecret(managedWorkerSecret, [
+        env.MANAGED_WORKER_SECRET,
+        env.MANAGED_WORKER_SECRET_PREVIOUS,
+      ])
+    ) {
       logger.error("[WorkerGroupTokenService] Managed secret mismatch", {
-        managedWorkerSecret,
         headers: this.sanitizeHeaders(request),
       });
       return;
@@ -256,7 +305,7 @@ export class WorkerGroupTokenService extends WithRunEngine {
         const workerGroup = await this.findWorkerGroup({ token });
 
         if (!workerGroup) {
-          logger.warn("[WorkerGroupTokenService] Worker group not found", { token });
+          logger.warn("[WorkerGroupTokenService] Worker group not found", { instanceName });
           return;
         }
 
@@ -285,6 +334,7 @@ export class WorkerGroupTokenService extends WithRunEngine {
           workloadType: workerGroup.workloadType,
           allowedSubscriptions: env.RUN_ENGINE_WORKER_QUEUE_SUBSCRIPTIONS[workerGroup.id] ?? [],
           mixedRuntime: env.RUN_ENGINE_MIXED_RUNTIME_WORKER_GROUPS.includes(workerGroup.id),
+          tokenValidUntil: workerGroup.tokenValidUntil,
         });
       }
     );
@@ -296,6 +346,15 @@ export class WorkerGroupTokenService extends WithRunEngine {
       return;
     }
 
+    if (result.val?.tokenValidUntil && result.val.tokenValidUntil.getTime() <= Date.now()) {
+      await authenticatedWorkerInstanceCache.authenticatedWorkerInstance.remove(cacheKey.join("-"));
+      logger.warn("[WorkerGroupTokenService] Previous worker group token has expired", {
+        instanceName,
+        workerGroupId: result.val.workerGroupId,
+      });
+      return;
+    }
+
     return result.val;
   }
 
@@ -368,16 +427,8 @@ export class WorkerGroupTokenService extends WithRunEngine {
     }
   }
 
-  private sanitizeHeaders(request: Request, skipHeaders = ["authorization"]) {
-    const sanitizedHeaders: Partial<Record<string, string>> = {};
-
-    for (const [key, value] of request.headers.entries()) {
-      if (!skipHeaders.includes(key.toLowerCase())) {
-        sanitizedHeaders[key] = value;
-      }
-    }
-
-    return sanitizedHeaders;
+  private sanitizeHeaders(request: Request) {
+    return sanitizeWorkerHeaders(request.headers);
   }
 }
 
@@ -391,6 +442,7 @@ export type AuthenticatedWorkerInstanceOptions = WithRunEngineOptions<
   WorkerQueueConsumerOptions & {
     type: WorkerInstanceGroupType;
     name: string;
+    tokenValidUntil?: Date;
   }
 >;
 
@@ -400,6 +452,7 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
   readonly workerGroupId: string;
   readonly workerInstanceId: string;
   readonly masterQueue: string;
+  readonly tokenValidUntil?: Date;
   private readonly queueConsumer: WorkerQueueConsumer;
 
   // FIXME: Required for unmanaged workers
@@ -413,6 +466,7 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
     this.workerGroupId = opts.workerGroupId;
     this.workerInstanceId = opts.workerInstanceId;
     this.masterQueue = opts.masterQueue;
+    this.tokenValidUntil = opts.tokenValidUntil;
     this.queueConsumer = createWorkerQueueConsumer(opts);
 
     if (refusedSubscriptions(opts).length > 0 && !warnedRefusedGroups.has(opts.workerGroupId)) {

apps/webapp/test/managedWorkerSecret.test.ts

@@ -0,0 +1,28 @@
+import { describe, expect, it } from "vitest";
+import { matchesAnyManagedWorkerSecret } from "~/v3/services/worker/managedWorkerSecret.server";
+
+describe("matchesAnyManagedWorkerSecret", () => {
+  it("accepts the current secret", () => {
+    expect(matchesAnyManagedWorkerSecret("current", ["current", undefined])).toBe(true);
+  });
+
+  it("accepts the previous secret while one is configured", () => {
+    expect(matchesAnyManagedWorkerSecret("previous", ["current", "previous"])).toBe(true); // Catches: old supervisors 401 mid-rotation
+  });
+
+  it("rejects a secret that matches neither", () => {
+    expect(matchesAnyManagedWorkerSecret("other", ["current", "previous"])).toBe(false);
+  });
+
+  it("rejects a length-mismatched secret without throwing", () => {
+    expect(matchesAnyManagedWorkerSecret("curr", ["current"])).toBe(false); // Catches: timingSafeEqual length throw
+  });
+
+  it("rejects an empty presented secret even when no candidate is set", () => {
+    expect(matchesAnyManagedWorkerSecret("", [undefined, undefined])).toBe(false); // Catches: unset env matching empty header
+  });
+
+  it("never treats an empty candidate as a match", () => {
+    expect(matchesAnyManagedWorkerSecret("", ["", "current"])).toBe(false); // Catches: empty PREVIOUS accepting empty header
+  });
+});