Strengthen financial reconciliation checks
This commit is contained in:
Executable
+29
@@ -0,0 +1,29 @@
|
||||
import { randomUUID, createHash } from 'node:crypto';
|
||||
import type { Pool } from 'pg';
|
||||
import { z } from 'zod';
|
||||
import { CUSTODY_WRITE_LOCK } from './postgres-schema.ts';
|
||||
import { auditPostgres } from './postgres-custody.ts';
|
||||
|
||||
/** Operator-only acknowledgement AFTER externally reviewed repair/restore; never repairs or resets a journal. */
|
||||
export async function resolveCustodyIncident(pool: Pool, evidenceId: string): Promise<{ resolutionId: string }> {
|
||||
z.string().uuid().parse(evidenceId);
|
||||
const c = await pool.connect();
|
||||
try {
|
||||
await c.query('BEGIN');
|
||||
await c.query("SET LOCAL statement_timeout='10s'; SET LOCAL lock_timeout='3s'; SET LOCAL idle_in_transaction_session_timeout='10s'; SET LOCAL synchronous_commit=on");
|
||||
await c.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
||||
const allowed = await c.query("SELECT pg_has_role(current_user,nspowner,'MEMBER') AS owns FROM pg_namespace WHERE nspname='wk_custody'");
|
||||
if (allowed.rows[0]?.owns !== true) throw new Error('Schema-owner approval required for integrity resolution');
|
||||
const incident = await c.query("SELECT value FROM wk_custody.custody_meta WHERE key='integrity_incident_v1' FOR UPDATE");
|
||||
if (!incident.rowCount) throw new Error('No unresolved custody integrity incident');
|
||||
// Delete is UNCOMMITTED and exclusively locked. Any audit failure rolls it back, leaving the hold intact.
|
||||
await c.query("DELETE FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'");
|
||||
await auditPostgres(c);
|
||||
const resolutionId = randomUUID();
|
||||
await c.query('INSERT INTO wk_custody.custody_meta VALUES($1,$2)', [`integrity_resolution:${resolutionId}`,
|
||||
JSON.stringify({ version: 1, evidenceId, resolvedAt: Date.now(),
|
||||
incidentDigest: createHash('sha256').update(incident.rows[0].value).digest('hex') })]);
|
||||
await c.query('COMMIT'); return { resolutionId };
|
||||
} catch (error) { await c.query('ROLLBACK').catch(() => {}); throw error; }
|
||||
finally { c.release(); }
|
||||
}
|
||||
Executable
+199
@@ -0,0 +1,199 @@
|
||||
import { z } from 'zod';
|
||||
import type { PoolClient, QueryResult } from 'pg';
|
||||
import { ADDITIONAL_MARKET_FEE } from '../../shared/market.ts';
|
||||
import { parseAtomicXmr } from '../../shared/xmr-amount.ts';
|
||||
import { allocateXmrSale } from '../../shared/xmr-pricing.ts';
|
||||
import { validateXmrAddress } from './xmr.ts';
|
||||
|
||||
/** Codes only: never attach customer IDs, payloads, SQL errors or addresses to operational diagnostics. */
|
||||
export class CustodyIntegrityError extends Error {
|
||||
readonly code: string;
|
||||
constructor(code: string) { super(`Custody integrity check failed: ${code}`); this.code = code; }
|
||||
}
|
||||
function fail(code: string): never { throw new CustodyIntegrityError(code); }
|
||||
const uuid = z.string().uuid(), hex = z.string().regex(/^[a-f0-9]{64}$/);
|
||||
const integer = z.number().int().nonnegative().safe();
|
||||
const atomic = z.string().max(22).refine(s => { try { parseAtomicXmr(s); return true; } catch { return false; } });
|
||||
const positive = atomic.refine(s => { try { return parseAtomicXmr(s) > 0n; } catch { return false; } });
|
||||
const wallet = z.string().regex(/^[1-9A-HJ-NP-Za-km-z]{95}$/);
|
||||
const listingSchema = z.object({ id: uuid, seller: hex, cardHash: hex, sequence: integer,
|
||||
title: z.string().min(1).max(200), priceAtomicXmr: positive, feeBps: z.number().int().min(0).max(5000),
|
||||
hostFeeAddress: wallet.optional(), status: z.enum(['open', 'reserved', 'cancelled', 'sold']), createdAt: integer }).strict();
|
||||
const orderSchema = z.object({ id: uuid, listingId: uuid, seller: hex, buyer: hex, cardHash: hex, sequence: integer,
|
||||
grossAtomicXmr: positive, sellerAtomicXmr: atomic, authorAtomicXmr: atomic, hostAtomicXmr: atomic,
|
||||
authorAddress: wallet, hostFeeAddress: wallet.optional(), status: z.enum(['awaiting-lock', 'cancelled', 'settled']), createdAt: integer }).strict();
|
||||
const withdrawalSchema = z.object({ id: uuid, actor: hex, destination: z.literal('mock-external'), amountAtomicXmr: positive,
|
||||
maxFeeAtomicXmr: atomic, status: z.enum(['reserved', 'prepared', 'complete', 'cancelled']),
|
||||
attemptId: uuid.optional(), transactionId: z.string().regex(/^fixture:[a-f0-9]{64}$/).optional(),
|
||||
feeAtomicXmr: atomic.optional(), createdAt: integer }).strict();
|
||||
type Entry = { account: string; debit: string; credit: string };
|
||||
type Journal = { id: string; kind: string; lines: Entry[] };
|
||||
function parse<T>(schema: z.ZodType<T>, data: unknown, code: string): T {
|
||||
const result = schema.safeParse(data); return result.success ? result.data : fail(code);
|
||||
}
|
||||
function same(a: unknown, b: unknown): boolean {
|
||||
// Compare parsed records without depending on JSONB object key ordering.
|
||||
if (a === b) return true;
|
||||
if (!a || !b || typeof a !== 'object' || typeof b !== 'object') return false;
|
||||
const x = a as Record<string, unknown>, y = b as Record<string, unknown>;
|
||||
const keys = Object.keys(x).filter(k => x[k] !== undefined).sort();
|
||||
return keys.join('\0') === Object.keys(y).filter(k => y[k] !== undefined).sort().join('\0') && keys.every(k => same(x[k], y[k]));
|
||||
}
|
||||
function json(text: string | null): unknown { try { return JSON.parse(text ?? ''); } catch { return fail('outbox-encoding'); } }
|
||||
const d = (account: string, amount: string): Entry => ({ account, debit: amount, credit: '0' });
|
||||
const c = (account: string, amount: string): Entry => ({ account, debit: '0', credit: amount });
|
||||
async function journals(client: PoolClient, ids: string[]): Promise<Map<string, Journal>> {
|
||||
const result = await client.query(`SELECT j.id,j.kind,jsonb_agg(jsonb_build_object('account',e.account,
|
||||
'debit',e.debit::text,'credit',e.credit::text) ORDER BY e.position) FILTER(WHERE e.journal IS NOT NULL) AS lines
|
||||
FROM wk_custody.journals j LEFT JOIN wk_custody.entries e ON e.journal=j.id
|
||||
WHERE j.id=ANY($1::text[]) GROUP BY j.id`, [ids]);
|
||||
return new Map(result.rows.map(r => [r.id, r as Journal]));
|
||||
}
|
||||
function expectJournal(all: Map<string, Journal>, id: string, kind: string, expected: Entry[] | null) {
|
||||
const got = all.get(id);
|
||||
if (expected === null) { if (got) fail('unexpected-terminal-posting'); return; }
|
||||
if (!got || got.kind !== kind || !Array.isArray(got.lines)) fail('missing-lifecycle-posting');
|
||||
const normalized = (lines: Entry[]) => lines.filter(l => BigInt(l.debit) !== 0n || BigInt(l.credit) !== 0n)
|
||||
.map(l => JSON.stringify([l.account, l.debit, l.credit])).sort();
|
||||
if (!same(normalized(got.lines), normalized(expected))) fail('lifecycle-posting-mismatch');
|
||||
}
|
||||
async function validateWallets(values: (string | undefined)[]) {
|
||||
// Bounded per-batch cache, not a growing process-wide map of customer data.
|
||||
for (const address of new Set(values.filter((v): v is string => !!v))) {
|
||||
try { await validateXmrAddress(address); } catch { fail('fee-recipient-invalid'); }
|
||||
}
|
||||
}
|
||||
|
||||
type OrderRow = { id: string; buyer: string; seller: string; status: string; data: unknown;
|
||||
listing_data: unknown; listing_status: string; listing_card: string; balance: string | null;
|
||||
normal: string | null; box_kind: string | null; box_status: string | null; box_data: string | null };
|
||||
async function checkOrders(client: PoolClient, rows: OrderRow[]) {
|
||||
const posting = await journals(client, rows.flatMap(r => [`reserve:${r.id}`, `cancel:${r.id}`, `settle:${r.id}`]));
|
||||
const wallets: string[] = [];
|
||||
for (const row of rows) {
|
||||
const o = parse(orderSchema, row.data, 'order-record-invalid'), l = parse(listingSchema, row.listing_data, 'order-listing-invalid');
|
||||
if (o.id !== row.id || o.buyer !== row.buyer || o.seller !== row.seller || o.status !== row.status ||
|
||||
l.status !== row.listing_status || l.cardHash !== row.listing_card || o.listingId !== l.id || o.seller !== l.seller ||
|
||||
o.cardHash !== l.cardHash || o.sequence !== l.sequence || o.grossAtomicXmr !== l.priceAtomicXmr || o.seller === o.buyer ||
|
||||
(o.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address) !== (l.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address)) fail('order-terms-mismatch');
|
||||
const split = allocateXmrSale(BigInt(l.priceAtomicXmr), l.feeBps);
|
||||
if (o.sellerAtomicXmr !== split.sellerBeforeNetworkCosts.toString() || o.hostAtomicXmr !== split.hosting.toString() ||
|
||||
o.authorAtomicXmr !== split.author.toString() || o.authorAddress !== ADDITIONAL_MARKET_FEE.address) fail('order-allocation-mismatch');
|
||||
wallets.push(o.authorAddress, o.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address);
|
||||
const account = `customer:${o.buyer}:purchase:${o.id}`, available = `customer:${o.buyer}:available`;
|
||||
if (row.normal !== 'credit' || row.balance !== (o.status === 'awaiting-lock' ? o.grossAtomicXmr : '0')) fail('order-reservation-mismatch');
|
||||
const expectedStatus = o.status === 'awaiting-lock' ? 'reserved' : o.status === 'settled' ? 'sold' : 'cancelled';
|
||||
if (l.status !== expectedStatus || row.box_kind !== 'nft-lock') fail('order-state-mismatch');
|
||||
if (o.status === 'settled') {
|
||||
if (row.box_status !== 'fixture-complete' || !/^fixture:[a-f0-9]{64}$/.test(row.box_data ?? '')) fail('settlement-evidence-mismatch');
|
||||
} else {
|
||||
const original = parse(orderSchema, json(row.box_data), 'order-outbox-invalid');
|
||||
if (row.box_status !== (o.status === 'cancelled' ? 'cancelled' : 'blocked') ||
|
||||
!same(original, { ...o, status: 'awaiting-lock' })) fail('order-outbox-mismatch');
|
||||
}
|
||||
expectJournal(posting, `reserve:${o.id}`, 'purchase-reservation', [d(available, o.grossAtomicXmr), c(account, o.grossAtomicXmr)]);
|
||||
expectJournal(posting, `cancel:${o.id}`, 'purchase-cancellation', o.status === 'cancelled' ? [d(account, o.grossAtomicXmr), c(available, o.grossAtomicXmr)] : null);
|
||||
expectJournal(posting, `settle:${o.id}`, 'mock-sale-settlement', o.status === 'settled' ? [d(account, o.grossAtomicXmr),
|
||||
c(`customer:${o.seller}:available`, o.sellerAtomicXmr), c(`author:${o.authorAddress}`, o.authorAtomicXmr),
|
||||
c(o.hostFeeAddress ? `revenue:host:${o.hostFeeAddress}` : 'revenue:host', o.hostAtomicXmr)] : null);
|
||||
}
|
||||
await validateWallets(wallets);
|
||||
}
|
||||
const orderSelect = `SELECT o.id::text,o.buyer,o.seller,o.status,o.data,l.data AS listing_data,l.status AS listing_status,
|
||||
l.card AS listing_card,a.balance::text,a.normal,b.kind AS box_kind,b.status AS box_status,b.data AS box_data
|
||||
FROM wk_custody.orders o LEFT JOIN wk_custody.listings l ON l.id::text=o.data->>'listingId'
|
||||
LEFT JOIN wk_custody.accounts a ON a.id='customer:'||o.buyer||':purchase:'||o.id::text
|
||||
LEFT JOIN wk_custody.outbox b ON b.id=o.id`;
|
||||
|
||||
type WithdrawalRow = { id: string; actor: string; status: string; data: unknown; balance: string | null;
|
||||
normal: string | null; box_kind: string | null; box_status: string | null; box_data: string | null };
|
||||
async function checkWithdrawals(client: PoolClient, rows: WithdrawalRow[]) {
|
||||
const posting = await journals(client, rows.flatMap(r => [`withdraw:${r.id}`, `withdraw-cancel:${r.id}`, `withdraw-confirm:${r.id}`]));
|
||||
for (const row of rows) {
|
||||
const w = parse(withdrawalSchema, row.data, 'withdrawal-record-invalid');
|
||||
if (w.id !== row.id || w.actor !== row.actor || w.status !== row.status) fail('withdrawal-identity-mismatch');
|
||||
const total = (BigInt(w.amountAtomicXmr) + BigInt(w.maxFeeAtomicXmr)).toString();
|
||||
const active = w.status === 'reserved' || w.status === 'prepared';
|
||||
if (row.normal !== 'credit' || row.balance !== (active ? total : '0')) fail('withdrawal-reservation-mismatch');
|
||||
const prepared = w.status === 'prepared' || w.status === 'complete';
|
||||
if (prepared ? !w.attemptId || !w.transactionId || w.feeAtomicXmr === undefined || BigInt(w.feeAtomicXmr) > BigInt(w.maxFeeAtomicXmr)
|
||||
: w.attemptId !== undefined || w.transactionId !== undefined || w.feeAtomicXmr !== undefined) fail('withdrawal-attempt-invalid');
|
||||
const status = w.status === 'reserved' ? 'blocked' : w.status === 'cancelled' ? 'cancelled'
|
||||
: w.status === 'prepared' ? 'fixture-prepared' : 'fixture-complete';
|
||||
const box = parse(withdrawalSchema, json(row.box_data), 'withdrawal-outbox-invalid');
|
||||
if (row.box_kind !== 'withdrawal' || row.box_status !== status || !same(box, { ...w, status: prepared ? 'prepared' : 'reserved' })) fail('withdrawal-outbox-mismatch');
|
||||
const account = `customer:${w.actor}:withdrawal:${w.id}`, available = `customer:${w.actor}:available`;
|
||||
expectJournal(posting, `withdraw:${w.id}`, 'withdrawal-reservation', [d(available, total), c(account, total)]);
|
||||
expectJournal(posting, `withdraw-cancel:${w.id}`, 'withdrawal-cancellation', w.status === 'cancelled' ? [d(account, total), c(available, total)] : null);
|
||||
expectJournal(posting, `withdraw-confirm:${w.id}`, 'mock-withdrawal', w.status === 'complete' ? [d(account, total),
|
||||
c('treasury:mock', (BigInt(w.amountAtomicXmr) + BigInt(w.feeAtomicXmr!)).toString()),
|
||||
c(available, (BigInt(w.maxFeeAtomicXmr) - BigInt(w.feeAtomicXmr!)).toString())] : null);
|
||||
}
|
||||
}
|
||||
const withdrawalSelect = `SELECT w.id::text,w.actor,w.status,w.data,a.balance::text,a.normal,
|
||||
b.kind AS box_kind,b.status AS box_status,b.data AS box_data FROM wk_custody.withdrawals w
|
||||
LEFT JOIN wk_custody.accounts a ON a.id='customer:'||w.actor||':withdrawal:'||w.id::text
|
||||
LEFT JOIN wk_custody.outbox b ON b.id=w.id`;
|
||||
|
||||
/** Targeted mutation checks share the same transaction as postings/receipts. No external side effects. */
|
||||
export async function auditOrder(client: PoolClient, id: string) {
|
||||
const result = await client.query(`${orderSelect} WHERE o.id=$1`, [id]);
|
||||
if (result.rowCount !== 1) fail('order-record-missing'); await checkOrders(client, result.rows);
|
||||
}
|
||||
export async function auditWithdrawal(client: PoolClient, id: string) {
|
||||
const result = await client.query(`${withdrawalSelect} WHERE w.id=$1`, [id]);
|
||||
if (result.rowCount !== 1) fail('withdrawal-record-missing'); await checkWithdrawals(client, result.rows);
|
||||
}
|
||||
/** Caller supplies a coherent transaction. Keyset pages bound application memory, not total audit duration. */
|
||||
export async function auditBusinessLiabilities(client: PoolClient): Promise<void> {
|
||||
const deadline = performance.now() + 15000;
|
||||
const budget = () => { if (performance.now() > deadline) throw new Error('Custody business audit budget exceeded'); };
|
||||
let after: string | null = null;
|
||||
for (;;) {
|
||||
budget();
|
||||
const result: QueryResult<OrderRow> = await client.query(`${orderSelect} WHERE ($1::uuid IS NULL OR o.id>$1::uuid) ORDER BY o.id LIMIT 128`, [after]);
|
||||
await checkOrders(client, result.rows); if (result.rows.length < 128) break; after = result.rows.at(-1)!.id;
|
||||
}
|
||||
after = null;
|
||||
for (;;) {
|
||||
budget();
|
||||
const result: QueryResult<WithdrawalRow> = await client.query(`${withdrawalSelect} WHERE ($1::uuid IS NULL OR w.id>$1::uuid) ORDER BY w.id LIMIT 128`, [after]);
|
||||
await checkWithdrawals(client, result.rows); if (result.rows.length < 128) break; after = result.rows.at(-1)!.id;
|
||||
}
|
||||
after = null;
|
||||
for (;;) {
|
||||
budget();
|
||||
const result: QueryResult<{ id: string; card: string; status: string; data: unknown }> = await client.query('SELECT id::text,card,status,data FROM wk_custody.listings WHERE ($1::uuid IS NULL OR id>$1::uuid) ORDER BY id LIMIT 128', [after]);
|
||||
const wallets: string[] = [];
|
||||
for (const row of result.rows) {
|
||||
const l = parse(listingSchema, row.data, 'listing-record-invalid');
|
||||
if (row.id !== l.id || row.card !== l.cardHash || row.status !== l.status) fail('listing-identity-mismatch');
|
||||
wallets.push(l.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address);
|
||||
}
|
||||
await validateWallets(wallets);
|
||||
if (result.rows.length < 128) break; after = result.rows.at(-1)!.id;
|
||||
}
|
||||
// Check both directions: a correct order cannot excuse orphan reserves, tasks or terminal postings.
|
||||
const checks: [string, string][] = [
|
||||
['orphan-purchase-reserve', `SELECT 1 FROM wk_custody.accounts a WHERE a.id LIKE 'customer:%:purchase:%'
|
||||
AND NOT EXISTS(SELECT 1 FROM wk_custody.orders o WHERE a.id='customer:'||o.buyer||':purchase:'||o.id::text) LIMIT 1`],
|
||||
['orphan-withdrawal-reserve', `SELECT 1 FROM wk_custody.accounts a WHERE a.id LIKE 'customer:%:withdrawal:%'
|
||||
AND NOT EXISTS(SELECT 1 FROM wk_custody.withdrawals w WHERE a.id='customer:'||w.actor||':withdrawal:'||w.id::text) LIMIT 1`],
|
||||
['orphan-outbox', `SELECT 1 FROM wk_custody.outbox b WHERE NOT EXISTS(SELECT 1 FROM wk_custody.orders o WHERE o.id=b.id AND b.kind='nft-lock')
|
||||
AND NOT EXISTS(SELECT 1 FROM wk_custody.withdrawals w WHERE w.id=b.id AND b.kind='withdrawal') LIMIT 1`],
|
||||
['listing-order-mismatch', `SELECT 1 FROM wk_custody.listings l LEFT JOIN wk_custody.orders o ON o.data->>'listingId'=l.id::text
|
||||
WHERE (l.status IN ('reserved','sold') AND o.id IS NULL) OR l.id::text IS DISTINCT FROM l.data->>'id'
|
||||
OR l.status IS DISTINCT FROM l.data->>'status' OR l.card IS DISTINCT FROM l.data->>'cardHash' LIMIT 1`],
|
||||
['duplicate-listing-orders', `SELECT 1 FROM wk_custody.orders GROUP BY data->>'listingId' HAVING count(*)<>1 LIMIT 1`],
|
||||
['orphan-lifecycle-posting', `SELECT 1 FROM wk_custody.journals j WHERE
|
||||
(j.kind IN ('purchase-reservation','purchase-cancellation','mock-sale-settlement') AND NOT EXISTS
|
||||
(SELECT 1 FROM wk_custody.orders o WHERE j.id IN ('reserve:'||o.id::text,'cancel:'||o.id::text,'settle:'||o.id::text))) OR
|
||||
(j.kind IN ('withdrawal-reservation','withdrawal-cancellation','mock-withdrawal') AND NOT EXISTS
|
||||
(SELECT 1 FROM wk_custody.withdrawals w WHERE j.id IN ('withdraw:'||w.id::text,'withdraw-cancel:'||w.id::text,'withdraw-confirm:'||w.id::text))) LIMIT 1`],
|
||||
['duplicate-prepared-attempt', `SELECT 1 FROM wk_custody.withdrawals WHERE status IN ('prepared','complete')
|
||||
GROUP BY data->>'attemptId' HAVING count(*)>1 LIMIT 1`],
|
||||
['duplicate-prepared-transaction', `SELECT 1 FROM wk_custody.withdrawals WHERE status IN ('prepared','complete')
|
||||
GROUP BY data->>'transactionId' HAVING count(*)>1 LIMIT 1`],
|
||||
];
|
||||
for (const [code, sql] of checks) { budget(); if ((await client.query(sql)).rowCount) fail(code); }
|
||||
}
|
||||
@@ -11,6 +11,7 @@ import { CUSTODY_POLICY, MOCK_CREDIT_ATOMIC, custodyCommandSchema, custodyMessag
|
||||
import { CustodyError } from './custody.ts';
|
||||
import { CUSTODY_WRITE_LOCK, verifyCustodySchema } from './postgres-schema.ts';
|
||||
import { resolvedMarketPolicy, validateMarketPolicy, permitsMarketAction, type MarketPolicy } from './policy.ts';
|
||||
import { CustodyIntegrityError, auditBusinessLiabilities, auditOrder, auditWithdrawal } from './custody-integrity.ts';
|
||||
|
||||
export type CustodyOptions = { feeBps: number; card(hash: string, owner: string, sequence: number): { title: string } | undefined; now?: () => number } & MarketPolicy;
|
||||
type Result = CustodyActivity | CustodyListing | CustodyOrder | CustodyWithdrawal;
|
||||
@@ -27,7 +28,18 @@ export class PostgresCustody {
|
||||
private readonly pool: Pool;
|
||||
private readonly options: CustodyOptions;
|
||||
private active = 0;
|
||||
private constructor(pool: Pool, options: CustodyOptions, id: string) { this.pool = pool; this.options = options; this.id = id; }
|
||||
private readonly auditTimer: ReturnType<typeof setInterval>;
|
||||
private auditInFlight: Promise<CustodyAudit> | undefined;
|
||||
private integrityHold = false;
|
||||
private auditState: { status: 'ok' | 'failed'; lastAttemptAt: number; lastSuccessAt: number; failureCode: string | null };
|
||||
private constructor(pool: Pool, options: CustodyOptions, id: string) {
|
||||
this.pool = pool; this.options = options; this.id = id;
|
||||
this.auditState = { status: 'ok', lastAttemptAt: Date.now(), lastSuccessAt: Date.now(), failureCode: null };
|
||||
this.auditTimer = setInterval(() => { void this.audit().catch(() => {
|
||||
console.error('Custody financial audit unavailable or inconsistent; mutations held pending operator review.');
|
||||
}); }, 60000);
|
||||
this.auditTimer.unref();
|
||||
}
|
||||
static async open(config: PoolConfig, options: CustodyOptions): Promise<PostgresCustody> {
|
||||
if (IS_MAINNET) throw new Error('Custody sandbox cannot run on mainnet');
|
||||
await validateMarketPolicy(options);
|
||||
@@ -61,12 +73,23 @@ export class PostgresCustody {
|
||||
}
|
||||
await auditPostgres(c);
|
||||
await c.query('COMMIT');
|
||||
} catch (e) { await c.query('ROLLBACK').catch(() => {}); throw e; }
|
||||
} catch (e) {
|
||||
await c.query('ROLLBACK').catch(() => {});
|
||||
if (e instanceof CustodyIntegrityError) await recordIntegrityIncident(c, e.code).catch(() => {
|
||||
console.error('Custody integrity incident could not be persisted; fence all custody writers and inspect database availability.');
|
||||
});
|
||||
throw e;
|
||||
}
|
||||
finally { c.release(); }
|
||||
return new PostgresCustody(pool, options, id);
|
||||
} catch (e) { await pool.end(); throw e; }
|
||||
}
|
||||
close(): Promise<void> { return this.pool.end(); }
|
||||
async close(): Promise<void> { clearInterval(this.auditTimer); await this.auditInFlight?.catch(() => {}); await this.pool.end(); }
|
||||
diagnostics() {
|
||||
const stale = Date.now() - this.auditState.lastSuccessAt > 120000;
|
||||
return { backend: 'postgres', mutationHold: this.integrityHold || this.auditState.status !== 'ok' || stale,
|
||||
integrityHold: this.integrityHold, audit: { ...this.auditState, stale } };
|
||||
}
|
||||
info(): CustodyInfo {
|
||||
return { enabled: true, mode: 'sandbox', policy: CUSTODY_POLICY, network: NETWORK, chainId: CHAIN_ID,
|
||||
custodianId: this.id, feeBps: this.options.feeBps, ...resolvedMarketPolicy(this.options), liveDeposits: false, liveWithdrawals: false, nftSettlement: false };
|
||||
@@ -75,17 +98,40 @@ export class PostgresCustody {
|
||||
check(this.active < 16, 'Custody database busy; retry the original command shortly');
|
||||
this.active++;
|
||||
let c: PoolClient | undefined;
|
||||
let mutationSavepoint = false;
|
||||
try {
|
||||
c = await this.pool.connect();
|
||||
await c.query(write ? 'BEGIN' : 'BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');
|
||||
await c.query("SET LOCAL statement_timeout='10s'; SET LOCAL lock_timeout='3s'; SET LOCAL idle_in_transaction_session_timeout='10s'; SET LOCAL synchronous_commit=on");
|
||||
// One writer per financial domain initially; other readers remain nonblocking. Lock is database-scoped,
|
||||
// shared by every node/process, migrations and import, not an in-process promise queue.
|
||||
if (write) await c.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
||||
if (write) {
|
||||
await c.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
||||
await c.query('SAVEPOINT custody_mutation'); mutationSavepoint = true;
|
||||
}
|
||||
const result = await fn(c);
|
||||
await c.query('COMMIT');
|
||||
return result;
|
||||
} catch (e) {
|
||||
if (e instanceof CustodyIntegrityError) {
|
||||
this.integrityHold = true;
|
||||
this.auditState = { ...this.auditState, status: 'failed', lastAttemptAt: Date.now(), failureCode: e.code };
|
||||
// Roll back business changes but retain the cross-process writer lock until the incident is committed.
|
||||
// Read-only/background audit failures require a separate marker transaction; DB/process loss can still
|
||||
// prevent persistence, so external writer fencing remains an incident-response requirement.
|
||||
if (c) {
|
||||
try {
|
||||
if (mutationSavepoint) await c.query('ROLLBACK TO SAVEPOINT custody_mutation');
|
||||
else await c.query('ROLLBACK');
|
||||
await recordIntegrityIncident(c, e.code);
|
||||
if (mutationSavepoint) await c.query('COMMIT');
|
||||
} catch {
|
||||
await c.query('ROLLBACK').catch(() => {});
|
||||
console.error('Custody integrity incident could not be persisted; fence all custody writers and inspect database availability.');
|
||||
}
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
await c?.query('ROLLBACK').catch(() => {});
|
||||
// Never retry after an ambiguous COMMIT. Signed receipts resolve the SAME nonce on a client retry.
|
||||
if (e instanceof CustodyError) throw e;
|
||||
@@ -147,13 +193,21 @@ export class PostgresCustody {
|
||||
const key = `${cmd.actor}:${cmd.nonce}`, hash = createHash('sha256').update(message).digest('hex');
|
||||
const old = await c.query('SELECT digest,data FROM wk_custody.receipts WHERE id=$1', [key]);
|
||||
if (old.rowCount) { check(old.rows[0].digest === hash, 'Nonce already used for different custody terms'); return old.rows[0].data as Result; }
|
||||
const incident = await c.query("SELECT 1 FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'");
|
||||
check(!incident.rowCount, 'Custody integrity incident is unresolved; operator review required');
|
||||
check(!this.diagnostics().mutationHold, 'Custody financial reconciliation requires operator review; balances remain readable and original receipts can be replayed');
|
||||
check(permitsMarketAction(this.options, a.type), 'New market trading is disabled by this host; existing balances and resolution remain available');
|
||||
// Sandbox admission budget matches SQLite. Receipt/history are NEVER pruned; cancellation stays available.
|
||||
if (!['cancel-listing', 'cancel-reservation', 'cancel-withdrawal'].includes(a.type)) {
|
||||
const count = await c.query('SELECT count(*)::integer AS n FROM (SELECT 1 FROM wk_custody.receipts LIMIT 10000) b');
|
||||
check(count.rows[0].n < 10000, 'Custody sandbox command budget reached');
|
||||
}
|
||||
// Verify existing liabilities BEFORE resolving them, and validate the new business state before COMMIT.
|
||||
if (a.type === 'cancel-reservation') await auditOrder(c, a.orderId);
|
||||
if (a.type === 'cancel-withdrawal') await auditWithdrawal(c, a.withdrawalId);
|
||||
const result = await this.mutate(c, cmd);
|
||||
if (a.type === 'reserve' || a.type === 'cancel-reservation') await auditOrder(c, (result as CustodyOrder).id);
|
||||
if (a.type === 'withdraw' || a.type === 'cancel-withdrawal') await auditWithdrawal(c, (result as CustodyWithdrawal).id);
|
||||
await c.query('INSERT INTO wk_custody.receipts VALUES($1,$2,$3)', [key, hash, JSON.stringify(result)]);
|
||||
return result;
|
||||
});
|
||||
@@ -226,7 +280,19 @@ export class PostgresCustody {
|
||||
}
|
||||
throw new CustodyError('Unsupported custody action');
|
||||
}
|
||||
audit(): Promise<CustodyAudit> { return this.transaction(false, auditPostgres); }
|
||||
audit(): Promise<CustodyAudit> {
|
||||
if (this.auditInFlight) return this.auditInFlight;
|
||||
this.auditState.lastAttemptAt = Date.now();
|
||||
const pending = this.transaction(false, auditPostgres).then(result => {
|
||||
this.auditState = { ...this.auditState, lastSuccessAt: Date.now(),
|
||||
status: this.integrityHold ? 'failed' : 'ok', failureCode: this.integrityHold ? this.auditState.failureCode : null };
|
||||
return result;
|
||||
}).catch(error => {
|
||||
this.auditState = { ...this.auditState, status: 'failed', failureCode: error instanceof CustodyIntegrityError ? error.code : 'audit-unavailable' };
|
||||
throw error;
|
||||
}).finally(() => { this.auditInFlight = undefined; });
|
||||
this.auditInFlight = pending; return pending;
|
||||
}
|
||||
importSource(): Promise<string | undefined> {
|
||||
return this.transaction(false, async c => (await c.query("SELECT value FROM wk_custody.custody_meta WHERE key='import_source'")).rows[0]?.value);
|
||||
}
|
||||
@@ -234,16 +300,24 @@ export class PostgresCustody {
|
||||
|
||||
/** Exact NUMERIC/text aggregation in a coherent transaction; no Number conversion of money. */
|
||||
export async function auditPostgres(c: PoolClient): Promise<CustodyAudit> {
|
||||
const incident = await c.query("SELECT 1 FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'");
|
||||
if (incident.rowCount) throw new CustodyIntegrityError('unresolved-integrity-incident');
|
||||
const bad = await c.query(`SELECT 1 FROM wk_custody.journals j LEFT JOIN wk_custody.entries e ON e.journal=j.id
|
||||
GROUP BY j.id HAVING count(e.position)=0 OR sum(e.debit-e.credit)<>0 LIMIT 1`);
|
||||
if (bad.rowCount) throw new Error('Unbalanced PostgreSQL custody history');
|
||||
if (bad.rowCount) throw new CustodyIntegrityError('unbalanced-journal');
|
||||
const mismatch = await c.query(`SELECT 1 FROM wk_custody.accounts a LEFT JOIN wk_custody.entries e ON e.account=a.id
|
||||
GROUP BY a.id HAVING a.balance<>COALESCE(sum(CASE WHEN a.normal='debit' THEN e.debit-e.credit ELSE e.credit-e.debit END),0) LIMIT 1`);
|
||||
if (mismatch.rowCount) throw new Error('PostgreSQL custody balance projection mismatch');
|
||||
if (mismatch.rowCount) throw new CustodyIntegrityError('balance-projection-mismatch');
|
||||
const totals = await c.query(`SELECT COALESCE(sum(balance) FILTER(WHERE normal='debit'),0)::text AS assets,
|
||||
COALESCE(sum(balance) FILTER(WHERE normal='credit'),0)::text AS claims FROM wk_custody.accounts`);
|
||||
if (totals.rows[0].assets !== totals.rows[0].claims) throw new Error('PostgreSQL custody backing mismatch');
|
||||
if (totals.rows[0].assets !== totals.rows[0].claims) throw new CustodyIntegrityError('accounting-backing-mismatch');
|
||||
await auditBusinessLiabilities(c);
|
||||
const n = await c.query('SELECT count(*)::text AS n FROM wk_custody.journals');
|
||||
if (BigInt(n.rows[0].n) > BigInt(Number.MAX_SAFE_INTEGER)) throw new Error('Journal count exceeds supported range');
|
||||
return { journals: Number(n.rows[0].n), assetsAtomicXmr: totals.rows[0].assets, claimsAndRevenueAtomicXmr: totals.rows[0].claims };
|
||||
}
|
||||
|
||||
async function recordIntegrityIncident(c: PoolClient, code: string): Promise<void> {
|
||||
await c.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2) ON CONFLICT(key) DO NOTHING',
|
||||
['integrity_incident_v1', JSON.stringify({ version: 1, code, detectedAt: Date.now() })]);
|
||||
}
|
||||
|
||||
@@ -6,7 +6,10 @@ const schema=z.object({schema:z.literal('wikideck-operations-v1'),observedAt:z.s
|
||||
peers:z.object({configured:nonnegative,connected:nonnegative}),
|
||||
eventLoop:z.object({meanMs:nonnegative}),disk:z.unknown(),diskSampledAt:nonnegative.nullable(),
|
||||
market:z.object({queued:nonnegative,lastReconciledAt:nonnegative.nullable(),
|
||||
orderStates:z.array(z.object({status:z.string(),count:nonnegative})).max(20)})});
|
||||
orderStates:z.array(z.object({status:z.string(),count:nonnegative})).max(20)}),
|
||||
custody:z.object({backend:z.string(),mutationHold:z.boolean().optional(),integrityHold:z.boolean().optional(),
|
||||
audit:z.object({status:z.enum(['ok','failed']),lastAttemptAt:nonnegative,lastSuccessAt:nonnegative,
|
||||
failureCode:z.string().nullable(),stale:z.boolean()}).optional()}).optional()});
|
||||
export function checkOperations(raw:unknown,now=Date.now()){
|
||||
const s=schema.parse(raw),warnings:string[]=[];
|
||||
const observation=Date.parse(s.observedAt);
|
||||
@@ -18,6 +21,11 @@ export function checkOperations(raw:unknown,now=Date.now()){
|
||||
if(s.eventLoop.meanMs>250)warnings.push('event-loop-delay');
|
||||
if(!s.market.lastReconciledAt||now-s.market.lastReconciledAt>60000)warnings.push('market-reconciliation-stale');
|
||||
if(s.market.orderStates.some(r=>(r.status==='review'||r.status==='refund-required')&&r.count>0))warnings.push('market-operator-review');
|
||||
if(s.custody?.backend==='postgres') {
|
||||
const audit=s.custody.audit;
|
||||
if(s.custody.mutationHold||s.custody.integrityHold||audit?.status==='failed')warnings.push('custody-financial-review');
|
||||
if(!audit||audit.stale||now-audit.lastSuccessAt>120000||audit.lastSuccessAt>now+60000)warnings.push('custody-audit-stale');
|
||||
}
|
||||
const disk=z.object({availableBytes:nonnegative}).safeParse(s.disk);
|
||||
if(!disk.success||!s.diskSampledAt||now-s.diskSampledAt>90000)warnings.push('disk-sample-unavailable-or-stale');
|
||||
else if(disk.data.availableBytes<512*1024*1024)warnings.push('disk-low');
|
||||
|
||||
@@ -225,7 +225,8 @@ const server=app.listen(port,host,()=>{
|
||||
const operations=startOperations(opsPort,()=>({network:NETWORK,paymentMode:'mock',
|
||||
ledger:ledger.diagnostics(),native:engine.diagnostics(),
|
||||
peers:{configured:peers.status().length,connected:peers.status().filter(p=>p.connected).length},
|
||||
rpc:{api:apiBudget.status(),p2p:peerBudget.status()},market:market.diagnostics()}),resolve(dataDir,PROFILE));
|
||||
rpc:{api:apiBudget.status(),p2p:peerBudget.status()},market:market.diagnostics(),
|
||||
custody: custody && 'diagnostics' in custody ? custody.diagnostics() : { backend: custody ? 'sqlite-sandbox' : 'off' }}),resolve(dataDir,PROFILE));
|
||||
server.maxConnections=256;
|
||||
server.headersTimeout=10000;
|
||||
server.requestTimeout=15000;
|
||||
|
||||
Reference in New Issue
Block a user