chore(run-engine): ship note + comment/format cleanup

1stvamp · 1stvamp · commit b748a4f9ed34 · 2026-07-24T17:51:00.000+01:00
Task 10 (final). Adds the .server-changes note (ships dark, off by default), fixes
three stale/misleading test comments (keyProducer var naming; the un-seeded test 4
comment; the flag-off KEYS-scan comment now correctly credits redisTest flushall),
and applies format. No production logic changed.
diff --git a/.server-changes/2026-07-24-ck-fair-scheduling.md b/.server-changes/2026-07-24-ck-fair-scheduling.md
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+The run queue can now schedule runs across concurrency key variants fairly, so one tenant or key with a large backlog can't starve runs waiting on other keys. This is opt-in via a flag and off by default, so nothing changes unless it's enabled.
diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts
@@ -127,7 +127,8 @@ describe("CK virtual-time (SFQ) dequeue", () => {
       const lightVariant = variantName("light");
       const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(heavyVariant);
 
-      // Task 4 (enqueue registration) isn't done yet; seed both variants at tag 0.
+      // Explicit seed is redundant now that enqueue registers variants at the
+      // floor itself; kept as a belt-and-braces fixture.
       await queue.redis.zadd(ckVtimeKey, 0, heavyVariant, 0, lightVariant);
 
       const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
@@ -537,8 +538,9 @@ describe("CK virtual-time (SFQ) dequeue", () => {
         const t0 = Date.now() - 100_000;
 
         // Enqueue on a variant but do NOT register it in ckVtime (simulating an
-        // enqueue from old code / before Task 4). Two messages so the variant
-        // survives its first serve and we can observe it was registered.
+        // enqueue from old code that predates enqueue-time registration, e.g.
+        // during a rolling deploy). Two messages so the variant survives its
+        // first serve and we can observe it was registered.
         for (let i = 0; i < 2; i++) {
           await queue.enqueueMessage({
             env: authenticatedEnvDev,
@@ -897,7 +899,11 @@ describe("CK virtual-time (SFQ) dequeue", () => {
           for (let i = 0; i < 3; i++) {
             await queue.enqueueMessage({
               env: authenticatedEnvDev,
-              message: makeMessage({ runId: `r-${ck}-${i}`, concurrencyKey: ck, timestamp: head + i }),
+              message: makeMessage({
+                runId: `r-${ck}-${i}`,
+                concurrencyKey: ck,
+                timestamp: head + i,
+              }),
               workerQueue: authenticatedEnvDev.id,
               skipDequeueProcessing: true,
             });
@@ -944,9 +950,9 @@ describe("CK virtual-time (SFQ) dequeue", () => {
 
         // After the whole mixed sequence (enqueues, batched dequeues, a nack,
         // acks, one message still in flight so the keyspace is non-empty) no
-        // vtime state exists at all: no :ckVtime, no :ckVtimeFloor. ioredis
-        // prepends the keyPrefix to the KEYS pattern, so this scans exactly
-        // this test's keyspace.
+        // vtime state exists at all: no :ckVtime, no :ckVtimeFloor. The KEYS
+        // scan is safe here because redisTest runs flushall before each test,
+        // so the DB only holds this test's keys.
         const allKeys = await queue.redis.keys("*");
         expect(allKeys.length).toBeGreaterThan(0);
         expect(allKeys.filter((k) => k.includes("ckVtime"))).toEqual([]);
diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtimeFairness.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtimeFairness.test.ts
@@ -250,96 +250,96 @@ describe("CK virtual-time fairness on the real batched dequeue path", () => {
   // The head-age starvation the spike measured appears on this path when env
   // capacity serializes the calls (limit 1): flag OFF then always picks the
   // globally oldest head, which is heavy for its whole backlog.
-  redisTest("ckSkew: light keys stop waiting behind the heavy backlog", async ({
-    redisContainer,
-  }) => {
-    const t0 = Date.now() - 500_000;
-    const messages: ScenarioMessage[] = [];
-    for (let i = 0; i < 120; i++) {
-      messages.push({ runId: `heavy-${i}`, ck: "heavy", timestamp: t0 });
-    }
-    for (let i = 0; i < 10; i++) {
-      for (let k = 0; k < 4; k++) {
-        messages.push({
-          runId: `light${k}-${i}`,
-          ck: `light${k}`,
-          timestamp: t0 + 10_000 + i * 4 + k,
-        });
+  redisTest(
+    "ckSkew: light keys stop waiting behind the heavy backlog",
+    async ({ redisContainer }) => {
+      const t0 = Date.now() - 500_000;
+      const messages: ScenarioMessage[] = [];
+      for (let i = 0; i < 120; i++) {
+        messages.push({ runId: `heavy-${i}`, ck: "heavy", timestamp: t0 });
       }
+      for (let i = 0; i < 10; i++) {
+        for (let k = 0; k < 4; k++) {
+          messages.push({
+            runId: `light${k}-${i}`,
+            ck: `light${k}`,
+            timestamp: t0 + 10_000 + i * 4 + k,
+          });
+        }
+      }
+      const scenario: Scenario = {
+        name: "ckSkew",
+        messages,
+        envConcurrencyLimit: 1,
+        holdSteps: 3,
+        maxSteps: 1_000,
+      };
+
+      const on = await runScenario(redisContainer, scenario, true);
+      const off = await runScenario(redisContainer, scenario, false);
+
+      assertConservation(scenario, on, off);
+
+      const isLight = (ck: string) => ck.startsWith("light");
+      const onWait = meanWait(on, isLight);
+      const offWait = meanWait(off, isLight);
+      debugLog("ckSkew", { onWait, offWait, ratio: onWait / offWait });
+
+      // Heavy's wait may rise under the fair order; that is expected and not
+      // asserted down.
+      expect(onWait).toBeLessThanOrEqual(0.3 * offWait);
     }
-    const scenario: Scenario = {
-      name: "ckSkew",
-      messages,
-      envConcurrencyLimit: 1,
-      holdSteps: 3,
-      maxSteps: 1_000,
-    };
-
-    const on = await runScenario(redisContainer, scenario, true);
-    const off = await runScenario(redisContainer, scenario, false);
-
-    assertConservation(scenario, on, off);
-
-    const isLight = (ck: string) => ck.startsWith("light");
-    const onWait = meanWait(on, isLight);
-    const offWait = meanWait(off, isLight);
-    debugLog("ckSkew", { onWait, offWait, ratio: onWait / offWait });
-
-    // Heavy's wait may rise under the fair order; that is expected and not
-    // asserted down.
-    expect(onWait).toBeLessThanOrEqual(0.3 * offWait);
-  });
+  );
 
   // ckTrickle (spike shape): one bulk key with a 120-message backlog on an old
   // shared head, two trickle keys with 15 messages each on later heads. Same
   // serialized contention regime as ckSkew, same assertion.
-  redisTest("ckTrickle: trickle keys stop waiting behind the bulk backlog", async ({
-    redisContainer,
-  }) => {
-    const t0 = Date.now() - 500_000;
-    const messages: ScenarioMessage[] = [];
-    for (let i = 0; i < 120; i++) {
-      messages.push({ runId: `bulk-${i}`, ck: "bulk", timestamp: t0 });
-    }
-    for (let i = 0; i < 15; i++) {
-      for (let k = 0; k < 2; k++) {
-        messages.push({
-          runId: `trickle${k}-${i}`,
-          ck: `trickle${k}`,
-          timestamp: t0 + 10_000 + i * 2 + k,
-        });
+  redisTest(
+    "ckTrickle: trickle keys stop waiting behind the bulk backlog",
+    async ({ redisContainer }) => {
+      const t0 = Date.now() - 500_000;
+      const messages: ScenarioMessage[] = [];
+      for (let i = 0; i < 120; i++) {
+        messages.push({ runId: `bulk-${i}`, ck: "bulk", timestamp: t0 });
       }
-    }
-    const scenario: Scenario = {
-      name: "ckTrickle",
-      messages,
-      envConcurrencyLimit: 1,
-      holdSteps: 3,
-      maxSteps: 1_000,
-    };
+      for (let i = 0; i < 15; i++) {
+        for (let k = 0; k < 2; k++) {
+          messages.push({
+            runId: `trickle${k}-${i}`,
+            ck: `trickle${k}`,
+            timestamp: t0 + 10_000 + i * 2 + k,
+          });
+        }
+      }
+      const scenario: Scenario = {
+        name: "ckTrickle",
+        messages,
+        envConcurrencyLimit: 1,
+        holdSteps: 3,
+        maxSteps: 1_000,
+      };
 
-    const on = await runScenario(redisContainer, scenario, true);
-    const off = await runScenario(redisContainer, scenario, false);
+      const on = await runScenario(redisContainer, scenario, true);
+      const off = await runScenario(redisContainer, scenario, false);
 
-    assertConservation(scenario, on, off);
+      assertConservation(scenario, on, off);
 
-    const isTrickle = (ck: string) => ck.startsWith("trickle");
-    const onWait = meanWait(on, isTrickle);
-    const offWait = meanWait(off, isTrickle);
-    debugLog("ckTrickle", { onWait, offWait, ratio: onWait / offWait });
+      const isTrickle = (ck: string) => ck.startsWith("trickle");
+      const onWait = meanWait(on, isTrickle);
+      const offWait = meanWait(off, isTrickle);
+      debugLog("ckTrickle", { onWait, offWait, ratio: onWait / offWait });
 
-    expect(onWait).toBeLessThanOrEqual(0.3 * offWait);
-  });
+      expect(onWait).toBeLessThanOrEqual(0.3 * offWait);
+    }
+  );
 
   // ckSybil (spike shape, the case per-key caps cannot fix): 20 attacker keys
   // with 8 messages each, all on older heads, and 1 light key with 10 newer
   // messages. 21 variants against a batch of 10 exercises the batched path
   // properly: flag OFF walks the age order and only reaches the light key when
   // the attackers are nearly drained; flag ON serves the light key from the
   // floor on its first fair round.
-  redisTest("ckSybil: many attacker keys cannot starve a light key", async ({
-    redisContainer,
-  }) => {
+  redisTest("ckSybil: many attacker keys cannot starve a light key", async ({ redisContainer }) => {
     const t0 = Date.now() - 500_000;
     const messages: ScenarioMessage[] = [];
     for (let i = 0; i < 8; i++) {
@@ -397,72 +397,74 @@ describe("CK virtual-time fairness on the real batched dequeue path", () => {
 
   // ckBalanced (spike shape, no-harm check): 4 symmetric keys with 25 messages
   // each. The fair order must not make the symmetric case worse.
-  redisTest("ckBalanced: fair order does not hurt the symmetric case", async ({
-    redisContainer,
-  }) => {
-    const t0 = Date.now() - 500_000;
-    const cks = ["bal0", "bal1", "bal2", "bal3"];
-    const messages: ScenarioMessage[] = [];
-    for (let i = 0; i < 25; i++) {
-      for (let k = 0; k < cks.length; k++) {
-        messages.push({
-          runId: `${cks[k]}-${i}`,
-          ck: cks[k]!,
-          timestamp: t0 + i * 4 + k,
-        });
+  redisTest(
+    "ckBalanced: fair order does not hurt the symmetric case",
+    async ({ redisContainer }) => {
+      const t0 = Date.now() - 500_000;
+      const cks = ["bal0", "bal1", "bal2", "bal3"];
+      const messages: ScenarioMessage[] = [];
+      for (let i = 0; i < 25; i++) {
+        for (let k = 0; k < cks.length; k++) {
+          messages.push({
+            runId: `${cks[k]}-${i}`,
+            ck: cks[k]!,
+            timestamp: t0 + i * 4 + k,
+          });
+        }
       }
-    }
-    const scenario: Scenario = {
-      name: "ckBalanced",
-      messages,
-      envConcurrencyLimit: 4,
-      holdSteps: 3,
-      maxSteps: 500,
-    };
+      const scenario: Scenario = {
+        name: "ckBalanced",
+        messages,
+        envConcurrencyLimit: 4,
+        holdSteps: 3,
+        maxSteps: 500,
+      };
 
-    const on = await runScenario(redisContainer, scenario, true);
-    const off = await runScenario(redisContainer, scenario, false);
+      const on = await runScenario(redisContainer, scenario, true);
+      const off = await runScenario(redisContainer, scenario, false);
 
-    assertConservation(scenario, on, off);
+      assertConservation(scenario, on, off);
 
-    const maxPerKeyMeanWait = (result: ScenarioResult) =>
-      Math.max(...cks.map((ck) => meanWait(result, (c) => c === ck)));
+      const maxPerKeyMeanWait = (result: ScenarioResult) =>
+        Math.max(...cks.map((ck) => meanWait(result, (c) => c === ck)));
 
-    const onMax = maxPerKeyMeanWait(on);
-    const offMax = maxPerKeyMeanWait(off);
-    debugLog("ckBalanced", { onMax, offMax, ratio: onMax / offMax });
+      const onMax = maxPerKeyMeanWait(on);
+      const offMax = maxPerKeyMeanWait(off);
+      debugLog("ckBalanced", { onMax, offMax, ratio: onMax / offMax });
 
-    expect(onMax).toBeLessThanOrEqual(1.25 * offMax);
-  });
+      expect(onMax).toBeLessThanOrEqual(1.25 * offMax);
+    }
+  );
 
   // ckHeavyIdle (spike shape, work conservation): a single key with 60
   // messages and nothing else contending. Any extra step to drain under the
   // fair order is a work-conservation bug, so the step counts must be exactly
   // equal.
-  redisTest("ckHeavyIdle: a lone key drains in exactly the same steps", async ({
-    redisContainer,
-  }) => {
-    const t0 = Date.now() - 500_000;
-    const messages: ScenarioMessage[] = [];
-    for (let i = 0; i < 60; i++) {
-      messages.push({ runId: `solo-${i}`, ck: "solo", timestamp: t0 + i });
-    }
-    const scenario: Scenario = {
-      name: "ckHeavyIdle",
-      messages,
-      envConcurrencyLimit: 25,
-      holdSteps: 3,
-      maxSteps: 300,
-    };
+  redisTest(
+    "ckHeavyIdle: a lone key drains in exactly the same steps",
+    async ({ redisContainer }) => {
+      const t0 = Date.now() - 500_000;
+      const messages: ScenarioMessage[] = [];
+      for (let i = 0; i < 60; i++) {
+        messages.push({ runId: `solo-${i}`, ck: "solo", timestamp: t0 + i });
+      }
+      const scenario: Scenario = {
+        name: "ckHeavyIdle",
+        messages,
+        envConcurrencyLimit: 25,
+        holdSteps: 3,
+        maxSteps: 300,
+      };
 
-    const on = await runScenario(redisContainer, scenario, true);
-    const off = await runScenario(redisContainer, scenario, false);
+      const on = await runScenario(redisContainer, scenario, true);
+      const off = await runScenario(redisContainer, scenario, false);
 
-    assertConservation(scenario, on, off);
+      assertConservation(scenario, on, off);
 
-    debugLog("ckHeavyIdle", { onDrainStep: on.drainStep, offDrainStep: off.drainStep });
+      debugLog("ckHeavyIdle", { onDrainStep: on.drainStep, offDrainStep: off.drainStep });
 
-    expect(on.drainStep).toBeGreaterThanOrEqual(0);
-    expect(on.drainStep).toBe(off.drainStep);
-  });
+      expect(on.drainStep).toBeGreaterThanOrEqual(0);
+      expect(on.drainStep).toBe(off.drainStep);
+    }
+  );
 });
diff --git a/internal-packages/run-engine/src/run-queue/tests/keyProducer.test.ts b/internal-packages/run-engine/src/run-queue/tests/keyProducer.test.ts
@@ -434,17 +434,17 @@ describe("KeyProducer", () => {
   });
 
   it("produces ckVtime keys from a CK variant queue name", () => {
-    const keys = new RunQueueFullKeyProducer();
+    const keyProducer = new RunQueueFullKeyProducer();
     const q = "{org:o1}:proj:p1:env:e1:queue:task/my-task:ck:tenant-a";
-    expect(keys.ckVtimeKeyFromQueue(q)).toBe(
+    expect(keyProducer.ckVtimeKeyFromQueue(q)).toBe(
       "{org:o1}:proj:p1:env:e1:queue:task/my-task:ckVtime"
     );
-    expect(keys.ckVtimeFloorKeyFromQueue(q)).toBe(
+    expect(keyProducer.ckVtimeFloorKeyFromQueue(q)).toBe(
       "{org:o1}:proj:p1:env:e1:queue:task/my-task:ckVtimeFloor"
     );
     // ck wildcard and base-queue inputs normalise the same way
-    expect(keys.ckVtimeKeyFromQueue(q.replace(":ck:tenant-a", ":ck:*"))).toBe(
-      keys.ckVtimeKeyFromQueue(q)
+    expect(keyProducer.ckVtimeKeyFromQueue(q.replace(":ck:tenant-a", ":ck:*"))).toBe(
+      keyProducer.ckVtimeKeyFromQueue(q)
     );
   });
 });
