diff --git a/Dockerfile b/Dockerfile index bd8cd21..cb82f3b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -43,6 +43,7 @@ COPY shared ./shared COPY scripts/node-backup.ts ./scripts/node-backup.ts COPY scripts/ops-check.ts ./scripts/ops-check.ts COPY scripts/custody-db.ts ./scripts/custody-db.ts +COPY scripts/monero-observer-worker.ts ./scripts/monero-observer-worker.ts COPY scripts/custody-deployment-init.ts ./scripts/custody-deployment-init.ts COPY scripts/custody-deployment-check.ts ./scripts/custody-deployment-check.ts COPY --from=randomx-build /native/build/wikideck-randomx ./native/wikideck-randomx diff --git a/docs/production-plan.md b/docs/production-plan.md index 71bb1a2..b29a835 100755 --- a/docs/production-plan.md +++ b/docs/production-plan.md @@ -32,6 +32,8 @@ Status: active engineering plan; **not launch or payment authorization**. Update **Offline deposit-review operator commands (2026-10-21):** `custody-db` now has explicit prepare/stage/release/reorg-hold/review-resolution actions, with exact argument-count checks and idempotent coordinator calls. They consume an owner-protected review-policy file and bounded signed-approval files; reviewer private keys are never loaded by the app. POSIX owner-only permissions and no-follow opens are required; these file operations refuse Windows until ACL verification is implemented. The existing `IS_MAINNET` guard keeps the CLI sandbox-only. This is an operator rehearsal interface, not a public deposit endpoint or live payment activation. Added argument-parser and protected-file tests; full offline suite passed **194/194** and PostgreSQL integration **36/36**. +**Separate Monero observation worker (2026-10-21):** added a standalone opt-in process, durable bounded catch-up (100,000-block pages, four pages per batch, 1,000-block overlap), PostgreSQL session-lock leadership, restart resumption, and redacted aggregate progress logs. It persists observations only: it has no journal, balance, credit, release, address-allocation, or transfer capability. It is not started by the API and is not included in the sandbox stack as a running service. A separate role grant is available; on mainnet the `custody-db` CLI now permits only this exact observer-role grant and continues to reject migrations, audits, coordinator grants, and all financial actions. The operator must separately create a non-admin, no-membership database login, provide its protected PostgreSQL URL and the protected mTLS client/observer configuration, and run the worker under a supervised POSIX service. This is not yet a complete deployable Monero topology: no mainnet Wallet RPC/gateway/network-policy service, secret provisioning, monitoring destination, key lifecycle, nonempty owned-output test, or deposit-credit automation is provided. Validation: production test-image build succeeded; offline suite **195/195** and disposable PostgreSQL suite **37/37** passed. Worker evidence is synthetic catch-up/restart/leader fencing and journal exclusion only; it does not authorize mainnet deposits or payments. + ## 1A. MVP release boundary — proposed safe first production-hosted release The apartment/single-operator constraint changes the rollout and availability claims, not the safety evidence required to hold customer funds or valuable assets. Recommended MVP: **a public, clearly labeled non-monetary beta on the existing testnet**, with real XMR completely disabled, no custodial balances/deposits/withdrawals, and marketplace checkout/delivery disabled for valuable assets. Keep mining, browsing and wallet use only if they are explicitly described as experimental and users accept that data/ownership may be lost. This would be a production-hosted beta, **not mainnet production**, a custody MVP, or an independently secured chain. diff --git a/scripts/custody-db.ts b/scripts/custody-db.ts index d371d0d..799372e 100755 --- a/scripts/custody-db.ts +++ b/scripts/custody-db.ts @@ -2,25 +2,27 @@ import { Pool } from 'pg'; import { open, unlink, mkdir } from 'node:fs/promises'; import { resolve, join } from 'node:path'; import { postgresConfig } from '../server/market/postgres-config.ts'; -import { migrateCustody, grantCustodyRuntime, grantMoneroDepositCreditRuntime } from '../server/market/postgres-schema.ts'; +import { migrateCustody, grantCustodyRuntime, grantMoneroObserverRuntime, + grantMoneroDepositCreditRuntime } from '../server/market/postgres-schema.ts'; import { importSqliteCustody } from '../server/market/custody-import.ts'; import { PostgresCustody } from '../server/market/postgres-custody.ts'; import { IS_MAINNET } from '../shared/network.ts'; import { resolveCustodyIncident } from '../server/market/custody-incident.ts'; import { loadMoneroDepositReviewApproval, loadMoneroDepositReviewPolicy } from '../server/market/monero-deposit-review-config.ts'; -import { parseCustodyDbArguments } from '../server/market/custody-db-args.ts'; +import { assertCustodyDbActionAllowedOnNetwork, parseCustodyDbArguments } from '../server/market/custody-db-args.ts'; import { prepareDepositCreditReview, postReviewedDepositPendingCredit, prepareDepositCreditReleaseReview, releaseReviewedDepositCredit, reconcileDepositCreditReorgHolds, prepareDepositReorgResolution, resolveReleasedDepositReorgHold, type DepositCreditApproval, type DepositCreditReleaseApproval, type DepositReorgResolutionApproval } from '../server/market/monero-deposit-credit.ts'; // Explicit operator task, never invoked by node startup. No passwords or database URLs printed. -if (IS_MAINNET) throw new Error('Only sandbox financial schema operations are implemented'); const commandArgs = process.argv.slice(2); const depositActions = ['deposit-prepare-credit', 'deposit-stage-credit', 'deposit-prepare-release', 'deposit-release', 'deposit-reconcile-holds', 'deposit-prepare-resolve-hold', 'deposit-resolve-hold']; -const custodyActions = ['migrate', 'grant-runtime', 'grant-monero-credit-runtime', 'import-sqlite', 'audit', 'resolve-integrity']; +const custodyActions = ['migrate', 'grant-runtime', 'grant-monero-observer-runtime', 'grant-monero-credit-runtime', + 'import-sqlite', 'audit', 'resolve-integrity']; const { action, argument, extraArgument } = parseCustodyDbArguments(commandArgs); +assertCustodyDbActionAllowedOnNetwork(action, IS_MAINNET); if (!custodyActions.includes(action) && !depositActions.includes(action)) throw new Error('Unsupported custody operation'); try { const config = await postgresConfig(); @@ -32,6 +34,10 @@ try { try { if (action === 'migrate') { await migrateCustody(pool); console.log('Custody PostgreSQL schema verified/migrated. No funds activated.'); } if (action === 'grant-runtime') { await grantCustodyRuntime(pool, argument ?? ''); console.log('Runtime grants applied; no credentials printed.'); } + if (action === 'grant-monero-observer-runtime') { + await grantMoneroObserverRuntime(pool, argument ?? ''); + console.log('Separate Monero observation-only grants applied; journal access and deposit posting remain denied.'); + } if (action === 'grant-monero-credit-runtime') { await grantMoneroDepositCreditRuntime(pool, argument ?? ''); console.log('Separate reviewed pending-credit coordinator grants applied; no credentials printed and no live deposits enabled.'); diff --git a/scripts/monero-observer-worker.ts b/scripts/monero-observer-worker.ts new file mode 100755 index 0000000..cf67c43 --- /dev/null +++ b/scripts/monero-observer-worker.ts @@ -0,0 +1,54 @@ +import { Pool } from 'pg'; +import { createMoneroWalletObserver, loadMoneroObserverConfig, moneroWalletFingerprint } from '../server/market/monero-observer.ts'; +import { loadMoneroMtlsTransport, loadMoneroMtlsTransportConfig } from '../server/market/monero-mtls-transport.ts'; +import { postgresConfig } from '../server/market/postgres-config.ts'; +import { runMoneroObserverWorker } from '../server/market/monero-observer-worker.ts'; + +if (process.platform === 'win32') throw new Error('Monero observer worker requires a POSIX deployment with verifiable protected-file permissions'); +if (process.env.WIKIDECK_MONERO_OBSERVER_WORKER !== '1') throw new Error('Set WIKIDECK_MONERO_OBSERVER_WORKER=1 to start the private observer worker'); + +const observerConfig = await loadMoneroObserverConfig(); +const transportConfig = await loadMoneroMtlsTransportConfig(); +if (new URL(observerConfig.url).href !== new URL(transportConfig.url).href) { + throw new Error('Observer and mTLS transport endpoints do not match'); +} +const transport = await loadMoneroMtlsTransport(transportConfig); +const scanner = createMoneroWalletObserver(observerConfig, transport); +const primaryAddress = await scanner.verifyMainnetWallet(); +const walletFingerprint = await moneroWalletFingerprint(primaryAddress); +const pool = new Pool(await postgresConfig()); +pool.on('error', () => console.error(JSON.stringify({ component: 'monero-observer', event: 'database-unavailable' }))); +const abort = new AbortController(); +const stop = () => abort.abort(); +process.once('SIGINT', stop); +process.once('SIGTERM', stop); + +try { + await runMoneroObserverWorker(pool, walletFingerprint, scanner, { + pollIntervalMs: parsePollInterval(process.env.WIKIDECK_MONERO_OBSERVER_POLL_MS), + signal: abort.signal, + onBatch(result) { + console.log(JSON.stringify({ component: 'monero-observer', event: 'scan-complete', + scanHeight: result.scanHeight, walletHeight: result.walletHeight, caughtUp: result.caughtUp, + pages: result.pages.length, observed: result.pages.reduce((sum, page) => sum + page.observed, 0), + changed: result.pages.reduce((sum, page) => sum + page.changed, 0), + reappeared: result.pages.reduce((sum, page) => sum + page.reappeared, 0), + notSeen: result.pages.reduce((sum, page) => sum + page.notSeen, 0) })); + }, + onFailure() { console.error(JSON.stringify({ component: 'monero-observer', event: 'scan-failed', action: 'retrying' })); }, + }); +} finally { + process.off('SIGINT', stop); + process.off('SIGTERM', stop); + await pool.end(); +} + +function parsePollInterval(value: string | undefined): number { + if (value === undefined) return 30_000; + if (!/^[1-9]\d{0,5}$/.test(value)) throw new Error('Invalid Monero observer polling interval'); + const interval = Number(value); + if (!Number.isSafeInteger(interval) || interval < 1_000 || interval > 15 * 60_000) { + throw new Error('Invalid Monero observer polling interval'); + } + return interval; +} diff --git a/server/market/custody-db-args.ts b/server/market/custody-db-args.ts index 5a4c852..ae6cd9f 100755 --- a/server/market/custody-db-args.ts +++ b/server/market/custody-db-args.ts @@ -1,6 +1,7 @@ export const CUSTODY_DB_ACTION_ARGUMENT_COUNTS = { migrate: 0, 'grant-runtime': 1, + 'grant-monero-observer-runtime': 1, 'grant-monero-credit-runtime': 1, 'import-sqlite': 1, audit: 0, @@ -17,6 +18,12 @@ export const CUSTODY_DB_ACTION_ARGUMENT_COUNTS = { export type CustodyDbAction = keyof typeof CUSTODY_DB_ACTION_ARGUMENT_COUNTS; export type CustodyDbArguments = { action: CustodyDbAction; argument?: string; extraArgument?: string }; +export function assertCustodyDbActionAllowedOnNetwork(action: CustodyDbAction, isMainnet: boolean): void { + if (isMainnet && action !== 'grant-monero-observer-runtime') { + throw new Error('Only the separately scoped Monero observation-role grant is allowed on mainnet'); + } +} + export function parseCustodyDbArguments(args: readonly string[]): CustodyDbArguments { const [action, ...values] = args; if (!action || !Object.hasOwn(CUSTODY_DB_ACTION_ARGUMENT_COUNTS, action)) { diff --git a/server/market/monero-observation-store.ts b/server/market/monero-observation-store.ts index dbc96ed..7b9edab 100755 --- a/server/market/monero-observation-store.ts +++ b/server/market/monero-observation-store.ts @@ -66,7 +66,7 @@ function rowDigest(observation: MoneroIncomingObservation): string { export async function persistMoneroObservationScan(pool: Pool, input: PersistObservationScanInput): Promise { const parsed = scanInputSchema.parse(input); const page = parsed.page; - if (page.fromHeight > page.throughHeight || page.throughHeight !== page.walletHeight || + if (page.fromHeight > page.throughHeight || page.throughHeight > page.walletHeight || page.throughHeight - page.fromHeight > MAX_SCAN_RANGE || parsed.expectedScanHeight - page.fromHeight > MAX_OVERLAP || page.fromHeight > parsed.expectedScanHeight || page.throughHeight < parsed.expectedScanHeight) { throw new Error('Monero observation scan range is not contiguous with its durable cursor'); diff --git a/server/market/monero-observer-worker.ts b/server/market/monero-observer-worker.ts new file mode 100755 index 0000000..97f5bf2 --- /dev/null +++ b/server/market/monero-observer-worker.ts @@ -0,0 +1,103 @@ +import type { Pool } from 'pg'; +import { randomUUID } from 'node:crypto'; +import { z } from 'zod'; +import { MONERO_OBSERVER_WORKER_LOCK } from './postgres-schema.ts'; +import { persistMoneroObservationScan, type PersistObservationScanResult } from './monero-observation-store.ts'; +import type { MoneroObservationScanner } from './monero-observer.ts'; + +const fingerprintSchema = z.string().regex(/^[a-f0-9]{64}$/); +export const MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE = 100_000; +export const MONERO_OBSERVER_OVERLAP_BLOCKS = 1_000; +export const MONERO_OBSERVER_MAX_PAGES_PER_BATCH = 4; + +export type MoneroObserverBatchResult = { + walletFingerprint: string; + pages: PersistObservationScanResult[]; + scanHeight: number; + walletHeight: number; + caughtUp: boolean; +}; + +/** + * Perform a small bounded catch-up batch. The database observation role supplies the durable cursor; + * the scanner never receives a spend or address-allocation capability. + */ +export async function runMoneroObserverBatch(pool: Pool, walletFingerprint: string, scanner: MoneroObservationScanner, + options: { now?: () => number; maxPages?: number } = {}): Promise { + fingerprintSchema.parse(walletFingerprint); + const maxPages = options.maxPages ?? MONERO_OBSERVER_MAX_PAGES_PER_BATCH; + if (!Number.isSafeInteger(maxPages) || maxPages < 1 || maxPages > MONERO_OBSERVER_MAX_PAGES_PER_BATCH) { + throw new Error('Invalid Monero observer page budget'); + } + const pages: PersistObservationScanResult[] = []; + let walletHeight = 0, scanHeight = 0; + for (let pageNumber = 0; pageNumber < maxPages; pageNumber++) { + const state = await pool.query<{ scan_height: string }>( + 'SELECT scan_height::text FROM wk_custody.monero_observer_state WHERE wallet_fingerprint=$1', [walletFingerprint]); + const expectedScanHeight = state.rowCount ? Number(state.rows[0].scan_height) : 0; + if (!Number.isSafeInteger(expectedScanHeight) || expectedScanHeight < 0) throw new Error('Invalid durable Monero observer cursor'); + const fromHeight = Math.max(0, expectedScanHeight - MONERO_OBSERVER_OVERLAP_BLOCKS); + const page = await scanner.scanIncoming(fromHeight, MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE); + if (page.network !== 'mainnet' || page.fromHeight !== fromHeight || page.throughHeight < expectedScanHeight || + page.throughHeight - page.fromHeight >= MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE) { + throw new Error('Monero observer returned a page outside the worker’s bounded cursor window'); + } + const result = await persistMoneroObservationScan(pool, { scanId: randomUUID(), walletFingerprint, + expectedScanHeight, scannedAt: (options.now ?? Date.now)(), page }); + pages.push(result); scanHeight = result.scanHeight; walletHeight = page.walletHeight; + if (page.throughHeight >= page.walletHeight) break; + } + return { walletFingerprint, pages, scanHeight, walletHeight, caughtUp: scanHeight >= walletHeight }; +} + +export type ObserverWorkerOptions = { + pollIntervalMs?: number; + onBatch?: (result: MoneroObserverBatchResult) => void; + onFailure?: () => void; + signal?: AbortSignal; +}; + +function sleep(ms: number, signal?: AbortSignal): Promise { + if (signal?.aborted) return Promise.resolve(); + return new Promise(resolve => { + const timer = setTimeout(done, ms); timer.unref(); + function done() { clearTimeout(timer); signal?.removeEventListener('abort', done); resolve(); } + signal?.addEventListener('abort', done, { once: true }); + }); +} + +/** + * Explicit opt-in worker loop. A PostgreSQL session advisory lock prevents duplicate scanner processes; + * scan failures are redacted and retried with bounded backoff. No journal/credit/release APIs are called. + */ +export async function runMoneroObserverWorker(pool: Pool, walletFingerprint: string, scanner: MoneroObservationScanner, + options: ObserverWorkerOptions = {}): Promise { + fingerprintSchema.parse(walletFingerprint); + const pollIntervalMs = options.pollIntervalMs ?? 30_000; + if (!Number.isSafeInteger(pollIntervalMs) || pollIntervalMs < 1_000 || pollIntervalMs > 15 * 60_000) { + throw new Error('Invalid Monero observer polling interval'); + } + const leader = await pool.connect(); + let acquired = false; + let retryDelay = pollIntervalMs; + try { + const lock = await leader.query<{ locked: boolean }>('SELECT pg_try_advisory_lock($1::bigint) AS locked', [MONERO_OBSERVER_WORKER_LOCK]); + acquired = lock.rows[0]?.locked === true; + if (!acquired) throw new Error('A Monero observer worker already holds the database lock'); + while (!options.signal?.aborted) { + try { + const result = await runMoneroObserverBatch(pool, walletFingerprint, scanner); + options.onBatch?.(result); + retryDelay = pollIntervalMs; + if (!result.caughtUp) continue; + } catch { + options.onFailure?.(); + retryDelay = Math.min(Math.max(pollIntervalMs, retryDelay * 2), 5 * 60_000); + } + await sleep(retryDelay, options.signal); + } + } finally { + if (acquired) await leader.query('SELECT pg_advisory_unlock($1::bigint)', [MONERO_OBSERVER_WORKER_LOCK]).catch(() => {}); + leader.release(); + } +} diff --git a/server/market/monero-observer.ts b/server/market/monero-observer.ts index cebaddc..f0784bd 100755 --- a/server/market/monero-observer.ts +++ b/server/market/monero-observer.ts @@ -8,6 +8,7 @@ const MAX_CONFIG_BYTES = 4096; const MAX_RESPONSE_BYTES = 1024 * 1024; const RPC_TIMEOUT_MS = 5000; const MAX_TRANSFERS = 4096; +const MAX_SCAN_RANGE = 100_000; const configSchema = z.object({ network: z.literal('mainnet'), url: z.string().min(1).max(2048), @@ -83,6 +84,9 @@ export type MoneroObservationPage = { primaryAddress: string; incoming: MoneroIncomingObservation[]; }; +export type MoneroObservationScanner = { + scanIncoming(fromHeight: number, maxBlocks?: number): Promise; +}; export type MoneroFixtureTransport = (url: string, init: RequestInit) => Promise; class MoneroWalletRpcError extends Error { @@ -215,15 +219,17 @@ export function createMoneroWalletObserver(config: z.infer, const result = addressResultSchema.parse(await rpc('get_address', { account_index: 0 })); return validateXmrAddress(result.address); }, - async scanIncoming(fromHeight: number): Promise { - if (!Number.isSafeInteger(fromHeight) || fromHeight < 0) throw new Error('Invalid Monero scan height'); + async scanIncoming(fromHeight: number, maxBlocks = MAX_SCAN_RANGE): Promise { + if (!Number.isSafeInteger(fromHeight) || fromHeight < 0 || !Number.isSafeInteger(maxBlocks) || + maxBlocks < 1 || maxBlocks > MAX_SCAN_RANGE) throw new Error('Invalid Monero scan range'); const [heightRaw, primaryAddress] = await Promise.all([ rpc('get_height'), this.verifyMainnetWallet(), ]); const walletHeight = integer(heightResultSchema.parse(heightRaw).height, BigInt(Number.MAX_SAFE_INTEGER), 'height'); if (fromHeight > walletHeight) throw new Error('Monero observer cursor is ahead of wallet height'); + const throughHeight = Math.min(walletHeight, fromHeight + maxBlocks - 1); const transferRaw = transfersResultSchema.parse(await rpc('get_transfers', { - in: true, filter_by_height: true, min_height: fromHeight, max_height: walletHeight, + in: true, filter_by_height: true, min_height: fromHeight, max_height: throughHeight, })); if (transferRaw.in.length >= MAX_TRANSFERS) throw new Error('Monero transfer page reached its safety limit; refusing to advance scan cursor'); const transferMetadata = new Map>(); @@ -264,7 +270,7 @@ export function createMoneroWalletObserver(config: z.infer, if (!metadata) continue; const height = integer(output.block_height, BigInt(Number.MAX_SAFE_INTEGER), 'output height'); const expectedHeight = integer(metadata.height, BigInt(Number.MAX_SAFE_INTEGER), 'transfer height'); - if (height !== expectedHeight || height < fromHeight || height > walletHeight) continue; + if (height !== expectedHeight || height < fromHeight || height > throughHeight) continue; const amount = BigInt(output.amount); if (amount <= 0n || amount > XMR_RPC_UINT64_MAX) throw new Error('Invalid incoming Monero output amount'); const globalOutputIndex = integer(output.global_index, BigInt(Number.MAX_SAFE_INTEGER), 'global output index'); @@ -290,7 +296,7 @@ export function createMoneroWalletObserver(config: z.infer, if (totals.get(key) !== BigInt(metadata.amount)) throw new Error('Wallet transaction amount does not match its owned output set'); await validateXmrAddress(metadata.address); } - return { network: 'mainnet', walletHeight, fromHeight, throughHeight: walletHeight, primaryAddress, incoming }; + return { network: 'mainnet', walletHeight, fromHeight, throughHeight, primaryAddress, incoming }; }, }; } diff --git a/server/market/postgres-schema.ts b/server/market/postgres-schema.ts index 4a43176..0c62a02 100755 --- a/server/market/postgres-schema.ts +++ b/server/market/postgres-schema.ts @@ -3,6 +3,7 @@ import type { Pool, PoolClient } from 'pg'; export const CUSTODY_PG_VERSION = 7; export const CUSTODY_WRITE_LOCK = '7140824361'; +export const MONERO_OBSERVER_WORKER_LOCK = '7140824362'; export const CUSTODY_SCHEMA_V1 = ` CREATE SCHEMA IF NOT EXISTS wk_custody; REVOKE ALL ON SCHEMA wk_custody FROM PUBLIC; @@ -395,6 +396,10 @@ export async function grantCustodyRuntime(pool: Pool, role: string): Promise {}); throw e; } diff --git a/tests/custody-db-args.test.ts b/tests/custody-db-args.test.ts index 36f1922..fbe142b 100755 --- a/tests/custody-db-args.test.ts +++ b/tests/custody-db-args.test.ts @@ -1,6 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { parseCustodyDbArguments } from '../server/market/custody-db-args.ts'; +import { assertCustodyDbActionAllowedOnNetwork, parseCustodyDbArguments } from '../server/market/custody-db-args.ts'; test('custody database CLI parses two-argument deposit operations without shifting arguments', () => { assert.deepEqual(parseCustodyDbArguments(['deposit-prepare-credit', 'a'.repeat(64), `${'b'.repeat(64)}:${'c'.repeat(64)}`]), { @@ -9,6 +9,17 @@ test('custody database CLI parses two-argument deposit operations without shifti assert.deepEqual(parseCustodyDbArguments(['deposit-prepare-resolve-hold', 'credit-id', 'hold-id']), { action: 'deposit-prepare-resolve-hold', argument: 'credit-id', extraArgument: 'hold-id', }); + assert.deepEqual(parseCustodyDbArguments(['grant-monero-observer-runtime', 'monero_observer']), { + action: 'grant-monero-observer-runtime', argument: 'monero_observer', extraArgument: undefined, + }); +}); + +test('mainnet custody CLI permits only the observation-role grant', () => { + assert.doesNotThrow(() => assertCustodyDbActionAllowedOnNetwork('grant-monero-observer-runtime', true)); + for (const action of ['migrate', 'grant-runtime', 'grant-monero-credit-runtime', 'audit', 'deposit-release'] as const) { + assert.throws(() => assertCustodyDbActionAllowedOnNetwork(action, true), /only the separately scoped/i); + assert.doesNotThrow(() => assertCustodyDbActionAllowedOnNetwork(action, false)); + } }); test('custody database CLI rejects unknown, missing, empty, and excess arguments', () => { diff --git a/tests/monero-observer.test.ts b/tests/monero-observer.test.ts index 040ba4a..1d7e1c6 100755 --- a/tests/monero-observer.test.ts +++ b/tests/monero-observer.test.ts @@ -234,7 +234,7 @@ test('observer refuses non-loopback plaintext, URL credentials and malformed hei assert.throws(() => createMoneroWalletObserver({ ...config, url: 'http://wallet.example/json_rpc' }, transport), /HTTPS/); assert.throws(() => createMoneroWalletObserver({ ...config, url: 'http://localhost/json_rpc' }, transport), /HTTPS/); assert.throws(() => createMoneroWalletObserver({ ...config, url: 'https://user:pass@wallet.example/json_rpc' }, transport), /bare/); - await assert.rejects(observer.scanIncoming(-1), /height/); + await assert.rejects(observer.scanIncoming(-1), /scan range/); await assert.rejects(observer.scanIncoming(101), /ahead/); }); diff --git a/tests/postgres/custody.test.ts b/tests/postgres/custody.test.ts index 2ed1531..8f065be 100755 --- a/tests/postgres/custody.test.ts +++ b/tests/postgres/custody.test.ts @@ -18,6 +18,7 @@ import { migrateCustody, grantCustodyRuntime, grantMoneroObserverRuntime, CUSTOD grantMoneroDepositCreditRuntime, CUSTODY_MIGRATION_V2, CUSTODY_SCHEMA_V2_DIGEST } from '../../server/market/postgres-schema.ts'; import { persistMoneroObservationScan } from '../../server/market/monero-observation-store.ts'; import { moneroWalletFingerprint } from '../../server/market/monero-observer.ts'; +import { runMoneroObserverBatch, runMoneroObserverWorker } from '../../server/market/monero-observer-worker.ts'; import { depositCreditApprovalMessage, depositCreditReleaseApprovalMessage, depositReorgResolutionMessage, postReviewedDepositPendingCredit, prepareDepositCreditReleaseReview, prepareDepositCreditReview, prepareDepositReorgResolution, reconcileDepositCreditReorgHolds, releaseReviewedDepositCredit, resolveReleasedDepositReorgHold, @@ -413,6 +414,51 @@ test('Monero scan persistence is idempotent, reorg-visible and never posts custo } finally { await custodyRuntime.end(); await runtime.end(); } }); +test('Monero observer worker catches up through bounded durable windows, resumes, and admits one leader', async t => { + const f = await fixture(t); + await f.admin.query("DO $$ BEGIN IF NOT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='monero_worker_test') THEN CREATE ROLE monero_worker_test LOGIN; END IF; END $$"); + await grantMoneroObserverRuntime(f.admin, 'monero_worker_test'); + const workerPool = new Pool({ ...config, user: 'monero_worker_test' }); + const secondWorkerPool = new Pool({ ...config, user: 'monero_worker_test' }); + const wallet = '7'.repeat(64), walletHeight = 250_005, now = Date.now(); + const requested: number[] = []; + const scanner = { + async scanIncoming(fromHeight: number, maxBlocks = 100_000) { + requested.push(fromHeight); + const throughHeight = Math.min(walletHeight, fromHeight + maxBlocks - 1); + return { network: 'mainnet' as const, walletHeight, fromHeight, throughHeight, + primaryAddress: ADDITIONAL_MARKET_FEE.address, incoming: [] }; + }, + }; + try { + const failedScanner = { scanIncoming: async () => { throw new Error('fixture RPC failure'); } }; + await assert.rejects(runMoneroObserverBatch(workerPool, wallet, failedScanner)); + assert.equal((await f.admin.query('SELECT count(*)::int AS n FROM wk_custody.monero_observer_state WHERE wallet_fingerprint=$1', + [wallet])).rows[0].n, 0, 'failed RPC must not create or advance durable cursor state'); + + const batch = await runMoneroObserverBatch(workerPool, wallet, scanner, { now: () => now }); + assert.equal(batch.pages.length, 3, 'catch-up is split into a bounded number of pages'); + assert.equal(batch.caughtUp, true); + assert.equal(batch.scanHeight, walletHeight); + assert.deepEqual(requested, [0, 98_999, 197_998]); + const restarted = await runMoneroObserverBatch(secondWorkerPool, wallet, scanner, { now: () => now + 1, maxPages: 1 }); + assert.equal(restarted.pages.length, 1); + assert.equal(restarted.scanHeight, walletHeight); + assert.equal(requested.at(-1), walletHeight - 1_000, 'restart resumes at a bounded reorg overlap'); + + const controller = new AbortController(); + let signalReady!: () => void; + const batchObserved = new Promise(resolve => { signalReady = resolve; }); + const firstWorker = runMoneroObserverWorker(workerPool, wallet, scanner, { pollIntervalMs: 60_000, + signal: controller.signal, onBatch: () => signalReady() }); + await batchObserved; + await assert.rejects(runMoneroObserverWorker(secondWorkerPool, wallet, scanner, { pollIntervalMs: 60_000 }), /already holds/); + controller.abort(); await firstWorker; + assert.equal((await f.admin.query('SELECT count(*)::int AS n FROM wk_custody.journals')).rows[0].n, 0, + 'observer worker only persists chain observations and never posts money'); + } finally { await secondWorkerPool.end(); await workerPool.end(); } +}); + test('reviewed Monero outputs post once to pending liability and reorgs create holds without reversal', async t => { const f = await fixture(t); await f.admin.query(`DO $$ BEGIN