diff --git a/.changeset/batch-deferred-redemption.md b/.changeset/batch-deferred-redemption.md new file mode 100644 index 000000000..1f3ae6992 --- /dev/null +++ b/.changeset/batch-deferred-redemption.md @@ -0,0 +1,14 @@ +--- +'@cashu/coco-core': minor +--- + +Add batched redemption for deferred receives: `redeemDeferred()` settles each viable +(mint, unit) group with one swap whose single NUT-02 fee is deterministically apportioned +across the members, while every member still finalizes as its own operation with its own +event and history entry. Groups below the combined fee stay queued. Queued members whose +inputs are already spent at the mint roll back before batching so one poisoned proof +cannot wedge the queue; on terminal mint errors spent members settle or roll back +individually while unspent members return to the queue, and a fresh receive that batched +with the queue falls back to a solo receive instead of failing. Recovery of an +interrupted batch re-executes the stored outputs only after verifying they still satisfy +the swap equation against a freshly computed fee. diff --git a/.changeset/cached-keyset-token-decode.md b/.changeset/cached-keyset-token-decode.md new file mode 100644 index 000000000..c77d65f67 --- /dev/null +++ b/.changeset/cached-keyset-token-decode.md @@ -0,0 +1,7 @@ +--- +'@cashu/coco-core': patch +--- + +Token decoding for a known mint now falls back to cached keysets when the mint refresh +fails (e.g. offline), and mint fetch failures are preserved as the `cause` of the thrown +`TokenValidationError` instead of being swallowed. diff --git a/.changeset/defer-dust-offline-receives.md b/.changeset/defer-dust-offline-receives.md new file mode 100644 index 000000000..41b856cab --- /dev/null +++ b/.changeset/defer-dust-offline-receives.md @@ -0,0 +1,12 @@ +--- +'@cashu/coco-core': major +--- + +Defer dust and unreachable-mint receives instead of failing. `prepare` now transitions +init → `deferred` (reasons `dust` / `mint-unreachable`) and emits a new +`receive-op:deferred` event rather than throwing and deleting the operation. +`wallet.receive()` / `ReceiveOperationService.receive()` now return the finalized or +deferred operation (previously `Promise`), `ops.receive.prepare()` can return a +deferred operation callers must branch on, and deferred operations can be cancelled via +rollback. Payment-request attempts whose child receive defers rest in `receiving` until a +later redemption sweep settles them. diff --git a/.changeset/deferred-receive-contract-tests.md b/.changeset/deferred-receive-contract-tests.md new file mode 100644 index 000000000..ae0d7f1ff --- /dev/null +++ b/.changeset/deferred-receive-contract-tests.md @@ -0,0 +1,6 @@ +--- +'@cashu/coco-adapter-tests': patch +--- + +Add repository contract coverage for deferred receive operations: deferred round-trips, +`batchId` round-trips, and `getPending` including executing and deferred states. diff --git a/.changeset/deferred-receive-indexeddb.md b/.changeset/deferred-receive-indexeddb.md new file mode 100644 index 000000000..7bb53777e --- /dev/null +++ b/.changeset/deferred-receive-indexeddb.md @@ -0,0 +1,6 @@ +--- +'@cashu/coco-indexeddb': minor +--- + +Persist deferred receive operations (`deferred` state, `deferredReason`, `batchId`) and +include them in pending queries. No Dexie schema version bump is required. diff --git a/.changeset/deferred-receive-ops-api.md b/.changeset/deferred-receive-ops-api.md new file mode 100644 index 000000000..8d8fe99c9 --- /dev/null +++ b/.changeset/deferred-receive-ops-api.md @@ -0,0 +1,7 @@ +--- +'@cashu/coco-core': minor +--- + +Expose deferred receives through the ops API: `ops.receive.listDeferred()`, +`ops.receive.redeemDeferred(filter?)`, `cancel()` now accepts deferred operations, and +`listInFlight()` includes deferred alongside executing operations. diff --git a/.changeset/deferred-receive-react-hook.md b/.changeset/deferred-receive-react-hook.md new file mode 100644 index 000000000..188fb20bc --- /dev/null +++ b/.changeset/deferred-receive-react-hook.md @@ -0,0 +1,7 @@ +--- +'@cashu/coco-react': minor +--- + +Surface deferred receives in `useReceiveOperation`: the hook binds deferred prepare +results, observes `receive-op:deferred` events, treats cancelled deferred operations like +cancelled inits, and adds `listDeferred()` / `redeemDeferred(filter?)` passthroughs. diff --git a/.changeset/deferred-receive-sql-storage.md b/.changeset/deferred-receive-sql-storage.md new file mode 100644 index 000000000..9ab94f6af --- /dev/null +++ b/.changeset/deferred-receive-sql-storage.md @@ -0,0 +1,10 @@ +--- +'@cashu/coco-sql-storage': minor +'@cashu/coco-sqlite': patch +'@cashu/coco-sqlite-bun': patch +'@cashu/coco-expo-sqlite': patch +--- + +Persist deferred receive operations: migration `037_receive_operations_deferred` rebuilds +`coco_cashu_receive_operations` with a `deferred` state, `deferredReason`, and `batchId` +columns, and pending queries now include deferred operations. diff --git a/.changeset/deferred-receive-state.md b/.changeset/deferred-receive-state.md new file mode 100644 index 000000000..a488acca4 --- /dev/null +++ b/.changeset/deferred-receive-state.md @@ -0,0 +1,9 @@ +--- +'@cashu/coco-core': major +--- + +Add a `deferred` state to the receive operation saga. Receives that cannot be settled yet +(dust below the swap fee, unreachable mints) are now modeled as +`DeferredReceiveOperation` with a `deferredReason`, and batched redemptions link members via +`batchId`. The `ReceiveOperationState` union and `ReceiveOperation` discriminated union are +widened, so downstream exhaustive state handling must account for `deferred`. diff --git a/.changeset/deferred-redemption-triggers.md b/.changeset/deferred-redemption-triggers.md new file mode 100644 index 000000000..d299dd79e --- /dev/null +++ b/.changeset/deferred-redemption-triggers.md @@ -0,0 +1,9 @@ +--- +'@cashu/coco-core': minor +--- + +Trigger deferred receive redemption automatically: an incoming `receive()` drains queued +deferred operations of the same mint and unit into its own batched swap (this is how +queued dust becomes redeemable), and the receive recovery sweep — already run at startup +and via `ops.receive.recovery.run()` — finishes by attempting to redeem every queued +group, tolerating unreachable mints. diff --git a/.changeset/typed-keypair-not-found.md b/.changeset/typed-keypair-not-found.md new file mode 100644 index 000000000..0cae9fdf7 --- /dev/null +++ b/.changeset/typed-keypair-not-found.md @@ -0,0 +1,6 @@ +--- +'@cashu/coco-core': minor +--- + +Throw a typed `KeyPairNotFoundError` (carrying the missing public key) from key ring +signing instead of a plain `Error`, so callers can branch on missing-key failures. diff --git a/CONTEXT.md b/CONTEXT.md index 78a7c7d97..e26a820a0 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -88,6 +88,18 @@ The time after which a quote can no longer receive a new payment. Expiry does no value that was already paid before expiry. _Avoid_: Claim deadline, quote invalidity +**Deferred Receive**: +A persisted receive operation whose redemption is postponed until it can be settled +fee-efficiently or its prerequisites exist (dust below the swap fee, or an unreachable +mint). +_Avoid_: Queued token, pending receive, receive later table + +**Batch Redemption**: +Settling several deferred receives with one mint swap whose single fee is apportioned +across the members. Each member still finalizes as its own operation with its own history +entry. +_Avoid_: Sweep, merge, combined receive + **Background Watcher**: A session-scoped automatic observer that keeps wallet state progressing without a direct caller waiting on a specific result. Disabling a Background Watcher does not disable explicit caller diff --git a/packages/adapter-tests/src/index.ts b/packages/adapter-tests/src/index.ts index 3bbe722ac..79be6c916 100644 --- a/packages/adapter-tests/src/index.ts +++ b/packages/adapter-tests/src/index.ts @@ -394,6 +394,15 @@ export function createDummyReceiveOperation(): ReceiveOperation { } satisfies ReceiveOperation; } +export function createDummyDeferredReceiveOperation(): ReceiveOperation { + return { + ...createDummyReceiveOperation(), + id: 'receive-op-deferred', + state: 'deferred', + deferredReason: 'dust', + } as ReceiveOperation; +} + export function createDummyPaymentRequestReceiveOperation( overrides?: Partial, ): PaymentRequestReceiveOperation { @@ -1118,6 +1127,82 @@ export async function runReceiveOperationRepositoryContract( await dispose(); } }); + + it('round-trips deferred operations without prepared data', async () => { + const { repositories, dispose } = await options.createRepositories(); + try { + const operation = createDummyDeferredReceiveOperation(); + await repositories.receiveOperationRepository.create(operation); + + const stored = await repositories.receiveOperationRepository.getById(operation.id); + + expect(stored).toBeDefined(); + expect(stored!.state).toBe('deferred'); + if (stored!.state === 'deferred') { + expect(stored!.deferredReason).toBe('dust'); + } + expect(stored!.amount.equals(Amount.from(3))).toBe(true); + expect(stored!.inputProofs).toHaveLength(2); + } finally { + await dispose(); + } + }); + + it('round-trips batchId on executing operations', async () => { + const { repositories, dispose } = await options.createRepositories(); + try { + const operation = { + ...createDummyReceiveOperation(), + id: 'receive-op-batch', + state: 'executing', + fee: Amount.from(1), + outputData: { keep: [], send: [] }, + batchId: 'batch-1', + } satisfies ReceiveOperation; + await repositories.receiveOperationRepository.create(operation); + + const stored = await repositories.receiveOperationRepository.getById(operation.id); + + expect(stored).toBeDefined(); + expect(stored!.batchId).toBe('batch-1'); + } finally { + await dispose(); + } + }); + + it('returns executing and deferred operations from getPending', async () => { + const { repositories, dispose } = await options.createRepositories(); + try { + const repo = repositories.receiveOperationRepository; + await repo.create({ ...createDummyReceiveOperation(), id: 'receive-op-init' }); + await repo.create(createDummyDeferredReceiveOperation()); + await repo.create({ + ...createDummyReceiveOperation(), + id: 'receive-op-executing', + state: 'executing', + fee: Amount.from(1), + outputData: { keep: [], send: [] }, + } satisfies ReceiveOperation); + await repo.create({ + ...createDummyReceiveOperation(), + id: 'receive-op-finalized', + state: 'finalized', + fee: Amount.from(1), + outputData: { keep: [], send: [] }, + } satisfies ReceiveOperation); + + const pending = await repo.getPending(); + + expect( + pending + .map((op) => op.id) + .sort() + .join(','), + ).toBe('receive-op-deferred,receive-op-executing'); + } finally { + await dispose(); + } + }); }); } diff --git a/packages/adapter-tests/src/integration.ts b/packages/adapter-tests/src/integration.ts index b47442542..3bd805d59 100644 --- a/packages/adapter-tests/src/integration.ts +++ b/packages/adapter-tests/src/integration.ts @@ -3041,6 +3041,146 @@ export async function runIntegrationTests { + let repositoriesDispose: (() => Promise) | undefined; + let repositories: Repositories | undefined; + + beforeEach(async () => { + const created = await createRepositories(); + repositories = created.repositories; + repositoriesDispose = created.dispose; + mgr = await initializeCoco({ + repo: created.repositories, + seedGetter, + logger, + }); + + await mgr.mint.addMint(mintUrl, { trusted: true }); + await mintAmount(mgr!, mintUrl, 200, testUnit); + }); + + afterEach(async () => { + if (repositoriesDispose) { + await repositoriesDispose(); + repositoriesDispose = undefined; + } + repositories = undefined; + }); + + const sendToken = async (amount: number): Promise => { + const preparedSend = await mgr!.ops.send.prepare({ + mintUrl, + amount: testAmount(amount), + }); + const { token } = await mgr!.ops.send.execute(preparedSend.id); + return token; + }; + + it('defers a dust token and settles it together with the next receive', async () => { + // Send both tokens up front so the spendable balance baseline is + // not disturbed between the two receives. + const dustToken = await sendToken(1); + const secondToken = await sendToken(32); + const balanceBefore = await getMintSpendableBalance(mgr!, mintUrl, testUnit); + + const deferredEventPromise = waitForEvent<{ + operationId: string; + operation: { state: string }; + }>(mgr!, 'receive-op:deferred', (payload) => payload.operation.state === 'deferred'); + + const dustResult = await mgr!.wallet.receive(dustToken); + if (dustResult.state === 'finalized') { + // The mint charges no input fees (e.g. the custom-unit run), so + // dust below the fee cannot exist; nothing to assert here. + return; + } + + expect(dustResult.state).toBe('deferred'); + expect(dustResult.deferredReason).toBe('dust'); + const deferredEvent = await deferredEventPromise; + expect(deferredEvent.operationId).toBe(dustResult.id); + + const inFlight = await mgr!.ops.receive.listInFlight(); + expect(inFlight.map((op) => op.id)).toContain(dustResult.id); + const queued = await mgr!.ops.receive.listDeferred(); + expect(queued.map((op) => op.id)).toContain(dustResult.id); + + // No history entry and no balance change while queued. + const history = await mgr!.history.getPaginatedHistory(0, 50); + expect( + history.some((entry) => 'operationId' in entry && entry.operationId === dustResult.id), + ).toBe(false); + expect(await getMintSpendableBalance(mgr!, mintUrl, testUnit)).toBe(balanceBefore); + + // A second receive for the same mint and unit drains the queue. + const finalizedIds = new Set(); + const bothFinalized = new Promise((resolve) => { + const unsubscribe = mgr!.on('receive-op:finalized', ({ operationId }) => { + finalizedIds.add(operationId); + if (finalizedIds.size >= 2) { + unsubscribe(); + resolve(); + } + }); + }); + + const secondResult = await mgr!.wallet.receive(secondToken); + expect(secondResult.state).toBe('finalized'); + await bothFinalized; + + const dustAfter = await mgr!.ops.receive.get(dustResult.id); + expect(dustAfter?.state).toBe('finalized'); + expect(dustAfter?.batchId).toBeDefined(); + expect(dustAfter?.batchId).toBe(secondResult.batchId!); + + // Two independent history entries, one per operation. + const historyAfter = await mgr!.history.getPaginatedHistory(0, 50); + const receiveEntries = historyAfter.filter( + (entry) => + 'operationId' in entry && + (entry.operationId === dustResult.id || entry.operationId === secondResult.id), + ); + expect(receiveEntries.length).toBe(2); + + // 33 in, one 1-unit batch fee: net +32. + expect(await getMintSpendableBalance(mgr!, mintUrl, testUnit)).toBe(balanceBefore + 32); + }, 30000); + + it('keeps queued dust across a restart and drains it with a later receive', async () => { + const dustToken = await sendToken(1); + const laterToken = await sendToken(32); + const balanceBefore = await getMintSpendableBalance(mgr!, mintUrl, testUnit); + + const dustResult = await mgr!.wallet.receive(dustToken); + if (dustResult.state === 'finalized') { + // Fee-free mint: dust cannot exist. + return; + } + expect(dustResult.deferredReason).toBe('dust'); + + await mgr!.pauseSubscriptions(); + await mgr!.dispose(); + + // Restarting on the same repositories runs the recovery sweep. A + // lone dust operation stays below the fee, so it must survive the + // restart still queued rather than being cleaned up or rolled back. + mgr = await initializeCoco({ + repo: repositories!, + seedGetter, + logger, + }); + + const afterRestart = await mgr!.ops.receive.get(dustResult.id); + expect(afterRestart?.state).toBe('deferred'); + + // A fresh receive after the restart drains the persisted queue. + const laterResult = await mgr!.wallet.receive(laterToken); + expect(laterResult.state).toBe('finalized'); + expect((await mgr!.ops.receive.get(dustResult.id))?.state).toBe('finalized'); + expect(await getMintSpendableBalance(mgr!, mintUrl, testUnit)).toBe(balanceBefore + 32); + }, 30000); + }); + describe('Wallet Restore', () => { it('should sweep a mint from another seed', async () => { const { repositories, dispose } = await createRepositories(); diff --git a/packages/core/adapter.ts b/packages/core/adapter.ts index f9678cb27..ccc19d6c0 100644 --- a/packages/core/adapter.ts +++ b/packages/core/adapter.ts @@ -64,6 +64,8 @@ export type { PaymentRequestReceiveOperation, PaymentRequestReceiveState, PaymentRequestReceiveTransport, + DeferredReceiveOperation, + DeferredReceiveReason, ReceiveOperation, ReceiveOperationState, SendMethod, diff --git a/packages/core/api/ReceiveOpsApi.ts b/packages/core/api/ReceiveOpsApi.ts index f966ac66a..4017ec72d 100644 --- a/packages/core/api/ReceiveOpsApi.ts +++ b/packages/core/api/ReceiveOpsApi.ts @@ -1,5 +1,6 @@ import type { Token } from '@cashu/cashu-ts'; import type { + DeferredReceiveOperation, FinalizedReceiveOperation, PreparedReceiveOperation, ReceiveOperation, @@ -47,8 +48,14 @@ export class ReceiveOpsApi { /** * Decodes and validates a token, then prepares a receive operation without * executing it. + * + * Returns a deferred operation instead when the receive cannot be settled + * yet (dust below the swap fee, or an unreachable mint); callers must branch + * on `state` before executing. */ - async prepare(input: PrepareReceiveInput): Promise { + async prepare( + input: PrepareReceiveInput, + ): Promise { const initOp = await this.receiveOperationService.init(input.token); return this.receiveOperationService.prepare(initOp); } @@ -80,7 +87,21 @@ export class ReceiveOpsApi { return this.receiveOperationService.getPreparedOperations(); } - /** Lists receive operations that are currently in flight. */ + /** Lists receive operations queued for later redemption. */ + async listDeferred(): Promise { + return this.receiveOperationService.getDeferredOperations(); + } + + /** + * Attempts to redeem deferred receive operations now, batched per mint and + * unit. Groups that are still below the swap fee stay deferred. Useful when + * connectivity returns; the recovery sweep also runs this automatically. + */ + async redeemDeferred(filter?: { mintUrl?: string; unit?: string }): Promise { + return this.receiveOperationService.redeemDeferred(filter); + } + + /** Lists receive operations that are currently in flight (executing or deferred). */ async listInFlight(): Promise { return this.receiveOperationService.getPendingOperations(); } @@ -104,13 +125,17 @@ export class ReceiveOpsApi { /** * Cancels a receive operation that has not completed yet. * - * Only `init` and `prepared` receive operations can be cancelled. + * Only `init`, `prepared`, and `deferred` receive operations can be cancelled. */ async cancel(operationId: string, reason?: string): Promise { const operation = await this.requireOperation(operationId); - if (operation.state !== 'init' && operation.state !== 'prepared') { + if ( + operation.state !== 'init' && + operation.state !== 'prepared' && + operation.state !== 'deferred' + ) { throw new Error( - `Cannot cancel operation in state '${operation.state}'. Expected 'init' or 'prepared'.`, + `Cannot cancel operation in state '${operation.state}'. Expected 'init', 'prepared', or 'deferred'.`, ); } diff --git a/packages/core/api/WalletApi.ts b/packages/core/api/WalletApi.ts index b79679697..a30818b5c 100644 --- a/packages/core/api/WalletApi.ts +++ b/packages/core/api/WalletApi.ts @@ -12,6 +12,10 @@ import type { TokenService, } from '@core/services'; import type { ReceiveOperationService } from '../operations/receive/ReceiveOperationService'; +import type { + DeferredReceiveOperation, + FinalizedReceiveOperation, +} from '../operations/receive/ReceiveOperation'; import type { Logger } from '../logging/Logger.ts'; import { WalletBalancesApi } from './WalletBalancesApi.ts'; import { DEFAULT_UNIT, normalizeUnit, normalizeUnitList } from '../amounts.ts'; @@ -64,10 +68,17 @@ export class WalletApi { /** * Receive a token in one shot. * + * Returns the finalized operation, or a deferred operation when the receive + * cannot be settled yet (dust below the swap fee, or an unreachable mint); + * deferred receives are redeemed later, batched with other queued proofs of + * the same mint and unit. + * * For a multi-step receive flow (review fees/amounts before committing), * use `manager.ops.receive.prepare()` and `manager.ops.receive.execute()`. */ - async receive(token: Token | string): Promise { + async receive( + token: Token | string, + ): Promise { return this.receiveOperationService.receive(token); } diff --git a/packages/core/events/types.ts b/packages/core/events/types.ts index 56f8d4055..93a4bc8b8 100644 --- a/packages/core/events/types.ts +++ b/packages/core/events/types.ts @@ -52,6 +52,12 @@ export interface CoreEvents { operationId: string; operation: ReceiveOperation; }; + /** Emitted when a receive operation is deferred for later redemption */ + 'receive-op:deferred': { + mintUrl: string; + operationId: string; + operation: ReceiveOperation; + }; /** Emitted when receive operation is finalized */ 'receive-op:finalized': { mintUrl: string; diff --git a/packages/core/index.ts b/packages/core/index.ts index b3a6910e6..77b1a2736 100644 --- a/packages/core/index.ts +++ b/packages/core/index.ts @@ -75,6 +75,8 @@ export type { InitReceiveOperation, PreparedReceiveOperation, ExecutingReceiveOperation, + DeferredReceiveOperation, + DeferredReceiveReason, FinalizedReceiveOperation, RolledBackReceiveOperation, ReceiveOperation, diff --git a/packages/core/models/Error.ts b/packages/core/models/Error.ts index d333ac9ce..45d79035e 100644 --- a/packages/core/models/Error.ts +++ b/packages/core/models/Error.ts @@ -30,6 +30,18 @@ export class KeysetSyncError extends Error { } } +/** + * This error is thrown when a signing key pair is not present in the key ring. + */ +export class KeyPairNotFoundError extends Error { + readonly publicKey: string; + constructor(publicKey: string, message?: string) { + super(message ?? `Key pair not found for public key: ${publicKey.substring(0, 8)}...`); + this.name = 'KeyPairNotFoundError'; + this.publicKey = publicKey; + } +} + export class ProofValidationError extends Error { constructor(message: string) { super(message); diff --git a/packages/core/operations/index.ts b/packages/core/operations/index.ts index 7ffc9b8fb..f09db963d 100644 --- a/packages/core/operations/index.ts +++ b/packages/core/operations/index.ts @@ -12,6 +12,11 @@ export type { } from './mint/MintMethodHandler.ts'; export { MintOperationService } from './mint/MintOperationService.ts'; export * from './send'; -export type { ReceiveOperation, ReceiveOperationState } from './receive/ReceiveOperation.ts'; +export type { + DeferredReceiveOperation, + DeferredReceiveReason, + ReceiveOperation, + ReceiveOperationState, +} from './receive/ReceiveOperation.ts'; export { ReceiveOperationService } from './receive/ReceiveOperationService.ts'; export * from './paymentRequestReceive'; diff --git a/packages/core/operations/receive/ReceiveOperation.ts b/packages/core/operations/receive/ReceiveOperation.ts index 40aacd914..17d64207b 100644 --- a/packages/core/operations/receive/ReceiveOperation.ts +++ b/packages/core/operations/receive/ReceiveOperation.ts @@ -3,15 +3,34 @@ * * init ──► prepared ──► executing ──► finalized * │ │ │ + * │ │ ├──► deferred (batch member returned to queue) + * │ │ │ * └─────────┴────────────┴──► rolled_back + * │ + * └──► deferred ──► executing (batch redemption) * * - init: Operation created, token decoded/validated * - prepared: Fees calculated, outputs created, ready to execute * - executing: Receive in progress (mint interaction) + * - deferred: Redemption postponed (dust or unreachable mint) until it can be + * settled fee-efficiently or its prerequisites exist * - finalized: Proofs saved, operation complete * - rolled_back: Operation failed or aborted before completion */ -export type ReceiveOperationState = 'init' | 'prepared' | 'executing' | 'finalized' | 'rolled_back'; +export type ReceiveOperationState = + | 'init' + | 'prepared' + | 'executing' + | 'deferred' + | 'finalized' + | 'rolled_back'; + +/** + * Why a receive operation was deferred: + * - dust: input value does not cover the swap fee on its own + * - mint-unreachable: mint or keyset data could not be fetched (e.g. offline) + */ +export type DeferredReceiveReason = 'dust' | 'mint-unreachable'; import type { Amount, Proof } from '@cashu/cashu-ts'; import { getSecretsFromSerializedOutputData, type SerializedOutputData } from '../../utils'; @@ -64,6 +83,13 @@ interface ReceiveOperationBase { /** Optional origin metadata for receives created by higher-level sagas. */ source?: ReceiveOperationSource; + + /** + * Groups operations redeemed together in a single batched swap. + * Only set once a deferred operation enters batch redemption; batch members + * must never be re-executed solo because their fee was apportioned batch-wide. + */ + batchId?: string; } /** @@ -102,6 +128,18 @@ export interface ExecutingReceiveOperation extends ReceiveOperationBase, Prepare state: 'executing'; } +/** + * Deferred state - redemption postponed until it can be settled fee-efficiently + * or its prerequisites exist. Carries no PreparedData; fees and outputs are + * recomputed at redemption time. + */ +export interface DeferredReceiveOperation extends ReceiveOperationBase { + state: 'deferred'; + + /** Why redemption was postponed */ + deferredReason: DeferredReceiveReason; +} + /** * Finalized state - proofs saved, operation complete */ @@ -127,6 +165,7 @@ export type ReceiveOperation = | InitReceiveOperation | PreparedReceiveOperation | ExecutingReceiveOperation + | DeferredReceiveOperation | FinalizedReceiveOperation | RolledBackReceiveOperation; @@ -164,6 +203,10 @@ export function isExecutingOperation(op: ReceiveOperation): op is ExecutingRecei return op.state === 'executing'; } +export function isDeferredOperation(op: ReceiveOperation): op is DeferredReceiveOperation { + return op.state === 'deferred'; +} + export function isFinalizedOperation(op: ReceiveOperation): op is FinalizedReceiveOperation { return op.state === 'finalized'; } @@ -173,7 +216,7 @@ export function isRolledBackOperation(op: ReceiveOperation): op is RolledBackRec } export function hasPreparedData(op: ReceiveOperation): op is PreparedOrLaterOperation { - return op.state !== 'init'; + return op.state !== 'init' && op.state !== 'deferred'; } export function isTerminalOperation(op: ReceiveOperation): op is TerminalReceiveOperation { diff --git a/packages/core/operations/receive/ReceiveOperationService.ts b/packages/core/operations/receive/ReceiveOperationService.ts index a9ea7bfdd..6af151785 100644 --- a/packages/core/operations/receive/ReceiveOperationService.ts +++ b/packages/core/operations/receive/ReceiveOperationService.ts @@ -14,10 +14,14 @@ import { serializeOutputData, deserializeOutputData, computeYHexForSecrets, + type SerializedOutputData, } from '../../utils'; import { UnknownMintError, + KeysetSyncError, + MintFetchError, MintOperationError, + NetworkError, ProofValidationError, OperationInProgressError, } from '../../models/Error'; @@ -27,6 +31,8 @@ import type { InitReceiveOperation, PreparedReceiveOperation, PreparedOrLaterOperation, + DeferredReceiveOperation, + DeferredReceiveReason, ExecutingReceiveOperation, FinalizedReceiveOperation, RolledBackReceiveOperation, @@ -40,11 +46,21 @@ import type { ProofService } from '../../services/ProofService'; import type { TokenService } from '../../services/TokenService'; import type { WalletService } from '../../services/WalletService'; import { createReceiveOperation, getOutputProofSecrets } from './ReceiveOperation'; +import { apportionReceiveFee } from './apportionFee'; import type { ReceiveOperationRepository, ProofRepository } from '../../repositories'; import { OperationIdLock } from '../OperationIdLock'; import { MintScopedLock } from '../MintScopedLock'; import { DEFAULT_UNIT, normalizeUnit } from '../../amounts.ts'; +/** A deferred (or incoming init) operation participating in a batch redemption. */ +interface BatchMember { + operation: InitReceiveOperation | DeferredReceiveOperation; + signedProofs: Proof[]; + /** Reason the member returns to the queue when the batch fails non-fatally. */ + requeueReason: DeferredReceiveReason; + releaseLock: () => void; +} + const NON_TERMINAL_RECEIVE_MINT_ERROR_CODES = new Set([ // 11003 is special for receive recovery: the mint may already have accepted and // signed our outputs even though the client saw an error, so we keep executing @@ -78,7 +94,11 @@ export class ReceiveOperationService { private readonly operationIdLock = new OperationIdLock(); /** Lock for the global recovery process */ private recoveryLock: Promise | null = null; - /** In-memory lock to serialize deterministic-output derivation (counter) per mint */ + /** + * In-memory lock to serialize deterministic-output derivation (counter) per + * mint. Shared with the other operation services via the Manager so batch + * redemption serializes against every other counter consumer. + */ private readonly mintScopedLock: MintScopedLock; constructor( @@ -171,15 +191,19 @@ export class ReceiveOperationService { /** * Prepare the operation by calculating fees and creating deterministic outputs. * Transitions init -> prepared and stores outputData for crash recovery. + * Transitions init -> deferred instead when the receive cannot be settled yet + * (dust below the swap fee, or an unreachable mint). */ - async prepare(operation: InitReceiveOperation): Promise { + async prepare( + operation: InitReceiveOperation, + ): Promise { const releaseLock = await this.acquireOperationLock(operation.id); try { // Serialize per-mint so concurrent receives on the same keyset cannot read the // same NUT-13 counter and derive colliding deterministic outputs. Mirrors the // send/melt/mint services, which already hold this lock across counter usage. const releaseMintLock = await this.mintScopedLock.acquire(operation.mintUrl); - let prepared: PreparedReceiveOperation; + let prepared: PreparedReceiveOperation | DeferredReceiveOperation; try { const current = await this.receiveOperationRepository.getById(operation.id); if (!current) { @@ -202,12 +226,15 @@ export class ReceiveOperationService { } // Emit outside the mint lock so a listener cannot extend or re-enter the - // per-mint critical section. Mirrors the send service. - await this.eventBus.emit('receive-op:prepared', { - mintUrl: prepared.mintUrl, - operationId: prepared.id, - operation: prepared, - }); + // per-mint critical section. Mirrors the send service. Deferred results + // already emitted receive-op:deferred inside markAsDeferred. + if (prepared.state === 'prepared') { + await this.eventBus.emit('receive-op:prepared', { + mintUrl: prepared.mintUrl, + operationId: prepared.id, + operation: prepared, + }); + } return prepared; } finally { @@ -218,20 +245,25 @@ export class ReceiveOperationService { /** Internal prepare logic used by prepare(), separated for error handling. */ private async prepareInternal( operation: InitReceiveOperation, - ): Promise { + ): Promise { if (!operation.inputProofs || operation.inputProofs.length === 0) { throw new ProofValidationError('Receive operation has no input proofs'); } const { mintUrl } = operation; - const { wallet } = await this.walletService.getWalletWithActiveKeysetId( - mintUrl, - operation.unit, - ); + let wallet; + try { + ({ wallet } = await this.walletService.getWalletWithActiveKeysetId(mintUrl, operation.unit)); + } catch (e) { + if (this.isMintUnreachableError(e)) { + return this.markAsDeferred(operation, 'mint-unreachable'); + } + throw e; + } const fee = wallet.getFeesForProofs(operation.inputProofs); if (operation.amount.lessThanOrEqual(fee)) { - throw new ProofValidationError('Receive amount is not sufficient after fees'); + return this.markAsDeferred(operation, 'dust'); } const keepAmount = operation.amount.subtract(fee); @@ -354,11 +386,49 @@ export class ReceiveOperationService { /** * High-level receive method that orchestrates init → prepare → execute. * This is the primary entry point used by WalletApi. + * Returns the deferred operation when the receive cannot be settled yet. + * + * When deferred operations are already queued for the same mint and unit, + * the incoming receive drains the queue: it settles together with them in + * one batched swap (this is also how queued dust becomes redeemable). */ - async receive(token: Token | string): Promise { + async receive( + token: Token | string, + ): Promise { const initOp = await this.init(token); + + const hasQueuedGroupMembers = ( + await this.receiveOperationRepository.getByMintUrl(initOp.mintUrl) + ).some((op) => op.state === 'deferred' && op.unit === initOp.unit); + if (hasQueuedGroupMembers) { + try { + const batched = await this.redeemDeferredGroup(initOp.mintUrl, initOp.unit, initOp); + if (batched) { + return batched; + } + } catch (e) { + // A failed batch must not fail a token that is receivable on its + // own: fall back to the solo path when the operation is untouched, + // or report it queued when the failure returned it to the queue. + const current = await this.receiveOperationRepository.getById(initOp.id); + if (current?.state === 'deferred') { + return current; + } + if (!current || current.state !== 'init') { + throw e; + } + this.logger?.warn('Batched receive failed, retrying solo', { + operationId: initOp.id, + error: e instanceof Error ? e.message : String(e), + }); + } + } + const preparedOp = await this.prepare(initOp); - await this.execute(preparedOp); + if (preparedOp.state === 'deferred') { + return preparedOp; + } + return await this.execute(preparedOp); } /** @@ -462,7 +532,17 @@ export class ReceiveOperationService { } const executingOps = await this.receiveOperationRepository.getByState('executing'); + const soloOps = executingOps.filter((op) => !op.batchId); + const batchGroups = new Map(); for (const op of executingOps) { + if (op.state === 'executing' && op.batchId) { + const group = batchGroups.get(op.batchId) ?? []; + group.push(op); + batchGroups.set(op.batchId, group); + } + } + + for (const op of soloOps) { let didRecover = false; try { const current = await this.receiveOperationRepository.getById(op.id); @@ -487,6 +567,22 @@ export class ReceiveOperationService { } } + for (const [batchId, group] of batchGroups) { + try { + await this.recoverBatchGroup(batchId, group); + executingCount += group.length; + } catch (e) { + this.logger?.error('Error recovering batched receive operations', { + batchId, + error: e instanceof Error ? e.message : String(e), + }); + } + } + + // Finally attempt to redeem whatever is queued; groups that are still + // below the fee or unreachable simply stay deferred. + await this.redeemDeferred(); + this.logger?.info('Receive recovery completed', { initOperations: initCount, executingOperations: executingCount, @@ -567,6 +663,17 @@ export class ReceiveOperationService { const allSpent = inputStates.every((s) => s.state === 'SPENT'); if (allUnspent) { + if (executing.batchId) { + // A batch member must never be re-executed solo: its outputs were + // built from a batch-wide fee apportionment and a solo swap would + // not balance. The batch group recovery sweep re-executes it. + this.logger?.debug('Batched receive member left for batch group recovery', { + operationId: executing.id, + batchId: executing.batchId, + }); + return; + } + if (!executing.outputData) { await this.markAsRolledBack(executing, 'Recovered: missing output data for receive'); return; @@ -597,6 +704,16 @@ export class ReceiveOperationService { return; } + if (executing.batchId) { + // Spent batch members settle individually from their own outputData; + // zero-keep members finalize without any outputs to recover. + await this.settleSpentBatchMember( + executing, + 'Recovered: input proofs spent without recoverable outputs', + ); + return; + } + if (!executing.outputData) { await this.markAsRolledBack(executing, 'Recovered: missing output data for receive'); return; @@ -732,19 +849,87 @@ export class ReceiveOperationService { return finalized; } + /** True for failures that indicate the mint could not be reached at all. */ + private isMintUnreachableError(error: unknown): boolean { + return ( + error instanceof MintFetchError || + error instanceof KeysetSyncError || + error instanceof NetworkError + ); + } + + /** + * Persist deferred state and emit the operation deferred event. + * Accepts executing operations so failed batch members can return to the queue; + * prepared data and batch linkage are intentionally dropped in that case. + */ + private async markAsDeferred( + op: InitReceiveOperation | DeferredReceiveOperation | ExecutingReceiveOperation, + deferredReason: DeferredReceiveReason, + ): Promise { + const deferred: DeferredReceiveOperation = { + id: op.id, + state: 'deferred', + deferredReason, + mintUrl: op.mintUrl, + unit: op.unit, + amount: op.amount, + inputProofs: op.inputProofs, + createdAt: op.createdAt, + updatedAt: Date.now(), + error: op.error, + source: op.source, + }; + await this.receiveOperationRepository.update(deferred); + await this.eventBus.emit('receive-op:deferred', { + mintUrl: deferred.mintUrl, + operationId: deferred.id, + operation: deferred, + }); + + this.logger?.info('Receive operation deferred', { + operationId: deferred.id, + mintUrl: deferred.mintUrl, + deferredReason, + amount: deferred.amount, + proofCount: deferred.inputProofs.length, + }); + + return deferred; + } + /** * Persist rolled back state with error context. + * Accepts deferred operations (e.g. queued members whose inputs turn out + * spent); they were never prepared, so an empty prepared payload is + * persisted to satisfy the terminal row shape. */ private async markAsRolledBack( - op: PreparedOrLaterOperation, + op: PreparedOrLaterOperation | DeferredReceiveOperation, error: string, ): Promise { - const rolledBack: RolledBackReceiveOperation = { - ...op, - state: 'rolled_back', - updatedAt: Date.now(), - error, - }; + const rolledBack: RolledBackReceiveOperation = + op.state === 'deferred' + ? { + id: op.id, + state: 'rolled_back', + mintUrl: op.mintUrl, + unit: op.unit, + amount: op.amount, + inputProofs: op.inputProofs, + createdAt: op.createdAt, + updatedAt: Date.now(), + error, + source: op.source, + fee: Amount.zero(), + outputData: serializeOutputData({ keep: [], send: [] }), + } + : { + ...op, + state: 'rolled_back', + updatedAt: Date.now(), + error, + }; await this.receiveOperationRepository.update(rolledBack); await this.eventBus.emit('receive-op:rolled-back', { mintUrl: rolledBack.mintUrl, @@ -805,6 +990,583 @@ export class ReceiveOperationService { return ops.filter((op): op is PreparedReceiveOperation => op.state === 'prepared'); } + /** + * Get all deferred operations. + */ + async getDeferredOperations(): Promise { + const ops = await this.receiveOperationRepository.getByState('deferred'); + return ops.filter((op): op is DeferredReceiveOperation => op.state === 'deferred'); + } + + /** + * Attempt to redeem deferred operations, batched per mint and unit. + * + * Each viable group (combined amount above the combined fee) is settled with + * ONE swap whose single fee is apportioned across the members; every member + * still finalizes as its own operation with its own event and history entry. + * Groups that stay below the combined fee remain deferred. Failures are + * logged per group and never abort the sweep. + */ + async redeemDeferred(filter?: { mintUrl?: string; unit?: string }): Promise { + const deferredOps = await this.getDeferredOperations(); + const groups = new Map(); + for (const op of deferredOps) { + if (filter?.mintUrl && normalizeMintUrl(filter.mintUrl) !== op.mintUrl) continue; + if (filter?.unit && normalizeUnit(filter.unit) !== op.unit) continue; + groups.set(`${op.mintUrl}::${op.unit}`, { mintUrl: op.mintUrl, unit: op.unit }); + } + + for (const { mintUrl, unit } of groups.values()) { + try { + await this.redeemDeferredGroup(mintUrl, unit); + } catch (e) { + this.logger?.warn('Deferred receive redemption failed for group, will retry later', { + mintUrl, + unit, + error: e instanceof Error ? e.message : String(e), + }); + } + } + } + + /** + * Redeem the deferred operations of one (mintUrl, unit) group in a single + * batched swap, optionally including an incoming init operation so a fresh + * receive can drain the queue it batches with. + * + * Returns the incoming operation's outcome when one was provided (finalized, + * or deferred when the group is still below the fee), or null when the + * incoming operation could not be processed (caller falls back to the solo + * path). Without an incoming operation the return value is null. + */ + async redeemDeferredGroup( + mintUrl: string, + unit: string, + incoming?: InitReceiveOperation, + ): Promise { + const releaseMintLock = await this.mintScopedLock.acquire(mintUrl); + const members: BatchMember[] = []; + try { + const candidates = (await this.receiveOperationRepository.getByMintUrl(mintUrl)) + .filter((op): op is DeferredReceiveOperation => op.state === 'deferred') + .filter((op) => op.unit === unit) + .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)); + + for (const candidate of candidates) { + const member = await this.collectBatchMember(candidate); + if (member) { + members.push(member); + } + } + + await this.dropUnredeemableBatchMembers(mintUrl, members); + + if (incoming) { + const releaseLock = await this.acquireOperationLock(incoming.id); + members.push({ + operation: incoming, + signedProofs: incoming.inputProofs, + requeueReason: 'dust', + releaseLock, + }); + } + + if (members.length === 0) { + return null; + } + + return await this.executeBatch(mintUrl, unit, members, incoming?.id); + } finally { + for (const member of members) { + member.releaseLock(); + } + releaseMintLock(); + } + } + + /** + * Validate queued members' inputs with the mint before batching. The batch + * swap is atomic, so one already-spent input (e.g. a sender double-spent a + * queued token) would fail redemption for every member on every attempt. + * Members with spent inputs roll back terminally; members with pending + * inputs stay queued but sit out this round. Best-effort: when the state + * check itself fails, the batch proceeds and the swap outcome decides. + * Dropped members are released and removed from `members` in place. + */ + private async dropUnredeemableBatchMembers( + mintUrl: string, + members: BatchMember[], + ): Promise { + const queued = members.filter((member) => member.operation.state === 'deferred'); + if (queued.length === 0) { + return; + } + + let states: CashuProofState[]; + try { + states = await this.checkProofStatesWithMint( + mintUrl, + queued.flatMap((member) => member.signedProofs), + ); + } catch { + return; + } + const stateByY = new Map(states.map((state) => [state.Y, state.state])); + + for (const member of queued) { + const yHexes = computeYHexForSecrets(member.signedProofs.map((proof) => proof.secret)); + const memberStates = yHexes.map((y) => stateByY.get(y)); + + if (memberStates.some((state) => state === 'SPENT')) { + if (member.operation.state === 'deferred') { + await this.markAsRolledBack(member.operation, 'Receive inputs are already spent'); + } + } else if (memberStates.some((state) => state === 'PENDING')) { + this.logger?.debug('Queued receive inputs pending elsewhere, skipping this batch', { + operationId: member.operation.id, + }); + } else { + continue; + } + + member.releaseLock(); + members.splice(members.indexOf(member), 1); + } + } + + /** + * Lock and validate one deferred operation for batch membership. + * Returns null (member stays deferred) when it is busy or changed state. + * + * The isLocked check must stay non-blocking: prepare()/execute() acquire + * their operation lock before any mint-scoped work, while the batch path + * already holds the mint lock here — blocking on a busy operation would be + * an ABBA deadlock between the two paths. + */ + private async collectBatchMember( + candidate: DeferredReceiveOperation, + ): Promise { + if (this.operationIdLock.isLocked(candidate.id)) { + return null; + } + const releaseLock = await this.acquireOperationLock(candidate.id); + + const current = await this.receiveOperationRepository.getById(candidate.id); + if (!current || current.state !== 'deferred') { + releaseLock(); + return null; + } + + return { + operation: current, + signedProofs: current.inputProofs, + requeueReason: current.deferredReason, + releaseLock, + }; + } + + /** + * Execute one batched swap for the locked members: apportion the single fee, + * create per-member outputs, swap once, then finalize each member. + */ + private async executeBatch( + mintUrl: string, + unit: string, + members: BatchMember[], + incomingId?: string, + ): Promise { + const { wallet } = await this.walletService.getWalletWithActiveKeysetId(mintUrl, unit); + + const allInputs = members.flatMap((member) => member.signedProofs); + const totalAmount = Amount.sum(members.map((member) => member.operation.amount)); + const fee = wallet.getFeesForProofs(allInputs); + + if (totalAmount.lessThanOrEqual(fee)) { + this.logger?.debug('Deferred receive group below combined fee, leaving queued', { + mintUrl, + unit, + totalAmount, + fee, + memberCount: members.length, + }); + const incoming = members.find((member) => member.operation.id === incomingId); + if (incoming && incoming.operation.state === 'init') { + return await this.markAsDeferred(incoming.operation, 'dust'); + } + return null; + } + + const shares = apportionReceiveFee( + members.map((member) => ({ id: member.operation.id, amount: member.operation.amount })), + fee, + ); + + const batchId = generateSubId(); + const executingMembers: ExecutingReceiveOperation[] = []; + for (const member of members) { + const share = shares.get(member.operation.id); + if (!share) { + throw new Error(`Missing fee share for batch member ${member.operation.id}`); + } + + let outputData: SerializedOutputData; + if (share.keepAmount.isZero()) { + outputData = serializeOutputData({ keep: [], send: [] }); + } else { + const outputResult = await this.proofService.createOutputsAndIncrementCounters( + mintUrl, + { + keep: { amount: share.keepAmount, unit }, + send: { amount: Amount.zero(), unit }, + }, + {}, + ); + if (!outputResult.keep || outputResult.keep.length === 0) { + throw new Error('Failed to create deterministic outputs for receive'); + } + outputData = serializeOutputData({ keep: outputResult.keep, send: [] }); + } + + executingMembers.push({ + id: member.operation.id, + state: 'executing', + mintUrl, + unit, + amount: member.operation.amount, + inputProofs: member.signedProofs, + createdAt: member.operation.createdAt, + updatedAt: Date.now(), + error: member.operation.error, + source: member.operation.source, + fee: share.feeShare, + outputData, + batchId, + }); + } + + for (const executing of executingMembers) { + await this.receiveOperationRepository.update(executing); + } + + this.logger?.info('Redeeming deferred receives in one batch', { + mintUrl, + unit, + batchId, + memberCount: executingMembers.length, + totalAmount, + fee, + }); + + try { + const allKeepOutputs = executingMembers.flatMap( + (executing) => deserializeOutputData(executing.outputData).keep, + ); + const newProofs = await wallet.receive( + { mint: mintUrl, proofs: allInputs, unit }, + undefined, + { + type: 'custom', + data: allKeepOutputs, + }, + ); + + const finalized = await this.finalizeBatchMembers(mintUrl, unit, executingMembers, newProofs); + return incomingId ? (finalized.get(incomingId) ?? null) : null; + } catch (e) { + await this.handleBatchFailure(mintUrl, members, executingMembers, e); + throw e; + } + } + + /** + * Split the proofs returned by a batched swap back to their members by + * output secret, save them per member, and finalize each member. + */ + private async finalizeBatchMembers( + mintUrl: string, + unit: string, + executingMembers: ExecutingReceiveOperation[], + newProofs: Proof[], + ): Promise> { + const finalized = new Map(); + for (const executing of executingMembers) { + const memberSecrets = new Set(getOutputProofSecrets(executing)); + const memberProofs = newProofs.filter((proof) => memberSecrets.has(proof.secret)); + if (memberProofs.length > 0) { + await this.proofService.saveProofs( + mintUrl, + mapProofToCoreProof(mintUrl, 'ready', memberProofs, { + unit, + createdByOperationId: executing.id, + }), + ); + } + finalized.set(executing.id, await this.markAsFinalized(executing)); + } + return finalized; + } + + /** + * Settle a failed batch swap: on terminal mint errors each member is checked + * against the mint (spent inputs settle or roll back individually, unspent + * members return to the deferred queue so one poisoned member cannot wedge + * it); transient failures keep members executing for crash recovery. + */ + private async handleBatchFailure( + mintUrl: string, + members: BatchMember[], + executingMembers: ExecutingReceiveOperation[], + error: unknown, + ): Promise { + const rollbackReason = this.getRollbackReasonForReceiveFailure(error); + if (!rollbackReason) { + this.logger?.warn('Batched receive swap failed transiently, members left executing', { + mintUrl, + batchId: executingMembers[0]?.batchId, + error: error instanceof Error ? error.message : String(error), + }); + return; + } + + const membersById = new Map(members.map((member) => [member.operation.id, member])); + for (const executing of executingMembers) { + try { + const inputStates = await this.checkProofStatesWithMint(mintUrl, executing.inputProofs); + const allSpent = + inputStates.length > 0 && inputStates.every((state) => state.state === 'SPENT'); + const original = membersById.get(executing.id); + if (allSpent) { + await this.settleSpentBatchMember(executing, rollbackReason); + } else if (original?.operation.state === 'init') { + // A fresh receive only joined the batch opportunistically; restore + // its init snapshot so receive() can retry it solo instead of + // parking a token that is receivable on its own. + await this.receiveOperationRepository.update({ + ...original.operation, + updatedAt: Date.now(), + }); + } else { + await this.markAsDeferred(executing, original?.requeueReason ?? 'dust'); + } + } catch (memberError) { + this.logger?.warn('Could not settle batch member after failed swap, left executing', { + operationId: executing.id, + error: memberError instanceof Error ? memberError.message : String(memberError), + }); + } + } + } + + /** + * Settle a batch member whose inputs are spent at the mint: recover its own + * outputs when possible, otherwise roll it back. + */ + private async settleSpentBatchMember( + executing: ExecutingReceiveOperation, + rollbackReason: string, + ): Promise { + const outputSecrets = getOutputProofSecrets(executing); + if (outputSecrets.length === 0) { + // Zero-keep member: its whole value was its fee share, nothing to recover. + await this.markAsFinalized(executing); + return; + } + + await this.proofService.recoverProofsFromOutputData(executing.mintUrl, executing.outputData, { + unit: executing.unit, + createdByOperationId: executing.id, + }); + if (await this.hasSavedOutputs(executing)) { + await this.markAsFinalized(executing); + return; + } + await this.markAsRolledBack(executing, rollbackReason); + } + + /** + * Recover the still-executing members of an interrupted batch redemption. + * + * The batch swap is atomic at the mint, so the members' input states decide + * together: all spent means the swap happened (restore each member from its + * own outputData), all unspent means it did not (re-execute the combined + * swap from the stored per-member outputData). Members whose inputs diverge + * (e.g. a sender double-spent queued dust) settle or return to the queue + * individually. + */ + private async recoverBatchGroup( + batchId: string, + group: ExecutingReceiveOperation[], + ): Promise { + const sorted = [...group].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)); + const releases: (() => void)[] = []; + try { + for (const op of sorted) { + if (this.operationIdLock.isLocked(op.id)) { + this.logger?.debug('Batch member busy, skipping batch recovery this round', { + batchId, + operationId: op.id, + }); + return; + } + releases.push(await this.acquireOperationLock(op.id)); + } + + const members: ExecutingReceiveOperation[] = []; + for (const op of sorted) { + const current = await this.receiveOperationRepository.getById(op.id); + if (current && current.state === 'executing' && current.batchId === batchId) { + members.push(current as ExecutingReceiveOperation); + } + } + if (members.length === 0) { + return; + } + + const pending: ExecutingReceiveOperation[] = []; + for (const member of members) { + if (getOutputProofSecrets(member).length > 0 && (await this.hasSavedOutputs(member))) { + await this.markAsFinalized(member); + } else { + pending.push(member); + } + } + if (pending.length === 0) { + return; + } + + const spent: ExecutingReceiveOperation[] = []; + const unspent: ExecutingReceiveOperation[] = []; + for (const member of pending) { + let inputStates: CashuProofState[]; + try { + inputStates = await this.checkProofStatesWithMint(member.mintUrl, member.inputProofs); + } catch (e) { + this.logger?.warn('Could not reach mint for batch recovery, will retry later', { + batchId, + operationId: member.id, + }); + return; + } + const allSpent = + inputStates.length > 0 && inputStates.every((state) => state.state === 'SPENT'); + const allUnspent = inputStates.every((state) => state.state === 'UNSPENT'); + if (allSpent) { + spent.push(member); + } else if (allUnspent) { + unspent.push(member); + } else { + this.logger?.warn('Batch member inputs not conclusive, retry later', { + batchId, + operationId: member.id, + }); + return; + } + } + + for (const member of spent) { + await this.settleSpentBatchMember( + member, + 'Recovered: batch inputs spent without recoverable outputs', + ); + } + + if (unspent.length === 0) { + return; + } + + if (spent.length > 0) { + // The swap can only have partially spent inputs when a third party + // spent some member's inputs; the surviving members return to the + // queue and get re-batched with a fresh fee. + for (const member of unspent) { + await this.markAsDeferred(member, 'dust'); + } + return; + } + + await this.reExecuteBatch(batchId, unspent); + } finally { + for (const release of releases) { + release(); + } + } + } + + /** + * Re-execute an interrupted batch swap from the stored per-member + * outputData. The stored outputs still balance because the fee depends only + * on the unchanged inputs. + */ + private async reExecuteBatch( + batchId: string, + members: ExecutingReceiveOperation[], + ): Promise { + const first = members[0]; + if (!first) { + return; + } + const { mintUrl, unit } = first; + const { wallet } = await this.walletService.getWalletWithActiveKeysetId(mintUrl, unit); + const allInputs = members.flatMap((member) => member.inputProofs); + const allKeepOutputs = members.flatMap( + (member) => deserializeOutputData(member.outputData).keep, + ); + + // The stored outputs only balance when every member of the original + // apportionment is present and the keyset fee is unchanged. A crash + // between persisting members can leave a subset whose outputs no longer + // satisfy the swap equation; requeue instead of trusting the mint to + // reject the unbalanced swap. + const fee = wallet.getFeesForProofs(allInputs); + const outputTotal = Amount.sum(allKeepOutputs.map((output) => output.blindedMessage.amount)); + if (!outputTotal.add(fee).equals(sumProofs(allInputs))) { + this.logger?.warn('Interrupted batch outputs do not balance, requeueing members', { + mintUrl, + batchId, + memberCount: members.length, + outputTotal, + fee, + }); + for (const member of members) { + await this.markAsDeferred(member, 'dust'); + } + return; + } + + this.logger?.info('Re-executing interrupted batched receive', { + mintUrl, + batchId, + memberCount: members.length, + }); + + try { + const newProofs = await wallet.receive( + { mint: mintUrl, proofs: allInputs, unit }, + undefined, + { + type: 'custom', + data: allKeepOutputs, + }, + ); + await this.finalizeBatchMembers(mintUrl, unit, members, newProofs); + } catch (e) { + const rollbackReason = this.getRollbackReasonForReceiveFailure(e); + if (!rollbackReason) { + this.logger?.warn('Batch re-execution failed transiently, will retry later', { + batchId, + error: e instanceof Error ? e.message : String(e), + }); + return; + } + // A terminal failure (e.g. keyset fees changed while interrupted) sends + // the members back to the queue; the next round re-batches with fresh + // fees and outputs. + for (const member of members) { + await this.markAsDeferred(member, 'dust'); + } + } + } + /** * Rollback a receive operation. * Only allowed for operations in 'init' or 'prepared' state. @@ -828,9 +1590,11 @@ export class ReceiveOperationService { throw new Error(`Cannot rollback operation in state ${operation.state}`); case 'init': + case 'deferred': await this.receiveOperationRepository.delete(operation.id); this.logger?.info('Receive operation cancelled', { operationId, + state: operation.state, reason: reason ?? 'User cancelled receive operation', }); return; diff --git a/packages/core/operations/receive/apportionFee.ts b/packages/core/operations/receive/apportionFee.ts new file mode 100644 index 000000000..7bd88105b --- /dev/null +++ b/packages/core/operations/receive/apportionFee.ts @@ -0,0 +1,69 @@ +import { Amount } from '@cashu/cashu-ts'; + +export interface ApportionableReceive { + /** Operation id the share is keyed by */ + id: string; + + /** The operation's input amount */ + amount: Amount; +} + +export interface ApportionedReceiveShare { + /** Portion of the batch fee charged to this operation */ + feeShare: Amount; + + /** Output value kept for this operation (amount - feeShare) */ + keepAmount: Amount; +} + +/** + * Deterministically apportion a single batched swap fee across member + * operations. A batch swap pays one ceil'd fee for all inputs combined + * (NUT-02), so members cannot each subtract their own solo fee; instead the + * batch fee is charged to the largest members first, letting small (dust) + * members keep their full value whenever possible. + * + * Guarantees, independent of input order: + * - every share satisfies 0 <= feeShare <= amount (no Amount underflow) + * - sum(feeShare) === fee + * - sum(keepAmount) === sum(amount) - fee + * + * Throws when the fee exceeds the combined amount; callers must check batch + * viability (total > fee) before apportioning. + */ +export function apportionReceiveFee( + operations: ApportionableReceive[], + fee: Amount, +): Map { + const shares = new Map(); + if (operations.length === 0) { + if (!fee.isZero()) { + throw new Error('Cannot apportion a non-zero fee across zero operations'); + } + return shares; + } + + const total = Amount.sum(operations.map((op) => op.amount)); + if (total.lessThan(fee)) { + throw new Error( + `Batch fee (${fee.toString()}) exceeds combined receive amount (${total.toString()})`, + ); + } + + const sorted = [...operations].sort((a, b) => { + const byAmountDesc = b.amount.compareTo(a.amount); + if (byAmountDesc !== 0) { + return byAmountDesc; + } + return a.id < b.id ? -1 : a.id > b.id ? 1 : 0; + }); + + let remainingFee = fee; + for (const op of sorted) { + const feeShare = remainingFee.lessThanOrEqual(op.amount) ? remainingFee : op.amount; + remainingFee = remainingFee.subtract(feeShare); + shares.set(op.id, { feeShare, keepAmount: op.amount.subtract(feeShare) }); + } + + return shares; +} diff --git a/packages/core/repositories/index.ts b/packages/core/repositories/index.ts index c985f58b9..b01d613d4 100644 --- a/packages/core/repositories/index.ts +++ b/packages/core/repositories/index.ts @@ -310,7 +310,7 @@ export interface ReceiveOperationRepository { /** Get all receive operations in a specific state */ getByState(state: ReceiveOperationState): Promise; - /** Get all pending operations (state in ['executing']) */ + /** Get all pending operations (state in ['executing', 'deferred']) */ getPending(): Promise; /** Get all operations for a specific mint */ diff --git a/packages/core/repositories/memory/MemoryReceiveOperationRepository.ts b/packages/core/repositories/memory/MemoryReceiveOperationRepository.ts index 3f5bb9541..b2d0e123a 100644 --- a/packages/core/repositories/memory/MemoryReceiveOperationRepository.ts +++ b/packages/core/repositories/memory/MemoryReceiveOperationRepository.ts @@ -39,7 +39,7 @@ export class MemoryReceiveOperationRepository implements ReceiveOperationReposit async getPending(): Promise { const results: ReceiveOperation[] = []; for (const op of this.operations.values()) { - if (op.state === 'executing') { + if (op.state === 'executing' || op.state === 'deferred') { results.push({ ...op }); } } diff --git a/packages/core/services/KeyRingService.ts b/packages/core/services/KeyRingService.ts index b7a7e9cde..2ac637ea5 100644 --- a/packages/core/services/KeyRingService.ts +++ b/packages/core/services/KeyRingService.ts @@ -2,6 +2,7 @@ import type { Proof } from '@cashu/cashu-ts'; import type { Logger } from '@core/logging'; import type { KeyRingRepository } from '@core/repositories'; import type { Keypair, KeypairPurpose } from '@core/models/Keypair'; +import { KeyPairNotFoundError } from '@core/models/Error'; import { schnorr, secp256k1 } from '@noble/curves/secp256k1.js'; import { bytesToHex } from '@noble/curves/utils.js'; import { sha256 } from '@noble/hashes/sha2.js'; @@ -122,9 +123,8 @@ export class KeyRingService { } const keyPair = await this.keyRingRepository.getPersistedKeyPair(publicKey, 'p2pk'); if (!keyPair) { - const publicKeyPreview = publicKey.substring(0, 8); this.logger?.error('Key pair not found', { publicKey }); - throw new Error(`Key pair not found for public key: ${publicKeyPreview}...`); + throw new KeyPairNotFoundError(publicKey); } const message = new TextEncoder().encode(proof.secret); const signature = schnorr.sign(sha256(message), keyPair.secretKey); diff --git a/packages/core/services/MintService.ts b/packages/core/services/MintService.ts index 9e902cdbf..a89e4c28a 100644 --- a/packages/core/services/MintService.ts +++ b/packages/core/services/MintService.ts @@ -176,6 +176,22 @@ export class MintService { return await this.mintRepo.isTrustedMint(normalizeMintUrl(mintUrl)); } + /** + * Get a known mint and its cached keysets without any mint interaction. + * Returns null when the mint is not known locally. + */ + async getKnownMintWithKeysets( + mintUrl: string, + ): Promise<{ mint: Mint; keysets: Keyset[] } | null> { + mintUrl = normalizeMintUrl(mintUrl); + const mint = await this.mintRepo.getMintByUrl(mintUrl).catch(() => null); + if (!mint) { + return null; + } + const keysets = await this.keysetRepo.getKeysetsByMintUrl(mint.mintUrl); + return { mint, keysets }; + } + async ensureUpdatedMint(mintUrl: string): Promise<{ mint: Mint; keysets: Keyset[] }> { mintUrl = normalizeMintUrl(mintUrl); let mint = await this.mintRepo.getMintByUrl(mintUrl).catch(() => null); diff --git a/packages/core/services/PaymentRequestReceiveService.ts b/packages/core/services/PaymentRequestReceiveService.ts index 6e26a7999..69153c269 100644 --- a/packages/core/services/PaymentRequestReceiveService.ts +++ b/packages/core/services/PaymentRequestReceiveService.ts @@ -540,6 +540,14 @@ export class PaymentRequestReceiveService { await this.resumeInitChildReceive(attempt, receiveOperation, { ignoreMissingTransportHandler: true, }); + } else if (receiveOperation.state === 'deferred') { + // Deferred children are redeemed by the receive redemption sweep; the + // attempt intentionally rests in 'receiving' until then. + this.logger?.debug('Payment request attempt waiting on deferred child receive', { + attemptId: attempt.id, + receiveOperationId: receiveOperation.id, + deferredReason: receiveOperation.deferredReason, + }); } } @@ -748,6 +756,16 @@ export class PaymentRequestReceiveService { }); const preparedReceive = await this.receiveOperationService.prepare(initReceive); + if (preparedReceive.state === 'deferred') { + // The attempt rests in 'receiving' until a later redemption sweep + // finalizes the deferred child receive. + this.logger?.info('Payment request child receive deferred', { + attemptId: attempt.id, + receiveOperationId: preparedReceive.id, + deferredReason: preparedReceive.deferredReason, + }); + return { operation, attempt, receiveOperation: preparedReceive }; + } const netAmount = preparedReceive.amount.subtract(preparedReceive.fee); attempt = await this.updateAttempt({ ...attempt, @@ -1134,6 +1152,14 @@ export class PaymentRequestReceiveService { ): Promise { try { const preparedReceive = await this.receiveOperationService.prepare(receiveOperation); + if (preparedReceive.state === 'deferred') { + this.logger?.info('Payment request child receive deferred during resume', { + attemptId: attempt.id, + receiveOperationId: preparedReceive.id, + deferredReason: preparedReceive.deferredReason, + }); + return; + } const netAmount = preparedReceive.amount.subtract(preparedReceive.fee); const updatedAttempt = await this.updateAttempt({ ...attempt, diff --git a/packages/core/services/TokenService.ts b/packages/core/services/TokenService.ts index 4f9947365..c8574113c 100644 --- a/packages/core/services/TokenService.ts +++ b/packages/core/services/TokenService.ts @@ -36,13 +36,24 @@ export class TokenService { const { keysets } = await this.mintService.ensureUpdatedMint(mintUrl); mintKeysets = keysets; } catch (err) { - const errMsg = err instanceof Error ? err.message : 'Unable to retrieve mint keysets'; - this.logger?.warn('Failed to get updated keysets for mint', { - token, - mintUrl, - err: errMsg, - }); - throw new TokenValidationError(errMsg); + // A known mint's cached keysets are still good enough to decode a token + // offline; only fail when there is no local keyset knowledge at all. + const cached = await this.mintService.getKnownMintWithKeysets(mintUrl).catch(() => null); + if (cached && cached.keysets.length > 0) { + this.logger?.warn('Mint refresh failed, decoding token with cached keysets', { + mintUrl, + err: err instanceof Error ? err.message : String(err), + }); + mintKeysets = cached.keysets; + } else { + const errMsg = err instanceof Error ? err.message : 'Unable to retrieve mint keysets'; + this.logger?.warn('Failed to get updated keysets for mint', { + token, + mintUrl, + err: errMsg, + }); + throw new TokenValidationError(errMsg, err); + } } try { diff --git a/packages/core/test/unit/KeyRingService.test.ts b/packages/core/test/unit/KeyRingService.test.ts index 8c8a04d42..66fc72ad0 100644 --- a/packages/core/test/unit/KeyRingService.test.ts +++ b/packages/core/test/unit/KeyRingService.test.ts @@ -1,6 +1,7 @@ import { Amount } from '@cashu/cashu-ts'; import { describe, it, beforeEach, expect } from 'bun:test'; import { KeyRingService } from '../../services/KeyRingService.ts'; +import { KeyPairNotFoundError } from '../../models/Error.ts'; import { SeedService } from '../../services/SeedService.ts'; import { MemoryKeyRingRepository } from '../../repositories/memory/MemoryKeyRingRepository.ts'; import { bytesToHex } from '@noble/curves/utils.js'; @@ -461,6 +462,25 @@ describe('KeyRingService', () => { ); }); + it('throws a typed KeyPairNotFoundError carrying the missing public key', async () => { + const proof: Proof = { + id: 'keyset123', + amount: Amount.from(64), + secret: 'my-secret-string', + C: '0000000000000000000000000000000000000000000000000000000000000000', + }; + + const fakePublicKey = '02' + '00'.repeat(32); + + try { + await service.signProof(proof, fakePublicKey); + throw new Error('Expected signProof to reject'); + } catch (error) { + expect(error).toBeInstanceOf(KeyPairNotFoundError); + expect((error as KeyPairNotFoundError).publicKey).toBe(fakePublicKey); + } + }); + it('signs different proofs with different signatures', async () => { const kp = await service.generateNewKeyPair(); diff --git a/packages/core/test/unit/MemoryReceiveOperationRepository.test.ts b/packages/core/test/unit/MemoryReceiveOperationRepository.test.ts new file mode 100644 index 000000000..61835eb82 --- /dev/null +++ b/packages/core/test/unit/MemoryReceiveOperationRepository.test.ts @@ -0,0 +1,96 @@ +import { Amount } from '@cashu/cashu-ts'; +import type { Proof } from '@cashu/cashu-ts'; +import { describe, it, beforeEach, expect } from 'bun:test'; +import type { + DeferredReceiveOperation, + ReceiveOperation, +} from '../../operations/receive/ReceiveOperation'; +import { MemoryReceiveOperationRepository } from '../../repositories/memory/MemoryReceiveOperationRepository'; + +describe('MemoryReceiveOperationRepository', () => { + const mintUrl = 'https://mint.test'; + + let repo: MemoryReceiveOperationRepository; + + const makeProof = (secret: string): Proof => + ({ + id: 'keyset-1', + amount: Amount.from(1), + secret, + C: `C_${secret}`, + }) as Proof; + + const makeOperation = ( + id: string, + state: ReceiveOperation['state'], + extra?: Partial, + ): ReceiveOperation => + ({ + id, + state, + mintUrl, + unit: 'sat', + amount: Amount.from(1), + inputProofs: [makeProof(`${id}-p1`)], + createdAt: Date.now(), + updatedAt: Date.now(), + ...extra, + }) as ReceiveOperation; + + beforeEach(() => { + repo = new MemoryReceiveOperationRepository(); + }); + + it('round-trips a deferred operation with its reason', async () => { + const deferred = makeOperation('op-deferred', 'deferred', { + deferredReason: 'dust', + } as Partial); + + await repo.create(deferred); + + const stored = (await repo.getById('op-deferred')) as DeferredReceiveOperation; + expect(stored.state).toBe('deferred'); + expect(stored.deferredReason).toBe('dust'); + expect(stored.amount).toEqual(Amount.from(1)); + }); + + it('round-trips batchId on an executing operation', async () => { + const executing = makeOperation('op-batch', 'executing', { + batchId: 'batch-1', + } as Partial); + + await repo.create(executing); + + const stored = await repo.getById('op-batch'); + expect(stored?.batchId).toBe('batch-1'); + }); + + it('getPending returns executing and deferred operations only', async () => { + await repo.create(makeOperation('op-init', 'init')); + await repo.create( + makeOperation('op-deferred', 'deferred', { + deferredReason: 'mint-unreachable', + } as Partial), + ); + await repo.create(makeOperation('op-executing', 'executing')); + await repo.create(makeOperation('op-finalized', 'finalized')); + + const pending = await repo.getPending(); + + expect(pending.map((op) => op.id).sort()).toEqual(['op-deferred', 'op-executing']); + }); + + it('getByState filters deferred operations', async () => { + await repo.create( + makeOperation('op-deferred', 'deferred', { + deferredReason: 'mint-unreachable', + } as Partial), + ); + await repo.create(makeOperation('op-executing', 'executing')); + + const deferred = await repo.getByState('deferred'); + + expect(deferred.length).toBe(1); + expect(deferred[0]?.id).toBe('op-deferred'); + }); +}); diff --git a/packages/core/test/unit/PaymentRequestReceiveService.test.ts b/packages/core/test/unit/PaymentRequestReceiveService.test.ts index da021f30a..bcaa6a206 100644 --- a/packages/core/test/unit/PaymentRequestReceiveService.test.ts +++ b/packages/core/test/unit/PaymentRequestReceiveService.test.ts @@ -565,6 +565,34 @@ describe('PaymentRequestReceiveService', () => { ); }); + it('leaves the attempt receiving when the child receive is deferred', async () => { + (receiveOperationService.prepare as ReturnType).mockImplementation( + async (operation: InitReceiveOperation) => ({ + ...operation, + state: 'deferred', + deferredReason: 'dust', + }), + ); + + const operation = await service.create({ + amount: Amount.from(100), + mints: [mintUrl], + requestId: 'request-id', + }); + + const result = await service.claimPayload(operation.id, createPayload(), { + transport: 'inband', + transportMessageId: 'message-1', + }); + + expect(result.operation.state).toBe('active'); + expect(result.attempt.state).toBe('receiving'); + expect(result.attempt.fee).toBeUndefined(); + expect(result.attempt.netAmount).toBeUndefined(); + expect(result.receiveOperation?.state).toBe('deferred'); + expect(receiveOperationService.execute).not.toHaveBeenCalled(); + }); + it('claims custom-unit payloads through the child receive operation', async () => { const operation = await service.create({ amount: { amount: Amount.from(100), unit: 'USD' }, diff --git a/packages/core/test/unit/ReceiveOperationService.batchRecovery.test.ts b/packages/core/test/unit/ReceiveOperationService.batchRecovery.test.ts new file mode 100644 index 000000000..b8ab99a8d --- /dev/null +++ b/packages/core/test/unit/ReceiveOperationService.batchRecovery.test.ts @@ -0,0 +1,290 @@ +import { Amount, type Proof } from '@cashu/cashu-ts'; +import { describe, it, beforeEach, expect, mock, type Mock } from 'bun:test'; +import type { ExecutingReceiveOperation } from '../../operations/receive/ReceiveOperation'; +import { EventBus } from '../../events/EventBus'; +import type { CoreEvents } from '../../events/types'; +import type { MintAdapter } from '../../infra/MintAdapter'; +import type { MintService } from '../../services/MintService'; +import type { ProofService } from '../../services/ProofService'; +import { TokenService } from '../../services/TokenService'; +import type { WalletService } from '../../services/WalletService'; +import type { CoreProof } from '../../types'; +import { MintOperationError } from '../../models/Error'; +import { ReceiveOperationService } from '../../operations/receive/ReceiveOperationService'; +import { MemoryProofRepository } from '../../repositories/memory/MemoryProofRepository'; +import { MemoryReceiveOperationRepository } from '../../repositories/memory/MemoryReceiveOperationRepository'; + +describe('ReceiveOperationService - batch recovery', () => { + const mintUrl = 'https://mint.test'; + const keysetId = 'keyset-1'; + const batchId = 'batch-1'; + const decoder = new TextDecoder(); + + let receiveOpRepo: MemoryReceiveOperationRepository; + let proofRepo: MemoryProofRepository; + let proofService: ProofService; + let walletService: WalletService; + let mintAdapter: MintAdapter; + let eventBus: EventBus; + let service: ReceiveOperationService; + + let mockWalletReceive: Mock<(...args: any[]) => Promise>; + let mockCheckProofStates: Mock<(mintUrl: string, ys: string[]) => Promise<{ state: string }[]>>; + + const makeProof = (secret: string, amount = 1): Proof => + ({ + id: keysetId, + amount: Amount.from(amount), + secret, + C: `C_${secret}`, + }) as Proof; + + const makeSerializedOutputs = (secrets: string[], amount = 1) => ({ + keep: secrets.map((secret) => ({ + blindedMessage: { amount, id: keysetId, B_: `B_${secret}` }, + blindingFactor: '1234567890abcdef', + secret: Buffer.from(secret).toString('hex'), + })), + send: [], + }); + + const makeBatchMember = ( + id: string, + amount: number, + fee: number, + keepSecrets: string[], + ): ExecutingReceiveOperation => + ({ + id, + state: 'executing', + mintUrl, + unit: 'sat', + amount: Amount.from(amount), + inputProofs: [makeProof(`${id}-input`, amount)], + createdAt: Date.now() - 10000, + updatedAt: Date.now() - 10000, + fee: Amount.from(fee), + outputData: makeSerializedOutputs(keepSecrets, amount - fee), + batchId, + }) as ExecutingReceiveOperation; + + beforeEach(() => { + receiveOpRepo = new MemoryReceiveOperationRepository(); + proofRepo = new MemoryProofRepository(); + eventBus = new EventBus(); + + mockWalletReceive = mock( + async (_token: unknown, _config: unknown, outputType: { data: { secret: Uint8Array }[] }) => + outputType.data.map( + (output, i) => + ({ + id: keysetId, + amount: Amount.from(1), + secret: decoder.decode(output.secret), + C: `C_recovered_${i}`, + }) as Proof, + ), + ); + + mockCheckProofStates = mock(async (_mintUrl: string, ys: string[]) => + ys.map(() => ({ state: 'UNSPENT' })), + ); + + walletService = { + getWalletWithActiveKeysetId: mock(async () => ({ + wallet: { + unit: 'sat', + getFeesForProofs: mock(() => Amount.from(1)), + receive: mockWalletReceive, + }, + })), + } as unknown as WalletService; + + let sweepOutputCounter = 0; + proofService = { + prepareProofsForReceiving: mock(async (proofs: Proof[]) => proofs), + createOutputsAndIncrementCounters: mock(async () => ({ + keep: [ + { + blindedMessage: { amount: Amount.from(1), id: keysetId, B_: 'B_sweep' }, + blindingFactor: BigInt(1), + secret: new TextEncoder().encode(`sweep-out-${sweepOutputCounter++}`), + }, + ], + send: [], + })), + saveProofs: mock(async (targetMintUrl: string, proofs: CoreProof[]) => { + await proofRepo.saveProofs(targetMintUrl, proofs); + }), + recoverProofsFromOutputData: mock(async () => []), + } as unknown as ProofService; + + mintAdapter = { checkProofStates: mockCheckProofStates } as unknown as MintAdapter; + + const mintService = {} as MintService; + service = new ReceiveOperationService( + receiveOpRepo, + proofRepo, + proofService, + mintService, + walletService, + mintAdapter, + new TokenService(mintService), + eventBus, + ); + }); + + it('re-executes the combined swap when all batch inputs are unspent', async () => { + await receiveOpRepo.create(makeBatchMember('op-a', 5, 1, ['a-out'])); + await receiveOpRepo.create(makeBatchMember('op-b', 4, 0, ['b-out'])); + + await service.recoverPendingOperations(); + + expect(mockWalletReceive).toHaveBeenCalledTimes(1); + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('finalized'); + expect((await receiveOpRepo.getById('op-b'))?.state).toBe('finalized'); + + const savedA = await proofRepo.getProofsByOperationId(mintUrl, 'op-a'); + const savedB = await proofRepo.getProofsByOperationId(mintUrl, 'op-b'); + expect(savedA.map((proof) => proof.secret)).toEqual(['a-out']); + expect(savedB.map((proof) => proof.secret)).toEqual(['b-out']); + }); + + it('restores each member from its own outputData when all inputs are spent', async () => { + await receiveOpRepo.create(makeBatchMember('op-a', 5, 1, ['a-out'])); + await receiveOpRepo.create(makeBatchMember('op-b', 4, 0, ['b-out'])); + mockCheckProofStates.mockImplementation(async (_mintUrl: string, ys: string[]) => + ys.map(() => ({ state: 'SPENT' })), + ); + (proofService.recoverProofsFromOutputData as Mock).mockImplementation( + async ( + targetMintUrl: string, + outputData: { keep: { secret: string }[] }, + options: { createdByOperationId: string }, + ) => { + const proofs = outputData.keep.map((output) => ({ + id: keysetId, + amount: Amount.from(1), + secret: decoder.decode(Buffer.from(output.secret, 'hex')), + C: 'C_restored', + mintUrl: targetMintUrl, + unit: 'sat', + state: 'ready', + createdByOperationId: options.createdByOperationId, + })); + await proofRepo.saveProofs(targetMintUrl, proofs as CoreProof[]); + return proofs; + }, + ); + + await service.recoverPendingOperations(); + + expect(mockWalletReceive).not.toHaveBeenCalled(); + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('finalized'); + expect((await receiveOpRepo.getById('op-b'))?.state).toBe('finalized'); + }); + + it('rolls back spent members whose outputs cannot be recovered', async () => { + await receiveOpRepo.create(makeBatchMember('op-a', 5, 1, ['a-out'])); + mockCheckProofStates.mockImplementation(async (_mintUrl: string, ys: string[]) => + ys.map(() => ({ state: 'SPENT' })), + ); + + await service.recoverPendingOperations(); + + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('rolled_back'); + }); + + it('finalizes zero-keep members when the batch inputs are spent', async () => { + await receiveOpRepo.create(makeBatchMember('op-zero', 1, 1, [])); + mockCheckProofStates.mockImplementation(async (_mintUrl: string, ys: string[]) => + ys.map(() => ({ state: 'SPENT' })), + ); + + await service.recoverPendingOperations(); + + expect((await receiveOpRepo.getById('op-zero'))?.state).toBe('finalized'); + }); + + it('requeues unspent members when another member was spent externally', async () => { + await receiveOpRepo.create(makeBatchMember('op-doublespent', 4, 0, ['ds-out'])); + await receiveOpRepo.create(makeBatchMember('op-survivor', 5, 1, ['sv-out'])); + // op-doublespent's inputs are gone (sender reclaimed them); the survivor's + // inputs are untouched. + mockCheckProofStates.mockImplementation(async (_mintUrl: string, ys: string[]) => { + // Member input secrets are 'op-doublespent-input' / 'op-survivor-input'; + // checkProofStatesWithMint hashes them, so track call order instead: + // members are processed in sorted id order (doublespent first). + const call = mockCheckProofStates.mock.calls.length; + const state = call <= 1 ? 'SPENT' : 'UNSPENT'; + return ys.map(() => ({ state })); + }); + + await service.recoverPendingOperations(); + + expect((await receiveOpRepo.getById('op-doublespent'))?.state).toBe('rolled_back'); + // The survivor returns to the queue and the sweep at the end of the + // recovery run re-redeems it in a fresh batch. + const survivor = await receiveOpRepo.getById('op-survivor'); + expect(survivor?.state).toBe('finalized'); + expect(survivor?.batchId).toBeDefined(); + expect(survivor?.batchId).not.toBe(batchId); + }); + + it('requeues an incomplete batch group instead of re-executing unbalanced outputs', async () => { + // Crash mid-persist: only the largest member of a bigger batch reached + // 'executing'. Its stored outputs carry the whole batch fee (5 input, + // 2 fee share), so replaying them against a solo swap of its input at + // fee 1 would not balance (3 + 1 != 5). Recovery must requeue instead of + // replaying; the end-of-recovery sweep then redeems with fresh outputs. + await receiveOpRepo.create(makeBatchMember('op-a', 5, 2, ['a-out'])); + + await service.recoverPendingOperations(); + + const a = await receiveOpRepo.getById('op-a'); + expect(a?.state).toBe('finalized'); + expect(a?.batchId).toBeDefined(); + expect(a?.batchId).not.toBe(batchId); + if (a?.state === 'finalized') { + expect(a.fee).toEqual(Amount.from(1)); + } + expect(mockWalletReceive).toHaveBeenCalledTimes(1); + const outputs = (mockWalletReceive.mock.calls[0]?.[2] as { data: { secret: Uint8Array }[] }) + .data; + expect(outputs.map((output) => decoder.decode(output.secret))).toEqual(['sweep-out-0']); + }); + + it('requeues all members when re-execution fails terminally', async () => { + await receiveOpRepo.create(makeBatchMember('op-a', 5, 1, ['a-out'])); + await receiveOpRepo.create(makeBatchMember('op-b', 4, 0, ['b-out'])); + mockWalletReceive.mockImplementation(async () => { + throw new MintOperationError(11001, 'Transaction inputs do not balance'); + }); + + await service.recoverPendingOperations(); + + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('deferred'); + expect((await receiveOpRepo.getById('op-b'))?.state).toBe('deferred'); + }); + + it('leaves members executing when the mint is unreachable', async () => { + await receiveOpRepo.create(makeBatchMember('op-a', 5, 1, ['a-out'])); + mockCheckProofStates.mockImplementation(async () => { + throw new Error('Network timeout'); + }); + + await service.recoverPendingOperations(); + + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('executing'); + }); + + it('never solo-re-executes a batch member through recoverExecutingOperation', async () => { + const member = makeBatchMember('op-a', 5, 1, ['a-out']); + await receiveOpRepo.create(member); + + await service.recoverExecutingOperation(member); + + expect(mockWalletReceive).not.toHaveBeenCalled(); + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('executing'); + }); +}); diff --git a/packages/core/test/unit/ReceiveOperationService.recovery.test.ts b/packages/core/test/unit/ReceiveOperationService.recovery.test.ts index 10ac812a1..7241932d0 100644 --- a/packages/core/test/unit/ReceiveOperationService.recovery.test.ts +++ b/packages/core/test/unit/ReceiveOperationService.recovery.test.ts @@ -3,6 +3,7 @@ import type { InitReceiveOperation, PreparedReceiveOperation, ExecutingReceiveOperation, + ReceiveOperation, } from '../../operations/receive/ReceiveOperation'; import { getOutputProofSecrets } from '../../operations/receive/ReceiveOperation'; import { EventBus } from '../../events/EventBus'; @@ -113,6 +114,9 @@ describe('ReceiveOperationService - recoverPendingOperations', () => { getWalletWithActiveKeysetId: mock(async () => ({ wallet: { unit: 'sat', + // High fee keeps deferred groups below viability so the redemption + // sweep at the end of recovery leaves them queued. + getFeesForProofs: mock(() => Amount.from(1000)), receive: mockWalletReceive, }, })), @@ -162,6 +166,21 @@ describe('ReceiveOperationService - recoverPendingOperations', () => { expect(stored?.state).toBe('prepared'); }); + it('leaves deferred operations untouched by the recovery sweep cleanup', async () => { + const proofs = [makeProof('p1')]; + const op = { + ...makeInitOp('deferred-op', proofs), + state: 'deferred', + deferredReason: 'dust', + } as ReceiveOperation; + await receiveOpRepo.create(op); + + await service.recoverPendingOperations(); + + const stored = await receiveOpRepo.getById(op.id); + expect(stored?.state).toBe('deferred'); + }); + it('retries executing operations when all inputs are unspent', async () => { const proofs = [makeProof('p1'), makeProof('p2')]; const op = makeExecutingOp('exec-op', proofs); diff --git a/packages/core/test/unit/ReceiveOperationService.redeemDeferred.test.ts b/packages/core/test/unit/ReceiveOperationService.redeemDeferred.test.ts new file mode 100644 index 000000000..413a20944 --- /dev/null +++ b/packages/core/test/unit/ReceiveOperationService.redeemDeferred.test.ts @@ -0,0 +1,448 @@ +import { Amount, OutputData, type Proof, type Token } from '@cashu/cashu-ts'; +import { describe, it, beforeEach, expect, mock, type Mock } from 'bun:test'; +import type { + DeferredReceiveOperation, + InitReceiveOperation, +} from '../../operations/receive/ReceiveOperation'; +import { EventBus } from '../../events/EventBus'; +import type { CoreEvents } from '../../events/types'; +import type { MintAdapter } from '../../infra/MintAdapter'; +import type { MintService } from '../../services/MintService'; +import type { ProofService } from '../../services/ProofService'; +import { TokenService } from '../../services/TokenService'; +import type { WalletService } from '../../services/WalletService'; +import type { CoreProof } from '../../types'; +import { MintOperationError, NetworkError } from '../../models/Error'; +import { computeYHexForSecrets } from '../../utils'; +import { ReceiveOperationService } from '../../operations/receive/ReceiveOperationService'; +import { MemoryProofRepository } from '../../repositories/memory/MemoryProofRepository'; +import { MemoryReceiveOperationRepository } from '../../repositories/memory/MemoryReceiveOperationRepository'; + +describe('ReceiveOperationService - redeemDeferred', () => { + const mintUrl = 'https://mint.test'; + const keysetId = 'keyset-1'; + const decoder = new TextDecoder(); + + let receiveOpRepo: MemoryReceiveOperationRepository; + let proofRepo: MemoryProofRepository; + let proofService: ProofService; + let mintService: MintService; + let walletService: WalletService; + let mintAdapter: MintAdapter; + let tokenService: TokenService; + let eventBus: EventBus; + let service: ReceiveOperationService; + + let mockWalletReceive: Mock<(...args: any[]) => Promise>; + let mockGetFees: Mock<(proofs: Proof[]) => Amount>; + let mockCheckProofStates: Mock<(mintUrl: string, ys: string[]) => Promise<{ state: string }[]>>; + let savedProofBatches: { mintUrl: string; proofs: CoreProof[] }[]; + let outputCounter: number; + + const makeProof = (secret: string, amount = 1): Proof => + ({ + id: keysetId, + amount: Amount.from(amount), + secret, + C: `C_${secret}`, + }) as Proof; + + const makeOutput = (secret: string, amount: Amount): OutputData => + new OutputData( + { amount, id: keysetId, B_: `B_${secret}` }, + BigInt(1), + new TextEncoder().encode(secret), + ); + + const makeDeferredOp = ( + id: string, + amount: number, + reason: DeferredReceiveOperation['deferredReason'] = 'dust', + ): DeferredReceiveOperation => ({ + id, + state: 'deferred', + deferredReason: reason, + mintUrl, + unit: 'sat', + amount: Amount.from(amount), + inputProofs: [makeProof(`${id}-input`, amount)], + createdAt: Date.now() - 10000, + updatedAt: Date.now() - 10000, + }); + + beforeEach(() => { + receiveOpRepo = new MemoryReceiveOperationRepository(); + proofRepo = new MemoryProofRepository(); + eventBus = new EventBus(); + savedProofBatches = []; + outputCounter = 0; + + // Echo the custom output data back as freshly signed proofs so the split + // by output secret can be asserted. + mockWalletReceive = mock( + async (_token: unknown, _config: unknown, outputType: { data: OutputData[] }) => + outputType.data.map( + (output) => + ({ + id: keysetId, + amount: output.blindedMessage.amount, + secret: decoder.decode(output.secret), + C: `C_${decoder.decode(output.secret)}`, + }) as Proof, + ), + ); + mockGetFees = mock(() => Amount.from(1)); + mockCheckProofStates = mock(async (_mintUrl: string, ys: string[]) => + ys.map(() => ({ state: 'UNSPENT' })), + ); + + walletService = { + getWalletWithActiveKeysetId: mock(async () => ({ + wallet: { + unit: 'sat', + getFeesForProofs: mockGetFees, + receive: mockWalletReceive, + }, + })), + } as unknown as WalletService; + + proofService = { + prepareProofsForReceiving: mock(async (proofs: Proof[]) => proofs), + createOutputsAndIncrementCounters: mock( + async (_mintUrl: string, intents: { keep: { amount: Amount } }) => ({ + keep: [makeOutput(`out-${outputCounter++}`, intents.keep.amount)], + send: [], + }), + ), + saveProofs: mock(async (targetMintUrl: string, proofs: CoreProof[]) => { + savedProofBatches.push({ mintUrl: targetMintUrl, proofs }); + await proofRepo.saveProofs(targetMintUrl, proofs); + }), + recoverProofsFromOutputData: mock(async () => []), + } as unknown as ProofService; + + mintAdapter = { checkProofStates: mockCheckProofStates } as unknown as MintAdapter; + + mintService = { + isTrustedMint: mock(async () => true), + ensureUpdatedMint: mock(async () => ({ + mint: { url: mintUrl }, + keysets: [{ id: keysetId }], + })), + } as unknown as MintService; + + tokenService = new TokenService(mintService); + + service = new ReceiveOperationService( + receiveOpRepo, + proofRepo, + proofService, + mintService, + walletService, + mintAdapter, + tokenService, + eventBus, + ); + }); + + it('settles a queued dust op together with an incoming receive in one swap', async () => { + // The issue #46 scenario: 1 sat queued dust + incoming 32 sat, fee 1. + await receiveOpRepo.create(makeDeferredOp('op-dust', 1)); + const finalizedEvents: CoreEvents['receive-op:finalized'][] = []; + eventBus.on('receive-op:finalized', (payload) => { + finalizedEvents.push(payload); + }); + + const proofs = [makeProof('incoming-input', 32)]; + const token: Token = { mint: mintUrl, proofs } as Token; + const initOp = (await service.init(token)) as InitReceiveOperation; + const result = await service.redeemDeferredGroup(mintUrl, 'sat', initOp); + + expect(result?.state).toBe('finalized'); + if (result?.state !== 'finalized') { + throw new Error('Expected finalized incoming operation'); + } + expect(result.amount).toEqual(Amount.from(32)); + expect(mockWalletReceive).toHaveBeenCalledTimes(1); + + const dust = await receiveOpRepo.getById('op-dust'); + expect(dust?.state).toBe('finalized'); + if (dust?.state === 'finalized') { + // Fee is charged to the largest member; dust keeps its full value. + expect(dust.fee).toEqual(Amount.from(0)); + } + expect(result.fee).toEqual(Amount.from(1)); + expect(dust?.batchId).toBeDefined(); + expect(dust?.batchId).toBe(result.batchId!); + + expect(finalizedEvents.length).toBe(2); + + // Each member's new proofs are attributed to its own operation. + const dustBatch = savedProofBatches.find((batch) => + batch.proofs.some((proof) => proof.createdByOperationId === 'op-dust'), + ); + const incomingBatch = savedProofBatches.find((batch) => + batch.proofs.some((proof) => proof.createdByOperationId === result.id), + ); + expect(dustBatch?.proofs[0]?.amount).toEqual(Amount.from(1)); + expect(incomingBatch?.proofs[0]?.amount).toEqual(Amount.from(31)); + }); + + it('defers the incoming receive too when the combined group stays below the fee', async () => { + await receiveOpRepo.create(makeDeferredOp('op-dust', 1)); + mockGetFees.mockImplementation(() => Amount.from(2)); + + const proofs = [makeProof('incoming-input', 1)]; + const token: Token = { mint: mintUrl, proofs } as Token; + const initOp = (await service.init(token)) as InitReceiveOperation; + const result = await service.redeemDeferredGroup(mintUrl, 'sat', initOp); + + expect(result?.state).toBe('deferred'); + if (result?.state === 'deferred') { + expect(result.deferredReason).toBe('dust'); + } + expect((await receiveOpRepo.getById('op-dust'))?.state).toBe('deferred'); + expect(mockWalletReceive).not.toHaveBeenCalled(); + }); + + it('redeems a viable deferred group from the sweep without an incoming receive', async () => { + await receiveOpRepo.create(makeDeferredOp('op-a', 5)); + await receiveOpRepo.create(makeDeferredOp('op-b', 4)); + + await service.redeemDeferred(); + + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('finalized'); + expect((await receiveOpRepo.getById('op-b'))?.state).toBe('finalized'); + expect(mockWalletReceive).toHaveBeenCalledTimes(1); + }); + + it('keeps members executing when the batched swap fails transiently', async () => { + await receiveOpRepo.create(makeDeferredOp('op-a', 5)); + await receiveOpRepo.create(makeDeferredOp('op-b', 4)); + mockWalletReceive.mockImplementation(async () => { + throw new NetworkError('network timeout'); + }); + + await service.redeemDeferred(); + + const a = await receiveOpRepo.getById('op-a'); + const b = await receiveOpRepo.getById('op-b'); + expect(a?.state).toBe('executing'); + expect(b?.state).toBe('executing'); + expect(a?.batchId).toBe(b?.batchId!); + }); + + it('returns unspent members to the queue on a terminal mint error', async () => { + await receiveOpRepo.create(makeDeferredOp('op-healthy', 5, 'mint-unreachable')); + await receiveOpRepo.create(makeDeferredOp('op-poisoned', 4)); + mockWalletReceive.mockImplementation(async () => { + throw new MintOperationError(11001, 'Proofs already spent'); + }); + // The poisoned member's inputs are spent at the mint, the healthy one's + // are not. The double-spend lands after the pre-batch validation, so the + // first (batched) state check still reports everything unspent. + const poisonedY = computeYHexForSecrets(['op-poisoned-input'])[0]!; + let preflightDone = false; + mockCheckProofStates.mockImplementation(async (_mintUrl: string, ys: string[]) => { + if (!preflightDone) { + preflightDone = true; + return ys.map((y) => ({ Y: y, state: 'UNSPENT' })); + } + return ys.map((y) => ({ Y: y, state: y === poisonedY ? 'SPENT' : 'UNSPENT' })); + }); + + await service.redeemDeferred(); + + const healthy = await receiveOpRepo.getById('op-healthy'); + expect(healthy?.state).toBe('deferred'); + if (healthy?.state === 'deferred') { + expect(healthy.deferredReason).toBe('mint-unreachable'); + } + // Spent member with no recoverable outputs rolls back. + expect((await receiveOpRepo.getById('op-poisoned'))?.state).toBe('rolled_back'); + }); + + it('rolls back queued members whose inputs were spent before batching', async () => { + await receiveOpRepo.create(makeDeferredOp('op-healthy', 5)); + await receiveOpRepo.create(makeDeferredOp('op-poisoned', 4)); + const poisonedY = computeYHexForSecrets(['op-poisoned-input'])[0]!; + mockCheckProofStates.mockImplementation(async (_mintUrl: string, ys: string[]) => + ys.map((y) => ({ Y: y, state: y === poisonedY ? 'SPENT' : 'UNSPENT' })), + ); + + await service.redeemDeferred(); + + // The batch swap is atomic, so the spent member must not poison it: it + // rolls back terminally and the healthy member settles without it. + expect((await receiveOpRepo.getById('op-poisoned'))?.state).toBe('rolled_back'); + expect((await receiveOpRepo.getById('op-healthy'))?.state).toBe('finalized'); + expect(mockWalletReceive).toHaveBeenCalledTimes(1); + const swapToken = mockWalletReceive.mock.calls[0]?.[0] as { proofs: Proof[] }; + expect(swapToken.proofs.map((proof) => proof.secret)).toEqual(['op-healthy-input']); + }); + + it('leaves queued members with pending inputs out of the batch', async () => { + await receiveOpRepo.create(makeDeferredOp('op-healthy', 5)); + await receiveOpRepo.create(makeDeferredOp('op-pending', 4)); + const pendingY = computeYHexForSecrets(['op-pending-input'])[0]!; + mockCheckProofStates.mockImplementation(async (_mintUrl: string, ys: string[]) => + ys.map((y) => ({ Y: y, state: y === pendingY ? 'PENDING' : 'UNSPENT' })), + ); + + await service.redeemDeferred(); + + expect((await receiveOpRepo.getById('op-pending'))?.state).toBe('deferred'); + expect((await receiveOpRepo.getById('op-healthy'))?.state).toBe('finalized'); + expect(mockWalletReceive).toHaveBeenCalledTimes(1); + }); + + it('falls back to a solo receive when a poisoned batch fails around a fresh token', async () => { + await receiveOpRepo.create(makeDeferredOp('op-poisoned', 4)); + // The pre-batch validation misses the double-spend that lands right + // before the swap; the batched swap fails terminally. + const poisonedY = computeYHexForSecrets(['op-poisoned-input'])[0]!; + let preflightDone = false; + mockCheckProofStates.mockImplementation(async (_mintUrl: string, ys: string[]) => { + if (!preflightDone) { + preflightDone = true; + return ys.map((y) => ({ Y: y, state: 'UNSPENT' })); + } + return ys.map((y) => ({ Y: y, state: y === poisonedY ? 'SPENT' : 'UNSPENT' })); + }); + let firstSwap = true; + mockWalletReceive.mockImplementation( + async (_token: unknown, _config: unknown, outputType: { data: OutputData[] }) => { + if (firstSwap) { + firstSwap = false; + throw new MintOperationError(11001, 'Token already spent'); + } + return outputType.data.map( + (output) => + ({ + id: keysetId, + amount: output.blindedMessage.amount, + secret: decoder.decode(output.secret), + C: `C_${decoder.decode(output.secret)}`, + }) as Proof, + ); + }, + ); + + const token: Token = { mint: mintUrl, proofs: [makeProof('incoming-input', 32)] } as Token; + const result = await service.receive(token); + + // The fresh token is receivable on its own, so the failed batch must not + // fail it or park it in the queue. + expect(result.state).toBe('finalized'); + if (result.state === 'finalized') { + expect(result.amount).toEqual(Amount.from(32)); + expect(result.batchId).toBeUndefined(); + } + expect(mockWalletReceive).toHaveBeenCalledTimes(2); + expect((await receiveOpRepo.getById('op-poisoned'))?.state).toBe('rolled_back'); + }); + + it('finalizes zero-keep members without saving proofs', async () => { + // Two 1-sat dust ops with a combined fee of 1: one member keeps zero. + await receiveOpRepo.create(makeDeferredOp('op-a', 1)); + await receiveOpRepo.create(makeDeferredOp('op-b', 1)); + + await service.redeemDeferred(); + + const a = await receiveOpRepo.getById('op-a'); + const b = await receiveOpRepo.getById('op-b'); + expect(a?.state).toBe('finalized'); + expect(b?.state).toBe('finalized'); + + const zeroKeep = [a, b].find( + (op) => op?.state === 'finalized' && op.fee.equals(Amount.from(1)), + ); + expect(zeroKeep).toBeDefined(); + const zeroKeepProofs = savedProofBatches.filter((batch) => + batch.proofs.some((proof) => proof.createdByOperationId === zeroKeep!.id), + ); + expect(zeroKeepProofs.length).toBe(0); + }); + + it('receive() drains the queue by batching with the incoming token', async () => { + await receiveOpRepo.create(makeDeferredOp('op-dust', 1)); + + const proofs = [makeProof('incoming-input', 32)]; + const token: Token = { mint: mintUrl, proofs } as Token; + const result = await service.receive(token); + + expect(result.state).toBe('finalized'); + expect(result.batchId).toBeDefined(); + expect((await receiveOpRepo.getById('op-dust'))?.state).toBe('finalized'); + expect(mockWalletReceive).toHaveBeenCalledTimes(1); + }); + + it('receive() defers the incoming token when the drained group is still dust', async () => { + await receiveOpRepo.create(makeDeferredOp('op-dust', 1)); + mockGetFees.mockImplementation(() => Amount.from(2)); + + const proofs = [makeProof('incoming-input', 1)]; + const token: Token = { mint: mintUrl, proofs } as Token; + const result = await service.receive(token); + + expect(result.state).toBe('deferred'); + expect((await receiveOpRepo.getById('op-dust'))?.state).toBe('deferred'); + }); + + it('receive() takes the solo path when the queue is empty', async () => { + let preparedEventCount = 0; + eventBus.on('receive-op:prepared', () => { + preparedEventCount += 1; + }); + mockGetFees.mockImplementation(() => Amount.zero()); + + const proofs = [makeProof('incoming-input', 32)]; + const token: Token = { mint: mintUrl, proofs } as Token; + const result = await service.receive(token); + + expect(result.state).toBe('finalized'); + expect(result.batchId).toBeUndefined(); + // The solo saga emits receive-op:prepared; the batch path does not. + expect(preparedEventCount).toBe(1); + }); + + it('recovery sweep redeems viable queued groups and leaves unreachable ones queued', async () => { + await receiveOpRepo.create(makeDeferredOp('op-a', 5)); + await receiveOpRepo.create(makeDeferredOp('op-b', 4)); + + await service.recoverPendingOperations(); + + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('finalized'); + expect((await receiveOpRepo.getById('op-b'))?.state).toBe('finalized'); + }); + + it('recovery sweep tolerates an unreachable mint', async () => { + await receiveOpRepo.create(makeDeferredOp('op-a', 5)); + (walletService.getWalletWithActiveKeysetId as Mock).mockImplementation(async () => { + throw new NetworkError('offline'); + }); + + await service.recoverPendingOperations(); + + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('deferred'); + }); + + it('skips deferred operations that are currently locked', async () => { + await receiveOpRepo.create(makeDeferredOp('op-a', 5)); + await receiveOpRepo.create(makeDeferredOp('op-busy', 4)); + const release = await ( + service as unknown as { + acquireOperationLock: (id: string) => Promise<() => void>; + } + ).acquireOperationLock('op-busy'); + + try { + await service.redeemDeferred(); + } finally { + release(); + } + + expect((await receiveOpRepo.getById('op-a'))?.state).toBe('finalized'); + expect((await receiveOpRepo.getById('op-busy'))?.state).toBe('deferred'); + }); +}); diff --git a/packages/core/test/unit/ReceiveOperationService.test.ts b/packages/core/test/unit/ReceiveOperationService.test.ts index 9bf654d57..5a2356bf6 100644 --- a/packages/core/test/unit/ReceiveOperationService.test.ts +++ b/packages/core/test/unit/ReceiveOperationService.test.ts @@ -16,6 +16,8 @@ import { TokenService } from '../../services/TokenService'; import type { WalletService } from '../../services/WalletService'; import { OutputData, type Proof, type Token } from '@cashu/cashu-ts'; import { + KeyPairNotFoundError, + MintFetchError, MintOperationError, NetworkError, ProofValidationError, @@ -74,6 +76,16 @@ describe('ReceiveOperationService', () => { checkProofStates: mock(() => Promise.resolve([])), }) as unknown as MintAdapter; + const prepareExpectingPrepared = async ( + op: InitReceiveOperation, + ): Promise => { + const result = await service.prepare(op); + if (result.state !== 'prepared') { + throw new Error(`Expected prepared operation, got '${result.state}'`); + } + return result; + }; + beforeEach(() => { receiveOpRepo = new MemoryReceiveOperationRepository(); proofRepo = new MemoryProofRepository(); @@ -159,7 +171,7 @@ describe('ReceiveOperationService', () => { const token: Token = { mint: mintUrl, proofs } as Token; const initOp = await service.init(token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); expect(prepared.state).toBe('prepared'); expect(prepared.fee).toEqual(Amount.from(0)); @@ -228,7 +240,7 @@ describe('ReceiveOperationService', () => { lockedDuringEvent = service.isOperationLocked(operationId); }); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); expect(prepared.state).toBe('prepared'); expect(persistedState).toBe('prepared'); @@ -239,7 +251,7 @@ describe('ReceiveOperationService', () => { const proofs = [makeProof('p1')]; const token: Token = { mint: mintUrl, proofs } as Token; const initOp = await service.init(token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); let persistedState: string | undefined; let lockedDuringEvent = false; @@ -259,7 +271,7 @@ describe('ReceiveOperationService', () => { const proofs = [makeProof('p1')]; const token: Token = { mint: mintUrl, proofs } as Token; const initOp = await service.init(token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); let persistedState: string | undefined; let lockedDuringEvent = false; @@ -279,7 +291,7 @@ describe('ReceiveOperationService', () => { const proofs = [makeProof('p1')]; const token: Token = { mint: mintUrl, proofs } as Token; const initOp = await service.init(token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); const historyRepo = new MemoryHistoryRepository({ receiveOperationRepository: receiveOpRepo, }); @@ -357,7 +369,7 @@ describe('ReceiveOperationService', () => { expect(service.prepare(initOp)).rejects.toThrow(ProofValidationError); }); - it('prepare throws when fees consume the full amount', async () => { + it('prepare defers the operation when fees consume the full amount', async () => { const proofs = [makeProof('p1')]; const token: Token = { mint: mintUrl, proofs } as Token; const initOp = await service.init(token); @@ -370,13 +382,25 @@ describe('ReceiveOperationService', () => { }, })); - expect(service.prepare(initOp)).rejects.toThrow(ProofValidationError); + const deferred = await service.prepare(initOp); + + expect(deferred.state).toBe('deferred'); + if (deferred.state === 'deferred') { + expect(deferred.deferredReason).toBe('dust'); + } }); - it('prepare throws ProofValidationError when fees exceed the amount', async () => { + it('prepare defers as dust and emits receive-op:deferred when fees exceed the amount', async () => { const proofs = [makeProof('p1')]; const token: Token = { mint: mintUrl, proofs } as Token; const initOp = await service.init(token); + let deferredEvent: CoreEvents['receive-op:deferred'] | undefined; + let persistedState: string | undefined; + + eventBus.on('receive-op:deferred', async (payload) => { + deferredEvent = payload; + persistedState = (await receiveOpRepo.getById(payload.operationId))?.state; + }); (walletService.getWalletWithActiveKeysetId as Mock).mockImplementation(async () => ({ wallet: { @@ -386,13 +410,99 @@ describe('ReceiveOperationService', () => { }, })); - try { - await service.prepare(initOp); - throw new Error('Expected prepare to reject'); - } catch (error) { - expect(error).toBeInstanceOf(ProofValidationError); - expect((error as Error).message).toBe('Receive amount is not sufficient after fees'); + const deferred = await service.prepare(initOp); + + expect(deferred.state).toBe('deferred'); + expect(persistedState).toBe('deferred'); + expect(deferredEvent?.operationId).toBe(initOp.id); + expect(deferredEvent?.operation.state).toBe('deferred'); + + const stored = await receiveOpRepo.getById(initOp.id); + expect(stored?.state).toBe('deferred'); + }); + + it('init rejects and persists nothing when a P2PK signing key is missing', async () => { + const proofs = [makeProof('p1')]; + const token: Token = { mint: mintUrl, proofs } as Token; + + (proofService.prepareProofsForReceiving as Mock).mockImplementation(async () => { + throw new KeyPairNotFoundError('02abc'); + }); + + await expect(service.init(token)).rejects.toBeInstanceOf(KeyPairNotFoundError); + expect(await receiveOpRepo.getByState('init')).toEqual([]); + expect(await receiveOpRepo.getByState('deferred')).toEqual([]); + }); + + it('init still rejects other signing failures', async () => { + const proofs = [makeProof('p1')]; + const token: Token = { mint: mintUrl, proofs } as Token; + + (proofService.prepareProofsForReceiving as Mock).mockImplementation(async () => { + throw new ProofValidationError('Multisig is not supported'); + }); + + await expect(service.init(token)).rejects.toThrow('Multisig is not supported'); + expect(await receiveOpRepo.getByState('init')).toEqual([]); + expect(await receiveOpRepo.getByState('deferred')).toEqual([]); + }); + + it('prepare defers the operation when the mint is unreachable', async () => { + const proofs = [makeProof('p1')]; + const token: Token = { mint: mintUrl, proofs } as Token; + const initOp = await service.init(token); + + (walletService.getWalletWithActiveKeysetId as Mock).mockImplementation(async () => { + throw new MintFetchError('Failed to fetch mint info'); + }); + + const deferred = await service.prepare(initOp); + + expect(deferred.state).toBe('deferred'); + if (deferred.state === 'deferred') { + expect(deferred.deferredReason).toBe('mint-unreachable'); } + expect((await receiveOpRepo.getById(initOp.id))?.state).toBe('deferred'); + }); + + it('receive() returns the deferred operation for a dust token', async () => { + const proofs = [makeProof('p1')]; + const token: Token = { mint: mintUrl, proofs } as Token; + + (walletService.getWalletWithActiveKeysetId as Mock).mockImplementation(async () => ({ + wallet: { + unit: 'sat', + getFeesForProofs: mock(() => Amount.from(10)), + receive: mockWalletReceive, + }, + })); + + const result = await service.receive(token); + + expect(result.state).toBe('deferred'); + expect(mockWalletReceive).not.toHaveBeenCalled(); + expect((proofService.saveProofs as Mock).mock.calls.length).toBe(0); + }); + + it('rollback deletes a deferred operation', async () => { + const proofs = [makeProof('p1')]; + const token: Token = { mint: mintUrl, proofs } as Token; + const initOp = await service.init(token); + + (walletService.getWalletWithActiveKeysetId as Mock).mockImplementation(async () => ({ + wallet: { + unit: 'sat', + getFeesForProofs: mock(() => initOp.amount), + receive: mockWalletReceive, + }, + })); + + const deferred = await service.prepare(initOp); + expect(deferred.state).toBe('deferred'); + + await service.rollback(deferred.id, 'User cancelled deferred receive'); + + expect(await receiveOpRepo.getById(deferred.id)).toBeNull(); }); it('prepare throws when deterministic outputs are empty', async () => { @@ -411,7 +521,7 @@ describe('ReceiveOperationService', () => { it('execute throws when outputData is missing', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); const brokenPrepared = { ...prepared, outputData: undefined, @@ -424,7 +534,7 @@ describe('ReceiveOperationService', () => { it('rolls back executing receive operations on terminal NUT-03 mint errors', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); let rolledBackEvent: CoreEvents['receive-op:rolled-back'] | undefined; eventBus.on('receive-op:rolled-back', (payload) => { @@ -447,7 +557,7 @@ describe('ReceiveOperationService', () => { it('rolls back executing receive operations on terminal NUT-03 keyset errors', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); mockWalletReceive.mockImplementation(async () => { throw new MintOperationError(12001, 'Keyset is not known'); @@ -463,7 +573,7 @@ describe('ReceiveOperationService', () => { it('rolls back executing receive operations on generic mint protocol errors', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); mockWalletReceive.mockImplementation(async () => { throw new MintOperationError(0, 'Keyset unknown'); @@ -484,7 +594,7 @@ describe('ReceiveOperationService', () => { new HistoryService(historyRepo, eventBus); const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); expect(await historyRepo.getReceiveHistoryEntry(mintUrl, prepared.id)).toBeNull(); @@ -502,7 +612,7 @@ describe('ReceiveOperationService', () => { it('keeps executing when receive fails with recovery-sensitive outputs already signed', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); mockWalletReceive.mockImplementation(async () => { throw new MintOperationError(11003, 'Outputs already signed'); @@ -521,7 +631,7 @@ describe('ReceiveOperationService', () => { it(`rolls back when receive fails with non-spendable NUT-03 state ${code}`, async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); mockWalletReceive.mockImplementation(async () => { throw new MintOperationError(code, message); @@ -538,7 +648,7 @@ describe('ReceiveOperationService', () => { it('keeps executing on local validation failures after the mint call', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); mockWalletReceive.mockImplementation(async () => { throw new ProofValidationError('Invalid signature in receive response'); @@ -555,7 +665,7 @@ describe('ReceiveOperationService', () => { it('keeps executing on transient receive failures', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); mockWalletReceive.mockImplementation(async () => { throw new NetworkError('network timeout'); @@ -570,7 +680,7 @@ describe('ReceiveOperationService', () => { it('finalize is idempotent on an already finalized operation', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); const executing = { ...prepared, state: 'executing', @@ -601,7 +711,7 @@ describe('ReceiveOperationService', () => { it('uses batched proof lookup when checking whether outputs were already saved', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); const executing = { ...prepared, state: 'executing', @@ -638,7 +748,7 @@ describe('ReceiveOperationService', () => { it('finalize throws when operation is not executing', async () => { const proofs = [makeProof('p1')]; const initOp = await service.init({ mint: mintUrl, proofs } as Token); - const prepared = await service.prepare(initOp); + const prepared = await prepareExpectingPrepared(initOp); expect(service.finalize(prepared.id)).rejects.toThrow('Cannot finalize operation'); }); diff --git a/packages/core/test/unit/ReceiveOpsApi.test.ts b/packages/core/test/unit/ReceiveOpsApi.test.ts index efffd0864..8b508b66f 100644 --- a/packages/core/test/unit/ReceiveOpsApi.test.ts +++ b/packages/core/test/unit/ReceiveOpsApi.test.ts @@ -2,6 +2,7 @@ import { Amount } from '@cashu/cashu-ts'; import { beforeEach, describe, expect, it, mock } from 'bun:test'; import type { Token } from '@cashu/cashu-ts'; import type { + DeferredReceiveOperation, FinalizedReceiveOperation, InitReceiveOperation, PreparedReceiveOperation, @@ -32,6 +33,7 @@ describe('ReceiveOpsApi', () => { let initOperation: InitReceiveOperation; let preparedOperation: PreparedReceiveOperation; let executingOperation: ReceiveOperation; + let deferredOperation: DeferredReceiveOperation; let finalizedOperation: FinalizedReceiveOperation; beforeEach(() => { @@ -50,6 +52,11 @@ describe('ReceiveOpsApi', () => { ...preparedOperation, state: 'executing', }; + deferredOperation = { + ...initOperation, + state: 'deferred', + deferredReason: 'dust', + }; finalizedOperation = { ...preparedOperation, state: 'finalized', @@ -62,7 +69,9 @@ describe('ReceiveOpsApi', () => { execute: mock(async () => finalizedOperation), getOperation: mock(async () => preparedOperation), getPreparedOperations: mock(async () => [preparedOperation]), + getDeferredOperations: mock(async () => [deferredOperation]), getPendingOperations: mock(async () => [executingOperation]), + redeemDeferred: mock(async () => {}), finalize: mock(async () => {}), recoverPendingOperations: mock(async () => {}), recoverExecutingOperation: mock(async () => {}), @@ -130,7 +139,41 @@ describe('ReceiveOpsApi', () => { ).mockResolvedValueOnce(finalizedOperation); await expect(api.cancel(finalizedOperation.id)).rejects.toThrow( - "Expected 'init' or 'prepared'", + "Expected 'init', 'prepared', or 'deferred'", ); }); + + it('cancel allows deferred operations', async () => { + ( + receiveOperationService.getOperation as unknown as ReturnType + ).mockResolvedValueOnce(deferredOperation); + + await api.cancel(deferredOperation.id, 'cancel queued receive'); + + expect(receiveOperationService.rollback).toHaveBeenCalledWith('op-1', 'cancel queued receive'); + }); + + it('listDeferred delegates to the service', async () => { + const deferred = await api.listDeferred(); + + expect(receiveOperationService.getDeferredOperations).toHaveBeenCalledWith(); + expect(deferred).toEqual([deferredOperation]); + }); + + it('redeemDeferred passes the filter through', async () => { + await api.redeemDeferred({ mintUrl, unit: 'sat' }); + + expect(receiveOperationService.redeemDeferred).toHaveBeenCalledWith({ mintUrl, unit: 'sat' }); + }); + + it('refresh returns deferred operations as-is without recovery', async () => { + ( + receiveOperationService.getOperation as unknown as ReturnType + ).mockResolvedValueOnce(deferredOperation); + + const result = await api.refresh(deferredOperation.id); + + expect(receiveOperationService.recoverExecutingOperation).not.toHaveBeenCalled(); + expect(result).toBe(deferredOperation); + }); }); diff --git a/packages/core/test/unit/TokenService.test.ts b/packages/core/test/unit/TokenService.test.ts index 9d9842fb5..a97a949b5 100644 --- a/packages/core/test/unit/TokenService.test.ts +++ b/packages/core/test/unit/TokenService.test.ts @@ -2,6 +2,7 @@ import { Amount, type Token } from '@cashu/cashu-ts'; import { describe, expect, it, mock } from 'bun:test'; import { TokenService } from '../../services/TokenService.ts'; import type { MintService } from '../../services/MintService.ts'; +import { MintFetchError, TokenValidationError } from '../../models/Error.ts'; describe('TokenService', () => { const mintUrl = 'https://mint.test'; @@ -30,4 +31,62 @@ describe('TokenService', () => { expect(decoded.unit).toBe('sat'); }); + + it('decodes with cached keysets when the mint refresh fails for a known mint', async () => { + const mintService = { + ensureUpdatedMint: mock(async () => { + throw new MintFetchError(mintUrl, 'Failed to fetch mint info'); + }), + getKnownMintWithKeysets: mock(async () => ({ + mint: { mintUrl }, + keysets: [{ id: 'keyset-1', mintUrl, unit: 'usd', active: true, feePpk: 0 }], + })), + } as unknown as MintService; + const service = new TokenService(mintService); + const token: Token = { + mint: mintUrl, + proofs: [ + { + id: 'keyset-1', + amount: Amount.from(1), + secret: 'secret-1', + C: 'C-1', + }, + ], + }; + + const decoded = await service.decodeToken(token, mintUrl); + + expect(decoded.unit).toBe('usd'); + }); + + it('preserves the mint fetch failure as cause when no cached keysets exist', async () => { + const fetchError = new MintFetchError(mintUrl, 'Failed to fetch mint info'); + const mintService = { + ensureUpdatedMint: mock(async () => { + throw fetchError; + }), + getKnownMintWithKeysets: mock(async () => null), + } as unknown as MintService; + const service = new TokenService(mintService); + const token: Token = { + mint: mintUrl, + proofs: [ + { + id: 'keyset-1', + amount: Amount.from(1), + secret: 'secret-1', + C: 'C-1', + }, + ], + }; + + try { + await service.decodeToken(token, mintUrl); + throw new Error('Expected decodeToken to reject'); + } catch (error) { + expect(error).toBeInstanceOf(TokenValidationError); + expect((error as { cause?: unknown }).cause).toBe(fetchError); + } + }); }); diff --git a/packages/core/test/unit/apportionFee.test.ts b/packages/core/test/unit/apportionFee.test.ts new file mode 100644 index 000000000..eb469ea51 --- /dev/null +++ b/packages/core/test/unit/apportionFee.test.ts @@ -0,0 +1,95 @@ +import { Amount } from '@cashu/cashu-ts'; +import { describe, expect, it } from 'bun:test'; +import { + apportionReceiveFee, + type ApportionableReceive, +} from '../../operations/receive/apportionFee'; + +describe('apportionReceiveFee', () => { + const op = (id: string, amount: number): ApportionableReceive => ({ + id, + amount: Amount.from(amount), + }); + + const sumShares = (shares: Map) => ({ + fee: Amount.sum([...shares.values()].map((share) => share.feeShare)), + keep: Amount.sum([...shares.values()].map((share) => share.keepAmount)), + }); + + it('charges a single operation the whole fee', async () => { + const shares = apportionReceiveFee([op('a', 10)], Amount.from(1)); + + expect(shares.get('a')?.feeShare).toEqual(Amount.from(1)); + expect(shares.get('a')?.keepAmount).toEqual(Amount.from(9)); + }); + + it('charges the largest member first (queued dust + incoming token)', async () => { + // The user scenario from issue #46: a queued 1-sat dust proof batched with + // an incoming 32-sat token at a combined fee of 1 sat. + const shares = apportionReceiveFee([op('dust', 1), op('incoming', 32)], Amount.from(1)); + + expect(shares.get('incoming')?.feeShare).toEqual(Amount.from(1)); + expect(shares.get('incoming')?.keepAmount).toEqual(Amount.from(31)); + expect(shares.get('dust')?.feeShare).toEqual(Amount.from(0)); + expect(shares.get('dust')?.keepAmount).toEqual(Amount.from(1)); + }); + + it('spreads a fee larger than the largest member across several members', async () => { + const shares = apportionReceiveFee([op('a', 3), op('b', 2), op('c', 2)], Amount.from(4)); + + expect(shares.get('a')?.feeShare).toEqual(Amount.from(3)); + expect(shares.get('b')?.feeShare).toEqual(Amount.from(1)); + expect(shares.get('c')?.feeShare).toEqual(Amount.from(0)); + + const { fee, keep } = sumShares(shares); + expect(fee).toEqual(Amount.from(4)); + expect(keep).toEqual(Amount.from(3)); + }); + + it('allows members to keep zero when the fee consumes them', async () => { + const shares = apportionReceiveFee([op('a', 1), op('b', 1)], Amount.from(1)); + + expect(shares.get('a')?.keepAmount).toEqual(Amount.from(0)); + expect(shares.get('b')?.keepAmount).toEqual(Amount.from(1)); + }); + + it('preserves the invariants for many dust members', async () => { + const ops = Array.from({ length: 10 }, (_, i) => op(`dust-${i}`, 1)); + const shares = apportionReceiveFee(ops, Amount.from(1)); + + const { fee, keep } = sumShares(shares); + expect(fee).toEqual(Amount.from(1)); + expect(keep).toEqual(Amount.from(9)); + for (const share of shares.values()) { + expect(share.feeShare.lessThanOrEqual(Amount.from(1))).toBe(true); + } + }); + + it('is deterministic under input reordering', async () => { + const ops = [op('b', 2), op('a', 2), op('c', 7)]; + const shares = apportionReceiveFee(ops, Amount.from(3)); + const reordered = apportionReceiveFee([...ops].reverse(), Amount.from(3)); + + for (const [id, share] of shares) { + expect(reordered.get(id)?.feeShare).toEqual(share.feeShare); + expect(reordered.get(id)?.keepAmount).toEqual(share.keepAmount); + } + // Ties on amount are broken by id: 'a' pays before 'b'. + expect(shares.get('c')?.feeShare).toEqual(Amount.from(3)); + expect(shares.get('a')?.feeShare).toEqual(Amount.from(0)); + }); + + it('throws when the fee exceeds the combined amount', async () => { + expect(() => apportionReceiveFee([op('a', 1), op('b', 1)], Amount.from(3))).toThrow( + 'exceeds combined receive amount', + ); + }); + + it('handles a zero fee and empty input', async () => { + const shares = apportionReceiveFee([op('a', 5)], Amount.zero()); + expect(shares.get('a')?.keepAmount).toEqual(Amount.from(5)); + + expect(apportionReceiveFee([], Amount.zero()).size).toBe(0); + expect(() => apportionReceiveFee([], Amount.from(1))).toThrow('across zero operations'); + }); +}); diff --git a/packages/docs/pages/receive-operations.md b/packages/docs/pages/receive-operations.md index 92819141c..7bb82d99c 100644 --- a/packages/docs/pages/receive-operations.md +++ b/packages/docs/pages/receive-operations.md @@ -9,64 +9,144 @@ token details, recover after crashes, and avoid duplicate receives. The canonical API is exposed through `coco.ops.receive`: - `prepare({ token })` decodes and validates a token, calculates fees, and - creates deterministic receive outputs + creates deterministic receive outputs; returns a `deferred` operation instead + when the receive cannot be settled yet - `execute(operationOrId)` receives the prepared token and saves the new proofs - `get(operationId)` returns a persisted receive operation - `listPrepared()` lists receives waiting for user confirmation -- `listInFlight()` lists receives that may need recovery +- `listDeferred()` lists receives queued for later redemption +- `redeemDeferred(filter?)` attempts to redeem queued receives now, batched per + mint and unit +- `listInFlight()` lists receives that may need recovery (executing or deferred) - `refresh(operationId)` recovers an executing receive and returns the latest operation state -- `cancel(operationId, reason?)` rolls back an `init` or `prepared` receive +- `cancel(operationId, reason?)` rolls back an `init`, `prepared`, or `deferred` + receive ## Operation States Receive operations progress through the following states: -| State | Description | -| ------------- | ------------------------------------------------------------- | -| `init` | Token decoded and validated, but outputs are not prepared yet | -| `prepared` | Fees calculated, output data persisted, ready to execute | -| `executing` | Receive request is in progress at the mint | -| `finalized` | New proofs were saved locally | -| `rolled_back` | Operation was cancelled or could not be recovered | +| State | Description | +| ------------- | --------------------------------------------------------------------------------------- | +| `init` | Token decoded and validated, but outputs are not prepared yet | +| `prepared` | Fees calculated, output data persisted, ready to execute | +| `executing` | Receive request is in progress at the mint | +| `deferred` | Redemption postponed until it can be settled fee-efficiently or its prerequisites exist | +| `finalized` | New proofs were saved locally | +| `rolled_back` | Operation was cancelled or could not be recovered | ``` init -> prepared -> executing -> finalized + | | | + | | +-> deferred (batch member returned to queue) | | | +--------+-------------+-> rolled_back + | + +-> deferred -> executing (batch redemption) ``` ## Lifecycle Actions | Action | Valid input state | Resulting state | Use when | | ------------------------------ | ---------------------------------- | ------------------------------------------ | ------------------------------------------------------------------------ | -| `prepare({ token })` | none | `prepared` | You want to inspect token amount, mint, unit, and fees before receiving. | +| `prepare({ token })` | none | `prepared` or `deferred` | You want to inspect token amount, mint, unit, and fees before receiving. | | `execute(operationOrId)` | `prepared` | `finalized` | The user confirmed the receive and proofs should be saved. | +| `redeemDeferred(filter?)` | `deferred` | `finalized` per redeemed member | Connectivity returned or the user asks to retry queued receives. | | `refresh(operationId)` | any, actively recovers `executing` | latest stored state | You are resuming an operation after a crash or stale UI state. | -| `cancel(operationId, reason?)` | `init`, `prepared` | `rolled_back` or deleted when still `init` | The user abandons the receive before it completes. | +| `cancel(operationId, reason?)` | `init`, `prepared`, `deferred` | `rolled_back` or deleted when not prepared | The user abandons the receive before it completes. | ## Prepare -> Execute Flow ```ts const prepared = await coco.ops.receive.prepare({ token }); -console.log('Amount:', prepared.amount); -console.log('Mint:', prepared.mintUrl); -console.log('Fee:', prepared.fee); - -if (userConfirmed) { - const finalized = await coco.ops.receive.execute(prepared.id); - console.log('Received:', finalized.amount); +if (prepared.state === 'deferred') { + console.log('Receive queued:', prepared.deferredReason); } else { - await coco.ops.receive.cancel(prepared.id, 'User cancelled receive'); + console.log('Amount:', prepared.amount); + console.log('Mint:', prepared.mintUrl); + console.log('Fee:', prepared.fee); + + if (userConfirmed) { + const finalized = await coco.ops.receive.execute(prepared.id); + console.log('Received:', finalized.amount); + } else { + await coco.ops.receive.cancel(prepared.id, 'User cancelled receive'); + } } ``` +## Deferred Receives + +Some tokens cannot be settled at the moment they arrive. Instead of failing, +coco queues them as `deferred` operations with a `deferredReason`: + +- `dust` — the token's value does not cover the swap fee on its own + (NUT-02: `fee = ceil(sum(input_fee_ppk) / 1000)`, so a lone 1-sat proof at + 100 ppk would leave zero outputs) +- `mint-unreachable` — mint or keyset data could not be fetched (e.g. offline) + +### Batch Redemption + +Deferred receives are redeemed in batches per mint and unit. A batch settles +with **one** swap whose single fee is apportioned across the members +(largest first), so dust that could never pay its own fee rides along with +larger receives. Every member still finalizes as its own operation with its +own `receive-op:finalized` event and history entry. + +Redemption is attempted automatically: + +1. when a new receive arrives for the same mint and unit — the incoming token + drains the queue by batching with it (this is how queued dust becomes + redeemable), +2. at the end of the receive recovery sweep (`initializeCoco()` startup and + `coco.ops.receive.recovery.run()`), and +3. explicitly via `coco.ops.receive.redeemDeferred()`. + +Because the batch swap is atomic, queued members are validated with the mint +before each attempt: members whose inputs were spent in the meantime (e.g. the +sender double-spent a queued token) roll back terminally instead of poisoning +every future batch. A fresh receive that batched with the queue is never +failed by the queue — when the batch cannot settle, the incoming token falls +back to a solo receive. + +Groups whose combined value stays at or below the combined fee remain queued. +A future configuration may additionally hold redemption until the fee ceiling +(`floor(1000 / input_fee_ppk)` inputs per fee unit) is better utilized. + +> **Design note.** Issue [#46](https://github.com/cashubtc/coco/issues/46) +> sketched a separate `receive_later` table. Deferred receives are modeled as a +> state of the receive operation saga instead: the saga already provides +> durable persistence, crash recovery, locking, events, and the one-operation → +> one-history-entry projection that keeps batched redemptions independently +> auditable. + +```ts +const queued = await coco.ops.receive.listDeferred(); +console.log( + 'Queued:', + queued.map((op) => `${op.amount} (${op.deferredReason})`), +); + +// e.g. when connectivity returns: +await coco.ops.receive.redeemDeferred(); +``` + ## Recovery `initializeCoco()` runs receive recovery automatically. Recovery removes stale -`init` operations, leaves `prepared` operations for user decision, and tries to -complete or roll back `executing` operations based on mint state. +`init` operations, leaves `prepared` operations for user decision, tries to +complete or roll back `executing` operations based on mint state, and finishes +by attempting to redeem queued `deferred` operations. + +Interrupted batch redemptions recover as a group: when the batch inputs were +spent the members restore from their own output data, and when they were not +the combined swap is re-executed. A batch member is never re-executed alone +because its fee share only balances inside its batch; recovery verifies the +stored outputs still satisfy the swap equation (e.g. after a crash between +persisting members, or a keyset fee change) and requeues the members for a +fresh batch when they do not. Use `refresh(operationId)` for explicit recovery UI: @@ -89,6 +169,12 @@ coco.on('receive-op:prepared', ({ operationId, operation }) => { console.log('Receive prepared', operationId, operation.amount); }); +coco.on('receive-op:deferred', ({ operationId, operation }) => { + if (operation.state === 'deferred') { + console.log('Receive queued', operationId, operation.deferredReason); + } +}); + coco.on('receive-op:finalized', ({ operationId, operation }) => { console.log('Receive finalized', operationId, operation.amount); }); diff --git a/packages/docs/starting/sending-receiving.md b/packages/docs/starting/sending-receiving.md index 1d339c943..d3428f30e 100644 --- a/packages/docs/starting/sending-receiving.md +++ b/packages/docs/starting/sending-receiving.md @@ -11,7 +11,10 @@ passed as either an encoded string or a parsed `Token` object: ```ts const prepared = await coco.ops.receive.prepare({ token: 'cashuBpGF0gaJhaUgA...' }); -if (userConfirmed) { +if (prepared.state === 'deferred') { + // Queued for later redemption (dust below the swap fee, or unreachable + // mint); coco redeems it automatically once it can. +} else if (userConfirmed) { await coco.ops.receive.execute(prepared.id); } else { await coco.ops.receive.cancel(prepared.id); @@ -20,7 +23,11 @@ if (userConfirmed) { > **Note:** The mint must be trusted before receiving tokens. See [Adding a Mint](./adding-mints.md). -For the full receive lifecycle, see [Receive Operations](../pages/receive-operations.md). +Tokens that cannot be settled yet (for example a token too small to pay its own +swap fee, or one received while offline) become **deferred receives** that are +redeemed later, batched with other queued proofs of the same mint and unit. See +[Receive Operations](../pages/receive-operations.md) for the full lifecycle and +the deferred redemption rules. ### Events @@ -30,6 +37,10 @@ You can listen for receive events: coco.on('receive-op:finalized', ({ mintUrl, operation }) => { console.log(`Received ${operation.amount} ${operation.unit} from ${mintUrl}`); }); + +coco.on('receive-op:deferred', ({ mintUrl, operation }) => { + console.log(`Queued ${operation.amount} ${operation.unit} from ${mintUrl}`); +}); ``` ## Sending Tokens diff --git a/packages/indexeddb/src/lib/db.ts b/packages/indexeddb/src/lib/db.ts index 81dc5c70e..53ab54349 100644 --- a/packages/indexeddb/src/lib/db.ts +++ b/packages/indexeddb/src/lib/db.ts @@ -217,7 +217,7 @@ export interface ReceiveOperationRow { mintUrl: string; unit?: string | null; amount: string | number; - state: 'init' | 'prepared' | 'executing' | 'finalized' | 'rolled_back'; + state: 'init' | 'prepared' | 'executing' | 'deferred' | 'finalized' | 'rolled_back'; createdAt: number; updatedAt: number; error?: string | null; @@ -225,6 +225,8 @@ export interface ReceiveOperationRow { inputProofsJson?: string | null; outputDataJson?: string | null; sourceJson?: string | null; + deferredReason?: 'dust' | 'mint-unreachable' | null; + batchId?: string | null; } export interface PaymentRequestReceiveOperationRow { diff --git a/packages/indexeddb/src/repositories/ReceiveOperationRepository.ts b/packages/indexeddb/src/repositories/ReceiveOperationRepository.ts index d3c1a6992..c9240b23e 100644 --- a/packages/indexeddb/src/repositories/ReceiveOperationRepository.ts +++ b/packages/indexeddb/src/repositories/ReceiveOperationRepository.ts @@ -35,12 +35,17 @@ function rowToOperation(row: ReceiveOperationRow): ReceiveOperation { updatedAt: row.updatedAt * 1000, error: row.error ?? undefined, source: row.sourceJson ? JSON.parse(row.sourceJson) : undefined, + batchId: row.batchId ?? undefined, }; if (row.state === 'init') { return { ...base, state: 'init' }; } + if (row.state === 'deferred') { + return { ...base, state: 'deferred', deferredReason: row.deferredReason ?? 'dust' }; + } + const preparedData = { fee: deserializeAmount(assertFieldPresent(row.fee, 'fee', row.id)), outputData: row.outputDataJson ? JSON.parse(row.outputDataJson) : undefined, @@ -64,7 +69,7 @@ function operationToRow(op: ReceiveOperation): ReceiveOperationRow { const createdAtSeconds = Math.floor(op.createdAt / 1000); const updatedAtSeconds = Math.floor(op.updatedAt / 1000); - if (op.state === 'init') { + if (op.state === 'init' || op.state === 'deferred') { return { id: op.id, mintUrl: op.mintUrl, @@ -78,6 +83,8 @@ function operationToRow(op: ReceiveOperation): ReceiveOperationRow { inputProofsJson: JSON.stringify(op.inputProofs), outputDataJson: null, sourceJson: op.source ? JSON.stringify(op.source) : null, + deferredReason: op.state === 'deferred' ? op.deferredReason : null, + batchId: op.batchId ?? null, }; } @@ -94,6 +101,8 @@ function operationToRow(op: ReceiveOperation): ReceiveOperationRow { inputProofsJson: JSON.stringify(op.inputProofs), outputDataJson: op.outputData ? JSON.stringify(op.outputData) : null, sourceJson: op.source ? JSON.stringify(op.source) : null, + deferredReason: null, + batchId: op.batchId ?? null, }; } @@ -148,7 +157,7 @@ export class IdbReceiveOperationRepository implements ReceiveOperationRepository const rows = (await (this.db as any) .table('coco_cashu_receive_operations') .where('state') - .anyOf(['executing']) + .anyOf(['executing', 'deferred']) .toArray()) as ReceiveOperationRow[]; return rows.map(rowToOperation); } diff --git a/packages/react/src/lib/hooks/operationHooks.test.tsx b/packages/react/src/lib/hooks/operationHooks.test.tsx index 923e32eec..3cb9fb553 100644 --- a/packages/react/src/lib/hooks/operationHooks.test.tsx +++ b/packages/react/src/lib/hooks/operationHooks.test.tsx @@ -44,6 +44,7 @@ type PendingSendOperationRecord = SendExecuteResult['operation']; type ReceiveOps = Manager['ops']['receive']; type ReceivePrepareResult = Awaited>; +type PreparedReceiveRecord = Extract; type ReceiveExecuteResult = Awaited>; type ReceiveOperationRecord = NonNullable>>; @@ -129,6 +130,8 @@ function createReceiveManagerMock() { execute: vi.fn(), get: vi.fn(), listPrepared: vi.fn(), + listDeferred: vi.fn(), + redeemDeferred: vi.fn(), listInFlight: vi.fn(), refresh: vi.fn(), cancel: vi.fn(), @@ -270,8 +273,8 @@ function createSendExecuteResult(overrides: Partial = {}): Se } function createPreparedReceiveOperation( - overrides: Partial = {}, -): ReceivePrepareResult { + overrides: Partial = {}, +): PreparedReceiveRecord { return { id: 'receive-op-1', state: 'prepared', @@ -282,7 +285,7 @@ function createPreparedReceiveOperation( createdAt: 1_700_000_000_000, updatedAt: 1_700_000_000_000, fee: Amount.zero(), - outputData: {} as ReceivePrepareResult['outputData'], + outputData: {} as PreparedReceiveRecord['outputData'], ...overrides, }; } @@ -844,6 +847,37 @@ describe('useReceiveOperation', () => { expect(result.current.executeResult).toEqual(finalized); }); + it('binds a deferred prepare result and passes deferred list and redemption through', async () => { + const { manager, receive } = createReceiveManagerMock(); + const deferred = { + ...createInitReceiveOperation({ id: 'receive-op-deferred' }), + state: 'deferred', + deferredReason: 'dust', + } as ReceiveOperationRecord; + + receive.prepare.mockResolvedValue(deferred); + receive.listDeferred.mockResolvedValue([deferred]); + receive.redeemDeferred.mockResolvedValue(undefined); + + const { result } = renderHook(() => useReceiveOperation(), { + wrapper: createHookWrapper(manager), + }); + + await act(async () => { + await result.current.prepare(RECEIVE_PREPARE_INPUT); + }); + + expect(result.current.currentOperation).toEqual(deferred); + + await act(async () => { + expect(await result.current.listDeferred()).toEqual([deferred]); + await result.current.redeemDeferred({ mintUrl: MINT_URL }); + }); + + expect(receive.listDeferred).toHaveBeenCalled(); + expect(receive.redeemDeferred).toHaveBeenCalledWith({ mintUrl: MINT_URL }); + }); + it('accepts an initial operation-id binding, synchronizes after cancel, and surfaces errors after reset', async () => { const { manager, receive } = createReceiveManagerMock(); const loaded = createPreparedReceiveOperation({ id: 'receive-op-load' }); diff --git a/packages/react/src/lib/hooks/useReceiveOperation.ts b/packages/react/src/lib/hooks/useReceiveOperation.ts index 76d2f59ec..d39fba9b5 100644 --- a/packages/react/src/lib/hooks/useReceiveOperation.ts +++ b/packages/react/src/lib/hooks/useReceiveOperation.ts @@ -18,6 +18,9 @@ type ReceiveOps = Manager['ops']['receive']; export type ReceiveOperationPrepareInput = Parameters[0]; export type ReceiveOperationPrepareResult = Awaited>; export type ReceiveOperationExecuteResult = Awaited>; +export type ReceiveOperationListPreparedResult = Awaited>; +export type ReceiveOperationListDeferredResult = Awaited>; +export type ReceiveOperationRedeemDeferredFilter = Parameters[0]; export interface UseReceiveOperationResult extends OperationHookResult< ReceiveOperation, @@ -26,7 +29,9 @@ export interface UseReceiveOperationResult extends OperationHookResult< prepare(input: ReceiveOperationPrepareInput): Promise; execute(): Promise; cancel(): Promise; - listPrepared(): Promise; + listPrepared(): Promise; + listDeferred(): Promise; + redeemDeferred(filter?: ReceiveOperationRedeemDeferredFilter): Promise; listInFlight(): Promise; } @@ -118,6 +123,9 @@ export function useReceiveOperation( const unsubscribePrepared = manager.on('receive-op:prepared', ({ operation }) => { handleObservedOperation(operation); }); + const unsubscribeDeferred = manager.on('receive-op:deferred', ({ operation }) => { + handleObservedOperation(operation); + }); const unsubscribeFinalized = manager.on('receive-op:finalized', ({ operation }) => { handleObservedOperation(operation); }); @@ -127,6 +135,7 @@ export function useReceiveOperation( return () => { unsubscribePrepared(); + unsubscribeDeferred(); unsubscribeFinalized(); unsubscribeRolledBack(); }; @@ -187,7 +196,10 @@ export function useReceiveOperation( return; } - if (operationBeforeCancel?.state === 'init') { + if ( + operationBeforeCancel?.state === 'init' || + operationBeforeCancel?.state === 'deferred' + ) { bindOperation(null, { clearExecuteResult: true }); return; } @@ -197,10 +209,21 @@ export function useReceiveOperation( ); }, [bindOperation, getCurrentOperation, manager, runStatefulAction]); - const listPrepared = useCallback(async (): Promise => { + const listPrepared = useCallback(async (): Promise => { return manager.ops.receive.listPrepared(); }, [manager]); + const listDeferred = useCallback(async (): Promise => { + return manager.ops.receive.listDeferred(); + }, [manager]); + + const redeemDeferred = useCallback( + async (filter?: ReceiveOperationRedeemDeferredFilter): Promise => { + return manager.ops.receive.redeemDeferred(filter); + }, + [manager], + ); + const listInFlight = useCallback(async (): Promise => { return manager.ops.receive.listInFlight(); }, [manager]); @@ -222,6 +245,8 @@ export function useReceiveOperation( execute, cancel, listPrepared, + listDeferred, + redeemDeferred, listInFlight, reset: resetBoundOperation, }; diff --git a/packages/sql-storage/src/repositories/ReceiveOperationRepository.ts b/packages/sql-storage/src/repositories/ReceiveOperationRepository.ts index cf6093bb8..b508644c6 100644 --- a/packages/sql-storage/src/repositories/ReceiveOperationRepository.ts +++ b/packages/sql-storage/src/repositories/ReceiveOperationRepository.ts @@ -1,4 +1,5 @@ import type { + DeferredReceiveReason, ReceiveOperationRepository, ReceiveOperation, ReceiveOperationState, @@ -24,6 +25,8 @@ interface ReceiveOperationRow { inputProofsJson: string | null; outputDataJson: string | null; sourceJson: string | null; + deferredReason: DeferredReceiveReason | null; + batchId: string | null; } function parseInputProofs(inputProofsJson: string | null): ReceiveOperation['inputProofs'] { @@ -47,12 +50,17 @@ function rowToOperation(row: ReceiveOperationRow): ReceiveOperation { updatedAt: row.updatedAt * 1000, error: row.error ?? undefined, source: row.sourceJson ? JSON.parse(row.sourceJson) : undefined, + batchId: row.batchId ?? undefined, }; if (row.state === 'init') { return { ...base, state: 'init' }; } + if (row.state === 'deferred') { + return { ...base, state: 'deferred', deferredReason: row.deferredReason ?? 'dust' }; + } + const preparedData = { fee: deserializeAmount(assertFieldPresent(row.fee, 'fee', row.id)), outputData: row.outputDataJson ? JSON.parse(row.outputDataJson) : undefined, @@ -76,7 +84,7 @@ function operationToParams(op: ReceiveOperation): SqlValue[] { const createdAtSeconds = Math.floor(op.createdAt / 1000); const updatedAtSeconds = Math.floor(op.updatedAt / 1000); - if (op.state === 'init') { + if (op.state === 'init' || op.state === 'deferred') { return [ op.id, op.mintUrl, @@ -90,6 +98,8 @@ function operationToParams(op: ReceiveOperation): SqlValue[] { JSON.stringify(op.inputProofs), null, op.source ? JSON.stringify(op.source) : null, + op.state === 'deferred' ? op.deferredReason : null, + op.batchId ?? null, ]; } @@ -106,6 +116,8 @@ function operationToParams(op: ReceiveOperation): SqlValue[] { JSON.stringify(op.inputProofs), op.outputData ? JSON.stringify(op.outputData) : null, op.source ? JSON.stringify(op.source) : null, + null, + op.batchId ?? null, ]; } @@ -128,8 +140,8 @@ export class SqliteReceiveOperationRepository implements ReceiveOperationReposit const params = operationToParams(operation); await this.db.run( `INSERT INTO coco_cashu_receive_operations - (id, mintUrl, unit, amount, state, createdAt, updatedAt, error, fee, inputProofsJson, outputDataJson, sourceJson) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + (id, mintUrl, unit, amount, state, createdAt, updatedAt, error, fee, inputProofsJson, outputDataJson, sourceJson, deferredReason, batchId) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, params, ); } @@ -145,10 +157,10 @@ export class SqliteReceiveOperationRepository implements ReceiveOperationReposit const updatedAtSeconds = getUnixTimeSeconds(); - if (operation.state === 'init') { + if (operation.state === 'init' || operation.state === 'deferred') { await this.db.run( `UPDATE coco_cashu_receive_operations - SET state = ?, updatedAt = ?, error = ?, unit = ?, inputProofsJson = ?, sourceJson = ? + SET state = ?, updatedAt = ?, error = ?, unit = ?, fee = NULL, inputProofsJson = ?, outputDataJson = NULL, sourceJson = ?, deferredReason = ?, batchId = ? WHERE id = ?`, [ operation.state, @@ -157,13 +169,15 @@ export class SqliteReceiveOperationRepository implements ReceiveOperationReposit getOperationUnit(operation), JSON.stringify(operation.inputProofs), operation.source ? JSON.stringify(operation.source) : null, + operation.state === 'deferred' ? operation.deferredReason : null, + operation.batchId ?? null, operation.id, ], ); } else { await this.db.run( `UPDATE coco_cashu_receive_operations - SET state = ?, updatedAt = ?, error = ?, unit = ?, fee = ?, inputProofsJson = ?, outputDataJson = ?, sourceJson = ? + SET state = ?, updatedAt = ?, error = ?, unit = ?, fee = ?, inputProofsJson = ?, outputDataJson = ?, sourceJson = ?, deferredReason = NULL, batchId = ? WHERE id = ?`, [ operation.state, @@ -174,6 +188,7 @@ export class SqliteReceiveOperationRepository implements ReceiveOperationReposit JSON.stringify(operation.inputProofs), operation.outputData ? JSON.stringify(operation.outputData) : null, operation.source ? JSON.stringify(operation.source) : null, + operation.batchId ?? null, operation.id, ], ); @@ -198,7 +213,7 @@ export class SqliteReceiveOperationRepository implements ReceiveOperationReposit async getPending(): Promise { const rows = await this.db.all( - "SELECT * FROM coco_cashu_receive_operations WHERE state IN ('executing')", + "SELECT * FROM coco_cashu_receive_operations WHERE state IN ('executing', 'deferred')", ); return rows.map(rowToOperation); } diff --git a/packages/sql-storage/src/schema.ts b/packages/sql-storage/src/schema.ts index 8a83120f1..3d0bb7a0a 100644 --- a/packages/sql-storage/src/schema.ts +++ b/packages/sql-storage/src/schema.ts @@ -1446,6 +1446,49 @@ const MIGRATIONS: readonly Migration[] = [ ON coco_cashu_melt_quotes(mintUrl, quoteId); `, }, + { + // SQLite cannot alter a CHECK constraint, so widening the receive state + // union to include 'deferred' requires a table rebuild (same pattern as 024). + id: '037_receive_operations_deferred', + sql: ` + ALTER TABLE coco_cashu_receive_operations RENAME TO coco_cashu_receive_operations_legacy_states; + + CREATE TABLE coco_cashu_receive_operations ( + id TEXT PRIMARY KEY, + mintUrl TEXT NOT NULL, + amount TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('init', 'prepared', 'executing', 'deferred', 'finalized', 'rolled_back')), + createdAt INTEGER NOT NULL, + updatedAt INTEGER NOT NULL, + error TEXT, + fee TEXT, + inputProofsJson TEXT NOT NULL, + outputDataJson TEXT, + unit TEXT NOT NULL DEFAULT 'sat', + sourceJson TEXT, + deferredReason TEXT, + batchId TEXT + ); + + INSERT INTO coco_cashu_receive_operations ( + id, mintUrl, amount, state, createdAt, updatedAt, error, fee, + inputProofsJson, outputDataJson, unit, sourceJson + ) + SELECT + id, mintUrl, amount, state, createdAt, updatedAt, error, fee, + inputProofsJson, outputDataJson, unit, sourceJson + FROM coco_cashu_receive_operations_legacy_states; + + DROP TABLE coco_cashu_receive_operations_legacy_states; + + CREATE INDEX IF NOT EXISTS idx_coco_cashu_receive_operations_state + ON coco_cashu_receive_operations(state); + CREATE INDEX IF NOT EXISTS idx_coco_cashu_receive_operations_mint + ON coco_cashu_receive_operations(mintUrl); + CREATE INDEX IF NOT EXISTS idx_coco_cashu_receive_operations_createdAt + ON coco_cashu_receive_operations(createdAt DESC, id DESC); + `, + }, ]; // Export for testing diff --git a/packages/sql-storage/src/test/schema.test.ts b/packages/sql-storage/src/test/schema.test.ts index 293e64d39..d4a5d6516 100644 --- a/packages/sql-storage/src/test/schema.test.ts +++ b/packages/sql-storage/src/test/schema.test.ts @@ -43,6 +43,7 @@ const EXPECTED_MIGRATION_IDS = [ '034_clean_unquoted_mint_operations', '035_duplicate_quote_ids', '036_quote_identity_unique_indexes', + '037_receive_operations_deferred', ] as const; const RECEIVE_OPERATIONS_SQL = `