From 21a37be6c125acac88a7bd80587bbcfb12188c66 Mon Sep 17 00:00:00 2001 From: pavelravvich Date: Wed, 4 Nov 2026 02:35:52 +0200 Subject: [PATCH 1/3] =?UTF-8?q?fix(agbossjs/agboss#720):=20the=20child=20q?= =?UTF-8?q?ueue's=20claim=20marks=20exactly=20the=20child=20it=20answers?= =?UTF-8?q?=20=E2=80=94=20the=20choice=20is=20a=20MATERIALIZED=20CTE,=20ru?= =?UTF-8?q?n=20once?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `UPDATE … FROM (SELECT … LIMIT 1 FOR UPDATE SKIP LOCKED) first` was planned, on statistics taken while the queue held about one row, as a nested loop rescanning the subquery 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` for the pod and returned one. The others ran nowhere, and on the core's side a reclaim of them could keep a lease with no pass behind it (agbossjs/agboss#720). That is the queue right after a batch of one, such as the Kubernetes blueprint's frozen-pod step. The choice is now `WITH first AS MATERIALIZED (…) UPDATE … FROM first`: the CTE runs once and its rescans read its one row. tests/stores/child-execution-queue.test.ts analyzes a one-row queue, queues 48 children and claims once: 16 rows were marked before, exactly one now. Refs agbossjs/agboss#720 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012Kh8JmysuhZyQRSpNyJdPg --- README.md | 2 +- ...stgres-child-execution-queue.repository.ts | 20 +++++++++----- tests/stores/child-execution-queue.test.ts | 27 +++++++++++++++++++ 3 files changed, 42 insertions(+), 7 deletions(-) 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/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"), From 1e0b5fafb372552fbd8df750949926ed7192bbf7 Mon Sep 17 00:00:00 2001 From: pavelravvich Date: Wed, 4 Nov 2026 18:53:30 +0200 Subject: [PATCH 2/3] =?UTF-8?q?chore(agbossjs/agboss#720):=20pin=20the=20c?= =?UTF-8?q?ore=20at=2006c4bc1d=20=E2=80=94=20the=20claim=20contract=20case?= =?UTF-8?q?=20and=20fault=20scenario=2016?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012Kh8JmysuhZyQRSpNyJdPg --- .agboss-version.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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" } From ba2f143b1b2965342e2935bfa26be6f5cbbf4e7d Mon Sep 17 00:00:00 2001 From: pavelravvich Date: Wed, 4 Nov 2026 19:06:51 +0200 Subject: [PATCH 3/3] chore(agbossjs/agboss#720): the API report's line numbers after the claim's doc comment Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012Kh8JmysuhZyQRSpNyJdPg --- api/postgres.api.md | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) 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".