From 2a5b46413d51731fd16a6e445e41588b2d8347f9 Mon Sep 17 00:00:00 2001 From: pavelravvich Date: Thu, 5 Nov 2026 08:05:52 +0200 Subject: [PATCH] =?UTF-8?q?fix(agbossjs/agboss#771):=20the=20ledger's=20en?= =?UTF-8?q?tries=20are=20pruned=20under=20a=20mark=20=E2=80=94=20migration?= =?UTF-8?q?=200023;=20a=20save=20of=20a=20pruned=20entry=20refused,=20neve?= =?UTF-8?q?r=20counted=20twice;=20core=20at=20a8fe1124?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The core (agbossjs/agboss#771) dates every ledger entry by its `madeAt`, prunes them with `deleteEntriesByMadeAtBefore` and refuses a save of one made before the store's pruning mark that it no longer holds. - migration 0023_cost_entries_pruned: `agboss_cost_entries.made_at` (indexed; rows of before dated 'infinity': never pruned, never refused) and `agboss_cost_entries_pruned`, the one-row mark, seeded at '-infinity'. Its header corrects 0022's (kept as is: migrations are checksummed), which described ids derived from the step, a design that was dropped. - saveAll: one statement — the mark read FOR SHARE (a prune's UPDATE waits, or the save waits and reads the new mark), an entry made before it never inserted, one not held refusing the whole save (CostLedgerEntryExpiredError). A missing mark row fails the save with PostgresLedgerMarkMissingError and adds nothing. - deleteEntriesByMadeAtBefore: the mark raised (greatest) and the entries deleted in one statement. - tests: the contract's four pruning cases; 40 lost answers retried on two instances while a third prunes ten times count once each (red without FOR SHARE: $5.875 for $5); the mark row missing; 0023 on a database at 0022. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012Kh8JmysuhZyQRSpNyJdPg --- .agboss-version.json | 2 +- CHANGELOG.md | 9 ++ README.md | 81 ++++++------ api/postgres.api.md | 24 +++- migrations/0023_cost_entries_pruned.sql | 29 +++++ src/errors.ts | 17 +++ src/index.ts | 1 + src/stores/postgres-cost-ledger.repository.ts | 118 +++++++++++++----- tests/cli/bin.test.ts | 2 +- tests/cli/command.test.ts | 13 +- tests/connection/postgres-connection.test.ts | 13 +- .../migrations/cost-entries-migration.test.ts | 38 +++++- .../execution-ids-migration.test.ts | 3 +- .../execution-ids-outbound-migration.test.ts | 3 +- tests/migrations/files.test.ts | 7 +- .../lease-finished-migration.test.ts | 5 +- tests/migrations/migrate.test.ts | 16 ++- tests/migrations/nul-escape-migration.test.ts | 1 + .../outbound-calls-migration.test.ts | 3 +- .../pause-answers-by-tree-migration.test.ts | 3 +- .../pauses-by-tree-migration.test.ts | 4 +- tests/stores/cost-ledger.test.ts | 70 ++++++++++- tests/support/database.ts | 9 +- 23 files changed, 361 insertions(+), 110 deletions(-) create mode 100644 migrations/0023_cost_entries_pruned.sql diff --git a/.agboss-version.json b/.agboss-version.json index b79a155..9804e72 100644 --- a/.agboss-version.json +++ b/.agboss-version.json @@ -2,6 +2,6 @@ "repository": "agbossjs/agboss", "branch": "main", "source": "commit", - "frameworkSha": "0221a7c9df1ed1e3509ccb8b3b766edc47aabf69", + "frameworkSha": "a8fe11247f73083d428392f9e5d7600a7e6490e0", "version": "0.1.0" } diff --git a/CHANGELOG.md b/CHANGELOG.md index 01c67ca..590b15f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,15 @@ own process, no shared disk — share every store on one database. (`PostgresConnection.listen`): an idle pod claims at once instead of at its next sweep. Idle on an empty queue; a dropped connection reconnects and wakes once; a notification lost costs at most a sweep. +- **The ledger's entries are pruned** (agbossjs/agboss#771) — migration `0023_cost_entries_pruned`: + `agboss_cost_entries.made_at` (indexed; entries of before dated `infinity`, never pruned) and + `agboss_cost_entries_pruned`, the pruning mark. `deleteEntriesByMadeAtBefore` raises the mark and + deletes in one statement; `saveAll` reads the mark `FOR SHARE` in its own statement and refuses an + entry made before it that the table does not hold (`CostLedgerEntryExpiredError`, nothing added). + Without the mark's row a save fails with `PostgresLedgerMarkMissingError` + (`postgres.ledger-mark-missing`) and adds nothing. 0023's header also corrects 0022's, which + described dropped ids. + - **Every store of the core on PostgreSQL** (#482, #483; #1, #3, #5, #8) — `PostgresExecutionLeaseRepository`, `PostgresExecutionRecordRepository`, `PostgresExecutionTreeRepository`, `PostgresPausedExecutionRepository`, `PostgresChildExecutionQueueRepository`, diff --git a/README.md b/README.md index 6d00d07..e687e3f 100644 --- a/README.md +++ b/README.md @@ -74,21 +74,21 @@ import { AppEnvironment } from "./environments/app.environment.js"; export class DeskApplication {} ``` -| Store | Core contract | Tables | -| --------------------------------------- | ------------------------------------ | ---------------------------------------------------------------------- | -| `PostgresExecutionLeaseRepository` | `ExecutionLeaseRepository` | `agboss_execution_leases` | -| `PostgresExecutionRecordRepository` | `ExecutionRecordRepository` | `agboss_threads`, `agboss_execution_records`, … (`0002`) | -| `PostgresExecutionTreeRepository` | `ExecutionTreeRepository` | `agboss_execution_tree_keys` | -| `PostgresPausedExecutionRepository` | `PausedExecutionRepository` | `agboss_paused_executions`, `agboss_pauses` | -| `PostgresChildExecutionQueueRepository` | `ChildExecutionQueueRepository` | `agboss_child_execution_queue` | -| `PostgresCheckpointSaver` | LangGraph `BaseCheckpointSaver` | `checkpoints`, `checkpoint_blobs`, `checkpoint_writes` | -| `PostgresFencedEffectRepository` | `FencedEffectRepository` | `agboss_fenced_effect_marks`, `…_results`, `…_business_results` | -| `PostgresOutboundCallRepository` | `OutboundCallRepository` | `agboss_outbound_calls` (and the leases' `superseded`, `0014`) | -| `PostgresExecutionLeaseEventRepository` | `ExecutionLeaseEventRepository` | `agboss_execution_lease_events` | -| `PostgresCostLedgerRepository` | `CostLedgerRepository` | `agboss_cost_counters`, `agboss_cost_max_spend`, `agboss_cost_entries` | -| `PostgresExecutionHistoryRepository` | `ExecutionHistoryRepository` | `agboss_execution_history` | -| `PostgresRateLimitRepository` | `RateLimitRepository` | `agboss_rate_limit_keys`, `…_slots`, `…_window_buckets`, `…_pauses` | -| `PostgresMeshRequestRepository` | `MeshRequestRepository` | `agboss_mesh_requests` | +| Store | Core contract | Tables | +| --------------------------------------- | ------------------------------------ | ---------------------------------------------------------------------------------------------------- | +| `PostgresExecutionLeaseRepository` | `ExecutionLeaseRepository` | `agboss_execution_leases` | +| `PostgresExecutionRecordRepository` | `ExecutionRecordRepository` | `agboss_threads`, `agboss_execution_records`, … (`0002`) | +| `PostgresExecutionTreeRepository` | `ExecutionTreeRepository` | `agboss_execution_tree_keys` | +| `PostgresPausedExecutionRepository` | `PausedExecutionRepository` | `agboss_paused_executions`, `agboss_pauses` | +| `PostgresChildExecutionQueueRepository` | `ChildExecutionQueueRepository` | `agboss_child_execution_queue` | +| `PostgresCheckpointSaver` | LangGraph `BaseCheckpointSaver` | `checkpoints`, `checkpoint_blobs`, `checkpoint_writes` | +| `PostgresFencedEffectRepository` | `FencedEffectRepository` | `agboss_fenced_effect_marks`, `…_results`, `…_business_results` | +| `PostgresOutboundCallRepository` | `OutboundCallRepository` | `agboss_outbound_calls` (and the leases' `superseded`, `0014`) | +| `PostgresExecutionLeaseEventRepository` | `ExecutionLeaseEventRepository` | `agboss_execution_lease_events` | +| `PostgresCostLedgerRepository` | `CostLedgerRepository` | `agboss_cost_counters`, `agboss_cost_max_spend`, `agboss_cost_entries`, `agboss_cost_entries_pruned` | +| `PostgresExecutionHistoryRepository` | `ExecutionHistoryRepository` | `agboss_execution_history` | +| `PostgresRateLimitRepository` | `RateLimitRepository` | `agboss_rate_limit_keys`, `…_slots`, `…_window_buckets`, `…_pauses` | +| `PostgresMeshRequestRepository` | `MeshRequestRepository` | `agboss_mesh_requests` | With all of them listed, nothing of the application is per pod: budgets sum every pod's spend, a provider's limits hold across pods, `agboss records export` on any pod lists every pod's @@ -307,31 +307,32 @@ a pod never waits on another's migration at start (both are safe: they serialize - `agboss_schema_migrations (version, name, checksum, applied_at)` records what ran; an applied migration whose file changed fails the run (`PostgresMigrationChangedError`). -| Migration | Tables | -| ---------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `0001_execution_leases` | `agboss_execution_leases` | -| `0002_execution_records` | `agboss_threads`, `agboss_execution_records`, `agboss_execution_scores`, `agboss_execution_summaries`, `agboss_config_versions`, `agboss_pause_answers` | -| `0003_execution_trees` | `agboss_execution_tree_keys` | -| `0004_paused_executions` | `agboss_paused_executions`, `agboss_pauses` | -| `0005_child_execution_queue` | `agboss_child_execution_queue` | -| `0006_fenced_effects` | `agboss_fenced_effect_marks`, `agboss_fenced_effect_results` | -| `0007_execution_lease_events` | `agboss_execution_lease_events` | -| `0008_execution_lease_yield` | `agboss_execution_leases.yielded` (agbossjs/agboss#516) | -| `0009_cost_ledger` | `agboss_cost_counters`, `agboss_cost_max_spend` | -| `0010_execution_history` | `agboss_execution_history` | -| `0011_rate_limits` | `agboss_rate_limit_keys`, `agboss_rate_limit_slots`, `agboss_rate_limit_window_buckets`, `agboss_rate_limit_pauses` | -| `0012_mesh_requests` | `agboss_mesh_requests` | -| `0013_nul_escape` | none: doubles U+E000 in the columns the stores now escape (agbossjs/agboss#521) | -| `0014_execution_lease_superseded` | `agboss_execution_leases.superseded`: a lease an outbound call's check closed (agbossjs/agboss#508) | -| `0015_fenced_effect_business_keys` | `agboss_fenced_effect_business_results` (agbossjs/agboss#508) | -| `0016_outbound_calls` | `agboss_outbound_calls` (agbossjs/agboss#505, #508) | -| `0017_execution_lease_finished` | `agboss_execution_leases.finished`: a lease its holder finished, never taken again (agbossjs/agboss#520) | -| `0018_execution_ids` | `agboss_execution_ids`: every execution id handed out, the leases' of before (agbossjs/agboss#591) | -| `0019_execution_ids_from_outbound_calls` | none: the ids of before of every execution that made an outbound call (agbossjs/agboss#638) | -| `0020_pauses_by_tree` | `agboss_pauses` unique per tree and step key; the ids of before of every execution that paused, and its tree (agbossjs/agboss#663) | -| `0021_pause_answers_by_tree` | `agboss_pause_answers` keyed per tree and step key: another tree's answer under the same step key is kept (agbossjs/agboss#676) | -| `0022_cost_ledger_entries` | `agboss_cost_entries`: each spend's `ledgerId`, so a step run again counts its spend once (agbossjs/agboss#727) | -| LangGraph | `checkpoints`, `checkpoint_blobs`, `checkpoint_writes`, `checkpoint_migrations` | +| Migration | Tables | +| ---------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `0001_execution_leases` | `agboss_execution_leases` | +| `0002_execution_records` | `agboss_threads`, `agboss_execution_records`, `agboss_execution_scores`, `agboss_execution_summaries`, `agboss_config_versions`, `agboss_pause_answers` | +| `0003_execution_trees` | `agboss_execution_tree_keys` | +| `0004_paused_executions` | `agboss_paused_executions`, `agboss_pauses` | +| `0005_child_execution_queue` | `agboss_child_execution_queue` | +| `0006_fenced_effects` | `agboss_fenced_effect_marks`, `agboss_fenced_effect_results` | +| `0007_execution_lease_events` | `agboss_execution_lease_events` | +| `0008_execution_lease_yield` | `agboss_execution_leases.yielded` (agbossjs/agboss#516) | +| `0009_cost_ledger` | `agboss_cost_counters`, `agboss_cost_max_spend` | +| `0010_execution_history` | `agboss_execution_history` | +| `0011_rate_limits` | `agboss_rate_limit_keys`, `agboss_rate_limit_slots`, `agboss_rate_limit_window_buckets`, `agboss_rate_limit_pauses` | +| `0012_mesh_requests` | `agboss_mesh_requests` | +| `0013_nul_escape` | none: doubles U+E000 in the columns the stores now escape (agbossjs/agboss#521) | +| `0014_execution_lease_superseded` | `agboss_execution_leases.superseded`: a lease an outbound call's check closed (agbossjs/agboss#508) | +| `0015_fenced_effect_business_keys` | `agboss_fenced_effect_business_results` (agbossjs/agboss#508) | +| `0016_outbound_calls` | `agboss_outbound_calls` (agbossjs/agboss#505, #508) | +| `0017_execution_lease_finished` | `agboss_execution_leases.finished`: a lease its holder finished, never taken again (agbossjs/agboss#520) | +| `0018_execution_ids` | `agboss_execution_ids`: every execution id handed out, the leases' of before (agbossjs/agboss#591) | +| `0019_execution_ids_from_outbound_calls` | none: the ids of before of every execution that made an outbound call (agbossjs/agboss#638) | +| `0020_pauses_by_tree` | `agboss_pauses` unique per tree and step key; the ids of before of every execution that paused, and its tree (agbossjs/agboss#663) | +| `0021_pause_answers_by_tree` | `agboss_pause_answers` keyed per tree and step key: another tree's answer under the same step key is kept (agbossjs/agboss#676) | +| `0022_cost_ledger_entries` | `agboss_cost_entries`: each tallied spend's `ledgerId` — the id of the pass that made it — so a write retried after a lost answer counts once (agbossjs/agboss#727; its header's "a step run again saves under the same ids" is corrected by 0023) | +| `0023_cost_entries_pruned` | `agboss_cost_entries.made_at` and `agboss_cost_entries_pruned`, the pruning mark: entries made longer ago than `ledgerEntryRetention` are pruned by the sweep, a save of one it may have pruned refused (agbossjs/agboss#771; entries of before dated "never") | +| LangGraph | `checkpoints`, `checkpoint_blobs`, `checkpoint_writes`, `checkpoint_migrations` | ## Text with a NUL character diff --git a/api/postgres.api.md b/api/postgres.api.md index 1b622d2..ca2a776 100644 --- a/api/postgres.api.md +++ b/api/postgres.api.md @@ -172,6 +172,8 @@ export interface PostgresConnectionSettings { export class PostgresCostLedgerRepository extends CostLedgerRepository { constructor(connection: PostgresConnection); // (undocumented) + deleteEntriesByMadeAtBefore(cutoff: Date): Promise; + // (undocumented) findMaxSpendByAccount(account: BudgetAccount): Promise; // (undocumented) findSpentByCounter(counter: CostCounter): Promise; @@ -327,6 +329,15 @@ export class PostgresFencedEffectRepository extends FencedEffectRepository(key: FencedEffectKey, work: FencedEffectWork): Promise>; } +// @public +export class PostgresLedgerMarkMissingError extends AGBossError { + constructor(); + // (undocumented) + static readonly code: string; + // (undocumented) + name: string; +} + // @public export class PostgresListener { close(): Promise; @@ -514,6 +525,8 @@ export interface SchemaVersions { // src/errors.ts:76:3 - (ae-undocumented) Missing documentation for "migration". // src/errors.ts:92:3 - (ae-undocumented) Missing documentation for "code". // src/errors.ts:93:12 - (ae-undocumented) Missing documentation for "name". +// src/errors.ts:103:3 - (ae-undocumented) Missing documentation for "code". +// src/errors.ts:104:12 - (ae-undocumented) Missing documentation for "name". // src/migrations/migrate.ts:23:3 - (ae-undocumented) Missing documentation for "version". // src/migrations/migrate.ts:24:3 - (ae-undocumented) Missing documentation for "name". // src/stores/postgres-checkpoint-saver.ts:44:9 - (ae-undocumented) Missing documentation for "getTuple". @@ -528,11 +541,12 @@ export interface SchemaVersions { // src/stores/postgres-child-execution-queue.repository.ts:176:9 - (ae-undocumented) Missing documentation for "completeByTenantIdAndExecutionId". // src/stores/postgres-child-execution-queue.repository.ts:197:9 - (ae-undocumented) Missing documentation for "deleteAllByTenantIdAndRootExecutionId". // src/stores/postgres-child-execution-queue.repository.ts:208:9 - (ae-undocumented) Missing documentation for "findQueueDepth". -// src/stores/postgres-cost-ledger.repository.ts:81:3 - (ae-undocumented) Missing documentation for "ping". -// src/stores/postgres-cost-ledger.repository.ts:85:9 - (ae-undocumented) Missing documentation for "findSpentByCounter". -// src/stores/postgres-cost-ledger.repository.ts:95:9 - (ae-undocumented) Missing documentation for "saveAll". -// src/stores/postgres-cost-ledger.repository.ts:109:9 - (ae-undocumented) Missing documentation for "findMaxSpendByAccount". -// src/stores/postgres-cost-ledger.repository.ts:119:9 - (ae-undocumented) Missing documentation for "saveMaxSpend". +// src/stores/postgres-cost-ledger.repository.ts:131:3 - (ae-undocumented) Missing documentation for "ping". +// src/stores/postgres-cost-ledger.repository.ts:135:9 - (ae-undocumented) Missing documentation for "findSpentByCounter". +// src/stores/postgres-cost-ledger.repository.ts:145:9 - (ae-undocumented) Missing documentation for "saveAll". +// src/stores/postgres-cost-ledger.repository.ts:165:9 - (ae-undocumented) Missing documentation for "deleteEntriesByMadeAtBefore". +// src/stores/postgres-cost-ledger.repository.ts:169:9 - (ae-undocumented) Missing documentation for "findMaxSpendByAccount". +// src/stores/postgres-cost-ledger.repository.ts:179:9 - (ae-undocumented) Missing documentation for "saveMaxSpend". // src/stores/postgres-execution-history.repository.ts:48:3 - (ae-undocumented) Missing documentation for "ping". // src/stores/postgres-execution-history.repository.ts:52:9 - (ae-undocumented) Missing documentation for "save". // src/stores/postgres-execution-history.repository.ts:63:9 - (ae-undocumented) Missing documentation for "findByTenantIdAndExecutionId". diff --git a/migrations/0023_cost_entries_pruned.sql b/migrations/0023_cost_entries_pruned.sql new file mode 100644 index 0000000..0600481 --- /dev/null +++ b/migrations/0023_cost_entries_pruned.sql @@ -0,0 +1,29 @@ +-- agbossjs/agboss#771: the ledger's entries are pruned. This also corrects 0022's header, kept as +-- it is because a migration is checksummed: an entry's ledgerId is an id of the pass that made the +-- spend (a random prefix of the pass and a number), never derived from where it was made. A step run +-- again after a crash or a takeover makes its spend again under new ids, and counts again. Only +-- the pass that made a spend saves its entry — its write and its retries after a lost answer —, so +-- an entry is needed only while that pass may still retry. +-- +-- Each entry is dated by its record's madeAt. The recovery sweep prunes those made before +-- now − ledgerEntryRetention: it raises the pruning mark in agboss_cost_entries_pruned (never +-- lowers it) and deletes the entries made before it, in one statement. A save reads the mark FOR +-- SHARE in its own statement and refuses an entry made before it that the table does not hold +-- (it may have been counted and pruned since), so a retry is never counted twice. +-- +-- An entry written before this migration has no date: it is dated 'infinity' — never pruned, never +-- refused (its pass's retry finds it held). Delete those rows by hand once no pass from before the +-- upgrade runs. +ALTER TABLE agboss_cost_entries ADD COLUMN made_at timestamptz; +UPDATE agboss_cost_entries SET made_at = 'infinity'; +ALTER TABLE agboss_cost_entries ALTER COLUMN made_at SET NOT NULL; +CREATE INDEX agboss_cost_entries_made_at ON agboss_cost_entries (made_at); + +CREATE TABLE agboss_cost_entries_pruned ( + id boolean PRIMARY KEY DEFAULT true CHECK (id), + pruned_before timestamptz NOT NULL +); +INSERT INTO agboss_cost_entries_pruned (id, pruned_before) VALUES (true, '-infinity'); + +COMMENT ON TABLE agboss_cost_entries IS + 'agbossjs/agboss#727, #771: one row per ledger entry (a tallied spend), by the ledgerId of the pass that made it, dated by its madeAt; pruned under agboss_cost_entries_pruned — a save of an entry made before the mark that is not held is refused'; diff --git a/src/errors.ts b/src/errors.ts index 6d56d08..da76a54 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -92,3 +92,20 @@ export class PostgresSettingsError extends AGBossError { static override readonly code: string = "postgres.settings-invalid"; override name = "PostgresSettingsError"; } + +/** + * The ledger's pruning mark (agbossjs/agboss#771, the one row of `agboss_cost_entries_pruned` that + * migration 0023 seeds) is gone — deleted by hand: without it a save cannot tell an entry the ledger + * pruned from a new one, so it adds nothing and fails. Put the row back with a mark no lower than + * any prune's cutoff (`now − ledgerEntryRetention` covers every prune the sweep made). + */ +export class PostgresLedgerMarkMissingError extends AGBossError { + static override readonly code: string = "postgres.ledger-mark-missing"; + override name = "PostgresLedgerMarkMissingError"; + + constructor() { + super( + "the ledger's pruning mark is missing (agboss_cost_entries_pruned has no row): nothing was added — insert it with a mark no lower than any prune's cutoff, e.g. INSERT INTO agboss_cost_entries_pruned VALUES (true, now())", + ); + } +} diff --git a/src/index.ts b/src/index.ts index 4f54dd8..f4e9f9a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -18,6 +18,7 @@ export { type MigrationReport, } from "./migrations/migrate.js"; export { + PostgresLedgerMarkMissingError, PostgresMigrationChangedError, PostgresSchemaMissingError, PostgresSchemaTooNewError, diff --git a/src/stores/postgres-cost-ledger.repository.ts b/src/stores/postgres-cost-ledger.repository.ts index 8457ecc..3e45f07 100644 --- a/src/stores/postgres-cost-ledger.repository.ts +++ b/src/stores/postgres-cost-ledger.repository.ts @@ -1,5 +1,6 @@ import { BudgetScope, + CostLedgerEntryExpiredError, CostLedgerRepository, Injectable, type BudgetAccount, @@ -10,48 +11,96 @@ import { usd, type Usd } from "@agboss/core/units"; import { z } from "zod"; import { PostgresConnection } from "../connection/postgres-connection.js"; import { changesOf, firstRowOf } from "../connection/query.js"; +import { PostgresLedgerMarkMissingError } from "../errors.js"; /** * Adds to every counter of the arrays (one counter per index), in one statement: `$5` — the cost of - * the records without a `ledgerId` — plus the cost of each entry of `$6` / `$7` (ids, costs) whose id - * this statement inserts (agbossjs/agboss#727: an id already there, or inserted by a save that - * commits first, adds nothing; the unique key makes the later save wait for the earlier one). A - * counter given twice counts twice (grouped first — one upsert may not touch a row twice), and the - * rows are upserted in key order, so two saves of overlapping counters lock them in one order and - * never deadlock. + * the records without a `ledgerId` — plus the cost of each entry of `$6` / `$7` / `$8` (ids, costs, + * `madeAt`s) whose id this statement inserts (agbossjs/agboss#727: an id already there, or inserted + * by a save that commits first, adds nothing; the unique key makes the later save wait for the + * earlier one). The pruning mark (agbossjs/agboss#771) is read `FOR SHARE`: a prune's `UPDATE` of it + * waits for this statement, or this one waits for the prune and reads the new mark (a locked row is + * read at its latest version). An entry made before the mark is never inserted; one the table does + * not hold is `expired`, and then nothing of the save is added — the statement answers the expired + * ids; without the mark's row (its migration seeds it) nothing is added and the save fails. A + * counter given twice counts twice (grouped first — one upsert may not touch a row twice), + * and the rows are upserted in key order, so two saves of overlapping counters lock them in one + * order and never deadlock. */ -const ADD = `WITH inserted AS ( - INSERT INTO agboss_cost_entries (ledger_id) SELECT unnest($6::text[]) +const ADD = `WITH mark AS ( + SELECT pruned_before FROM agboss_cost_entries_pruned WHERE id FOR SHARE + ), given AS ( + SELECT * FROM unnest($6::text[], $7::double precision[], $8::timestamptz[]) + AS entry (ledger_id, cost, made_at) + ), expired AS ( + SELECT given.ledger_id FROM given, mark + WHERE given.made_at < mark.pruned_before + AND NOT EXISTS (SELECT 1 FROM agboss_cost_entries held WHERE held.ledger_id = given.ledger_id) + ), inserted AS ( + INSERT INTO agboss_cost_entries (ledger_id, made_at) + SELECT given.ledger_id, given.made_at FROM given, mark + WHERE given.made_at >= mark.pruned_before AND NOT EXISTS (SELECT 1 FROM expired) ON CONFLICT (ledger_id) DO NOTHING RETURNING ledger_id ), spent AS ( - SELECT $5::double precision + COALESCE(sum(entry.cost), 0) AS usd - FROM unnest($6::text[], $7::double precision[]) AS entry (ledger_id, cost) - WHERE entry.ledger_id IN (SELECT ledger_id FROM inserted) + SELECT $5::double precision + COALESCE(sum(given.cost), 0) AS usd + FROM given WHERE given.ledger_id IN (SELECT ledger_id FROM inserted) + ), counted AS ( + INSERT INTO agboss_cost_counters AS counter + (budget, scope, scope_key, period, spent_usd) + SELECT budget, scope, scope_key, period, count(*) * (SELECT usd FROM spent) + FROM unnest($1::text[], $2::text[], $3::text[], $4::text[]) + AS counted (budget, scope, scope_key, period) + WHERE NOT EXISTS (SELECT 1 FROM expired) AND EXISTS (SELECT 1 FROM mark) + GROUP BY budget, scope, scope_key, period + ORDER BY budget, scope, scope_key, period + ON CONFLICT (budget, scope, scope_key, period) + DO UPDATE SET spent_usd = counter.spent_usd + EXCLUDED.spent_usd ) - INSERT INTO agboss_cost_counters AS counter - (budget, scope, scope_key, period, spent_usd) - SELECT budget, scope, scope_key, period, count(*) * (SELECT usd FROM spent) - FROM unnest($1::text[], $2::text[], $3::text[], $4::text[]) - AS given (budget, scope, scope_key, period) - GROUP BY budget, scope, scope_key, period - ORDER BY budget, scope, scope_key, period - ON CONFLICT (budget, scope, scope_key, period) - DO UPDATE SET spent_usd = counter.spent_usd + EXCLUDED.spent_usd`; + SELECT EXISTS (SELECT 1 FROM mark) AS marked, + (SELECT array_agg(ledger_id) FROM expired) AS expired, + CASE WHEN EXISTS (SELECT 1 FROM expired) THEN (SELECT pruned_before FROM mark) END + AS pruned_before`; + +/** + * Prunes the entries (agbossjs/agboss#771), in one statement: the mark raised to `$1` (never + * lowered) under its row lock — a save holding it `FOR SHARE` finishes first — then every entry + * made before the mark deleted. + */ +const PRUNE = `WITH mark AS ( + UPDATE agboss_cost_entries_pruned SET pruned_before = greatest(pruned_before, $1::timestamptz) + WHERE id RETURNING pruned_before + ) + DELETE FROM agboss_cost_entries WHERE made_at < (SELECT pruned_before FROM mark)`; + +/** What a save answers: the entries it refused and the mark, or none (both `null`). */ +const ExpiredRow = z.object({ + marked: z.boolean(), + expired: z.array(z.string()).nullable(), + pruned_before: z.date().nullable(), +}); /** The entries of `records` (one per `ledgerId`, the first), and what the others cost. */ function entriesOf(records: readonly UsageRecord[]): { readonly ids: string[]; readonly costs: number[]; + readonly madeAt: Date[]; readonly unnamed: number; } { - const entries = new Map(); + const entries = new Map(); let unnamed = 0; - for (const { ledgerId, costUsd } of records) { - if (ledgerId === undefined) unnamed += costUsd; - else if (!entries.has(ledgerId)) entries.set(ledgerId, costUsd); + for (const record of records) { + if (record.ledgerId === undefined) unnamed += record.costUsd; + else if (!entries.has(record.ledgerId)) entries.set(record.ledgerId, record); } - return { ids: [...entries.keys()], costs: [...entries.values()], unnamed }; + const kept = [...entries.values()]; + return { + ids: [...entries.keys()], + costs: kept.map((record) => record.costUsd), + // an entry without a date is the oldest (agbossjs/agboss#771) + madeAt: kept.map((record) => new Date(record.madeAt ?? 0)), + unnamed, + }; } const SpentRow = z.object({ spent_usd: z.number() }); @@ -63,8 +112,9 @@ const scopeOf = (account: BudgetAccount): string => z.enum(BudgetScope).parse(ac /** * `CostLedgerRepository` on PostgreSQL (agbossjs/agboss#483): one row per counter — budget, scope, * scope key and period — so the instances of an application share a tenant's day. `saveAll` is - * one statement: the entries' ids inserted into `agboss_cost_entries` (a repeated one adds nothing, - * agbossjs/agboss#727), then `INSERT … ON CONFLICT DO UPDATE SET spent_usd = spent_usd + …` of every + * one statement: the entries' ids inserted into `agboss_cost_entries` with their dates (a repeated + * one adds nothing, agbossjs/agboss#727; one made before the pruning mark and not held refuses the + * save, agbossjs/agboss#771), then `INSERT … ON CONFLICT DO UPDATE SET spent_usd = spent_usd + …` of every * counter it is given: an addition under the row's own lock, so two instances' concurrent saves * lose nothing. No other * locking: as on SQLite, the core accepts a small overrun of executions racing past one check @@ -94,8 +144,8 @@ export class PostgresCostLedgerRepository extends CostLedgerRepository { async saveAll(counters: readonly CostCounter[], records: readonly UsageRecord[]): Promise { if (counters.length === 0 || records.length === 0) return; - const { ids, costs, unnamed } = entriesOf(records); - await changesOf(this.connection.pool, ADD, [ + const { ids, costs, madeAt, unnamed } = entriesOf(records); + const row = await firstRowOf(this.connection.pool, ADD, [ counters.map((counter) => counter.budget), counters.map(scopeOf), counters.map((counter) => counter.key), @@ -103,7 +153,17 @@ export class PostgresCostLedgerRepository extends CostLedgerRepository { unnamed, ids, costs, + madeAt, ]); + const { marked, expired, pruned_before: prunedBefore } = ExpiredRow.parse(row); + // without its mark the ledger cannot tell a pruned entry: nothing was added (agbossjs/agboss#771) + if (!marked) throw new PostgresLedgerMarkMissingError(); + if (expired !== null && prunedBefore !== null) + throw new CostLedgerEntryExpiredError({ ledgerIds: expired, prunedBefore }); + } + + async deleteEntriesByMadeAtBefore(cutoff: Date): Promise { + await changesOf(this.connection.pool, PRUNE, [cutoff]); } async findMaxSpendByAccount(account: BudgetAccount): Promise { diff --git a/tests/cli/bin.test.ts b/tests/cli/bin.test.ts index b770b87..6e31101 100644 --- a/tests/cli/bin.test.ts +++ b/tests/cli/bin.test.ts @@ -26,7 +26,7 @@ describe("the agboss-postgres bin (a Kubernetes Job)", () => { ); expect(first.stdout).toContain("applied 0013_nul_escape"); - expect(second.stdout).toMatch(/nothing to apply: t_\w+ is at migration 22/); + expect(second.stdout).toMatch(/nothing to apply: t_\w+ is at migration 23/); expect(usage).toMatchObject({ code: 2, stderr: expect.stringContaining("usage: agboss-postgres"), diff --git a/tests/cli/command.test.ts b/tests/cli/command.test.ts index 6f588e6..079df51 100644 --- a/tests/cli/command.test.ts +++ b/tests/cli/command.test.ts @@ -23,7 +23,7 @@ describe("agboss-postgres migrate", () => { expect(await runCommand(["migrate", "--url", url], first)).toBe(ExitCode.Ok); expect(await runCommand(["migrate", "--url", url], second)).toBe(ExitCode.Ok); - expect(first.lines.slice(0, 22)).toEqual([ + expect(first.lines.slice(0, 23)).toEqual([ expect.stringMatching(/^applied 0001_execution_leases \(t_/), expect.stringMatching(/^applied 0002_execution_records/), expect.stringMatching(/^applied 0003_execution_trees/), @@ -46,10 +46,11 @@ describe("agboss-postgres migrate", () => { expect.stringMatching(/^applied 0020_pauses_by_tree/), expect.stringMatching(/^applied 0021_pause_answers_by_tree/), expect.stringMatching(/^applied 0022_cost_ledger_entries/), + expect.stringMatching(/^applied 0023_cost_entries_pruned/), ]); - expect(first.lines[22]).toMatch(/^applied \d+ LangGraph checkpoint migration\(s\)$/); + expect(first.lines[23]).toMatch(/^applied \d+ LangGraph checkpoint migration\(s\)$/); expect(second.lines).toEqual([ - expect.stringMatching(/^nothing to apply: t_\w+ is at migration 22$/), + expect.stringMatching(/^nothing to apply: t_\w+ is at migration 23$/), ]); expect([...first.errors, ...second.errors]).toEqual([]); }); @@ -59,7 +60,7 @@ describe("agboss-postgres migrate", () => { const io = ioOf({ DATABASE_URL: url }); expect(await runCommand(["migrate"], io)).toBe(ExitCode.Ok); - expect(io.lines).toHaveLength(23); + expect(io.lines).toHaveLength(24); }); it("neither --url nor DATABASE_URL: a usage error, exit code 2", async () => { @@ -103,13 +104,13 @@ describe("agboss-postgres migrate", () => { const url = await freshSchema(); await migrateWith( { url }, - migrationsIn(migrationsFolder({ "0023_later.sql": "CREATE TABLE later (id integer);" })), + migrationsIn(migrationsFolder({ "0024_later.sql": "CREATE TABLE later (id integer);" })), ); const io = ioOf(); expect(await runCommand(["migrate", "--url", url], io)).toBe(ExitCode.Failed); expect(io.errors[0]).toMatch( - /^agboss-postgres: the migration failed — the PostgreSQL schema is at migration 23/, + /^agboss-postgres: the migration failed — the PostgreSQL schema is at migration 24/, ); }); diff --git a/tests/connection/postgres-connection.test.ts b/tests/connection/postgres-connection.test.ts index 7109e0d..9b24e6d 100644 --- a/tests/connection/postgres-connection.test.ts +++ b/tests/connection/postgres-connection.test.ts @@ -108,7 +108,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { ); expect(error).toBeInstanceOf(PostgresSchemaMissingError); - expect(error).toMatchObject({ code: "postgres.schema-missing", applied: 0, expected: 22 }); + expect(error).toMatchObject({ code: "postgres.schema-missing", applied: 0, expected: 23 }); expect(String(error)).toContain('npx agboss-postgres migrate --url "$DATABASE_URL"'); expect(String(error)).toContain("postgresMigrateOnStart"); }); @@ -122,6 +122,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }), ), ); @@ -130,7 +131,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { rejectionOf(connection.onInit()), ); - expect(error).toMatchObject({ applied: 19, expected: 22 }); + expect(error).toMatchObject({ applied: 19, expected: 23 }); }); it("postgresMigrateOnStart: a missing schema is migrated at start", async () => { @@ -142,7 +143,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { await withConnection(new PostgresConnection(application), (connection) => connection.onInit()); - expect(await historyOf(url)).toHaveLength(22); + expect(await historyOf(url)).toHaveLength(23); }); it("postgresMigrateOnStart: two instances starting at once migrate it once", async () => { @@ -154,14 +155,14 @@ describe("PostgresConnection.onInit: the schema check at start", () => { await Promise.all(instances.map((connection) => connection.onInit())); await Promise.all(instances.map((connection) => connection.onDestroy())); - expect(await historyOf(url)).toHaveLength(22); + expect(await historyOf(url)).toHaveLength(23); }); it("a schema a newer package migrated fails with PostgresSchemaTooNewError, flag or not", async () => { const url = await freshSchema(); await migrateWith( { url }, - migrationsIn(migrationsFolder({ "0023_later.sql": "CREATE TABLE later (id integer);" })), + migrationsIn(migrationsFolder({ "0024_later.sql": "CREATE TABLE later (id integer);" })), ); for (const migrateOnStart of [false, true]) { @@ -170,7 +171,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { (connection) => rejectionOf(connection.onInit()), ); expect(error).toBeInstanceOf(PostgresSchemaTooNewError); - expect(error).toMatchObject({ applied: 23, expected: 22 }); + expect(error).toMatchObject({ applied: 24, expected: 23 }); expect(String(error)).toContain("the migrations are not rolled back"); } }); diff --git a/tests/migrations/cost-entries-migration.test.ts b/tests/migrations/cost-entries-migration.test.ts index 88f57af..b751f02 100644 --- a/tests/migrations/cost-entries-migration.test.ts +++ b/tests/migrations/cost-entries-migration.test.ts @@ -10,6 +10,8 @@ import { connectionTo } from "../support/stores.js"; /** * agbossjs/agboss#727: migration 0022 on a database migrated to 0021 — the counters written before * it keep what they hold, and from then on a save adds an entry with a `ledgerId` once. + * agbossjs/agboss#771: migration 0023 on a database at 0022 — an entry held before it is dated + * "never": no prune removes it, a save of it adds nothing. */ const tenantDay = { budget: "tenant-daily-budget", @@ -35,7 +37,12 @@ describe("migration 0022 adds the ledger's entries", () => { const url = await freshSchema(); await migrateWith( { url }, - migrationsIn(migrationsFolder({ "0022_cost_ledger_entries.sql": null })), + migrationsIn( + migrationsFolder({ + "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, + }), + ), ); await onSchema( url, @@ -49,7 +56,34 @@ describe("migration 0022 adds the ledger's entries", () => { await ledger.saveAll([tenantDay], [paid(0.25, "run-1:pay-action:1#0"), paid(0.5)]); await ledger.saveAll([tenantDay], [paid(0.25, "run-1:pay-action:1#0")]); - expect(report.applied.map((migration) => migration.version)).toEqual([22]); + expect(report.applied.map((migration) => migration.version)).toEqual([22, 23]); expect(await ledger.findSpentByCounter(tenantDay)).toBe(2.25); }); }); + +describe("migration 0023 dates the ledger's entries", () => { + it("#771: an entry held before it is never pruned and never counted again", async () => { + const url = await freshSchema(); + await migrateWith( + { url }, + migrationsIn(migrationsFolder({ "0023_cost_entries_pruned.sql": null })), + ); + const before = new PostgresCostLedgerRepository(connectionTo(url)); + await onSchema(url, "INSERT INTO agboss_cost_entries (ledger_id) VALUES ('pass-a:0')", []); + + const report = await migratePostgres({ url }); + await before.deleteEntriesByMadeAtBefore(new Date("2030-01-01T00:00:00.000Z")); + await before.saveAll( + [tenantDay], + [{ ...paid(0.25, "pass-a:0"), madeAt: "2026-11-04T08:00:00.000Z" }], + ); + + expect(report.applied.map((migration) => migration.version)).toEqual([23]); + expect(await before.findSpentByCounter(tenantDay)).toBe(0); + const rows = await onSchema( + url, + "SELECT made_at = 'infinity' AS never FROM agboss_cost_entries", + ); + expect(rows.rows).toEqual([{ never: true }]); + }); +}); diff --git a/tests/migrations/execution-ids-migration.test.ts b/tests/migrations/execution-ids-migration.test.ts index 2bee6cb..fe94efb 100644 --- a/tests/migrations/execution-ids-migration.test.ts +++ b/tests/migrations/execution-ids-migration.test.ts @@ -25,6 +25,7 @@ describe("migration 0018 keeps the ids of the trees already there", () => { "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }), ), ); @@ -47,7 +48,7 @@ describe("migration 0018 keeps the ids of the trees already there", () => { new TestApplicationContext({}, "instance-1"), ); - expect(report.applied.map((migration) => migration.version)).toEqual([18, 19, 20, 21, 22]); + expect(report.applied.map((migration) => migration.version)).toEqual([18, 19, 20, 21, 22, 23]); const { rows } = await onSchema( url, "SELECT execution_id FROM agboss_execution_ids ORDER BY execution_id", diff --git a/tests/migrations/execution-ids-outbound-migration.test.ts b/tests/migrations/execution-ids-outbound-migration.test.ts index 129367b..c8b48ff 100644 --- a/tests/migrations/execution-ids-outbound-migration.test.ts +++ b/tests/migrations/execution-ids-outbound-migration.test.ts @@ -24,6 +24,7 @@ describe("migration 0019 keeps the ids of every execution that made an outbound "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }); await migrateWith({ url }, migrationsIn(before)); await onSchema( @@ -51,7 +52,7 @@ describe("migration 0019 keeps the ids of every execution that made an outbound new TestApplicationContext({}, "instance-1"), ); - expect(report.applied.map((migration) => migration.version)).toEqual([18, 19, 20, 21, 22]); + expect(report.applied.map((migration) => migration.version)).toEqual([18, 19, 20, 21, 22, 23]); const { rows } = await onSchema( url, "SELECT execution_id, tenant_id FROM agboss_execution_ids ORDER BY execution_id", diff --git a/tests/migrations/files.test.ts b/tests/migrations/files.test.ts index 2a20af7..57d4724 100644 --- a/tests/migrations/files.test.ts +++ b/tests/migrations/files.test.ts @@ -34,8 +34,9 @@ describe("the package's migration files", () => { "20:pauses_by_tree", "21:pause_answers_by_tree", "22:cost_ledger_entries", + "23:cost_entries_pruned", ]); - expect(newestVersionOf(files)).toBe(22); + expect(newestVersionOf(files)).toBe(23); expect(files.every((file) => file.checksum === checksumOf(file.sql))).toBe(true); }); @@ -75,8 +76,8 @@ describe("the package's migration files", () => { .map(({ version, name, checksum }) => ({ version, name, checksum })); expect(pendingMigrations(files, applied).map((file) => file.version)).toEqual([ - 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, + 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, ]); - expect(pendingMigrations(files, [])).toHaveLength(22); + expect(pendingMigrations(files, [])).toHaveLength(23); }); }); diff --git a/tests/migrations/lease-finished-migration.test.ts b/tests/migrations/lease-finished-migration.test.ts index 205873f..34b3850 100644 --- a/tests/migrations/lease-finished-migration.test.ts +++ b/tests/migrations/lease-finished-migration.test.ts @@ -44,6 +44,7 @@ describe("migration 0017 keeps the leases written before it", () => { "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }), ), ); @@ -65,7 +66,9 @@ describe("migration 0017 keeps the leases written before it", () => { const report = await migratePostgres({ url }); await clock.advance(seconds(31)); - expect(report.applied.map((migration) => migration.version)).toEqual([17, 18, 19, 20, 21, 22]); + expect(report.applied.map((migration) => migration.version)).toEqual([ + 17, 18, 19, 20, 21, 22, 23, + ]); const { rows } = await onSchema( url, "SELECT root_execution_id, finished FROM agboss_execution_leases ORDER BY root_execution_id", diff --git a/tests/migrations/migrate.test.ts b/tests/migrations/migrate.test.ts index 576ade0..7317ec0 100644 --- a/tests/migrations/migrate.test.ts +++ b/tests/migrations/migrate.test.ts @@ -33,6 +33,7 @@ const PACKAGE_MIGRATIONS = [ "20:pauses_by_tree", "21:pause_answers_by_tree", "22:cost_ledger_entries", + "23:cost_entries_pruned", ]; /** The migrations after 0013, left out of a folder: a database migrated by the 0.1.0 package. */ @@ -46,6 +47,7 @@ const AFTER_0013 = { "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }; const rejectionOf = async (call: Promise): Promise => @@ -65,7 +67,7 @@ describe("migratePostgres (AC5)", () => { const second = await migratePostgres({ url }); expect(first.applied.map((m) => `${String(m.version)}:${m.name}`)).toEqual(PACKAGE_MIGRATIONS); - expect(first.version).toBe(22); + expect(first.version).toBe(23); expect(first.checkpointMigrations).toBeGreaterThan(0); expect(second).toEqual({ ...first, applied: [], checkpointMigrations: 0 }); expect(await schemaDumpOf(url)).toBe(dumped); @@ -83,6 +85,7 @@ describe("migratePostgres (AC5)", () => { "agboss_config_versions", "agboss_cost_counters", "agboss_cost_entries", + "agboss_cost_entries_pruned", "agboss_cost_max_spend", "agboss_execution_history", "agboss_execution_ids", @@ -119,7 +122,7 @@ describe("migratePostgres (AC5)", () => { const reports = await Promise.all([1, 2, 3].map(() => migratePostgres({ url }))); const applied = reports.map((report) => report.applied.length).sort(); - expect(applied).toEqual([0, 0, 22]); + expect(applied).toEqual([0, 0, 23]); expect(await historyOf(url)).toEqual(PACKAGE_MIGRATIONS); }); @@ -145,7 +148,7 @@ describe("migratePostgres (AC5)", () => { await holder.end(); expect(doneWhileLocked).toBe(false); - expect((await run).applied).toHaveLength(22); + expect((await run).applied).toHaveLength(23); }); it("a migration changed after it was applied fails the run, and nothing more is applied", async () => { @@ -193,20 +196,20 @@ describe("migratePostgres (AC5)", () => { it("a database a newer package migrated fails the run: PostgresSchemaTooNewError", async () => { const url = await freshSchema(); const newer = migrationsFolder({ - "0023_later_feature.sql": "CREATE TABLE later (id integer);", + "0024_later_feature.sql": "CREATE TABLE later (id integer);", }); await migrateWith({ url }, migrationsIn(newer)); const error = await rejectionOf(migratePostgres({ url })); expect(error).toBeInstanceOf(PostgresSchemaTooNewError); - expect(error).toMatchObject({ code: "postgres.schema-too-new", applied: 23, expected: 22 }); + expect(error).toMatchObject({ code: "postgres.schema-too-new", applied: 24, expected: 23 }); }); it("a failing migration is rolled back alone: the ones before it stay applied", async () => { const url = await freshSchema(); const broken = migrationsFolder({ - "0023_broken.sql": "CREATE TABLE half_made (id integer); SELECT no_such_function();", + "0024_broken.sql": "CREATE TABLE half_made (id integer); SELECT no_such_function();", }); await expect(migrateWith({ url }, migrationsIn(broken))).rejects.toThrow(/no_such_function/); @@ -231,6 +234,7 @@ describe("migratePostgres (AC5)", () => { { version: 20, name: "pauses_by_tree" }, { version: 21, name: "pause_answers_by_tree" }, { version: 22, name: "cost_ledger_entries" }, + { version: 23, name: "cost_entries_pruned" }, ]); expect(report.checkpointMigrations).toBe(0); }); diff --git a/tests/migrations/nul-escape-migration.test.ts b/tests/migrations/nul-escape-migration.test.ts index be8510b..5784a8f 100644 --- a/tests/migrations/nul-escape-migration.test.ts +++ b/tests/migrations/nul-escape-migration.test.ts @@ -31,6 +31,7 @@ describe("AC4: migration 0013 keeps the rows written before it", () => { "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }), ), ); diff --git a/tests/migrations/outbound-calls-migration.test.ts b/tests/migrations/outbound-calls-migration.test.ts index 1dec09d..bcfef1a 100644 --- a/tests/migrations/outbound-calls-migration.test.ts +++ b/tests/migrations/outbound-calls-migration.test.ts @@ -29,6 +29,7 @@ const BEFORE = { "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }; const leasesOn = (url: string, clock: VirtualClock, instanceId: string) => @@ -70,7 +71,7 @@ describe("migrations 0014–0016 keep the leases written before them", () => { await clock.advance(seconds(31)); expect(report.applied.map((migration) => migration.version)).toEqual([ - 14, 15, 16, 17, 18, 19, 20, 21, 22, + 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, ]); const { rows } = await onSchema( url, diff --git a/tests/migrations/pause-answers-by-tree-migration.test.ts b/tests/migrations/pause-answers-by-tree-migration.test.ts index 5d58f77..edd5242 100644 --- a/tests/migrations/pause-answers-by-tree-migration.test.ts +++ b/tests/migrations/pause-answers-by-tree-migration.test.ts @@ -62,6 +62,7 @@ describe("migration 0021 keeps one pause answer per pause of a tree", () => { migrationsFolder({ "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }), ), ); @@ -96,7 +97,7 @@ describe("migration 0021 keeps one pause answer per pause of a tree", () => { answerOf("manager-99", "2026-02-01T09:00:00.000Z"), ); - expect(report.applied.map((migration) => migration.version)).toEqual([21, 22]); + expect(report.applied.map((migration) => migration.version)).toEqual([21, 22, 23]); expect(await respondersOf(records, "old-root")).toEqual(["manager-17", "manager-18"]); expect(await respondersOf(records, "new-root")).toEqual(["manager-21"]); const { rows } = await onSchema( diff --git a/tests/migrations/pauses-by-tree-migration.test.ts b/tests/migrations/pauses-by-tree-migration.test.ts index f4d1eb2..1c3dd0c 100644 --- a/tests/migrations/pauses-by-tree-migration.test.ts +++ b/tests/migrations/pauses-by-tree-migration.test.ts @@ -56,6 +56,7 @@ async function schemaWithAnApprovedChildPause(): Promise { "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }), ), ); @@ -90,7 +91,7 @@ describe("migration 0020 keeps the ids of every execution that paused, and finds new TestApplicationContext({}, "instance-1"), ); - expect(report.applied.map((migration) => migration.version)).toEqual([20, 21, 22]); + expect(report.applied.map((migration) => migration.version)).toEqual([20, 21, 22, 23]); const { rows } = await onSchema( url, "SELECT execution_id, tenant_id FROM agboss_execution_ids ORDER BY execution_id", @@ -136,6 +137,7 @@ describe("migration 0020 keeps the ids of every execution that paused, and finds "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, "0022_cost_ledger_entries.sql": null, + "0023_cost_entries_pruned.sql": null, }), ), ); diff --git a/tests/stores/cost-ledger.test.ts b/tests/stores/cost-ledger.test.ts index 2a3289e..b408b3f 100644 --- a/tests/stores/cost-ledger.test.ts +++ b/tests/stores/cost-ledger.test.ts @@ -1,14 +1,20 @@ import { BudgetScope, type CostCounter, type UsageRecord } from "@agboss/core"; import { usd } from "@agboss/core/units"; import { describe, expect, it } from "vitest"; -import { PostgresCostLedgerRepository } from "../../src/index.js"; -import { onSchema, testDatabase } from "../support/database.js"; +import { + migratePostgres, + PostgresCostLedgerRepository, + PostgresLedgerMarkMissingError, +} from "../../src/index.js"; +import { freshSchema, onSchema, testDatabase } from "../support/database.js"; import { connectionTo } from "../support/stores.js"; /** * The cost ledger's edges beyond the contract: a counter given twice counts twice, nothing is * written without counters or calls, the period's rollover starts a new row, overlapping saves in - * opposite orders never deadlock, an unknown scope is refused, and a lost connection rejects. + * opposite orders never deadlock, an unknown scope is refused, and a lost connection rejects; + * agbossjs/agboss#771: lost answers retried while other instances prune never count twice, and a + * ledger whose pruning mark was deleted adds nothing and says why. */ const database = testDatabase(); @@ -116,3 +122,61 @@ describe("PostgresCostLedgerRepository", () => { await expect(down.saveMaxSpend(account, usd(1))).rejects.toThrow(); }); }); + +describe("PostgresCostLedgerRepository: pruning (agbossjs/agboss#771)", () => { + /** A schema of its own, migrated: the pruning mark is the schema's one row. */ + const migrated = async (): Promise => { + const url = await freshSchema(); + await migratePostgres({ url }); + return url; + }; + + const entry = (index: number): UsageRecord => ({ + ...call(0.125), + ledgerId: `pass-r:${String(index)}`, + madeAt: new Date(Date.parse("2026-10-18T08:00:00.000Z") + index * 1000).toISOString(), + }); + + it("lost answers retried on two instances while a third prunes again and again: each entry once", async () => { + const url = await migrated(); + const [first, second, sweeper] = [ledgerOn(url), ledgerOn(url), ledgerOn(url)]; + const entries = Array.from({ length: 40 }, (_, index) => entry(index)); + // every write landed; its answer was lost + for (const record of entries) await first.saveAll([day("race")], [record]); + + const outcomes = await Promise.allSettled([ + ...entries.map((record, index) => + (index % 2 === 0 ? first : second).saveAll([day("race")], [record]), + ), + ...Array.from({ length: 10 }, (_, step) => + sweeper.deleteEntriesByMadeAtBefore( + new Date(Date.parse("2026-10-18T08:00:00.000Z") + (step + 1) * 4000), + ), + ), + ]); + + for (const outcome of outcomes) + if (outcome.status === "rejected") + expect(outcome.reason).toMatchObject({ code: "cost-ledger.entry-expired" }); + expect(await first.findSpentByCounter(day("race"))).toBe(5); + const held = await onSchema(url, "SELECT count(*)::int AS n FROM agboss_cost_entries"); + expect(held.rows).toEqual([{ n: 0 }]); + }); + + it("without its pruning mark a save adds nothing and fails, naming the row", async () => { + const url = await migrated(); + const ledger = ledgerOn(url); + await onSchema(url, "DELETE FROM agboss_cost_entries_pruned"); + + await expect(ledger.saveAll([day("unmarked")], [entry(0), call(1)])).rejects.toBeInstanceOf( + PostgresLedgerMarkMissingError, + ); + await expect(ledger.saveAll([day("unmarked")], [call(1)])).rejects.toThrow( + /agboss_cost_entries_pruned has no row/, + ); + const counters = await onSchema(url, "SELECT count(*)::int AS n FROM agboss_cost_counters"); + expect(counters.rows).toEqual([{ n: 0 }]); + // a prune without it removes nothing + await ledger.deleteEntriesByMadeAtBefore(new Date("2026-10-18T09:00:00.000Z")); + }); +}); diff --git a/tests/support/database.ts b/tests/support/database.ts index 0f562f9..bec5660 100644 --- a/tests/support/database.ts +++ b/tests/support/database.ts @@ -93,7 +93,8 @@ export async function onSchema( * By `DELETE`, a table that references another first (agbossjs/agboss#760): the explorer resets * before every schedule it replays, and a `TRUNCATE` per table — new files for each, 330 ms a reset * on a loaded machine — took most of a 15-minute search; the tables hold a few rows, a `DELETE` of - * all of them is 9 ms. Sequences are kept, as `TRUNCATE` without `RESTART IDENTITY` kept them. + * all of them is 9 ms. Sequences are kept, as `TRUNCATE` without `RESTART IDENTITY` kept them; the + * ledger's pruning mark is seeded again, as the migration seeds it. */ export const truncateAll = (url: string): Promise => onSchema( @@ -117,7 +118,11 @@ export const truncateAll = (url: string): Promise => GROUP BY rel ORDER BY max(height), rel::regclass::text LOOP EXECUTE format('DELETE FROM %s', t); - END LOOP; END $$`, + END LOOP; + -- the ledger's pruning mark is one row the migration seeds (agbossjs/agboss#771) + IF to_regclass('agboss_cost_entries_pruned') IS NOT NULL THEN + INSERT INTO agboss_cost_entries_pruned VALUES (true, '-infinity'); + END IF; END $$`, ); const created: string[] = [];