fix(agbossjs/agboss#720): the child queue's claim marks exactly the child it answers — a MATERIALIZED CTE, run once; core at 06c4bc1d

.agboss-version.json

@@ -2,6 +2,6 @@
   "repository": "agbossjs/agboss",
   "branch": "main",
   "source": "commit",
-  "frameworkSha": "606370a0012178f69352275fc9a12860c524262f",
+  "frameworkSha": "06c4bc1df2880b0480475309ce998490dc4d0477",
   "version": "0.1.0"
 }

README.md

@@ -351,7 +351,7 @@ the tree over, so a pod never moves a tree whose lease it could not renew.
 | Lease         | `yield` / `reclaim`  | `UPDATE … WHERE holder AND lease_epoch AND expires_at > $now`: `yielded = true` / epoch + 1; 0 rows: lost                             |
 | Record        | `completeIfStatus`   | `UPDATE … WHERE status = $expected`                                                                                                   |
 | Tree          | `saveIfAbsent`       | `INSERT … ON CONFLICT DO NOTHING RETURNING`                                                                                           |
-| Child queue   | claim                | `UPDATE … FROM (SELECT … FOR UPDATE SKIP LOCKED LIMIT 1)`                                                                             |
+| Child queue   | claim                | `WITH … AS MATERIALIZED (SELECT … FOR UPDATE SKIP LOCKED LIMIT 1) UPDATE … FROM` — chosen once (agbossjs/agboss#720)                  |
 | Paused        | `Pending → …`        | `UPDATE … WHERE status = 'pending' RETURNING` (none → `NotPausedError` from the core)                                                 |
 | Fenced effect | `saveResultIfAbsent` | one transaction: lease row `FOR UPDATE`, mark upsert, key lookup, `await work(client)`, insert                                        |
 | Fenced effect | business key         | the same, the key's identity serialized by `pg_advisory_xact_lock` first (no lease row covers two trees)                              |

api/postgres.api.md

@@ -499,13 +499,13 @@ export interface SchemaVersions {
 // src/stores/postgres-checkpoint-saver.ts:55:9 - (ae-undocumented) Missing documentation for "put".
 // src/stores/postgres-checkpoint-saver.ts:65:9 - (ae-undocumented) Missing documentation for "putWrites".
 // src/stores/postgres-checkpoint-saver.ts:74:9 - (ae-undocumented) Missing documentation for "deleteThread".
-// src/stores/postgres-child-execution-queue.repository.ts:107:3 - (ae-undocumented) Missing documentation for "ping".
-// src/stores/postgres-child-execution-queue.repository.ts:111:9 - (ae-undocumented) Missing documentation for "saveIfAbsent".
-// src/stores/postgres-child-execution-queue.repository.ts:127:9 - (ae-undocumented) Missing documentation for "findByTenantIdAndExecutionId".
-// src/stores/postgres-child-execution-queue.repository.ts:139:9 - (ae-undocumented) Missing documentation for "claimFirstByStatusQueuedOrderByDepthDesc".
-// src/stores/postgres-child-execution-queue.repository.ts:147:9 - (ae-undocumented) Missing documentation for "completeByTenantIdAndExecutionId".
-// src/stores/postgres-child-execution-queue.repository.ts:168:9 - (ae-undocumented) Missing documentation for "deleteAllByTenantIdAndRootExecutionId".
-// src/stores/postgres-child-execution-queue.repository.ts:179:9 - (ae-undocumented) Missing documentation for "findQueueDepth".
+// src/stores/postgres-child-execution-queue.repository.ts:115:3 - (ae-undocumented) Missing documentation for "ping".
+// src/stores/postgres-child-execution-queue.repository.ts:119:9 - (ae-undocumented) Missing documentation for "saveIfAbsent".
+// src/stores/postgres-child-execution-queue.repository.ts:135:9 - (ae-undocumented) Missing documentation for "findByTenantIdAndExecutionId".
+// src/stores/postgres-child-execution-queue.repository.ts:147:9 - (ae-undocumented) Missing documentation for "claimFirstByStatusQueuedOrderByDepthDesc".
+// src/stores/postgres-child-execution-queue.repository.ts:155:9 - (ae-undocumented) Missing documentation for "completeByTenantIdAndExecutionId".
+// src/stores/postgres-child-execution-queue.repository.ts:176:9 - (ae-undocumented) Missing documentation for "deleteAllByTenantIdAndRootExecutionId".
+// src/stores/postgres-child-execution-queue.repository.ts:187:9 - (ae-undocumented) Missing documentation for "findQueueDepth".
 // src/stores/postgres-cost-ledger.repository.ts:56:3 - (ae-undocumented) Missing documentation for "ping".
 // src/stores/postgres-cost-ledger.repository.ts:60:9 - (ae-undocumented) Missing documentation for "findSpentByCounter".
 // src/stores/postgres-cost-ledger.repository.ts:70:9 - (ae-undocumented) Missing documentation for "saveAll".

src/stores/postgres-child-execution-queue.repository.ts

@@ -67,12 +67,19 @@ const SAVE = `INSERT INTO agboss_child_execution_queue AS queued (tenant_id, exe
 /**
  * Claims the first queued child — the deepest, then the longest waiting — in one statement: the
  * row is locked with `FOR UPDATE SKIP LOCKED` (a row another instance is claiming is passed over,
- * never waited for) and marked running for this instance.
+ * never waited for) and marked running for this instance. The choice is a `MATERIALIZED` CTE, run
+ * once (agbossjs/agboss#720): as a subquery joined to the table, a plan made on statistics of a
+ * nearly empty queue rescanned it once per row of the table — each rescan skipped the rows the
+ * statement had already marked and locked the next — so one claim marked many children `running`
+ * and answered one; the others ran nowhere.
  */
-const CLAIM = `UPDATE agboss_child_execution_queue AS queued
+const CLAIM = `WITH first AS MATERIALIZED (
+    SELECT tenant_id, execution_id FROM agboss_child_execution_queue WHERE status = 'queued'
+    ORDER BY depth DESC, queued_at LIMIT 1 FOR UPDATE SKIP LOCKED
+  )
+  UPDATE agboss_child_execution_queue AS queued
   SET status = 'running', instance_id = $1, claimed_at = $2
-  FROM (SELECT tenant_id, execution_id FROM agboss_child_execution_queue WHERE status = 'queued'
-        ORDER BY depth DESC, queued_at LIMIT 1 FOR UPDATE SKIP LOCKED) first
+  FROM first
   WHERE queued.tenant_id = first.tenant_id AND queued.execution_id = first.execution_id
   RETURNING queued.*`;
 
@@ -89,8 +96,9 @@ const DepthRow = z.object({ queued: z.number(), running: z.number(), oldest: z.d
 
 /**
  * `ChildExecutionQueueRepository` on PostgreSQL (#333 PR 2, #482): one row per child queued with
- * `ChildPlacement.AnyInstance`. A claim is one `UPDATE … FROM (SELECT … FOR UPDATE SKIP LOCKED
- * LIMIT 1)`, so of two instances exactly one claims a child and neither waits on the other's; a
+ * `ChildPlacement.AnyInstance`. A claim is one `WITH first AS MATERIALIZED (SELECT … FOR UPDATE
+ * SKIP LOCKED LIMIT 1) UPDATE … FROM first`, so of two instances exactly one claims a child, a claim
+ * marks exactly the child it answers, and neither waits on the other's; a
  * completion is one `UPDATE` matching this instance's running row. Times are the framework
  * clock's. Listed in `@Application({ providers })` with `postgresUrl`.
  */

tests/stores/child-execution-queue.test.ts

@@ -61,6 +61,33 @@ describe("PostgresChildExecutionQueueRepository: FOR UPDATE SKIP LOCKED", () =>
     expect(stored).toMatchObject({ status: ExecutionRecordStatus.Running });
   });
 
+  it("agbossjs/agboss#720: on statistics taken while the queue held one row, a claim marks exactly one child", async () => {
+    const connection = connectionTo(database.url(), 1);
+    const queue = new PostgresChildExecutionQueueRepository(
+      connection,
+      new VirtualClock(),
+      new TestApplicationContext({}, "instance-1"),
+    );
+    // the queue after a batch of one (the frozen pod's job): analyzed at one row, then a batch
+    await connection.pool.query("DELETE FROM agboss_child_execution_queue");
+    await queue.saveIfAbsent({ ...child(99), rootExecutionId: "root-0" });
+    await connection.pool.query("VACUUM ANALYZE agboss_child_execution_queue");
+    await queue.deleteAllByTenantIdAndRootExecutionId("default", "root-0");
+    const children = Array.from({ length: 48 }, (_, index) => ({
+      ...child(index),
+      executionId: `batch-${String(index)}`,
+    }));
+    for (const one of children) await queue.saveIfAbsent(one);
+
+    const claimed = await queue.claimFirstByStatusQueuedOrderByDepthDesc();
+
+    const { rows } = await connection.pool.query<{ readonly execution_id: string }>(
+      "SELECT execution_id FROM agboss_child_execution_queue WHERE status = 'running'",
+    );
+    expect(rows.map((row) => row.execution_id)).toEqual([claimed?.executionId]);
+    await connection.pool.end();
+  });
+
   it("a lost connection rejects the claim", async () => {
     const queue = new PostgresChildExecutionQueueRepository(
       connectionTo("postgres://postgres:pw@127.0.0.1:1/postgres"),