diff --git a/.agboss-version.json b/.agboss-version.json index 09bccea..f116762 100644 --- a/.agboss-version.json +++ b/.agboss-version.json @@ -2,6 +2,6 @@ "repository": "agbossjs/agboss", "branch": "main", "source": "commit", - "frameworkSha": "606370a0012178f69352275fc9a12860c524262f", + "frameworkSha": "06c4bc1df2880b0480475309ce998490dc4d0477", "version": "0.1.0" } diff --git a/README.md b/README.md index 7f32824..501ca67 100644 --- a/README.md +++ b/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) | diff --git a/api/postgres.api.md b/api/postgres.api.md index b1f3bcb..5a3a379 100644 --- a/api/postgres.api.md +++ b/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". diff --git a/src/stores/postgres-child-execution-queue.repository.ts b/src/stores/postgres-child-execution-queue.repository.ts index c82f4b0..e4abb6c 100644 --- a/src/stores/postgres-child-execution-queue.repository.ts +++ b/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`. */ diff --git a/tests/stores/child-execution-queue.test.ts b/tests/stores/child-execution-queue.test.ts index 9796e6b..10e7797 100644 --- a/tests/stores/child-execution-queue.test.ts +++ b/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"),