Skip to content

Commit 58c358e

Browse files
1stvampTrigger.dev RepoOps
authored andcommitted
feat(webapp): let one worker group carry container and compute consumers
Lets one worker group carry container and compute consumers side by side. Items on an `any` queue lane go to whichever runtime takes them first, and only runs that compute can start are given an `any` lane. Mono-RevId: d2ca67b6bfdd7ca8c940aeaa4471b6af4bab4386
1 parent bf4041c commit 58c358e

17 files changed

Lines changed: 1002 additions & 80 deletions

‎apps/webapp/app/env.server.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1304,6 +1304,17 @@ const EnvironmentSchema = z
13041304
RUN_ENGINE_DEQUEUE_DISABLED_WORKER_QUEUES: z.string().optional(),
13051305
// JSON map of worker-group IDs to allowed v2 subscriptions. Unlisted groups are denied.
13061306
RUN_ENGINE_WORKER_QUEUE_SUBSCRIPTIONS: WorkerQueueSubscriptionPolicyEnv,
1307+
// Comma-separated worker-group IDs that carry container and compute consumers side by side. A
1308+
// MICROVM group must be listed here to consume the container lanes its allowlist names.
1309+
RUN_ENGINE_MIXED_RUNTIME_WORKER_GROUPS: z
1310+
.string()
1311+
.default("")
1312+
.transform((raw) =>
1313+
raw
1314+
.split(",")
1315+
.map((id) => id.trim())
1316+
.filter(Boolean)
1317+
),
13071318
RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS: z.coerce.number().int().default(1000),
13081319
// Off by default. Enable on a single service (e.g. the engine worker) so only one
13091320
// instance reports worker queue length, rather than every replica.

