From 95bf197191fa14dc4681cc21c09f604f55f9aafc Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:16:46 +0200 Subject: [PATCH 1/3] 40947f62 - Create the missing Binance custody assets that wedged ledger posting (#4452) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(ledger): add the missing Binance custody assets for ONDO and ADA Every asset held on an exchange needs a second `asset` row with type='Custody' and blockchain=, next to its on-chain row. Both were missing for ONDO and ADA on Binance, so LedgerBootstrapService — which iterates over `asset` rows — could never create their CoA accounts. Without the account, exchangeAssetByCcy fail-closes with "Ledger account Binance/ONDO not found (CoA bootstrap missing)" and processForward breaks out of the batch, leaving the exchange_tx watermark frozen: every later row stays unposted, whatever exchange it belongs to. The recurring CoA bootstrap cron cannot heal this — it only creates accounts for `asset` rows that exist. ADA has the same gap and only escaped the wedge because its last Binance trade predates the ledger cutover. The migration is idempotent per uniqueName and takes a transaction-scoped advisory lock first: uniqueName is not DB-unique (the unique index covers (dexName, type, blockchain)) and there is no ENVIRONMENT guard, so two instances starting at once could otherwise both pass the idempotency check and have the second INSERT crash on the index. Prices come from each coin's on-chain row via subquery, with a fail-loud guard instead of a silent NULL priceRuleId. * fix(ledger): check the target row before the price source in the custody migration up() claimed to be idempotent per uniqueName but evaluated the price-source guard first, so a re-run against an already-created asset still threw once its source had been renamed or removed. Checking the target first — as AddSavingZchfAsset does — makes the claim true and keeps the guard on the path that actually inserts. Both assets stay independent: an existing ONDO row no longer skips ADA. Also narrows the guard's doc comment to what it really rejects, a missing or renamed source row. It does not require the source to carry a priceRuleId, and an empty one is legitimate — the seed ships Ethereum/ONDO exactly like that, and the subquery carries the NULL over on purpose. * docs(ledger): say which environments actually seed the custody rows The header claimed dev, loc and CI all mirror the two rows from migration/seed/asset.csv. Only LOC does: main.ts runs the seed exclusively for Environment.LOC and seed.js blocks everything else outright, so dev, CI and prod receive the rows from this migration alone. --- ...20000000-AddBinanceCustodyAssetsOndoAda.js | 148 +++++++ migration/seed/asset.csv | 2 + ...-custody-assets-ondo-ada.migration.spec.ts | 384 ++++++++++++++++++ 3 files changed, 534 insertions(+) create mode 100644 migration/1785320000000-AddBinanceCustodyAssetsOndoAda.js create mode 100644 src/shared/models/asset/__tests__/add-binance-custody-assets-ondo-ada.migration.spec.ts diff --git a/migration/1785320000000-AddBinanceCustodyAssetsOndoAda.js b/migration/1785320000000-AddBinanceCustodyAssetsOndoAda.js new file mode 100644 index 0000000000..ef71735bd4 --- /dev/null +++ b/migration/1785320000000-AddBinanceCustodyAssetsOndoAda.js @@ -0,0 +1,148 @@ +/** + * @typedef {import('typeorm').MigrationInterface} MigrationInterface + * @typedef {import('typeorm').QueryRunner} QueryRunner + */ + +/** + * Data migration that creates the missing Binance Custody master-data rows for ONDO and ADA + * (Binance/ONDO, Binance/ADA). + * + * Every asset DFX holds on an exchange needs a second `asset` row with type='Custody' and + * blockchain=, separate from the on-chain asset. LedgerBootstrapService can only + * create a ledger CoA account for rows that exist — without these Custody rows, exchange-tx + * ledger posting fail-closes with "Ledger account Binance/ONDO not found (CoA bootstrap + * missing)" and stalls the consumer watermark. + * + * Unlike prod-only wiring migrations (e.g. AddBankFrickCustodyAssets), these rows are pure + * master data with no LiquidityManagementRule, bank link, or external API side effect — they + * are correct and desired in every environment identically, so there is no ENVIRONMENT guard. + * Only LOC also mirrors them via migration/seed/asset.csv (ids 413/414); DEV/CI/PRD get them + * solely via this migration. + * + * Each asset is priced off a single on-chain source via subquery (no COALESCE fallback): + * Binance/ONDO ← Ethereum/ONDO + * Binance/ADA ← Cardano/ADA + * The fail-loud price-source guards only check that the source row exists (SELECT "id"); they + * do not check whether that row has a priceRuleId. A source with priceRuleId IS NULL is a + * legitimate state — migration/seed/asset.csv ships Ethereum/ONDO with an empty priceRuleId — + * and the INSERT subquery correctly and silently carries that NULL over, same as the precedent + * AddSavingZchfAsset. What the guards actually prevent is a missing or renamed source row. + * + * decimals/sortOrder stay NULL (omitted from the INSERT) — same as every existing Custody + * asset. refundEnabled is false (unlike the entity default of true), matching other live + * Custody assets in this system. + * + * up() acquires a transaction-scoped advisory lock before the idempotency checks and + * price-source guards: uniqueName is not DB-unique (only (dexName, type, blockchain) is), + * and this migration has no ENVIRONMENT guard, so it can run concurrently from multiple + * app instances starting at once. Without the lock, two concurrent runs could both pass + * the idempotency SELECT before either INSERTs, and the second INSERT would crash on the + * unique index instead of no-op'ing. The lock key is this migration's own timestamp — + * unique across migrations by naming convention, and far outside hashtext()'s 32-bit + * range, so it can never collide with the application's hashtext()-based advisory locks + * (setting.repository.ts, custody-account.service.ts, realunit.service.ts). There is no + * SET LOCAL lock_timeout — unlike AddBankFrickCustodyAssets this path has no bank-table + * UPDATE contending for row locks under load; the advisory lock alone serializes concurrent + * up() execution. + * + * up() is fully idempotent per uniqueName: each asset's existence check runs before its + * price-source guard, so a re-run against an already-created row never depends on the source + * asset still existing under the same name. down() has no lock (rollback is a deliberate + * manual Ops action, not a multi-instance boot race) and deletes both rows by uniqueName; + * the DELETE fails loud (FK violation) if a liquidity_balance / ledger row already + * references them — intentional: rolling back a used wiring is an Ops procedure, not a + * plain migration revert. + * + * @class + * @implements {MigrationInterface} + */ +module.exports = class AddBinanceCustodyAssetsOndoAda1785320000000 { + name = 'AddBinanceCustodyAssetsOndoAda1785320000000'; + + /** + * @param {QueryRunner} queryRunner + */ + async up(queryRunner) { + await queryRunner.query(`SELECT pg_advisory_xact_lock(1785320000000)`); + + // --- Binance/ONDO --- + // Idempotent: assets are keyed by the stable uniqueName (ids are env-specific). Checked + // before the price-source guard so a re-run against an already-created row never depends + // on the source asset still existing under the same name. + const ondoExisting = (await queryRunner.query(`SELECT "id" FROM "asset" WHERE "uniqueName" = 'Binance/ONDO'`)).at( + 0, + ); + if (!ondoExisting) { + // Fail-loud price-source guard: ONDO has exactly one price source (Ethereum/ONDO). + // No COALESCE fallback — an insert without a real price source would silently mask a + // missing/renamed source asset. + const ondoPriceSource = ( + await queryRunner.query(`SELECT "id" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO'`) + ).at(0); + if (!ondoPriceSource) { + throw new Error('Cannot create Binance/ONDO custody asset: price source Ethereum/ONDO not found'); + } + + await queryRunner.query(` + INSERT INTO "asset" + ("name", "uniqueName", "type", "blockchain", "category", "dexName", "financialType", + "buyable", "sellable", "cardBuyable", "cardSellable", "instantBuyable", "instantSellable", + "paymentEnabled", "refEnabled", "refundEnabled", "ikna", "personalIbanEnabled", "comingSoon", + "priceRuleId", "approxPriceChf", "approxPriceEur", "approxPriceUsd") + VALUES + ('ONDO', 'Binance/ONDO', 'Custody', 'Binance', 'Private', 'ONDO', 'Other', + false, false, false, false, false, false, + false, false, false, false, false, false, + (SELECT "priceRuleId" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO'), + (SELECT "approxPriceChf" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO'), + (SELECT "approxPriceEur" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO'), + (SELECT "approxPriceUsd" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO')) + `); + } + + // --- Binance/ADA --- + // Idempotent: assets are keyed by the stable uniqueName (ids are env-specific). Checked + // before the price-source guard so a re-run against an already-created row never depends + // on the source asset still existing under the same name. + const adaExisting = (await queryRunner.query(`SELECT "id" FROM "asset" WHERE "uniqueName" = 'Binance/ADA'`)).at(0); + if (!adaExisting) { + // Fail-loud price-source guard: ADA has exactly one price source (Cardano/ADA). + // No COALESCE fallback — an insert without a real price source would silently mask a + // missing/renamed source asset. + const adaPriceSource = ( + await queryRunner.query(`SELECT "id" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA'`) + ).at(0); + if (!adaPriceSource) { + throw new Error('Cannot create Binance/ADA custody asset: price source Cardano/ADA not found'); + } + + await queryRunner.query(` + INSERT INTO "asset" + ("name", "uniqueName", "type", "blockchain", "category", "dexName", "financialType", + "buyable", "sellable", "cardBuyable", "cardSellable", "instantBuyable", "instantSellable", + "paymentEnabled", "refEnabled", "refundEnabled", "ikna", "personalIbanEnabled", "comingSoon", + "priceRuleId", "approxPriceChf", "approxPriceEur", "approxPriceUsd") + VALUES + ('ADA', 'Binance/ADA', 'Custody', 'Binance', 'Private', 'ADA', 'Other', + false, false, false, false, false, false, + false, false, false, false, false, false, + (SELECT "priceRuleId" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA'), + (SELECT "approxPriceChf" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA'), + (SELECT "approxPriceEur" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA'), + (SELECT "approxPriceUsd" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA')) + `); + } + } + + /** + * @param {QueryRunner} queryRunner + */ + async down(queryRunner) { + // No advisory lock: rollback is a deliberate manual Ops action, not a multi-instance boot + // race the way up() is. No FK-safety check, matching AddBankFrickCustodyAssets: rolling + // back an asset already referenced by liquidity_balance / ledger rows is an Ops procedure, + // not a plain revert. This DELETE will fail loud (FK violation) if such a reference + // already exists — intentional. + await queryRunner.query(`DELETE FROM "asset" WHERE "uniqueName" IN ('Binance/ONDO', 'Binance/ADA')`); + } +}; diff --git a/migration/seed/asset.csv b/migration/seed/asset.csv index 09b49641de..72dac259e6 100644 --- a/migration/seed/asset.csv +++ b/migration/seed/asset.csv @@ -228,3 +228,5 @@ id,name,type,buyable,sellable,chainId,sellCommand,dexName,category,blockchain,un 410,EUR,Custody,FALSE,FALSE,,,EUR,Private,OlkyFrozen,OlkyFrozen/EUR,,FALSE,,1.17786809,FALSE,39,0.9287514723,FALSE,FALSE,FALSE,FALSE,EUR,,FALSE,0,0,1,TRUE 411,EUR,Custody,FALSE,FALSE,,,EUR,Private,Frick,Frick/EUR,,FALSE,,1.17786809,FALSE,39,0.9287514723,FALSE,FALSE,FALSE,FALSE,EUR,,FALSE,0,0,1,TRUE 412,CHF,Custody,FALSE,FALSE,,,CHF,Private,Frick,Frick/CHF,,FALSE,,1.268227427,FALSE,37,1,FALSE,FALSE,FALSE,FALSE,CHF,,FALSE,0,0,1.076714309,TRUE +413,ONDO,Custody,FALSE,FALSE,,,ONDO,Private,Binance,Binance/ONDO,,FALSE,,,FALSE,,,FALSE,FALSE,FALSE,FALSE,Other,,FALSE,0,0,,FALSE +414,ADA,Custody,FALSE,FALSE,,,ADA,Private,Binance,Binance/ADA,,FALSE,,0.3492050313,FALSE,63,0.2753489034,FALSE,FALSE,FALSE,FALSE,Other,,FALSE,0,0,0.2964721043,FALSE diff --git a/src/shared/models/asset/__tests__/add-binance-custody-assets-ondo-ada.migration.spec.ts b/src/shared/models/asset/__tests__/add-binance-custody-assets-ondo-ada.migration.spec.ts new file mode 100644 index 0000000000..d02a104f2b --- /dev/null +++ b/src/shared/models/asset/__tests__/add-binance-custody-assets-ondo-ada.migration.spec.ts @@ -0,0 +1,384 @@ +import { DataSource, QueryRunner } from 'typeorm'; + +const PG_URL = process.env.MIGRATION_TEST_PG; +const describeDb = PG_URL ? describe : describe.skip; +const SCHEMA = 'binance_custody_assets_ondo_ada_spec'; + +let AddBinanceCustodyAssetsOndoAda: new () => { + up(queryRunner: QueryRunner): Promise; + down(queryRunner: QueryRunner): Promise; +}; + +describe('AddBinanceCustodyAssetsOndoAda migration (SQL content)', () => { + beforeAll(() => { + // eslint-disable-next-line @typescript-eslint/no-require-imports + AddBinanceCustodyAssetsOndoAda = require('../../../../../migration/1785320000000-AddBinanceCustodyAssetsOndoAda'); + }); + + it('is idempotent: both assets already exist → no INSERT calls', async () => { + const migration = new AddBinanceCustodyAssetsOndoAda(); + const queryRunner = { + query: jest.fn(async (sql: string) => { + const s = sql.toLowerCase(); + if (s.includes('from "asset"') && s.includes(`'ethereum/ondo'`)) return [{ id: 1 }]; + if (s.includes('from "asset"') && s.includes(`'binance/ondo'`)) return [{ id: 10 }]; + if (s.includes('from "asset"') && s.includes(`'cardano/ada'`)) return [{ id: 2 }]; + if (s.includes('from "asset"') && s.includes(`'binance/ada'`)) return [{ id: 11 }]; + return []; + }), + }; + + await migration.up(queryRunner as unknown as QueryRunner); + + const calls = queryRunner.query.mock.calls as [string, unknown[]?][]; + expect(calls[0][0].toLowerCase()).toContain('pg_advisory_xact_lock'); + expect(calls.some(([statement]) => statement.toLowerCase().includes('insert'))).toBe(false); + }); + + it('inserts both Custody assets with single price-source subqueries, no COALESCE, no decimals/sortOrder', async () => { + const migration = new AddBinanceCustodyAssetsOndoAda(); + const queryRunner = { + query: jest.fn(async (sql: string) => { + const s = sql.toLowerCase(); + // price sources present + if (s.includes('from "asset"') && s.includes(`'ethereum/ondo'`) && !s.includes('insert')) { + return [{ id: 1 }]; + } + if (s.includes('from "asset"') && s.includes(`'cardano/ada'`) && !s.includes('insert')) { + return [{ id: 2 }]; + } + // idempotency: neither custody row exists yet + if (s.includes('from "asset"') && s.includes(`'binance/ondo'`) && !s.includes('insert')) { + return []; + } + if (s.includes('from "asset"') && s.includes(`'binance/ada'`) && !s.includes('insert')) { + return []; + } + return []; + }), + }; + + await migration.up(queryRunner as unknown as QueryRunner); + + const calls = queryRunner.query.mock.calls as [string, unknown[]?][]; + const sql = calls.map(([statement]) => statement).join('\n'); + + // Every query must be single-argument (no bound parameter arrays) + for (const call of calls) { + expect(call).toHaveLength(1); + } + + // Advisory lock must be the very first call (serializes concurrent multi-instance starts) + expect(calls[0][0]).toContain('pg_advisory_xact_lock(1785320000000)'); + expect(sql).toContain(`'ONDO'`); + expect(sql).toContain(`'Binance/ONDO'`); + expect(sql).toContain(`'ADA'`); + expect(sql).toContain(`'Binance/ADA'`); + expect(sql).toContain(`'Custody'`); + expect(sql).toContain(`'Private'`); + expect(sql).toContain(`'Binance'`); + expect(sql).toContain(`'Other'`); + expect(sql).toContain(`(SELECT "priceRuleId" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO')`); + expect(sql).toContain(`(SELECT "approxPriceChf" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO')`); + expect(sql).toContain(`(SELECT "approxPriceEur" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO')`); + expect(sql).toContain(`(SELECT "approxPriceUsd" FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO')`); + expect(sql).toContain(`(SELECT "priceRuleId" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA')`); + expect(sql).toContain(`(SELECT "approxPriceChf" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA')`); + expect(sql).toContain(`(SELECT "approxPriceEur" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA')`); + expect(sql).toContain(`(SELECT "approxPriceUsd" FROM "asset" WHERE "uniqueName" = 'Cardano/ADA')`); + expect(sql).not.toContain('COALESCE'); + expect(sql).not.toContain('"decimals"'); + expect(sql).not.toContain('"sortOrder"'); + // refundEnabled must be explicit false (entity default is true) + expect(sql).toMatch(/"refundEnabled"[\s\S]*false/); + }); + + it('throws when Ethereum/ONDO price source is missing and does not INSERT', async () => { + const migration = new AddBinanceCustodyAssetsOndoAda(); + const queryRunner = { + query: jest.fn(async (sql: string) => { + const s = sql.toLowerCase(); + if (s.includes('from "asset"') && s.includes(`'ethereum/ondo'`)) return []; + return []; + }), + }; + + await expect(migration.up(queryRunner as unknown as QueryRunner)).rejects.toThrow( + 'Cannot create Binance/ONDO custody asset: price source Ethereum/ONDO not found', + ); + + const calls = queryRunner.query.mock.calls as [string, unknown[]?][]; + expect(calls[0][0].toLowerCase()).toContain('pg_advisory_xact_lock'); + expect(calls.some(([statement]) => statement.toLowerCase().includes('insert'))).toBe(false); + }); + + it('throws when Cardano/ADA price source is missing and does not INSERT (ONDO already present)', async () => { + const migration = new AddBinanceCustodyAssetsOndoAda(); + const queryRunner = { + query: jest.fn(async (sql: string) => { + const s = sql.toLowerCase(); + // ONDO path: price source ok, custody row already exists → skip INSERT + if (s.includes('from "asset"') && s.includes(`'ethereum/ondo'`)) return [{ id: 1 }]; + if (s.includes('from "asset"') && s.includes(`'binance/ondo'`)) return [{ id: 10 }]; + // ADA path: price source missing + if (s.includes('from "asset"') && s.includes(`'cardano/ada'`)) return []; + return []; + }), + }; + + await expect(migration.up(queryRunner as unknown as QueryRunner)).rejects.toThrow( + 'Cannot create Binance/ADA custody asset: price source Cardano/ADA not found', + ); + + const calls = queryRunner.query.mock.calls as [string, unknown[]?][]; + expect(calls[0][0].toLowerCase()).toContain('pg_advisory_xact_lock'); + expect(calls.some(([statement]) => statement.toLowerCase().includes('insert'))).toBe(false); + }); +}); + +describeDb('AddBinanceCustodyAssetsOndoAda migration (real Postgres)', () => { + let dataSource: DataSource; + let queryRunner: QueryRunner; + + beforeAll(async () => { + // eslint-disable-next-line @typescript-eslint/no-require-imports + AddBinanceCustodyAssetsOndoAda = require('../../../../../migration/1785320000000-AddBinanceCustodyAssetsOndoAda'); + dataSource = new DataSource({ type: 'postgres', url: PG_URL }); + await dataSource.initialize(); + }); + + beforeEach(async () => { + queryRunner = dataSource.createQueryRunner(); + await queryRunner.connect(); + await queryRunner.query(`DROP SCHEMA IF EXISTS "${SCHEMA}" CASCADE`); + await queryRunner.query(`CREATE SCHEMA "${SCHEMA}"`); + await queryRunner.query(`SET search_path TO "${SCHEMA}"`); + + // Minimal fixture table — columns the migration touches, plus decimals/sortOrder so the + // post-condition NULL checks below have columns to assert against. refundEnabled mirrors + // the real entity default (true) so omitting it from INSERT would silently get the wrong value. + await queryRunner.query(` + CREATE TABLE "asset" ( + "id" SERIAL PRIMARY KEY, + "updated" TIMESTAMP NOT NULL DEFAULT now(), + "created" TIMESTAMP NOT NULL DEFAULT now(), + "name" varchar(256) NOT NULL, + "uniqueName" varchar(256) NOT NULL, + "type" varchar(256) NOT NULL, + "blockchain" varchar(256) NOT NULL, + "category" varchar(256) NOT NULL DEFAULT 'Public', + "dexName" varchar(256), + "financialType" varchar(256), + "buyable" boolean NOT NULL DEFAULT true, + "sellable" boolean NOT NULL DEFAULT true, + "cardBuyable" boolean NOT NULL DEFAULT true, + "cardSellable" boolean NOT NULL DEFAULT true, + "instantBuyable" boolean NOT NULL DEFAULT true, + "instantSellable" boolean NOT NULL DEFAULT true, + "paymentEnabled" boolean NOT NULL DEFAULT false, + "refEnabled" boolean NOT NULL DEFAULT false, + "refundEnabled" boolean NOT NULL DEFAULT true, + "ikna" boolean NOT NULL DEFAULT false, + "personalIbanEnabled" boolean NOT NULL DEFAULT false, + "comingSoon" boolean NOT NULL DEFAULT false, + "decimals" integer, + "sortOrder" integer, + "priceRuleId" integer, + "approxPriceChf" double precision, + "approxPriceEur" double precision, + "approxPriceUsd" double precision + ) + `); + + // Price-source fixtures. + await queryRunner.query(` + INSERT INTO "asset" + ("name", "uniqueName", "type", "blockchain", "category", "dexName", "financialType", + "buyable", "sellable", "cardBuyable", "cardSellable", "instantBuyable", "instantSellable", + "paymentEnabled", "refEnabled", "refundEnabled", "ikna", "personalIbanEnabled", "comingSoon", + "priceRuleId", "approxPriceChf", "approxPriceEur", "approxPriceUsd") + VALUES + ('ONDO', 'Ethereum/ONDO', 'Token', 'Ethereum', 'Public', 'ONDO', 'Other', + true, true, true, true, true, true, false, false, true, false, false, false, + 98, 0.8, 0.85, 1.0), + ('ADA', 'Cardano/ADA', 'Coin', 'Cardano', 'Public', 'ADA', 'Other', + true, true, true, true, true, true, false, false, true, false, false, false, + 63, 0.2753489034, 0.2964721043, 0.3492050313) + `); + }); + + afterEach(async () => { + if (queryRunner.isTransactionActive) await queryRunner.rollbackTransaction(); + await queryRunner.query(`SET search_path TO public`); + await queryRunner.query(`DROP SCHEMA IF EXISTS "${SCHEMA}" CASCADE`); + await queryRunner.release(); + }); + + afterAll(async () => { + if (dataSource?.isInitialized) await dataSource.destroy(); + }); + + it('creates Binance/ONDO and Binance/ADA priced off their sources, flags false', async () => { + const migration = new AddBinanceCustodyAssetsOndoAda(); + await migration.up(queryRunner); + + const rows = await queryRunner.query( + `SELECT "name", "uniqueName", "type", "category", "blockchain", "dexName", "financialType", + "buyable", "sellable", "cardBuyable", "cardSellable", "instantBuyable", "instantSellable", + "paymentEnabled", "refEnabled", "refundEnabled", "ikna", "personalIbanEnabled", "comingSoon", + "decimals", "sortOrder", "priceRuleId", "approxPriceChf", "approxPriceEur", "approxPriceUsd" + FROM "asset" WHERE "uniqueName" IN ('Binance/ONDO', 'Binance/ADA') ORDER BY "uniqueName"`, + ); + + expect(rows).toHaveLength(2); + expect(rows[0]).toMatchObject({ + name: 'ADA', + uniqueName: 'Binance/ADA', + type: 'Custody', + category: 'Private', + blockchain: 'Binance', + dexName: 'ADA', + financialType: 'Other', + buyable: false, + sellable: false, + cardBuyable: false, + cardSellable: false, + instantBuyable: false, + instantSellable: false, + paymentEnabled: false, + refEnabled: false, + refundEnabled: false, + ikna: false, + personalIbanEnabled: false, + comingSoon: false, + priceRuleId: 63, + approxPriceChf: 0.2753489034, + approxPriceEur: 0.2964721043, + approxPriceUsd: 0.3492050313, + }); + expect(rows[0].decimals).toBeNull(); + expect(rows[0].sortOrder).toBeNull(); + + expect(rows[1]).toMatchObject({ + name: 'ONDO', + uniqueName: 'Binance/ONDO', + type: 'Custody', + category: 'Private', + blockchain: 'Binance', + dexName: 'ONDO', + financialType: 'Other', + buyable: false, + sellable: false, + cardBuyable: false, + cardSellable: false, + instantBuyable: false, + instantSellable: false, + paymentEnabled: false, + refEnabled: false, + refundEnabled: false, + ikna: false, + personalIbanEnabled: false, + comingSoon: false, + priceRuleId: 98, + approxPriceChf: 0.8, + approxPriceEur: 0.85, + approxPriceUsd: 1.0, + }); + expect(rows[1].decimals).toBeNull(); + expect(rows[1].sortOrder).toBeNull(); + }); + + it('is idempotent: re-running up() does not create duplicate rows', async () => { + const migration = new AddBinanceCustodyAssetsOndoAda(); + await migration.up(queryRunner); + await migration.up(queryRunner); + + const count = ( + await queryRunner.query( + `SELECT COUNT(*)::int AS c FROM "asset" WHERE "uniqueName" IN ('Binance/ONDO', 'Binance/ADA')`, + ) + )[0].c; + + expect(count).toBe(2); + }); + + it('down() removes both custody rows and leaves price sources untouched', async () => { + const migration = new AddBinanceCustodyAssetsOndoAda(); + await migration.up(queryRunner); + await migration.down(queryRunner); + + const custody = await queryRunner.query( + `SELECT COUNT(*)::int AS c FROM "asset" WHERE "uniqueName" IN ('Binance/ONDO', 'Binance/ADA')`, + ); + const sources = await queryRunner.query( + `SELECT COUNT(*)::int AS c FROM "asset" WHERE "uniqueName" IN ('Ethereum/ONDO', 'Cardano/ADA')`, + ); + + expect(custody[0].c).toBe(0); + expect(sources[0].c).toBe(2); + }); + + it('throws when Ethereum/ONDO is missing and leaves no custody rows', async () => { + await queryRunner.query(`DELETE FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO'`); + + const migration = new AddBinanceCustodyAssetsOndoAda(); + await expect(migration.up(queryRunner)).rejects.toThrow( + 'Cannot create Binance/ONDO custody asset: price source Ethereum/ONDO not found', + ); + + const rows = await queryRunner.query( + `SELECT COUNT(*)::int AS c FROM "asset" WHERE "uniqueName" IN ('Binance/ONDO', 'Binance/ADA')`, + ); + expect(rows[0].c).toBe(0); + }); + + it('throws when Cardano/ADA is missing and does not insert Binance/ADA', async () => { + await queryRunner.query(`DELETE FROM "asset" WHERE "uniqueName" = 'Cardano/ADA'`); + + const migration = new AddBinanceCustodyAssetsOndoAda(); + await expect(migration.up(queryRunner)).rejects.toThrow( + 'Cannot create Binance/ADA custody asset: price source Cardano/ADA not found', + ); + + // ONDO is inserted before the ADA guard runs; ADA must not exist. This asserts up()'s + // internal order / fail-fast on a non-transactional QueryRunner (this suite's beforeEach + // never calls startTransaction(), so the insert auto-commits). The partial state seen here + // (ONDO present, ADA guard throwing) is never visible on disk in production: real deploys + // run migrations inside TypeORM's migration transaction (migrationsRun in + // src/config/config.ts; no migrationsTransactionMode override → default 'all'), so the + // whole batch is one transaction and a throw at the ADA guard rolls the ONDO insert back. + const ondo = await queryRunner.query(`SELECT COUNT(*)::int AS c FROM "asset" WHERE "uniqueName" = 'Binance/ONDO'`); + const ada = await queryRunner.query(`SELECT COUNT(*)::int AS c FROM "asset" WHERE "uniqueName" = 'Binance/ADA'`); + expect(ondo[0].c).toBe(1); + expect(ada[0].c).toBe(0); + }); + + it('target already exists but its price source was removed: up() resolves without throwing and does not insert a duplicate', async () => { + // Simulate a pre-existing Binance/ONDO row (as if a previous, successful run already + // created it) whose price source has since been renamed/removed. + await queryRunner.query(` + INSERT INTO "asset" + ("name", "uniqueName", "type", "blockchain", "category", "dexName", "financialType", + "buyable", "sellable", "cardBuyable", "cardSellable", "instantBuyable", "instantSellable", + "paymentEnabled", "refEnabled", "refundEnabled", "ikna", "personalIbanEnabled", "comingSoon", + "priceRuleId", "approxPriceChf", "approxPriceEur", "approxPriceUsd") + VALUES + ('ONDO', 'Binance/ONDO', 'Custody', 'Binance', 'Private', 'ONDO', 'Other', + false, false, false, false, false, false, + false, false, false, false, false, false, + 98, 0.8, 0.85, 1.0) + `); + await queryRunner.query(`DELETE FROM "asset" WHERE "uniqueName" = 'Ethereum/ONDO'`); + + const migration = new AddBinanceCustodyAssetsOndoAda(); + await expect(migration.up(queryRunner)).resolves.toBeUndefined(); + + const ondo = await queryRunner.query(`SELECT COUNT(*)::int AS c FROM "asset" WHERE "uniqueName" = 'Binance/ONDO'`); + expect(ondo[0].c).toBe(1); + + // ADA is unaffected by ONDO's missing price source — each asset's existence check and + // price-source guard are independent, so ADA is still created normally. + const ada = await queryRunner.query(`SELECT COUNT(*)::int AS c FROM "asset" WHERE "uniqueName" = 'Binance/ADA'`); + expect(ada[0].c).toBe(1); + }); +}); From e362e3a5437976884d3e09fd40b90e53df62a488 Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:23:46 +0200 Subject: [PATCH 2/3] 7392bb63 - Project asset prices in SQL instead of loading whole snapshots (#4464) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * perf(ledger): project asset prices in SQL instead of loading whole snapshots LedgerMarkService.preload loaded complete log rows and parsed each message in Node, only to keep three values per asset: created, assetId and priceChf. A snapshot carries 200 assets in an average 42 KB JSON document. Measured on the real two-day preload window in production: 930 rows and 38 MB transferred, of which 7.3 MB is the data actually used. The JSON.parse over those 38 MB runs on the event loop and is a direct cause of the API container sitting at a saturated core, which slows every request. The new repository method projects the prices in SQL via LATERAL jsonb_each and returns flat rows. The LIMIT and the keyset cursor stay on the log rows, not on the expanded result, so pagination semantics are unchanged — the cursor therefore carries the originating log id. getLatestMarks keeps using getFinancialLogs: it runs on the dailySample path, where the row count is already small and the added complexity would not pay for itself. Mark map contents, ordering and the handling of missing or non-finite prices are unchanged. * fix(ledger): keep every read log row countable and tighten price validation Review follow-ups on the SQL price projection. A log row whose assets yield no usable price used to vanish from the projection entirely. Overflow detection and keyset pagination count log rows, so a vanished row made them undercount: preload could miss that more data exists, and paginate could stop early — a silent loss of marks. The projection now uses LEFT JOIN LATERAL, so every read log row comes back even when it carries no usable price, and the counting is correct again. This also removes a related false end-of-data path in the stale-cursor guard. Price conversion now matches the previous Number.isFinite semantics: a price is only taken when the JSON value really is a number, so a string "1.25" is skipped as before rather than silently accepted by the float cast. Non-numeric asset keys no longer abort the whole query; they yield null and are skipped. Malformed message JSON now fails loudly instead of being swallowed — deliberate, and documented on the method. buildMarkMap keeps Number.isFinite as a second line of defence. The repository already nulls NaN and Infinity, but a NaN mark would corrupt a valuation silently, so that must not rest on a single guard. The test fake mirrors the SQL semantics instead of masking it, and three tests cover the cases: a log row without usable prices between two valid ones, a string price, and a non-numeric asset key. --- .../__tests__/ledger-mark.service.spec.ts | 296 +++++++++++++++--- .../services/ledger-mark.service.ts | 132 ++++++-- .../supporting/log/log.repository.ts | 107 +++++++ src/subdomains/supporting/log/log.service.ts | 11 +- 4 files changed, 468 insertions(+), 78 deletions(-) diff --git a/src/subdomains/core/accounting/services/__tests__/ledger-mark.service.spec.ts b/src/subdomains/core/accounting/services/__tests__/ledger-mark.service.spec.ts index 0b29127d25..b5ed996d57 100644 --- a/src/subdomains/core/accounting/services/__tests__/ledger-mark.service.spec.ts +++ b/src/subdomains/core/accounting/services/__tests__/ledger-mark.service.spec.ts @@ -5,6 +5,7 @@ import { TestUtil } from 'src/shared/utils/test.util'; import { Util } from 'src/shared/utils/util'; import { createCustomLog } from 'src/subdomains/supporting/log/__mocks__/log.entity.mock'; import { Log } from 'src/subdomains/supporting/log/log.entity'; +import { FinancialLogAssetPrice } from 'src/subdomains/supporting/log/log.repository'; import { LogService } from 'src/subdomains/supporting/log/log.service'; import { LedgerMarkService } from '../ledger-mark.service'; @@ -60,6 +61,50 @@ function fakeGetFinancialLogs(allRows: Log[]): FakeGetFinancialLogs { }; } +/** Matches `LogService.getFinancialLogAssetPrices` signature. */ +type FakeGetFinancialLogAssetPrices = ( + from?: Date, + to?: Date, + limit?: number, + after?: number, +) => Promise; + +/** Expand a Log fixture into the SQL projection shape (same LEFT JOIN LATERAL semantics as the repository). */ +function logToAssetPrices(row: Log): FinancialLogAssetPrice[] { + // message::jsonb aborts the whole query on malformed JSON in production (fail-loud, see + // log.repository.ts) — this fake must throw too, not swallow it into an empty projection (Finding 4). + const assets = (JSON.parse(row.message) as { assets?: Record }).assets; + const entries = assets ? Object.entries(assets) : []; + + // LEFT JOIN LATERAL preserves the log row even when jsonb_each yields no rows (empty/absent assets): + // exactly one null-price placeholder row so logId-based overflow counting still sees this log row. + if (!entries.length) { + return [{ created: row.created, assetId: null, priceChf: null, logId: row.id }]; + } + + return entries.map(([assetIdKey, assetLog]) => { + const priceChf = (assetLog as { priceChf?: unknown })?.priceChf; + return { + created: row.created, + assetId: /^[0-9]+$/.test(assetIdKey) ? Number(assetIdKey) : null, + priceChf: typeof priceChf === 'number' && Number.isFinite(priceChf) ? priceChf : null, + logId: row.id, + }; + }); +} + +/** + * Fake `getFinancialLogAssetPrices`: filters/limits LOG rows first (same as SQL subquery LIMIT), then expands. + * Cursor is the underlying log id — not a flat result-row index. + */ +function fakeGetFinancialLogAssetPrices(allRows: Log[]): FakeGetFinancialLogAssetPrices { + const filterLogs = fakeGetFinancialLogs(allRows); + return async (from?: Date, to?: Date, limit?: number, after?: number): Promise => { + const logs = await filterLogs(from, false, to, limit, after); + return logs.flatMap(logToAssetPrices); + }; +} + describe('LedgerMarkService', () => { let service: LedgerMarkService; let logService: LogService; @@ -142,7 +187,15 @@ describe('LedgerMarkService', () => { await service.getMarkAtWidened(5, asOf, 90); - expect(spy).toHaveBeenCalledWith(Util.daysBefore(90, asOf), true, asOf, Config.ledger.markPreloadMaxRows + 1); + // trailing undefined: the first page carries no cursor yet — asserted explicitly so a + // dropped or reordered cursor argument still fails this test. + expect(spy).toHaveBeenCalledWith( + Util.daysBefore(90, asOf), + true, + asOf, + Config.ledger.markPreloadMaxRows + 1, + undefined, + ); }); it('never returns a mark created after asOf', async () => { @@ -207,12 +260,14 @@ describe('LedgerMarkService', () => { it('returns the priceChf of the latest mark ≤ bookingDate (stage 2)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([ - financialLog(new Date('2026-06-01'), { '5': { priceChf: 50000 } }), - financialLog(new Date('2026-06-02'), { '5': { priceChf: 51000 } }), - financialLog(new Date('2026-06-03'), { '5': { priceChf: 52000 } }), - ]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([ + financialLog(new Date('2026-06-01'), { '5': { priceChf: 50000 } }), + financialLog(new Date('2026-06-02'), { '5': { priceChf: 51000 } }), + financialLog(new Date('2026-06-03'), { '5': { priceChf: 52000 } }), + ]), + ); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-03')); @@ -222,8 +277,10 @@ describe('LedgerMarkService', () => { it('returns undefined when no log row ≤ bookingDate exists (stage 3 → needsMark)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-05'), { '5': { priceChf: 50000 } })]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([financialLog(new Date('2026-06-05'), { '5': { priceChf: 50000 } })]), + ); const cache = await service.preload(new Date('2026-06-05'), new Date('2026-06-05')); @@ -232,8 +289,10 @@ describe('LedgerMarkService', () => { it('returns undefined when a log row exists but its assets JSON lacks the assetId (Minor R5-5)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-01'), { '7': { priceChf: 1.0 } })]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([financialLog(new Date('2026-06-01'), { '7': { priceChf: 1.0 } })]), + ); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); @@ -242,24 +301,51 @@ describe('LedgerMarkService', () => { it('skips non-finite priceChf entries (no phantom 0 mark)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-01'), { '5': { priceChf: NaN } })]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([financialLog(new Date('2026-06-01'), { '5': { priceChf: NaN } })]), + ); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); expect(cache.getMarkAt(5, new Date('2026-06-01'))).toBeUndefined(); }); - it('never throws on malformed message JSON (defensive parse)', async () => { + it('throws on malformed message JSON (fail-loud, matches `message::jsonb` in production SQL)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([createCustomLog({ created: new Date('2026-06-01'), message: 'not-json' })]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([createCustomLog({ created: new Date('2026-06-01'), message: 'not-json' })]), + ); + + await expect(service.preload(new Date('2026-06-01'), new Date('2026-06-01'))).rejects.toThrow(); + }); + + it('excludes a priceChf that is a JSON string, not a JSON number (matches the old Number.isFinite gate)', async () => { + const row = financialLog(new Date('2026-06-01'), { + '5': { priceChf: '1.25' as unknown as number }, + }); + + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockImplementation(fakeGetFinancialLogAssetPrices([row])); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); expect(cache.getMarkAt(5, new Date('2026-06-01'))).toBeUndefined(); }); + it('excludes a non-numeric asset key without aborting the rest of the projection', async () => { + const row = financialLog(new Date('2026-06-01'), { + abc: { priceChf: 10 }, + '5': { priceChf: 20 }, + }); + + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockImplementation(fakeGetFinancialLogAssetPrices([row])); + + const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); + + expect(cache.getMarkAt(5, new Date('2026-06-01'))).toBe(20); // numeric key still projected despite the sibling non-numeric key + }); + it('uses dailySample when the span exceeds the threshold (bounded preload)', async () => { const spy = jest .spyOn(logService, 'getFinancialLogs') @@ -272,39 +358,88 @@ describe('LedgerMarkService', () => { true, new Date('2026-06-10'), Config.ledger.markPreloadMaxRows + 1, + undefined, // first page carries no cursor ); }); it('uses the full minute-tick for fresh windows within the threshold', async () => { - const spy = jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-01'), { '5': { priceChf: 50000 } })]); + const spy = jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue([ + { + created: new Date('2026-06-01'), + assetId: 5, + priceChf: 50000, + logId: 1, + }, + ]); await service.preload(new Date('2026-06-01'), new Date('2026-06-01T06:00:00Z')); // < 2 days expect(spy).toHaveBeenCalledWith( new Date('2026-06-01'), - false, new Date('2026-06-01T06:00:00Z'), Config.ledger.markPreloadMaxRows + 1, + undefined, // first page carries no cursor ); }); it('passes to and limit (maxRows + 1) on the preload trigger read', async () => { const to = new Date('2026-06-02'); - const spy = jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-01'), { '5': { priceChf: 50000 } })]); + const spy = jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue([ + { + created: new Date('2026-06-01'), + assetId: 5, + priceChf: 50000, + logId: 1, + }, + ]); await service.preload(new Date('2026-06-01'), to); // Upper bound and row cap are enforced in SQL (no post-load JS filter); +1 keeps overflow detectable. - expect(spy).toHaveBeenCalledWith(new Date('2026-06-01'), false, to, Config.ledger.markPreloadMaxRows + 1); + expect(spy).toHaveBeenCalledWith(new Date('2026-06-01'), to, Config.ledger.markPreloadMaxRows + 1, undefined); }); - // §5.2 step 3 pagination backstop: when the first bounded read returns more than markPreloadMaxRows rows the service - // continues via keyset pages over id (created resolved in-DB). With markPreloadMaxRows=1 the first read (2 rows) - // trips the backstop; the probe's first maxRows rows are reused as page 1. + // Projection → same mark map semantics as the former full-Log path (assets, prices, order). + it('buildMarkMap from the asset-price projection yields the same marks as full log rows would', async () => { + const t0 = new Date('2026-06-01T00:00:00Z'); + const t1 = new Date('2026-06-01T01:00:00Z'); + const projection: FinancialLogAssetPrice[] = [ + { created: t0, assetId: 5, priceChf: 50000, logId: 10 }, + { created: t0, assetId: 6, priceChf: 1.5, logId: 10 }, + { created: t1, assetId: 5, priceChf: 51000, logId: 11 }, + ]; + + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue(projection); + + const cache = await service.preload(t0, t1); + + expect(cache.getMarkAt(5, t0)).toBe(50000); + expect(cache.getMarkAt(6, t0)).toBe(1.5); + expect(cache.getMarkAt(5, t1)).toBe(51000); + expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBe(50000); + }); + + it('omits assets with missing or non-finite projected prices (no 0-mark)', async () => { + const t = new Date('2026-06-01T00:00:00Z'); + // Repo/fake null non-finite / unusable fields; inject null placeholders so buildMarkMap's skip is covered. + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue([ + { created: t, assetId: 5, priceChf: 42, logId: 1 }, + { created: t, assetId: 6, priceChf: null, logId: 1 }, + { created: t, assetId: null, priceChf: 7, logId: 1 }, + // asset 8 absent entirely → no mark + ]); + + const cache = await service.preload(t, t); + + expect(cache.getMarkAt(5, t)).toBe(42); + expect(cache.getMarkAt(6, t)).toBeUndefined(); + expect(cache.getMarkAt(7, t)).toBeUndefined(); + expect(cache.getMarkAt(8, t)).toBeUndefined(); + }); + + // §5.2 step 3 pagination backstop: when the first bounded read returns more than markPreloadMaxRows log rows the + // service continues via keyset pages over log id. With markPreloadMaxRows=1 the first read (2 log rows) trips the + // backstop; the probe's first maxRows complete log groups are reused as page 1. describe('pagination backstop (rows > markPreloadMaxRows)', () => { let pagedService: LedgerMarkService; @@ -327,11 +462,13 @@ describe('LedgerMarkService', () => { const w1b = financialLog(new Date('2026-06-01T01:00:00Z'), { '5': { priceChf: 51000 } }); const w2 = financialLog(new Date('2026-06-01T02:00:00Z'), { '5': { priceChf: 52000 } }); - const spy = jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([w1a, w1b, w2])); + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([w1a, w1b, w2])); const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); - // the continuation windows were used (>1 getFinancialLogs call beyond the trigger read) + // the continuation windows were used (>1 projection call beyond the trigger read) expect(spy.mock.calls.length).toBeGreaterThan(1); // all three marks made it into the cache built from the paginated rows expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBe(50000); @@ -344,7 +481,7 @@ describe('LedgerMarkService', () => { const r1 = financialLog(new Date('2026-06-01T00:00:00Z'), { '5': { priceChf: 50000 } }); const r2 = financialLog(new Date('2026-06-01T01:00:00Z'), { '5': { priceChf: 51000 } }); - jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([r1, r2])); + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockImplementation(fakeGetFinancialLogAssetPrices([r1, r2])); const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); @@ -372,7 +509,9 @@ describe('LedgerMarkService', () => { message: JSON.stringify({ assets: { '6': { priceChf: 51000 } } }), }); - jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([rowA, rowB])); + jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([rowA, rowB])); const cache = await pagedService.preload(t0, new Date('2026-06-01T03:00:00Z')); @@ -387,8 +526,8 @@ describe('LedgerMarkService', () => { const to = new Date('2026-06-01T03:00:00Z'); const spy = jest - .spyOn(logService, 'getFinancialLogs') - .mockImplementation(fakeGetFinancialLogs([inRange, alsoInRange, afterTo])); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([inRange, alsoInRange, afterTo])); const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), to); @@ -397,14 +536,16 @@ describe('LedgerMarkService', () => { // too-late row must not leak into the cache (lookup at/after its created still shows the last in-range mark) expect(cache.getMarkAt(5, afterTo.created)).toBe(51000); expect(cache.getMarkAt(5, afterTo.created)).not.toBe(99999); - // every call passes the same upper bound + // every call passes the same upper bound (arg index 1 = `to` on getFinancialLogAssetPrices) for (const call of spy.mock.calls) { - expect(call[2]).toEqual(to); + expect(call[1]).toEqual(to); } }); it('returns an empty cache when no financial logs fall in the window', async () => { - const spy = jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([])); + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([])); const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); @@ -412,6 +553,32 @@ describe('LedgerMarkService', () => { expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBeUndefined(); }); + // Multi-asset log rows: overflow/page boundaries must cut on logId groups, never mid-snapshot by array index. + it('keeps every asset of a multi-asset log when maxRows=1 (no mid-group cut on logId)', async () => { + const r1 = financialLog(new Date('2026-06-01T00:00:00Z'), { + '5': { priceChf: 1 }, + '6': { priceChf: 2 }, + }); + const r2 = financialLog(new Date('2026-06-01T01:00:00Z'), { '7': { priceChf: 3 } }); + + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([r1, r2])); + + const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); + + // both logs fully present — slicing the flat projection at maxRows=1 would have dropped asset 6 + expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBe(1); + expect(cache.getMarkAt(6, new Date('2026-06-01T00:30:00Z'))).toBe(2); + expect(cache.getMarkAt(7, new Date('2026-06-01T01:30:00Z'))).toBe(3); + expect(spy.mock.calls.length).toBeGreaterThan(1); + // continuation cursor is a log id (number), not a flat result index + for (const call of spy.mock.calls) { + const afterArg = call[3]; + if (afterArg !== undefined) expect(typeof afterArg).toBe('number'); + } + }); + // §5.2 precision fix: log.created is timestamp(6) in Postgres (microsecond precision) but JS `Date` // only carries milliseconds - reading a row truncates e.g. ...841802 -> ...841. A (created, id)-Date // cursor sent back to the DB compares this truncated value against the SAME row's full-precision @@ -433,7 +600,9 @@ describe('LedgerMarkService', () => { // pagedService here = the pagination-backstop describe-block's service instance with // markPreloadMaxRows = 1 (see that block's beforeEach) - reuse it, do not rebuild a separate module. - const spy = jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs(rows)); + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices(rows)); const cache = await pagedService.preload(new Date(sameMs), new Date(sameMs)); @@ -442,10 +611,9 @@ describe('LedgerMarkService', () => { expect(cache.getMarkAt(7, new Date(sameMs))).toBe(3); expect(spy.mock.calls.length).toBeLessThan(10); // terminates - no infinite loop - // structural guarantee: the cursor argument passed to getFinancialLogs is always a plain id - // (number), never a Date - so there is no JS-truncated timestamp for the DB to compare at all. + // structural guarantee: the cursor argument is always a plain log id (number), never a Date for (const call of spy.mock.calls) { - const afterArg = call[4]; + const afterArg = call[3]; if (afterArg !== undefined) expect(typeof afterArg).toBe('number'); } }); @@ -480,7 +648,9 @@ describe('LedgerMarkService', () => { const r3 = financialLog(sameCreated, { '7': { priceChf: 3 } }); const r4 = financialLog(new Date('2026-06-01T18:00:00Z'), { '8': { priceChf: 4 } }); - const spy = jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([r1, r2, r3, r4])); + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([r1, r2, r3, r4])); const cache = await pagedService.preload(batchStart, to); @@ -493,7 +663,7 @@ describe('LedgerMarkService', () => { expect(spy.mock.calls.length).toBeLessThan(10); }); - // Overflow reuse: probe reads maxRows+1, reuses first maxRows as page 1, continues from that last id. + // Overflow reuse: probe reads maxRows+1, reuses first maxRows complete log groups as page 1, continues from that last logId. // Without reuse: 1 probe + 3 pages = 4 calls. With reuse: 1 probe + 2 continuation pages = 3. it('reuses the overflow probe as page 1 (no double-read of the first maxRows rows)', async () => { const batchStart = new Date('2026-06-01T00:00:00Z'); @@ -505,8 +675,8 @@ describe('LedgerMarkService', () => { const r5 = financialLog(new Date('2026-06-01T05:00:00Z'), { '5': { priceChf: 50 } }); const spy = jest - .spyOn(logService, 'getFinancialLogs') - .mockImplementation(fakeGetFinancialLogs([r1, r2, r3, r4, r5])); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([r1, r2, r3, r4, r5])); const cache = await pagedService.preload(batchStart, to); @@ -521,8 +691,44 @@ describe('LedgerMarkService', () => { }); }); + // Finding 1+2: a log row with no usable prices must still occupy a logId slot so overflow/pagination continue + // past it — otherwise later valid marks are silently dropped when maxRows=1. + describe('pagination continues past a log row with no usable prices (Finding 1+2 regression)', () => { + let pagedService: LedgerMarkService; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + TestUtil.provideConfig({ + ledger: { markPreloadMaxRows: 1, markPreloadDailySampleThresholdDays: 2 }, + }), + LedgerMarkService, + { provide: LogService, useValue: logService }, + ], + }).compile(); + pagedService = module.get(LedgerMarkService); + }); + + it('keeps every valid mark when an unusable-price row sits between two valid rows', async () => { + const r1 = financialLog(new Date('2026-06-01T00:00:00Z'), { '5': { priceChf: 1 } }); + const unusable = financialLog(new Date('2026-06-01T01:00:00Z'), {}); // no usable price at all + const r3 = financialLog(new Date('2026-06-01T02:00:00Z'), { '5': { priceChf: 3 } }); + + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([r1, unusable, r3])); + + const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); + + expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBe(1); + expect(cache.getMarkAt(5, new Date('2026-06-01T02:30:00Z'))).toBe(3); // r3 must not be silently dropped + expect(spy.mock.calls.length).toBeGreaterThan(1); // pagination actually continued past the unusable row + }); + }); + // dailySample=true pagination path: span > threshold so both getFinancialLogs branches' cursor logic is exercised - // under overflow (maxRows small enough that multi-page keyset runs). + // under overflow (maxRows small enough that multi-page keyset runs). Stays on getFinancialLogs (no dailySample + // parameter on getFinancialLogAssetPrices) — rare long-window path only. describe('pagination with dailySample=true (span > markPreloadDailySampleThresholdDays)', () => { let pagedService: LedgerMarkService; diff --git a/src/subdomains/core/accounting/services/ledger-mark.service.ts b/src/subdomains/core/accounting/services/ledger-mark.service.ts index 4aea62f446..fda1173e47 100644 --- a/src/subdomains/core/accounting/services/ledger-mark.service.ts +++ b/src/subdomains/core/accounting/services/ledger-mark.service.ts @@ -3,6 +3,7 @@ import { Config } from 'src/config/config'; import { Util } from 'src/shared/utils/util'; import { FinanceLog } from 'src/subdomains/supporting/log/dto/log.dto'; import { Log } from 'src/subdomains/supporting/log/log.entity'; +import { FinancialLogAssetPrice } from 'src/subdomains/supporting/log/log.repository'; import { LogService } from 'src/subdomains/supporting/log/log.service'; interface MarkPoint { @@ -80,6 +81,8 @@ export class LedgerMarkService { } // bounded, memoized youngest-mark map (≤ now). Ascending by created → the last finite write per asset wins → youngest. + // Stays on full getFinancialLogs (not the price projection): does not use buildMarkMap / preload pagination and only + // needs a flat Map over a short daily-sampled window. private async getLatestMarks(): Promise> { const now = Date.now(); if (this.latestMarks && now - this.latestMarks.loadedAt < LATEST_MARK_TTL_MS) return this.latestMarks.map; @@ -138,48 +141,117 @@ export class LedgerMarkService { /** * Bounded preload (§5.2, Hard Constraint #4): always limited by (batchStartDate, to) and maxRows. * Order is fixed — dailySample decision FIRST (avoids loading the full minute-tick), THEN upper-bound - * trimming, THEN the maxRows pagination backstop (keyset over id; created resolved in-DB). + * trimming, THEN the maxRows pagination backstop (keyset over log id; created resolved in-DB). + * + * Hot path (dailySample=false): SQL projects priceChf only (getFinancialLogAssetPrices). + * Rare long-window path (dailySample=true): still getFinancialLogs + local expansion to the same projection type. */ async preload(batchStartDate: Date, to: Date): Promise { const spanDays = Util.daysDiff(batchStartDate, to); const dailySample = spanDays > Config.ledger.markPreloadDailySampleThresholdDays; const maxRows = this.getMarkPreloadMaxRows(); - // +1 so probeRows.length > maxRows can still detect overflow when SQL already caps at maxRows - const probeRows = await this.logService.getFinancialLogs(batchStartDate, dailySample, to, maxRows + 1); + // +1 so unique log-id count > maxRows can still detect overflow when SQL already caps at maxRows log rows + const probeRows = await this.loadAssetPrices(batchStartDate, to, dailySample, maxRows + 1); const rows = - probeRows.length > maxRows - ? await this.paginate(batchStartDate, to, dailySample, probeRows.slice(0, maxRows)) + this.uniqueLogIds(probeRows).length > maxRows + ? await this.paginate(batchStartDate, to, dailySample, this.takeCompleteLogGroups(probeRows, maxRows)) : probeRows; return new LedgerMarkCache(this.buildMarkMap(rows)); } - // Keyset pages over id; never load everything into one heap (§5.2 step 3). - // `firstPage` reuses the rows preload() already read via the overflow probe (the same maxRows-sized - // first page a from-scratch pagination would produce, since both share the same filters/order/limit= - // maxRows and deterministic ORDER BY created ASC, id ASC) — so the probe read is not thrown away and - // re-fetched. Halves the data read on overflow, result set stays identical to full re-pagination. - private async paginate(batchStartDate: Date, to: Date, dailySample: boolean, firstPage: Log[]): Promise { + // Keyset pages over log id; never load everything into one heap (§5.2 step 3). + // Overflow/page sizes are measured in distinct logId groups (one FinancialDataLog snapshot), not flattened + // asset-result length — slicing by array index would cut mid-snapshot and drop assets silently. + // `firstPage` reuses the complete log groups preload() already read via the overflow probe. + private async paginate( + batchStartDate: Date, + to: Date, + dailySample: boolean, + firstPage: FinancialLogAssetPrice[], + ): Promise { const maxRows = this.getMarkPreloadMaxRows(); - const result: Log[] = [...firstPage]; - let after: number | undefined = firstPage[firstPage.length - 1]?.id; + const result: FinancialLogAssetPrice[] = [...firstPage]; + const firstPageLogIds = this.uniqueLogIds(firstPage); + let after: number | undefined = firstPageLogIds[firstPageLogIds.length - 1]; - // Keyset continuation: each page starts strictly after the last returned id. + // Keyset continuation: each page starts strictly after the last returned log id. while (true) { - const window = await this.logService.getFinancialLogs(batchStartDate, dailySample, to, maxRows, after); + const window = await this.loadAssetPrices(batchStartDate, to, dailySample, maxRows, after); if (!window.length) break; result.push(...window); - if (window.length < maxRows) break; + const windowLogIds = this.uniqueLogIds(window); + if (windowLogIds.length < maxRows) break; - after = window[window.length - 1].id; + after = windowLogIds[windowLogIds.length - 1]; } return result; } + // Hot path: SQL projection. dailySample path: full logs (day-group SQL not reimplemented) expanded locally. + private async loadAssetPrices( + from: Date, + to: Date, + dailySample: boolean, + limit?: number, + after?: number, + ): Promise { + if (dailySample) { + const logs = await this.logService.getFinancialLogs(from, true, to, limit, after); + return logs.flatMap((row) => this.rowToAssetPrices(row)); + } + return this.logService.getFinancialLogAssetPrices(from, to, limit, after); + } + + // Expand one Log into the same projection shape as getFinancialLogAssetPrices (dailySample adapter only). + // Mirrors the repository's LEFT JOIN LATERAL: every log row must yield at least one result row so that + // uniqueLogIds/overflow-detection below count log rows actually read, not just the ones carrying a usable + // mark (see PR review Finding 1+2) — an empty/absent `assets` object still emits one all-null placeholder + // row, and each present asset key emits its own row with assetId/priceChf nulled per-field when unusable + // (non-numeric key / non-finite price), rather than being skipped. + private rowToAssetPrices(row: Log): FinancialLogAssetPrice[] { + const assets = this.parseAssets(row.message); + const entries = assets ? Object.entries(assets) : []; + + if (!entries.length) { + return [{ created: row.created, assetId: null, priceChf: null, logId: row.id }]; + } + + return entries.map(([assetIdKey, assetLog]) => { + const priceChf = assetLog?.priceChf; + return { + created: row.created, + assetId: /^[0-9]+$/.test(assetIdKey) ? Number(assetIdKey) : null, + priceChf: typeof priceChf === 'number' && Number.isFinite(priceChf) ? priceChf : null, + logId: row.id, + }; + }); + } + + // Distinct logIds in first-seen order (SQL keeps all assets of one log contiguous). + // logId is present on every projection row, including null-price placeholders, so this counts log rows + // actually read — not only those that carried a usable mark (Finding 1+2). + private uniqueLogIds(rows: FinancialLogAssetPrice[]): number[] { + const ids: number[] = []; + const seen = new Set(); + for (const row of rows) { + if (seen.has(row.logId)) continue; + seen.add(row.logId); + ids.push(row.logId); + } + return ids; + } + + // Keep every projected asset belonging to the first `maxLogRows` distinct logIds (no mid-group cut). + private takeCompleteLogGroups(rows: FinancialLogAssetPrice[], maxLogRows: number): FinancialLogAssetPrice[] { + const allowed = new Set(this.uniqueLogIds(rows).slice(0, maxLogRows)); + return rows.filter((row) => allowed.has(row.logId)); + } + // Fail loud on a non-positive / non-integer markPreloadMaxRows (e.g. LEDGER_MARK_PRELOAD_MAX_ROWS=0 or // a broken env parse): LIMIT 0 / empty first page would otherwise silently build an empty cache. private getMarkPreloadMaxRows(): number { @@ -190,26 +262,22 @@ export class LedgerMarkService { return value; } - private buildMarkMap(rows: Log[]): Map { + private buildMarkMap(rows: FinancialLogAssetPrice[]): Map { const marks = new Map(); for (const row of rows) { - // tolerate parse/shape issues defensively — never throw, mirrors log-job getJsonValue - const assets = this.parseAssets(row.message); - if (!assets) continue; - - for (const [assetIdKey, assetLog] of Object.entries(assets)) { - const priceChf = assetLog?.priceChf; - if (!Number.isFinite(priceChf)) continue; - - const assetId = +assetIdKey; - const points = marks.get(assetId) ?? []; - points.push({ created: row.created, priceChf }); - marks.set(assetId, points); - } + // Repo / rowToAssetPrices keep a row per read log even without a usable mark (assetId/priceChf + // null) so overflow detection and keyset pagination count log rows correctly — skip those here. + // Number.isFinite is kept as a second line of defence: the repository already nulls NaN/Infinity, + // but a mark of NaN would silently corrupt a valuation, so it must not depend on one guard alone. + if (row.assetId == null || !Number.isFinite(row.priceChf)) continue; + + const points = marks.get(row.assetId) ?? []; + points.push({ created: row.created, priceChf: row.priceChf }); + marks.set(row.assetId, points); } - // rows arrive ascending by created (getFinancialLogs order); keep lists sorted for binary search + // rows arrive ascending by created (repository order); keep lists sorted for binary search for (const points of marks.values()) { points.sort((a, b) => a.created.getTime() - b.created.getTime()); } diff --git a/src/subdomains/supporting/log/log.repository.ts b/src/subdomains/supporting/log/log.repository.ts index fc55654bb8..6be2809ef7 100644 --- a/src/subdomains/supporting/log/log.repository.ts +++ b/src/subdomains/supporting/log/log.repository.ts @@ -19,6 +19,27 @@ import { MAX_VALIDITY_SWEEP_ROWS, } from './log.entity'; +/** One asset priceChf projected from a FinancialDataLog snapshot (no full message JSON). */ +export interface FinancialLogAssetPrice { + created: Date; + /** + * null when the JSON key in `assets` is not a plain non-negative-integer string (`^[0-9]+$`) — + * kept as a row with assetId=null rather than aborting the whole query via a failing `::int` cast. + */ + assetId: number | null; + /** + * null when this row carries no usable price: `assets` was empty/absent for the log row (one + * placeholder row per log row), the JSON value at `priceChf` was not a JSON number (e.g. the string + * "1.25"), or the numeric value was NaN/Infinity. + */ + priceChf: number | null; + /** id of the underlying log row — logId is ALWAYS present, on every row, even the null-price ones. + * Overflow detection / keyset cursor logic in LedgerMarkService counts distinct logId values across + * ALL returned rows (not just the ones with a usable price) to know how many log rows were actually + * read — see LedgerMarkService.uniqueLogIds. */ + logId: number; +} + @Injectable() export class LogRepository extends BaseRepository { constructor(manager: EntityManager) { @@ -217,6 +238,92 @@ export class LogRepository extends BaseRepository { return rows; } + /** + * SQL-side projection of priceChf per asset from FinancialDataLog snapshots. + * LIMIT/keyset apply to log rows (inner subquery), not to the expanded asset result — same page semantics as + * getFinancialLogs. Callers that only need marks avoid shipping/parsing the full message JSON. + * + * LEFT JOIN LATERAL (not an implicit CROSS JOIN) so every log row yields at least one result row — including + * when `assets` is empty/absent or a key/price is unusable (assetId/priceChf null). That keeps logId-based + * overflow detection and keyset pagination in LedgerMarkService correct (Finding 1+2: the old join+WHERE + * dropped unusable-price rows entirely, so uniqueLogIds under-counted and pagination/overflow stopped early). + * Invalid keys and non-number priceChf are nulled via CASE expressions rather than filtered in WHERE, so the + * row (and its logId) always remains. Malformed `message` JSON fails loud: `message::jsonb` aborts the whole + * query (intentional change vs the old JS path that try/caught per row). + */ + async getFinancialLogAssetPrices( + from?: Date, + to?: Date, + limit?: number, + after?: number, // id of the last LOG row of the previous page; same cursor semantics as getFinancialLogs + ): Promise { + const params: unknown[] = []; + let i = 1; + const conditions = [`system = $${i++}`, `subsystem = $${i++}`, `severity = $${i++}`, `valid = $${i++}`]; + params.push('LogService', FINANCIAL_DATA_LOG_SUBSYSTEM, LogSeverity.INFO, true); + + if (from) { + conditions.push(`created >= $${i++}`); + params.push(from); + } + if (to) { + conditions.push(`created <= $${i++}`); + params.push(to); + } + if (after != null) { + // Same row-value keyset as getFinancialLogs: created resolved in-DB at full precision. + conditions.push(`(created, id) > ((SELECT c.created FROM log c WHERE c.id = $${i}), $${i + 1})`); + params.push(after, after); + i += 2; + } + + let limitClause = ''; + if (limit != null) { + limitClause = `LIMIT $${i++}`; + params.push(limit); + } + + const sql = ` +SELECT l.created AS "created", + CASE WHEN kv.key ~ '^[0-9]+$' THEN kv.key::int ELSE NULL END AS "assetId", + CASE + WHEN jsonb_typeof(kv.value -> 'priceChf') = 'number' THEN (kv.value ->> 'priceChf')::float8 + ELSE NULL + END AS "priceChf", + l.id AS "logId" +FROM ( + SELECT id, created, message + FROM log + WHERE ${conditions.join(' AND ')} + ORDER BY created ASC, id ASC + ${limitClause} +) l +LEFT JOIN LATERAL jsonb_each(l.message::jsonb -> 'assets') kv ON true +ORDER BY l.created ASC, l.id ASC`; + + const raw = (await this.query(sql, params)) as { + created: Date | string; + assetId: number | string | null; + priceChf: number | string | null; + logId: number | string; + }[]; + + const rows: FinancialLogAssetPrice[] = raw.map((r) => { + const priceChf = r.priceChf == null ? null : Number(r.priceChf); + return { + created: r.created instanceof Date ? r.created : new Date(r.created), + assetId: r.assetId == null ? null : Number(r.assetId), + // float8 NaN/Infinity (e.g. an out-of-range numeric text) must be excluded the same way the old + // Number.isFinite gate excluded them — never surface as a phantom mark. + priceChf: priceChf != null && Number.isFinite(priceChf) ? priceChf : null, + logId: Number(r.logId), + }; + }); + + if (!rows.length && after != null) await this.assertEmptyResultIsEndOfData(after); + return rows; + } + // After an empty main-query result with a keyset cursor, fail loud when the cursor id is gone: the row-value // subquery would return NULL and `(created, id) > (NULL, :afterId)` is NULL in Postgres → WHERE excludes every // row → silent empty result that callers misread as end-of-data. Only invoked when the main query already diff --git a/src/subdomains/supporting/log/log.service.ts b/src/subdomains/supporting/log/log.service.ts index edc180269a..d122a4708e 100644 --- a/src/subdomains/supporting/log/log.service.ts +++ b/src/subdomains/supporting/log/log.service.ts @@ -13,7 +13,7 @@ import { LogSeverity, MAX_VALIDITY_SWEEP_ROWS, } from './log.entity'; -import { LogRepository } from './log.repository'; +import { FinancialLogAssetPrice, LogRepository } from './log.repository'; @Injectable() export class LogService { @@ -145,6 +145,15 @@ export class LogService { return this.logRepo.getFinancialLogs(from, dailySample, to, limit, after); } + async getFinancialLogAssetPrices( + from?: Date, + to?: Date, + limit?: number, + after?: number, + ): Promise { + return this.logRepo.getFinancialLogAssetPrices(from, to, limit, after); + } + async getLatestFinancialLog(): Promise { return this.logRepo.getLatestFinancialLog(); } From 52468f3cfa13ea1e6ab0866a8568c8ab89c9bfbe Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:25:37 +0200 Subject: [PATCH 3/3] 465d5198 - test(scrypt): cover the cooldown interval formula and the since hand-off (#4455) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(scrypt): cover the cooldown interval formula and the since hand-off Both properties were provably untested. Replacing `ageMs / 10` and the 30-minute cap with the flat one-minute floor left the suite green, and so did making getOrderStatus ignore its `since` argument and fetch the fixed 30-day window again — every existing cooldown test uses a freshly created order, where the interval is always the floor, and the adapter tests mock getOrderStatus away entirely. - pipeline service: a 100-minute-old order still owes its 10-minute wait after 9 minutes, and an 8-hour-old one is asked again after 31 because the cap holds it at 30 rather than the uncapped 48 - scrypt service: a caller-supplied `since` reaches the venue request as StartDate, and omitting it still yields the 30-day window * test(scrypt): pin the cooldown rate and cap from both sides Review follow-up. Both cooldown tests asserted only the lower side of the interval, so a mutation that ENLARGES it stayed green: with `ageMs / 5` the order simply remains in cooldown and a "still waiting" assertion keeps passing, and the cap test is dominated by the cap whatever the divisor. Each test now brackets its exact threshold. For a 100-minute-old order the wait is satisfied once elapsed >= (100 min + elapsed) / 10, i.e. at 11 min 6.7 s — hence one call at 11 minutes and two at 11 min 20 s. The cap is bracketed at 29 min 59 s and 30 min 1 s, which also pins the boundary against a cap set too low. * test(scrypt): land the cap check on the boundary itself The cap test bracketed 29 min 59 s and 30 min 1 s but never the boundary between them, which is the only place `<` and `<=` differ: at exactly 1_800_000 ms elapsed the production check admits the lookup, a `<=` variant would hold it back, and neither assertion point could tell them apart. Moving the second check onto 30 minutes exactly pins that too. * test(scrypt): pin the cap to the millisecond Checking at 29 min 59 s left the cap a whole second of room: any value between 1_799_001 and 1_799_999 ms produced the same two observations as the real 1_800_000, so a cap set slightly too low passed unnoticed. Checking one millisecond before the boundary closes that gap entirely. Also corrects the comment's arithmetic: 48 minutes is the uncapped wait at the first pass; by the boundary check the order has aged another 30 minutes and its uncapped wait is 51. Both are far past the cap, so the conclusion held, but the number did not. * test(scrypt): say what the cap check proves, more carefully The comment claimed the bracket leaves the cap "no room to sit anywhere else". That holds for every whole millisecond, which is every value the constant can actually take, but the test alone does not rule out a fractional one — so say whole-millisecond and mean it. Also fixes the grammar of the preceding clause. --- .../services/__tests__/scrypt.service.spec.ts | 30 ++++++++++++++ ...uidity-management-pipeline.service.spec.ts | 40 +++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/src/integration/exchange/services/__tests__/scrypt.service.spec.ts b/src/integration/exchange/services/__tests__/scrypt.service.spec.ts index d811cad5ca..177431005b 100644 --- a/src/integration/exchange/services/__tests__/scrypt.service.spec.ts +++ b/src/integration/exchange/services/__tests__/scrypt.service.spec.ts @@ -529,6 +529,36 @@ describe('ScryptService', () => { expect((service as any).executionReports.get('X')).toEqual(liveTerminal); }); + it('passes a caller-supplied since through to the venue request as StartDate', async () => { + // the caller side of this is covered elsewhere (scrypt.adapter.spec.ts); this covers the receiving side — + // a mutation that drops `since` and always fetches the fixed 30-day window would pass unnoticed otherwise + (instance as any).fetch.mockClear(); + const since = new Date('2026-07-01T00:00:00.000Z'); + + await (service as any).getOrderStatus('ord-since-1', since); + + expect((instance as any).fetch).toHaveBeenCalledWith( + ScryptMessageType.EXECUTION_REPORT, + expect.objectContaining({ StartDate: since.toISOString() }), + ); + }); + + it('falls back to the fixed thirty-day window when no since is given', async () => { + // guards the `since ?? Util.daysBefore(30)` default; the bound is derived from "now", so pin it from both + // sides instead of asserting an exact timestamp + (instance as any).fetch.mockClear(); + const earliest = Util.daysBefore(30); + + await (service as any).getOrderStatus('ord-since-2'); + + const latest = Util.daysBefore(30); + const [, filters] = (instance as any).fetch.mock.calls[0]; + const startDate = new Date((filters as Record).StartDate as string); + + expect(startDate.getTime()).toBeGreaterThanOrEqual(earliest.getTime()); + expect(startDate.getTime()).toBeLessThanOrEqual(latest.getTime()); + }); + it('constructor warm-up BalanceTransaction fetch goes through the terminal-aware guard', async () => { const now = new Date().toISOString(); const terminalRecord = { diff --git a/src/subdomains/core/liquidity-management/services/liquidity-management-pipeline.service.spec.ts b/src/subdomains/core/liquidity-management/services/liquidity-management-pipeline.service.spec.ts index d56e44de6d..3e835685a0 100644 --- a/src/subdomains/core/liquidity-management/services/liquidity-management-pipeline.service.spec.ts +++ b/src/subdomains/core/liquidity-management/services/liquidity-management-pipeline.service.spec.ts @@ -608,6 +608,46 @@ describe('LiquidityManagementPipelineService', () => { await service['resolveUncertainOrders'](); expect(service['uncertainResolveAttempts'].size).toBe(0); }); + + it('scales the cooldown interval with the order age, at the rate the formula states', async () => { + // Pins `ageMs / 10` from both sides. The wait for a 100-minute-old order is only satisfied once + // elapsed >= (100 min + elapsed) / 10, i.e. at 11 min 6.7 s — so 11 minutes is still inside it and + // 11 min 20 s is past it. A one-sided assertion would let the rate drift unnoticed: with `ageMs / 5` + // the order simply stays in cooldown and a lower-bound-only test keeps passing. + const resolveUncertainOrder = stubResolver(); + const order = uncertainOrder({ created: new Date(Date.now() - 100 * 60_000) }); + jest.spyOn(orderRepo, 'findBy').mockResolvedValue([order]); + + await service['resolveUncertainOrders'](); + + jest.advanceTimersByTime(11 * 60_000); + await service['resolveUncertainOrders'](); + expect(resolveUncertainOrder).toHaveBeenCalledTimes(1); + + jest.advanceTimersByTime(20_000); + await service['resolveUncertainOrders'](); + expect(resolveUncertainOrder).toHaveBeenCalledTimes(2); + }); + + it('caps the cooldown interval at thirty minutes no matter how old the order is', async () => { + // Pins the cap to the millisecond. An 8-hour-old order's uncapped wait would be 48 minutes at the + // first pass and 51 by the time of the boundary check — either way far past the cap, so a lookup at + // exactly 30 minutes can only come from it. Requiring no lookup a millisecond earlier leaves the cap + // no other whole-millisecond value to take, and landing on the boundary pins `<` against `<=`. + const resolveUncertainOrder = stubResolver(); + const order = uncertainOrder({ created: new Date(Date.now() - 8 * 60 * 60_000) }); + jest.spyOn(orderRepo, 'findBy').mockResolvedValue([order]); + + await service['resolveUncertainOrders'](); + + jest.advanceTimersByTime(30 * 60_000 - 1); + await service['resolveUncertainOrders'](); + expect(resolveUncertainOrder).toHaveBeenCalledTimes(1); + + jest.advanceTimersByTime(1); + await service['resolveUncertainOrders'](); + expect(resolveUncertainOrder).toHaveBeenCalledTimes(2); + }); }); });