From 31635fa618388a21e8d0498cc66f2cb6a92cf12f Mon Sep 17 00:00:00 2001 From: pavelravvich Date: Wed, 4 Nov 2026 20:57:38 +0200 Subject: [PATCH] =?UTF-8?q?fix(agbossjs/agboss#727):=20the=20ledger=20adds?= =?UTF-8?q?=20each=20entry=20once=20=E2=80=94=20migration=200022=20agboss?= =?UTF-8?q?=5Fcost=5Fentries;=20a=20two-process=20test=20of=20a=20sibling?= =?UTF-8?q?=20child's=20spend=20in=20flight;=20core=20at=2053216599?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The core writes each tallied spend to the ledger at once, under a ledgerId, and CostLedgerRepository.saveAll adds a record with a ledgerId once. PostgresCostLedgerRepository inserts the ids into agboss_cost_entries (migration 0022) in the statement that adds their cost to the counters: a repeated or racing save of an id adds nothing. tests/processes/spend-in-flight: two processes, a child holding $0.60 at A's gate, the other child on B held back at $1.20. Refs agbossjs/agboss#727, agbossjs/agboss#732 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012Kh8JmysuhZyQRSpNyJdPg --- .agboss-version.json | 2 +- README.md | 31 ++-- api/postgres.api.md | 10 +- migrations/0022_cost_ledger_entries.sql | 7 + src/stores/postgres-cost-ledger.repository.ts | 54 ++++-- tests/cli/bin.test.ts | 2 +- tests/cli/command.test.ts | 13 +- tests/connection/postgres-connection.test.ts | 13 +- tests/fixtures/process.application.ts | 3 + tests/fixtures/spend-in-flight.workflow.ts | 168 ++++++++++++++++++ .../migrations/cost-entries-migration.test.ts | 55 ++++++ .../execution-ids-migration.test.ts | 3 +- .../execution-ids-outbound-migration.test.ts | 3 +- tests/migrations/files.test.ts | 7 +- .../lease-finished-migration.test.ts | 3 +- 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 | 9 +- .../pauses-by-tree-migration.test.ts | 9 +- tests/processes/instance.ts | 3 + tests/processes/spend-in-flight.test.ts | 40 +++++ 22 files changed, 391 insertions(+), 64 deletions(-) create mode 100644 migrations/0022_cost_ledger_entries.sql create mode 100644 tests/fixtures/spend-in-flight.workflow.ts create mode 100644 tests/migrations/cost-entries-migration.test.ts create mode 100644 tests/processes/spend-in-flight.test.ts diff --git a/.agboss-version.json b/.agboss-version.json index f116762..c1ca508 100644 --- a/.agboss-version.json +++ b/.agboss-version.json @@ -2,6 +2,6 @@ "repository": "agbossjs/agboss", "branch": "main", "source": "commit", - "frameworkSha": "06c4bc1df2880b0480475309ce998490dc4d0477", + "frameworkSha": "53216599ef55ecb94cf0649d03bd6cac73cb8aab", "version": "0.1.0" } diff --git a/README.md b/README.md index 501ca67..cb3191d 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` | -| `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` | +| `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 @@ -322,6 +322,7 @@ a pod never waits on another's migration at start (both are safe: they serialize | `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` | ## Text with a NUL character diff --git a/api/postgres.api.md b/api/postgres.api.md index 5a3a379..09c0f32 100644 --- a/api/postgres.api.md +++ b/api/postgres.api.md @@ -506,11 +506,11 @@ export interface SchemaVersions { // src/stores/postgres-child-execution-queue.repository.ts:155:9 - (ae-undocumented) Missing documentation for "completeByTenantIdAndExecutionId". // src/stores/postgres-child-execution-queue.repository.ts:176:9 - (ae-undocumented) Missing documentation for "deleteAllByTenantIdAndRootExecutionId". // src/stores/postgres-child-execution-queue.repository.ts:187:9 - (ae-undocumented) Missing documentation for "findQueueDepth". -// src/stores/postgres-cost-ledger.repository.ts:56:3 - (ae-undocumented) Missing documentation for "ping". -// src/stores/postgres-cost-ledger.repository.ts:60:9 - (ae-undocumented) Missing documentation for "findSpentByCounter". -// src/stores/postgres-cost-ledger.repository.ts:70:9 - (ae-undocumented) Missing documentation for "saveAll". -// src/stores/postgres-cost-ledger.repository.ts:81:9 - (ae-undocumented) Missing documentation for "findMaxSpendByAccount". -// src/stores/postgres-cost-ledger.repository.ts:91:9 - (ae-undocumented) Missing documentation for "saveMaxSpend". +// 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-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/0022_cost_ledger_entries.sql b/migrations/0022_cost_ledger_entries.sql new file mode 100644 index 0000000..1274814 --- /dev/null +++ b/migrations/0022_cost_ledger_entries.sql @@ -0,0 +1,7 @@ +-- agbossjs/agboss#727: spend is written to the ledger the moment it is made, each record under its +-- ledgerId (the step's key, its call, its caller, its number there). A step run again after a crash +-- or a takeover saves its spend again under the same ids: the ledger adds an entry once — its id +-- inserted here in the statement that adds its cost to the counters. +CREATE TABLE agboss_cost_entries ( + ledger_id text PRIMARY KEY +); diff --git a/src/stores/postgres-cost-ledger.repository.ts b/src/stores/postgres-cost-ledger.repository.ts index 8d2420b..8457ecc 100644 --- a/src/stores/postgres-cost-ledger.repository.ts +++ b/src/stores/postgres-cost-ledger.repository.ts @@ -12,14 +12,26 @@ import { PostgresConnection } from "../connection/postgres-connection.js"; import { changesOf, firstRowOf } from "../connection/query.js"; /** - * Adds `$5` to every counter of the arrays (one counter per index) in one statement: 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. + * 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. */ -const ADD = `INSERT INTO agboss_cost_counters AS counter +const ADD = `WITH inserted AS ( + INSERT INTO agboss_cost_entries (ledger_id) SELECT unnest($6::text[]) + 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) + ) + INSERT INTO agboss_cost_counters AS counter (budget, scope, scope_key, period, spent_usd) - SELECT budget, scope, scope_key, period, count(*) * $5::double precision + 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 @@ -27,21 +39,34 @@ const ADD = `INSERT INTO agboss_cost_counters AS counter ON CONFLICT (budget, scope, scope_key, period) DO UPDATE SET spent_usd = counter.spent_usd + EXCLUDED.spent_usd`; +/** 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 unnamed: number; +} { + 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); + } + return { ids: [...entries.keys()], costs: [...entries.values()], unnamed }; +} + const SpentRow = z.object({ spent_usd: z.number() }); const MaxSpendRow = z.object({ max_spend_usd: z.number() }); /** The scope as stored: one of `BudgetScope`'s values (anything else is refused). */ const scopeOf = (account: BudgetAccount): string => z.enum(BudgetScope).parse(account.scope); -/** What the calls cost together, in USD. */ -const costOf = (records: readonly UsageRecord[]): number => - records.reduce((sum, record) => sum + record.costUsd, 0); - /** * `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 `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 + * 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 + * 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 * (nothing is reserved). A read is one `SELECT` of the counter's row (`usd(0)` when none). The * period is the core's (the `Clock`'s UTC day, week or month), passed as text. Every statement runs @@ -69,12 +94,15 @@ 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, [ counters.map((counter) => counter.budget), counters.map(scopeOf), counters.map((counter) => counter.key), counters.map((counter) => counter.period), - costOf(records), + unnamed, + ids, + costs, ]); } diff --git a/tests/cli/bin.test.ts b/tests/cli/bin.test.ts index a6b786b..b770b87 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 21/); + expect(second.stdout).toMatch(/nothing to apply: t_\w+ is at migration 22/); 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 f686402..6f588e6 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, 21)).toEqual([ + expect(first.lines.slice(0, 22)).toEqual([ expect.stringMatching(/^applied 0001_execution_leases \(t_/), expect.stringMatching(/^applied 0002_execution_records/), expect.stringMatching(/^applied 0003_execution_trees/), @@ -45,10 +45,11 @@ describe("agboss-postgres migrate", () => { expect.stringMatching(/^applied 0019_execution_ids_from_outbound_calls/), expect.stringMatching(/^applied 0020_pauses_by_tree/), expect.stringMatching(/^applied 0021_pause_answers_by_tree/), + expect.stringMatching(/^applied 0022_cost_ledger_entries/), ]); - expect(first.lines[21]).toMatch(/^applied \d+ LangGraph checkpoint migration\(s\)$/); + expect(first.lines[22]).toMatch(/^applied \d+ LangGraph checkpoint migration\(s\)$/); expect(second.lines).toEqual([ - expect.stringMatching(/^nothing to apply: t_\w+ is at migration 21$/), + expect.stringMatching(/^nothing to apply: t_\w+ is at migration 22$/), ]); expect([...first.errors, ...second.errors]).toEqual([]); }); @@ -58,7 +59,7 @@ describe("agboss-postgres migrate", () => { const io = ioOf({ DATABASE_URL: url }); expect(await runCommand(["migrate"], io)).toBe(ExitCode.Ok); - expect(io.lines).toHaveLength(22); + expect(io.lines).toHaveLength(23); }); it("neither --url nor DATABASE_URL: a usage error, exit code 2", async () => { @@ -102,13 +103,13 @@ describe("agboss-postgres migrate", () => { const url = await freshSchema(); await migrateWith( { url }, - migrationsIn(migrationsFolder({ "0022_later.sql": "CREATE TABLE later (id integer);" })), + migrationsIn(migrationsFolder({ "0023_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 22/, + /^agboss-postgres: the migration failed — the PostgreSQL schema is at migration 23/, ); }); diff --git a/tests/connection/postgres-connection.test.ts b/tests/connection/postgres-connection.test.ts index 9f75baa..7109e0d 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: 21 }); + expect(error).toMatchObject({ code: "postgres.schema-missing", applied: 0, expected: 22 }); expect(String(error)).toContain('npx agboss-postgres migrate --url "$DATABASE_URL"'); expect(String(error)).toContain("postgresMigrateOnStart"); }); @@ -121,6 +121,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { migrationsFolder({ "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, }), ), ); @@ -129,7 +130,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { rejectionOf(connection.onInit()), ); - expect(error).toMatchObject({ applied: 19, expected: 21 }); + expect(error).toMatchObject({ applied: 19, expected: 22 }); }); it("postgresMigrateOnStart: a missing schema is migrated at start", async () => { @@ -141,7 +142,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { await withConnection(new PostgresConnection(application), (connection) => connection.onInit()); - expect(await historyOf(url)).toHaveLength(21); + expect(await historyOf(url)).toHaveLength(22); }); it("postgresMigrateOnStart: two instances starting at once migrate it once", async () => { @@ -153,14 +154,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(21); + expect(await historyOf(url)).toHaveLength(22); }); it("a schema a newer package migrated fails with PostgresSchemaTooNewError, flag or not", async () => { const url = await freshSchema(); await migrateWith( { url }, - migrationsIn(migrationsFolder({ "0022_later.sql": "CREATE TABLE later (id integer);" })), + migrationsIn(migrationsFolder({ "0023_later.sql": "CREATE TABLE later (id integer);" })), ); for (const migrateOnStart of [false, true]) { @@ -169,7 +170,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => { (connection) => rejectionOf(connection.onInit()), ); expect(error).toBeInstanceOf(PostgresSchemaTooNewError); - expect(error).toMatchObject({ applied: 22, expected: 21 }); + expect(error).toMatchObject({ applied: 23, expected: 22 }); expect(String(error)).toContain("the migrations are not rolled back"); } }); diff --git a/tests/fixtures/process.application.ts b/tests/fixtures/process.application.ts index ce476e9..b777b81 100644 --- a/tests/fixtures/process.application.ts +++ b/tests/fixtures/process.application.ts @@ -46,6 +46,7 @@ import { SpendWorkflow, } from "./shared.workflow.js"; import { PaidHoldWorkflow } from "./paid-hold.workflow.js"; +import { PayItemsWorkflow } from "./spend-in-flight.workflow.js"; import { TakeoverWorkflow } from "./takeover.workflow.js"; /** @@ -111,6 +112,7 @@ export class ProcessObserver implements OnExecutionEnd, OnExecutionLeaseTakeover ChargeWorkflow, OrderChargeWorkflow, PaidHoldWorkflow, + PayItemsWorkflow, ], observers: [ProcessObserver], budgets: [ @@ -156,6 +158,7 @@ export { ItemWorkflow, LimitedWorkflow, PaidHoldWorkflow, + PayItemsWorkflow, SpendWorkflow, TakeoverWorkflow, }; diff --git a/tests/fixtures/spend-in-flight.workflow.ts b/tests/fixtures/spend-in-flight.workflow.ts new file mode 100644 index 0000000..18aa28d --- /dev/null +++ b/tests/fixtures/spend-in-flight.workflow.ts @@ -0,0 +1,168 @@ +import { + Action, + ApplicationContext, + ChildPlacement, + ExecutionContext, + listOf, + SpendRecorder, + start, + Workflow, + WorkflowFinish, + WorkflowStart, + type Produced, +} from "@agboss/core"; +import { usd } from "@agboss/core/units"; +import { DoneDto, JobDto, SummaryDto } from "./dto.js"; +import { atGate, emit } from "./process-log.js"; + +/** + * agbossjs/agboss#727 — a tree's spend in flight seen by its own children on the other process. + * One child per job, queued with `ChildPlacement.AnyInstance`; each child pays $0.60, tells the + * test and waits at its process's gate, then pays $0.30 twice. The execution may spend $1: with + * one child holding its $0.60 at one process's gate, the other child — on the other process — + * pays its $0.60 (it read $0.60 of $1) and is held back at its next step ($1.20), because the + * first child's spend is in the ledger the moment it was made, not at the end of its superstep. + */ + +@Action({ + name: "pay-first-action", + description: "Pays $0.60 for the job, then waits at its process's gate", + input: { job: JobDto }, + output: JobDto, + deps: [ApplicationContext, ExecutionContext, SpendRecorder], +}) +export class PayFirstAction { + constructor( + private readonly applicationContext: ApplicationContext, + private readonly executionContext: ExecutionContext, + private readonly spendRecorder: SpendRecorder, + ) {} + + async execute(input: { readonly job: Produced }): Promise { + const { name } = input.job.value; + this.spendRecorder.record({ service: "job-api", cost: usd(0.6) }); + emit("paying", { name, instanceId: this.applicationContext.instanceId }); + await atGate(this.executionContext.signal); + return { name }; + } +} + +/** A paid step after the first: $0.30, told to the test. */ +function paidStep( + step: string, + job: JobDto, + applicationContext: ApplicationContext, + spendRecorder: SpendRecorder, +): Promise { + emit("paid-step", { name: job.name, step, instanceId: applicationContext.instanceId }); + spendRecorder.record({ service: "job-api", cost: usd(0.3) }); + return Promise.resolve({ name: job.name }); +} + +@Action({ + name: "pay-second-action", + description: "Pays $0.30 for the job", + input: { job: JobDto }, + output: JobDto, + deps: [ApplicationContext, SpendRecorder], +}) +export class PaySecondAction { + constructor( + private readonly applicationContext: ApplicationContext, + private readonly spendRecorder: SpendRecorder, + ) {} + + execute(input: { readonly job: Produced }): Promise { + return paidStep("pay-second", input.job.value, this.applicationContext, this.spendRecorder); + } +} + +@Action({ + name: "pay-third-action", + description: "Pays $0.30 for the job", + input: { job: JobDto }, + output: JobDto, + deps: [ApplicationContext, SpendRecorder], +}) +export class PayThirdAction { + constructor( + private readonly applicationContext: ApplicationContext, + private readonly spendRecorder: SpendRecorder, + ) {} + + execute(input: { readonly job: Produced }): Promise { + return paidStep("pay-third", input.job.value, this.applicationContext, this.spendRecorder); + } +} + +@Action({ + name: "pay-done-action", + description: "The job is paid", + input: { job: JobDto }, + output: DoneDto, + deps: [ApplicationContext], +}) +export class PayDoneAction { + constructor(private readonly applicationContext: ApplicationContext) {} + + execute(input: { readonly job: Produced }): Promise { + return Promise.resolve({ name: input.job.value.name, by: this.applicationContext.instanceId }); + } +} + +@WorkflowStart({ name: "pay-item-workflow-start", description: "One job", input: JobDto }) +export class PayItemWorkflowStart {} + +@WorkflowFinish({ name: "pay-item-workflow-finish", description: "Paid", output: DoneDto }) +export class PayItemWorkflowFinish {} + +@Workflow({ + name: "pay-item-workflow", + version: "1.0.0", + flow: [ + start(PayItemWorkflowStart) + .returns(JobDto) + .toAction(PayFirstAction) + .returns(JobDto) + .toAction(PaySecondAction) + .returns(JobDto) + .toAction(PayThirdAction) + .returns(JobDto) + .toAction(PayDoneAction) + .returns(DoneDto) + .toFinish(PayItemWorkflowFinish), + ], +}) +export class PayItemWorkflow {} + +@WorkflowStart({ name: "pay-items-workflow-start", description: "The jobs", input: listOf(JobDto) }) +export class PayItemsWorkflowStart {} + +@WorkflowFinish({ + name: "pay-items-workflow-finish", + description: "Every job, paid", + input: { results: listOf(DoneDto) }, + output: SummaryDto, +}) +export class PayItemsWorkflowFinish { + finish(input: { readonly results: readonly Produced[] }): SummaryDto { + return { text: input.results.map((result) => result.value.name).join(", ") }; + } +} + +@Workflow({ + name: "pay-items-workflow", + version: "1.0.0", + flow: [ + start(PayItemsWorkflowStart) + .returns(listOf(JobDto)) + .toParallelWorkflows(PayItemWorkflow, { + maxItems: 2, + placement: ChildPlacement.AnyInstance, + }) + .join() + .returns(listOf(DoneDto)) + .toFinish(PayItemsWorkflowFinish), + ], +}) +export class PayItemsWorkflow {} diff --git a/tests/migrations/cost-entries-migration.test.ts b/tests/migrations/cost-entries-migration.test.ts new file mode 100644 index 0000000..88f57af --- /dev/null +++ b/tests/migrations/cost-entries-migration.test.ts @@ -0,0 +1,55 @@ +import { BudgetScope, type UsageRecord } from "@agboss/core"; +import { describe, expect, it } from "vitest"; +import { migratePostgres, PostgresCostLedgerRepository } from "../../src/index.js"; +import { migrationsIn } from "../../src/migrations/files.js"; +import { migrateWith } from "../../src/migrations/migrate.js"; +import { freshSchema, onSchema } from "../support/database.js"; +import { migrationsFolder } from "../support/migration-folders.js"; +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. + */ +const tenantDay = { + budget: "tenant-daily-budget", + scope: BudgetScope.Tenant, + key: "acme", + period: "2026-11-04", +}; + +const paid = (costUsd: number, ledgerId?: string): UsageRecord => ({ + caller: "action:pay-action", + model: "pay-api", + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheWriteTokens: 0, + costUsd, + costSource: "tool", + ...(ledgerId === undefined ? {} : { isRecordedElsewhere: true, ledgerId }), +}); + +describe("migration 0022 adds the ledger's entries", () => { + it("#727: a counter of before keeps its spend; an entry saved twice after it counts once", async () => { + const url = await freshSchema(); + await migrateWith( + { url }, + migrationsIn(migrationsFolder({ "0022_cost_ledger_entries.sql": null })), + ); + await onSchema( + url, + `INSERT INTO agboss_cost_counters (budget, scope, scope_key, period, spent_usd) + VALUES ($1, $2, $3, $4, 1.5)`, + [tenantDay.budget, tenantDay.scope, tenantDay.key, tenantDay.period], + ); + + const report = await migratePostgres({ url }); + const ledger = new PostgresCostLedgerRepository(connectionTo(url)); + 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(await ledger.findSpentByCounter(tenantDay)).toBe(2.25); + }); +}); diff --git a/tests/migrations/execution-ids-migration.test.ts b/tests/migrations/execution-ids-migration.test.ts index 685abd2..2bee6cb 100644 --- a/tests/migrations/execution-ids-migration.test.ts +++ b/tests/migrations/execution-ids-migration.test.ts @@ -24,6 +24,7 @@ describe("migration 0018 keeps the ids of the trees already there", () => { "0019_execution_ids_from_outbound_calls.sql": null, "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, }), ), ); @@ -46,7 +47,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]); + expect(report.applied.map((migration) => migration.version)).toEqual([18, 19, 20, 21, 22]); 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 2403781..129367b 100644 --- a/tests/migrations/execution-ids-outbound-migration.test.ts +++ b/tests/migrations/execution-ids-outbound-migration.test.ts @@ -23,6 +23,7 @@ describe("migration 0019 keeps the ids of every execution that made an outbound "0019_execution_ids_from_outbound_calls.sql": null, "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, }); await migrateWith({ url }, migrationsIn(before)); await onSchema( @@ -50,7 +51,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]); + expect(report.applied.map((migration) => migration.version)).toEqual([18, 19, 20, 21, 22]); 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 ddb67fb..2a20af7 100644 --- a/tests/migrations/files.test.ts +++ b/tests/migrations/files.test.ts @@ -33,8 +33,9 @@ describe("the package's migration files", () => { "19:execution_ids_from_outbound_calls", "20:pauses_by_tree", "21:pause_answers_by_tree", + "22:cost_ledger_entries", ]); - expect(newestVersionOf(files)).toBe(21); + expect(newestVersionOf(files)).toBe(22); expect(files.every((file) => file.checksum === checksumOf(file.sql))).toBe(true); }); @@ -74,8 +75,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, + 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, ]); - expect(pendingMigrations(files, [])).toHaveLength(21); + expect(pendingMigrations(files, [])).toHaveLength(22); }); }); diff --git a/tests/migrations/lease-finished-migration.test.ts b/tests/migrations/lease-finished-migration.test.ts index b36692c..205873f 100644 --- a/tests/migrations/lease-finished-migration.test.ts +++ b/tests/migrations/lease-finished-migration.test.ts @@ -43,6 +43,7 @@ describe("migration 0017 keeps the leases written before it", () => { "0019_execution_ids_from_outbound_calls.sql": null, "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, }), ), ); @@ -64,7 +65,7 @@ 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]); + expect(report.applied.map((migration) => migration.version)).toEqual([17, 18, 19, 20, 21, 22]); 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 38df818..576ade0 100644 --- a/tests/migrations/migrate.test.ts +++ b/tests/migrations/migrate.test.ts @@ -32,6 +32,7 @@ const PACKAGE_MIGRATIONS = [ "19:execution_ids_from_outbound_calls", "20:pauses_by_tree", "21:pause_answers_by_tree", + "22:cost_ledger_entries", ]; /** The migrations after 0013, left out of a folder: a database migrated by the 0.1.0 package. */ @@ -44,6 +45,7 @@ const AFTER_0013 = { "0019_execution_ids_from_outbound_calls.sql": null, "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, }; const rejectionOf = async (call: Promise): Promise => @@ -63,7 +65,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(21); + expect(first.version).toBe(22); expect(first.checkpointMigrations).toBeGreaterThan(0); expect(second).toEqual({ ...first, applied: [], checkpointMigrations: 0 }); expect(await schemaDumpOf(url)).toBe(dumped); @@ -80,6 +82,7 @@ describe("migratePostgres (AC5)", () => { "agboss_child_execution_queue", "agboss_config_versions", "agboss_cost_counters", + "agboss_cost_entries", "agboss_cost_max_spend", "agboss_execution_history", "agboss_execution_ids", @@ -116,7 +119,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, 21]); + expect(applied).toEqual([0, 0, 22]); expect(await historyOf(url)).toEqual(PACKAGE_MIGRATIONS); }); @@ -142,7 +145,7 @@ describe("migratePostgres (AC5)", () => { await holder.end(); expect(doneWhileLocked).toBe(false); - expect((await run).applied).toHaveLength(21); + expect((await run).applied).toHaveLength(22); }); it("a migration changed after it was applied fails the run, and nothing more is applied", async () => { @@ -190,20 +193,20 @@ describe("migratePostgres (AC5)", () => { it("a database a newer package migrated fails the run: PostgresSchemaTooNewError", async () => { const url = await freshSchema(); const newer = migrationsFolder({ - "0022_later_feature.sql": "CREATE TABLE later (id integer);", + "0023_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: 22, expected: 21 }); + expect(error).toMatchObject({ code: "postgres.schema-too-new", applied: 23, expected: 22 }); }); it("a failing migration is rolled back alone: the ones before it stay applied", async () => { const url = await freshSchema(); const broken = migrationsFolder({ - "0022_broken.sql": "CREATE TABLE half_made (id integer); SELECT no_such_function();", + "0023_broken.sql": "CREATE TABLE half_made (id integer); SELECT no_such_function();", }); await expect(migrateWith({ url }, migrationsIn(broken))).rejects.toThrow(/no_such_function/); @@ -227,6 +230,7 @@ describe("migratePostgres (AC5)", () => { { version: 19, name: "execution_ids_from_outbound_calls" }, { version: 20, name: "pauses_by_tree" }, { version: 21, name: "pause_answers_by_tree" }, + { version: 22, name: "cost_ledger_entries" }, ]); expect(report.checkpointMigrations).toBe(0); }); diff --git a/tests/migrations/nul-escape-migration.test.ts b/tests/migrations/nul-escape-migration.test.ts index 04212a6..be8510b 100644 --- a/tests/migrations/nul-escape-migration.test.ts +++ b/tests/migrations/nul-escape-migration.test.ts @@ -30,6 +30,7 @@ describe("AC4: migration 0013 keeps the rows written before it", () => { "0019_execution_ids_from_outbound_calls.sql": null, "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, }), ), ); diff --git a/tests/migrations/outbound-calls-migration.test.ts b/tests/migrations/outbound-calls-migration.test.ts index 85b8d55..1dec09d 100644 --- a/tests/migrations/outbound-calls-migration.test.ts +++ b/tests/migrations/outbound-calls-migration.test.ts @@ -28,6 +28,7 @@ const BEFORE = { "0019_execution_ids_from_outbound_calls.sql": null, "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, }; const leasesOn = (url: string, clock: VirtualClock, instanceId: string) => @@ -69,7 +70,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, + 14, 15, 16, 17, 18, 19, 20, 21, 22, ]); 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 8db318e..5d58f77 100644 --- a/tests/migrations/pause-answers-by-tree-migration.test.ts +++ b/tests/migrations/pause-answers-by-tree-migration.test.ts @@ -58,7 +58,12 @@ describe("migration 0021 keeps one pause answer per pause of a tree", () => { const url = await freshSchema(); await migrateWith( { url }, - migrationsIn(migrationsFolder({ "0021_pause_answers_by_tree.sql": null })), + migrationsIn( + migrationsFolder({ + "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, + }), + ), ); // the rows as the store of before wrote them: one per tenant and step key for (const [key, responder] of [ @@ -91,7 +96,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]); + expect(report.applied.map((migration) => migration.version)).toEqual([21, 22]); 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 1346bd6..f4d1eb2 100644 --- a/tests/migrations/pauses-by-tree-migration.test.ts +++ b/tests/migrations/pauses-by-tree-migration.test.ts @@ -52,7 +52,11 @@ async function schemaWithAnApprovedChildPause(): Promise { await migrateWith( { url }, migrationsIn( - migrationsFolder({ "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null }), + migrationsFolder({ + "0020_pauses_by_tree.sql": null, + "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, + }), ), ); // the row as the store of before wrote it: the record as jsonb, its status and resolution time @@ -86,7 +90,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]); + expect(report.applied.map((migration) => migration.version)).toEqual([20, 21, 22]); const { rows } = await onSchema( url, "SELECT execution_id, tenant_id FROM agboss_execution_ids ORDER BY execution_id", @@ -131,6 +135,7 @@ describe("migration 0020 keeps the ids of every execution that paused, and finds migrationsFolder({ "0020_pauses_by_tree.sql": null, "0021_pause_answers_by_tree.sql": null, + "0022_cost_ledger_entries.sql": null, }), ), ); diff --git a/tests/processes/instance.ts b/tests/processes/instance.ts index ae53474..17c6f8a 100644 --- a/tests/processes/instance.ts +++ b/tests/processes/instance.ts @@ -26,6 +26,7 @@ import { SpendWorkflowStart, } from "../fixtures/shared.workflow.js"; import { PaidHoldWorkflowStart } from "../fixtures/paid-hold.workflow.js"; +import { PayItemsWorkflowStart } from "../fixtures/spend-in-flight.workflow.js"; import { TakeoverWorkflowStart } from "../fixtures/takeover.workflow.js"; import { z } from "zod"; @@ -73,6 +74,8 @@ const STARTS: Readonly< app.execute(OrderChargeWorkflowStart, Job.parse(input), options), batch: (app, input, options) => app.execute(BatchWorkflowStart, z.array(Job).parse(input), options), + "pay-items": (app, input, options) => + app.execute(PayItemsWorkflowStart, z.array(Job).parse(input), options), }; /** Starts the named workflow with `input`, checked against its start's input, for `tenantId`. */ diff --git a/tests/processes/spend-in-flight.test.ts b/tests/processes/spend-in-flight.test.ts new file mode 100644 index 0000000..2cfb8b1 --- /dev/null +++ b/tests/processes/spend-in-flight.test.ts @@ -0,0 +1,40 @@ +/** + * agbossjs/agboss#727 — a tree's spend in flight, seen by its own children on the other process. + * Two `node` processes on one PostgreSQL, one step slot each; the execution may spend $1. A's + * child pays $0.60 and waits at A's gate; B, started then, claims the other child, which pays its + * $0.60 (the ledger held A's $0.60 by then: $0.60 of $1) and waits at B's gate. When B's gate + * opens, that child's next step is held back at $1.20: A's child's spend was in the ledger the + * moment it was made, though its step never ended. Before #727 it reached the ledger only at the + * end of its superstep, so B's child paid twice more and finished at $1.50 — the tree $1.80. + */ +import { describe, expect, it } from "vitest"; +import { testDatabase } from "../support/database.js"; +import { eventsOf, launch } from "./instances.js"; + +const database = testDatabase(); + +describe("agbossjs/agboss#727: a sibling child's spend on the other process", () => { + it("the child on B is held back at $1.20 while the child on A holds its $0.60", async () => { + const a = await launch(database.url()); + const run = a.send("execute", { + workflow: "pay-items", + input: [{ name: "x" }, { name: "y" }], + tenantId: "inflight", + }); + run.catch(() => undefined); + const first = await a.next((event) => event.event === "paying"); + const b = await launch(database.url()); + const second = await b.next((event) => event.event === "paying"); + expect(second.name).not.toBe(first.name); + + await b.send("open"); + + await expect(run).rejects.toMatchObject({ + name: "BudgetExceededError", + message: expect.stringMatching(/spent \$1\.2;/), + }); + // neither child paid past its first step: the one on B was held back at once + expect(eventsOf(b, "paid-step")).toEqual([]); + expect(eventsOf(a, "paid-step")).toEqual([]); + }, 120_000); +});