‎apps/webapp/app/runEngine/concerns/computeMigration.server.ts‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,17 @@ export function isOrgMigrated({
4343
return hashBucket(orgId) < pct;
4444
}
4545

46+
/**
47+
* Whether a run may go on an `any` lane. Either runtime takes an `any` item, so only a run that
48+
* compute can start qualifies: a compute-enabled org and a deployment with a template.
49+
*/
50+
export function allowsAnyCompat({
51+
templateKey,
52+
...decision
53+
}: MigrationDecisionInput & { templateKey: string | null | undefined }): boolean {
54+
return !!templateKey && isOrgMigrated(decision);
55+
}
56+
4657
type ResolveInput = MigrationDecisionInput & {
4758
baseWorkerQueue: string | undefined;
4859
baseEnableFastPath: boolean;
@@ -57,7 +68,7 @@ type ResolveInput = MigrationDecisionInput & {
5768
/**
5869
* Produce the target descriptor `{ workerQueue, region, enableFastPath }` for a
5970
* run. When the org is migrated and the region has a compute backing, the queue
60-
* and fast-path setting come from the MICROVM backing group; `region` is the geo
71+
* and fast-path setting come from the compute backing group; `region` is the geo
6172
* either way. Same-geo swap (us-east-1 -> us-east-1-next): any explicit placement
6273
* is a geography preference, honored by staying in-region. Applied after region
6374
* resolution, mirroring the scheduled-split. An environment compat of compute or container
@@ -81,6 +92,8 @@ export function resolveComputeMigration({
8192
enableFastPath: boolean;
8293
/** Set when the missing template key kept a migrated run on the base queue. */
8394
templateFallback?: true;
95+
/** Set when the run moved to the backing, which may be the base queue's own mixed group. */
96+
compute?: true;
8497
} {
8598
const passthrough = { workerQueue: baseWorkerQueue, region, enableFastPath: baseEnableFastPath };
8699
if (baseWorkerQueue === undefined) return passthrough;
@@ -94,5 +107,10 @@ export function resolveComputeMigration({
94107
: isOrgMigrated(decision);
95108
if (!migrated) return passthrough;
96109
if (!templateKey) return { ...passthrough, templateFallback: true };
97-
return { workerQueue: backing.workerQueue, region, enableFastPath: backing.enableFastPath };
110+
return {
111+
workerQueue: backing.workerQueue,
112+
region,
113+
enableFastPath: backing.enableFastPath,
114+
compute: true,
115+
};
98116
}

‎apps/webapp/app/runEngine/concerns/workerQueueAssignment.server.test.ts‎

Lines changed: 59 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
import { describe, expect, it } from "vitest";
2-
import { resolveComputeMigration } from "~/runEngine/concerns/computeMigration.server";
2+
import {
3+
allowsAnyCompat,
4+
resolveComputeMigration,
5+
} from "~/runEngine/concerns/computeMigration.server";
36
import {
47
environmentWorkerQueueOverride,
58
workerQueueForBirth,
@@ -16,6 +19,7 @@ const assignment = {
1619
splitEnabled: true,
1720
workerGroups: [{ masterQueue: "us-east-1-next", workloadType: "MICROVM" as const }],
1821
rootTriggerSource: "api",
22+
anyCompatAllowed: true,
1923
};
2024

2125
describe("workerQueueForBirth", () => {
@@ -242,18 +246,21 @@ describe("per-environment worker queue overrides", () => {
242246
environmentId,
243247
environmentType,
244248
});
249+
const decision = {
250+
planType: "free",
251+
orgId: "org_x",
252+
orgFeatureFlags,
253+
flags: { computeMigrationEnabled: migrationEnabled, computeMigrationFreePercentage: 100 },
254+
templateKey,
255+
};
245256
const migrated = resolveComputeMigration({
257+
...decision,
246258
baseWorkerQueue: "us-east-1",
247259
baseEnableFastPath: false,
248260
region: "us-east-1",
249261
backing,
250-
planType: "free",
251-
orgId: "org_x",
252-
orgFeatureFlags,
253-
flags: { computeMigrationEnabled: migrationEnabled, computeMigrationFreePercentage: 100 },
254262
envType: environmentType,
255263
environmentCompat: environmentOverride.compat,
256-
templateKey,
257264
});
258265
const workerQueue = workerQueueForBirth({
259266
workerQueue: migrated.workerQueue!,
@@ -266,6 +273,7 @@ describe("per-environment worker queue overrides", () => {
266273
workerGroups,
267274
rootTriggerSource: "api",
268275
templateFallback: migrated.templateFallback,
276+
anyCompatAllowed: allowsAnyCompat(decision),
269277
});
270278
return { workerQueue, enableFastPath: migrated.enableFastPath };
271279
}
@@ -341,6 +349,7 @@ describe("per-environment worker queue overrides", () => {
341349
splitEnabled: false,
342350
workerGroups,
343351
rootTriggerSource: "api",
352+
anyCompatAllowed: true,
344353
});
345354
expect(birth(true)).toBe("us-east-1:v2:ondemand:fresh:container:stable");
346355
expect(birth(false)).toBe("us-east-1-next");
@@ -372,10 +381,54 @@ describe("per-environment worker queue overrides", () => {
372381
splitEnabled: false,
373382
workerGroups: [{ masterQueue: "eu-central-1", workloadType: "CONTAINER" }],
374383
rootTriggerSource: "api",
384+
anyCompatAllowed: true,
375385
})
376386
).toBe("eu-central-1:v2:ondemand:fresh:compute:stable");
377387
});
378388

389+
const anyOrg = { [FEATURE_FLAG.workerQueueCompatibility]: "any" };
390+
const anyStaging = {
391+
[FEATURE_FLAG.workerQueueEnvironmentOverrides]: { STAGING: { compat: "any" } },
392+
};
393+
it.each([
394+
{ migrationEnabled: false, templateKey: "a".repeat(64), compat: "container" },
395+
{ migrationEnabled: false, templateKey: null, compat: "container" },
396+
{ migrationEnabled: true, templateKey: null, compat: "container" },
397+
{ migrationEnabled: true, templateKey: "a".repeat(64), compat: "any" },
398+
])(
399+
"gives the any lane only to a migrated org's keyed deployment (migrated=$migrationEnabled, key=$templateKey)",
400+
({ migrationEnabled, templateKey, compat }) => {
401+
for (const orgFeatureFlags of [anyOrg, anyStaging]) {
402+
expect(trigger({ orgFeatureFlags, migrationEnabled, templateKey }).workerQueue).toBe(
403+
`us-east-1:v2:ondemand:fresh:${compat}:stable`
404+
);
405+
}
406+
}
407+
);
408+
409+
it("puts a migrated org's keyless any-compat canary run on the stable container channel", () => {
410+
const canaryOrg = { [FEATURE_FLAG.workerQueueChannel]: "canary" };
411+
const canaryStaging = {
412+
[FEATURE_FLAG.workerQueueEnvironmentOverrides]: { STAGING: { channel: "canary" } },
413+
};
414+
const anyCanaryStaging = {
415+
[FEATURE_FLAG.workerQueueEnvironmentOverrides]: {
416+
STAGING: { compat: "any", channel: "canary" },
417+
},
418+
};
419+
for (const orgFeatureFlags of [
420+
{ ...anyOrg, ...canaryOrg },
421+
{ ...anyStaging, ...canaryOrg },
422+
{ ...anyOrg, ...canaryStaging },
423+
anyCanaryStaging,
424+
]) {
425+
expect(trigger({ orgFeatureFlags, migrationEnabled: true, templateKey: null })).toEqual({
426+
workerQueue: "us-east-1:v2:ondemand:fresh:container:stable",
427+
enableFastPath: false,
428+
});
429+
}
430+
});
431+
379432
it("leaves a channel-only override to the org's migration decision", () => {
380433
const canaryStaging = {
381434
[FEATURE_FLAG.workerQueueEnvironmentOverrides]: { STAGING: { channel: "canary" } },

‎apps/webapp/app/runEngine/concerns/workerQueueAssignment.server.ts‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ export function workerQueueForBirth({
5050
workerGroups,
5151
rootTriggerSource,
5252
templateFallback,
53+
compute,
54+
anyCompatAllowed,
5355
}: {
5456
workerQueue: string;
5557
region: string | undefined;
@@ -62,6 +64,10 @@ export function workerQueueForBirth({
6264
rootTriggerSource: string | undefined;
6365
/** The run stays off compute because its deployment has no template key. */
6466
templateFallback?: boolean;
67+
/** The run moved to its region's compute backing, so it defaults to compute in a mixed group. */
68+
compute?: boolean;
69+
/** The org is compute-enabled and the deployment has a template, so either runtime can run it. */
70+
anyCompatAllowed: boolean;
6571
}): string {
6672
if (envType === "DEVELOPMENT") {
6773
return workerQueue;
@@ -73,7 +79,8 @@ export function workerQueueForBirth({
7379
return workerQueueForRun({ workerQueue, rootTriggerSource, splitEnabled });
7480
}
7581

76-
const defaultCompatibility = group.workloadType === "MICROVM" ? "compute" : "container";
82+
const defaultCompatibility =
83+
compute || group.workloadType === "MICROVM" ? "compute" : "container";
7784
const compatibility = FeatureFlagCatalog[FEATURE_FLAG.workerQueueCompatibility].safeParse(
7885
environmentOverride.compat ??
7986
orgFeatureFlags?.[FEATURE_FLAG.workerQueueCompatibility] ??
@@ -82,7 +89,9 @@ export function workerQueueForBirth({
8289
const channel = FeatureFlagCatalog[FEATURE_FLAG.workerQueueChannel].safeParse(
8390
environmentOverride.channel ?? orgFeatureFlags?.[FEATURE_FLAG.workerQueueChannel] ?? "stable"
8491
);
85-
const compat = compatibility.success ? compatibility.data : defaultCompatibility;
92+
const parsed = compatibility.success ? compatibility.data : defaultCompatibility;
93+
// An `any` item goes to whichever runtime takes it first, so only a run compute can start is one.
94+
const compat = parsed === "any" && !anyCompatAllowed ? "container" : parsed;
8695
// Match the dequeue fallback: container consumers take only the stable channel.
8796
const keylessOnContainers = templateFallback && compat === "container";
8897
const assignment = formatWorkerQueue({

0 commit comments

Comments
 (0)