test(run-engine): guard the watcher first-write-wins rule and pin the edge-field split

Author: d-cs

internal-packages/run-engine/src/engine/waitpointCoordinator/keys.test.ts

@@ -46,7 +46,9 @@ describe("edgeField", () => {
   it("distinguishes index 0 from an absent index", () => {
     expect(edgeField("w_a", 0)).not.toBe(edgeField("w_a"));
   });
+});
 
+describe("waitpointIdFromEdgeField", () => {
   it("round-trips back to the waitpoint id", () => {
     for (const index of [undefined, null, 0, 7]) {
       expect(waitpointIdFromEdgeField(edgeField("w_a", index))).toBe("w_a");
@@ -56,6 +58,10 @@ describe("edgeField", () => {
   it("returns undefined for a field with no separator", () => {
     expect(waitpointIdFromEdgeField("nope")).toBeUndefined();
   });
+
+  it("splits on the last separator, tolerating a '#' inside the waitpoint id", () => {
+    expect(waitpointIdFromEdgeField("a#b#3")).toBe("a#b");
+  });
 });
 
 describe("watcherField", () => {

internal-packages/run-engine/src/engine/waitpointCoordinator/keys.ts

@@ -31,10 +31,11 @@ export function idempotencyKey(environmentId: string, key: string): string {
   return `wp:idem:{${environmentId}}:${key}`;
 }
 
-// "#" separates the id from the index. A waitpoint id and a run id never contain "#", so
-// the split is unambiguous. An absent index collapses onto the empty suffix, which is how
-// the partial unique index on a null batchIndex behaves; index 0 keeps its own field,
-// because "0" and "" are different strings.
+// "#" separates the id from the index. An absent index collapses onto the empty suffix,
+// which is how the partial unique index on a null batchIndex behaves; index 0 keeps its
+// own field, because "0" and "" are different strings. The split back to an id below is
+// taken from the LAST "#", not the first, so this stays unambiguous even if a waitpoint id
+// or a run id ever contains "#" itself.
 const SEPARATOR = "#";
 
 export function edgeField(waitpointId: string, batchIndex?: number | null): string {
@@ -45,6 +46,9 @@ export function watcherField(runId: string, batchIndex?: number | null): string
   return `${runId}${SEPARATOR}${batchIndex ?? ""}`;
 }
 
+// The last-"#" rule here is re-implemented as a Lua pattern in runClear (scripts.ts). This
+// function has no caller besides its own test, so that test is what pins the rule as a
+// specification the Lua mirrors, not just documentation of this helper.
 export function waitpointIdFromEdgeField(field: string): string | undefined {
   const separator = field.lastIndexOf(SEPARATOR);
   return separator === -1 ? undefined : field.slice(0, separator);

internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts

@@ -252,6 +252,33 @@ describe("registerOrReport", () => {
       await store.quit();
     }
   });
+
+  redisTest("keeps the first registration's watcher on a re-register", async ({ redisOptions }) => {
+    const store = coordinator(redisOptions);
+    try {
+      await store.createIfAbsent({ record: record("w_a"), status: "PENDING" });
+      await store.registerOrReport({
+        waitpointId: "w_a",
+        runId: "run_1",
+        spanIdToComplete: "span_first",
+        createdAt: NOW,
+      });
+      // Same run, same (absent) batch index, so the watcher field collides. HSETNX must
+      // not let this second registration overwrite the first one's span.
+      await store.registerOrReport({
+        waitpointId: "w_a",
+        runId: "run_1",
+        spanIdToComplete: "span_second",
+        createdAt: NOW,
+      });
+
+      const completed = await store.complete({ waitpointId: "w_a", completion: completion() });
+      expect(completed.watchers).toHaveLength(1);
+      expect(completed.watchers[0]!.spanIdToComplete).toBe("span_first");
+    } finally {
+      await store.quit();
+    }
+  });
 });
 
 describe("complete", () => {