fix(webapp,run-engine,scripts): harden v2 cross-table tests and tighten the recovery query

- updateMetadata cross-table test: wrap the body in try/finally so stopFlushing
  always runs and the flush loop cannot bleed into later tests on a failure path.
- cancelling cross-table cancel-cascade test: poll for the child CANCELED status
  with a deadline instead of a fixed 1s sleep, to de-flake it under slow CI.
- recover-stuck-runs: constrain each UNION branch by id = ANY(runIds) so the
  recovery query scans only candidate rows instead of unioning both full tables
  before the join.

Author: d-cs

apps/webapp/test/updateMetadata.test.ts

@@ -1306,98 +1306,105 @@ describe("UpdateMetadataService.call", () => {
         logLevel: "debug",
       });
 
-      const organization = await prisma.organization.create({
-        data: { title: "test", slug: "test" },
-      });
-      const project = await prisma.project.create({
-        data: { name: "test", slug: "test", organizationId: organization.id, externalRef: "test" },
-      });
-      const runtimeEnvironment = await prisma.runtimeEnvironment.create({
-        data: {
-          slug: "test",
-          type: "DEVELOPMENT",
-          projectId: project.id,
-          organizationId: organization.id,
-          apiKey: "test",
-          pkApiKey: "test",
-          shortcode: "test",
-        },
-      });
-
-      // Legacy parent (cuid id) lives in TaskRun. This is the mixed-window
-      // hierarchy: an org flips runTableV2 on while a pre-flip parent is live,
-      // and its post-flip child mints a ksuid into task_run_v2.
-      const parentId = RunId.generate();
-      expect(isKsuidId(parentId.id)).toBe(false);
-      const parentTaskRun = await prisma.taskRun.create({
-        data: {
-          id: parentId.id,
-          friendlyId: parentId.friendlyId,
-          taskIdentifier: "my-task",
-          payload: "{}",
-          payloadType: "application/json",
-          traceId: "t",
-          spanId: "s",
-          queue: "test",
-          runtimeEnvironmentId: runtimeEnvironment.id,
-          projectId: project.id,
-          organizationId: organization.id,
-          environmentType: "DEVELOPMENT",
-          engine: "V2",
-        },
-      });
-
-      // v2 child (ksuid id) lives in task_run_v2 and points at the legacy
-      // parent by the scalar parentTaskRunId (no cross-table FK).
-      const childId = RunId.generateKsuid();
-      expect(isKsuidId(childId.id)).toBe(true);
-      await prisma.taskRunV2.create({
-        data: {
-          id: childId.id,
-          friendlyId: childId.friendlyId,
-          taskIdentifier: "my-child-task",
-          payload: "{}",
-          payloadType: "application/json",
-          traceId: "t",
-          spanId: "s",
-          queue: "test",
-          runtimeEnvironmentId: runtimeEnvironment.id,
-          projectId: project.id,
-          organizationId: organization.id,
-          environmentType: "DEVELOPMENT",
-          engine: "V2",
-          parentTaskRunId: parentTaskRun.id,
-        },
-      });
-
-      // The child applies metadata.parent operations. Pre-fix, the table-bound
-      // parentTaskRun relation resolved null (parent is in the OTHER table), so
-      // the ops fell back to the child's own id — corrupting the child and
-      // never touching the parent.
-      await service.call(childId.id, {
-        parentOperations: [
-          { type: "set", key: "foo", value: "bar" },
-          { type: "append", key: "bar", value: "baz" },
-        ],
-      });
-
-      // Wait for the buffered operations to flush.
-      await setTimeout(1000);
-
-      // The PARENT (in TaskRun) must have received the operations.
-      const updatedParent = await prisma.taskRun.findFirst({ where: { id: parentTaskRun.id } });
-      expect(
-        await parsePacket({
-          data: updatedParent?.metadata ?? undefined,
-          dataType: updatedParent?.metadataType ?? "application/json",
-        })
-      ).toEqual({ foo: "bar", bar: ["baz"] });
-
-      // The CHILD (in task_run_v2) must NOT have been polluted with parent ops.
-      const updatedChild = await prisma.taskRunV2.findFirst({ where: { id: childId.id } });
-      expect(updatedChild?.metadata ?? null).toBeNull();
-
-      service.stopFlushing();
+      try {
+        const organization = await prisma.organization.create({
+          data: { title: "test", slug: "test" },
+        });
+        const project = await prisma.project.create({
+          data: {
+            name: "test",
+            slug: "test",
+            organizationId: organization.id,
+            externalRef: "test",
+          },
+        });
+        const runtimeEnvironment = await prisma.runtimeEnvironment.create({
+          data: {
+            slug: "test",
+            type: "DEVELOPMENT",
+            projectId: project.id,
+            organizationId: organization.id,
+            apiKey: "test",
+            pkApiKey: "test",
+            shortcode: "test",
+          },
+        });
+
+        // Legacy parent (cuid id) lives in TaskRun. This is the mixed-window
+        // hierarchy: an org flips runTableV2 on while a pre-flip parent is live,
+        // and its post-flip child mints a ksuid into task_run_v2.
+        const parentId = RunId.generate();
+        expect(isKsuidId(parentId.id)).toBe(false);
+        const parentTaskRun = await prisma.taskRun.create({
+          data: {
+            id: parentId.id,
+            friendlyId: parentId.friendlyId,
+            taskIdentifier: "my-task",
+            payload: "{}",
+            payloadType: "application/json",
+            traceId: "t",
+            spanId: "s",
+            queue: "test",
+            runtimeEnvironmentId: runtimeEnvironment.id,
+            projectId: project.id,
+            organizationId: organization.id,
+            environmentType: "DEVELOPMENT",
+            engine: "V2",
+          },
+        });
+
+        // v2 child (ksuid id) lives in task_run_v2 and points at the legacy
+        // parent by the scalar parentTaskRunId (no cross-table FK).
+        const childId = RunId.generateKsuid();
+        expect(isKsuidId(childId.id)).toBe(true);
+        await prisma.taskRunV2.create({
+          data: {
+            id: childId.id,
+            friendlyId: childId.friendlyId,
+            taskIdentifier: "my-child-task",
+            payload: "{}",
+            payloadType: "application/json",
+            traceId: "t",
+            spanId: "s",
+            queue: "test",
+            runtimeEnvironmentId: runtimeEnvironment.id,
+            projectId: project.id,
+            organizationId: organization.id,
+            environmentType: "DEVELOPMENT",
+            engine: "V2",
+            parentTaskRunId: parentTaskRun.id,
+          },
+        });
+
+        // The child applies metadata.parent operations. Pre-fix, the table-bound
+        // parentTaskRun relation resolved null (parent is in the OTHER table), so
+        // the ops fell back to the child's own id — corrupting the child and
+        // never touching the parent.
+        await service.call(childId.id, {
+          parentOperations: [
+            { type: "set", key: "foo", value: "bar" },
+            { type: "append", key: "bar", value: "baz" },
+          ],
+        });
+
+        // Wait for the buffered operations to flush.
+        await setTimeout(1000);
+
+        // The PARENT (in TaskRun) must have received the operations.
+        const updatedParent = await prisma.taskRun.findFirst({ where: { id: parentTaskRun.id } });
+        expect(
+          await parsePacket({
+            data: updatedParent?.metadata ?? undefined,
+            dataType: updatedParent?.metadataType ?? "application/json",
+          })
+        ).toEqual({ foo: "bar", bar: ["baz"] });
+
+        // The CHILD (in task_run_v2) must NOT have been polluted with parent ops.
+        const updatedChild = await prisma.taskRunV2.findFirst({ where: { id: childId.id } });
+        expect(updatedChild?.metadata ?? null).toBeNull();
+      } finally {
+        service.stopFlushing();
+      }
     }
   );
 });

