fix(agbossjs/agboss#771): the ledger's entries are pruned under a mark — migration 0023; core at a8fe1124

.agboss-version.json

@@ -2,6 +2,6 @@
   "repository": "agbossjs/agboss",
   "branch": "main",
   "source": "commit",
-  "frameworkSha": "0221a7c9df1ed1e3509ccb8b3b766edc47aabf69",
+  "frameworkSha": "a8fe11247f73083d428392f9e5d7600a7e6490e0",
   "version": "0.1.0"
 }

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`,

api/postgres.api.md

@@ -172,6 +172,8 @@ export interface PostgresConnectionSettings {
 export class PostgresCostLedgerRepository extends CostLedgerRepository {
     constructor(connection: PostgresConnection);
     // (undocumented)
+    deleteEntriesByMadeAtBefore(cutoff: Date): Promise<void>;
+    // (undocumented)
     findMaxSpendByAccount(account: BudgetAccount): Promise<Usd | undefined>;
     // (undocumented)
     findSpentByCounter(counter: CostCounter): Promise<Usd>;
@@ -327,6 +329,15 @@ export class PostgresFencedEffectRepository extends FencedEffectRepository<PoolC
     protected runInTransaction<TResult>(key: FencedEffectKey, work: FencedEffectWork<PoolClient, TResult>): Promise<FencedEffectOutcome<TResult>>;
 }
 
+// @public
+export class PostgresLedgerMarkMissingError extends AGBossError {
+    constructor();
+    // (undocumented)
+    static readonly code: string;
+    // (undocumented)
+    name: string;
+}
+
 // @public
 export class PostgresListener {
     close(): Promise<void>;
@@ -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".

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';

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())",
+    );
+  }
+}

src/index.ts

@@ -18,6 +18,7 @@ export {
   type MigrationReport,
 } from "./migrations/migrate.js";
 export {
+  PostgresLedgerMarkMissingError,
   PostgresMigrationChangedError,
   PostgresSchemaMissingError,
   PostgresSchemaTooNewError,

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<string, number>();
+  const entries = new Map<string, UsageRecord>();
   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,16 +144,26 @@ export class PostgresCostLedgerRepository extends CostLedgerRepository {
 
   async saveAll(counters: readonly CostCounter[], records: readonly UsageRecord[]): Promise<void> {
     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),
       counters.map((counter) => counter.period),
       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<void> {
+    await changesOf(this.connection.pool, PRUNE, [cutoff]);
   }
 
   async findMaxSpendByAccount(account: BudgetAccount): Promise<Usd | undefined> {

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"),