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[] = [];