internal-packages/run-engine/src/engine/tests/cancelling.test.ts

@@ -330,10 +330,14 @@ describe("RunEngine cancelling", () => {
           reason: "Cancelled by the user",
         });
 
-        // The child cancellation is enqueued as a job; give the worker a moment.
-        await setTimeout(1000);
-
-        const childData = await engine.getRunExecutionData({ runId: childRun.id });
+        // The child cancellation is enqueued as a job; wait for the worker to process it
+        // (poll instead of a fixed sleep so the test isn't flaky under slow CI).
+        let childData = await engine.getRunExecutionData({ runId: childRun.id });
+        const deadline = Date.now() + 5_000;
+        while (childData?.run.status !== "CANCELED" && Date.now() < deadline) {
+          await setTimeout(50);
+          childData = await engine.getRunExecutionData({ runId: childRun.id });
+        }
         expect(childData?.run.status).toBe("CANCELED");
       } finally {
         await engine.quit();

scripts/recover-stuck-runs.ts

@@ -220,9 +220,9 @@ async function main() {
           r."concurrencyKey"
         FROM "TaskRunExecutionSnapshot" s
         INNER JOIN (
-          SELECT id, "organizationId", "projectId", "runtimeEnvironmentId", "taskIdentifier", "queue", "concurrencyKey" FROM "TaskRun"
+          SELECT id, "organizationId", "projectId", "runtimeEnvironmentId", "taskIdentifier", "queue", "concurrencyKey" FROM "TaskRun" WHERE id = ANY(${runIds})
           UNION
-          SELECT id, "organizationId", "projectId", "runtimeEnvironmentId", "taskIdentifier", "queue", "concurrencyKey" FROM task_run_v2
+          SELECT id, "organizationId", "projectId", "runtimeEnvironmentId", "taskIdentifier", "queue", "concurrencyKey" FROM task_run_v2 WHERE id = ANY(${runIds})
         ) r ON r.id = s."runId"
         WHERE s."runId" = ANY(${runIds})
           AND s."isValid" = true