diff --git a/.changeset/safe-memory-transactions.md b/.changeset/safe-memory-transactions.md new file mode 100644 index 000000000..7bf3b4a5b --- /dev/null +++ b/.changeset/safe-memory-transactions.md @@ -0,0 +1,6 @@ +--- +'@cashu/coco-core': patch +--- + +Make in-memory repository transactions commit and roll back atomically while preserving stable +repository references. diff --git a/packages/core/repositories/memory/MemoryRepositories.ts b/packages/core/repositories/memory/MemoryRepositories.ts index 65b72e67f..382dd41da 100644 --- a/packages/core/repositories/memory/MemoryRepositories.ts +++ b/packages/core/repositories/memory/MemoryRepositories.ts @@ -37,6 +37,62 @@ import { MemoryPaymentRequestReceiveOperationRepository, } from './MemoryPaymentRequestReceiveRepository'; +type MutableContainer = Map | unknown[]; + +interface MutableContainerSnapshot { + repository: object; + property: string; + value: MutableContainer; +} + +function cloneTransactionValue(value: T, seen = new Map()): T { + if (typeof value !== 'object' || value === null) return value; + if (seen.has(value)) return seen.get(value) as T; + if (value instanceof Uint8Array) return value.slice() as T; + if (value instanceof Map) { + const clone = new Map(); + seen.set(value, clone); + for (const [key, item] of value) { + clone.set(cloneTransactionValue(key, seen), cloneTransactionValue(item, seen)); + } + return clone as T; + } + if (Array.isArray(value)) { + const clone: unknown[] = []; + seen.set(value, clone); + clone.push(...value.map((item) => cloneTransactionValue(item, seen))); + return clone as T; + } + const clone = Object.create(Object.getPrototypeOf(value)) as Record; + seen.set(value, clone); + for (const [key, item] of Object.entries(value)) { + clone[key] = cloneTransactionValue(item, seen); + } + return clone as T; +} + +function snapshotMutableContainers(repositories: object[]): MutableContainerSnapshot[] { + const snapshots: MutableContainerSnapshot[] = []; + for (const repository of new Set(repositories)) { + for (const [property, value] of Object.entries(repository)) { + if (value instanceof Map || Array.isArray(value)) { + snapshots.push({ + repository, + property, + value: cloneTransactionValue(value), + }); + } + } + } + return snapshots; +} + +function restoreMutableContainers(snapshots: MutableContainerSnapshot[]): void { + for (const snapshot of snapshots) { + Reflect.set(snapshot.repository, snapshot.property, cloneTransactionValue(snapshot.value)); + } +} + export class MemoryRepositories implements Repositories { mintRepository: MintRepository; keyRingRepository: KeyRingRepository; @@ -54,6 +110,11 @@ export class MemoryRepositories implements Repositories { receiveOperationRepository: ReceiveOperationRepository; paymentRequestReceiveOperationRepository: PaymentRequestReceiveOperationRepository; paymentRequestReceiveAttemptRepository: PaymentRequestReceiveAttemptRepository; + private readonly transactionScope: RepositoryTransactionScope; + private transactionTail: Promise = Promise.resolve(); + private activeOperations = 0; + private operationsDrained: Promise = Promise.resolve(); + private releaseOperationsDrained?: () => void; constructor() { this.mintRepository = new MemoryMintRepository(); @@ -85,6 +146,46 @@ export class MemoryRepositories implements Repositories { new MemoryPaymentRequestReceiveOperationRepository(); this.paymentRequestReceiveAttemptRepository = new MemoryPaymentRequestReceiveAttemptRepository(); + + this.transactionScope = { + mintRepository: this.mintRepository, + keyRingRepository: this.keyRingRepository, + counterRepository: this.counterRepository, + keysetRepository: this.keysetRepository, + proofRepository: this.proofRepository, + mintQuoteRepository: this.mintQuoteRepository, + legacyMintQuoteRepository: this.legacyMintQuoteRepository, + meltQuoteRepository: this.meltQuoteRepository, + historyRepository: this.historyRepository, + sendOperationRepository: this.sendOperationRepository, + meltOperationRepository: this.meltOperationRepository, + authSessionRepository: this.authSessionRepository, + mintOperationRepository: this.mintOperationRepository, + receiveOperationRepository: this.receiveOperationRepository, + paymentRequestReceiveOperationRepository: this.paymentRequestReceiveOperationRepository, + paymentRequestReceiveAttemptRepository: this.paymentRequestReceiveAttemptRepository, + }; + + this.mintRepository = this.serializeRepository(this.mintRepository); + this.keyRingRepository = this.serializeRepository(this.keyRingRepository); + this.counterRepository = this.serializeRepository(this.counterRepository); + this.keysetRepository = this.serializeRepository(this.keysetRepository); + this.proofRepository = this.serializeRepository(this.proofRepository); + this.mintQuoteRepository = this.serializeRepository(this.mintQuoteRepository); + this.legacyMintQuoteRepository = this.serializeRepository(this.legacyMintQuoteRepository); + this.meltQuoteRepository = this.serializeRepository(this.meltQuoteRepository); + this.historyRepository = this.serializeRepository(this.historyRepository); + this.sendOperationRepository = this.serializeRepository(this.sendOperationRepository); + this.meltOperationRepository = this.serializeRepository(this.meltOperationRepository); + this.authSessionRepository = this.serializeRepository(this.authSessionRepository); + this.mintOperationRepository = this.serializeRepository(this.mintOperationRepository); + this.receiveOperationRepository = this.serializeRepository(this.receiveOperationRepository); + this.paymentRequestReceiveOperationRepository = this.serializeRepository( + this.paymentRequestReceiveOperationRepository, + ); + this.paymentRequestReceiveAttemptRepository = this.serializeRepository( + this.paymentRequestReceiveAttemptRepository, + ); } async init(): Promise { @@ -92,6 +193,64 @@ export class MemoryRepositories implements Repositories { } async withTransaction(fn: (repos: RepositoryTransactionScope) => Promise): Promise { - return fn(this); + const previousTransaction = this.transactionTail; + let releaseTransaction!: () => void; + const currentTransaction = new Promise((resolve) => { + releaseTransaction = resolve; + }); + this.transactionTail = previousTransaction.then(() => currentTransaction); + + await previousTransaction; + await this.operationsDrained; + try { + const snapshots = snapshotMutableContainers( + Object.values(this.transactionScope).filter( + (value): value is object => typeof value === 'object', + ), + ); + try { + return await fn(this.transactionScope); + } catch (error) { + restoreMutableContainers(snapshots); + throw error; + } + } finally { + releaseTransaction(); + } + } + + private serializeRepository(repository: T): T { + return new Proxy(repository, { + get: (target, property, receiver) => { + const value = Reflect.get(target, property, receiver); + if (typeof value !== 'function') return value; + return (...args: unknown[]) => + this.runRepositoryOperation(() => Reflect.apply(value, target, args) as unknown); + }, + }); + } + + private async runRepositoryOperation(fn: () => T | Promise): Promise { + while (true) { + const transaction = this.transactionTail; + await transaction; + if (transaction === this.transactionTail) break; + } + + if (this.activeOperations === 0) { + this.operationsDrained = new Promise((resolve) => { + this.releaseOperationsDrained = resolve; + }); + } + this.activeOperations += 1; + try { + return await fn(); + } finally { + this.activeOperations -= 1; + if (this.activeOperations === 0) { + this.releaseOperationsDrained?.(); + this.releaseOperationsDrained = undefined; + } + } } } diff --git a/packages/core/test/unit/MemoryRepositories.test.ts b/packages/core/test/unit/MemoryRepositories.test.ts new file mode 100644 index 000000000..626a6acb0 --- /dev/null +++ b/packages/core/test/unit/MemoryRepositories.test.ts @@ -0,0 +1,124 @@ +import { describe, expect, it } from 'bun:test'; +import type { Mint } from '../../models/Mint'; +import { MemoryRepositories } from '../../repositories/memory'; + +function createMint(mintUrl: string, trusted = true): Mint { + return { + mintUrl, + name: 'Test Mint', + mintInfo: { + name: 'Test Mint', + pubkey: 'pubkey', + version: '1.0', + contact: {}, + nuts: {}, + } as Mint['mintInfo'], + trusted, + createdAt: 0, + updatedAt: 0, + }; +} + +function createDeferred() { + let resolve!: () => void; + const promise = new Promise((resolver) => { + resolve = resolver; + }); + return { promise, resolve } as const; +} + +describe('MemoryRepositories transactions', () => { + it('commits staged creates, updates, and deletes across repositories', async () => { + const repositories = new MemoryRepositories(); + const mintRepository = repositories.mintRepository; + const authSessionRepository = repositories.authSessionRepository; + const updatedMint = createMint('https://updated-mint.test', false); + const createdMint = createMint('https://created-mint.test'); + const deletedSession = { + mintUrl: 'https://session-mint.test', + accessToken: 'access-token', + expiresAt: 1_730_000_000, + }; + + await mintRepository.addOrUpdateMint(updatedMint); + await authSessionRepository.saveSession(deletedSession); + + const result = await repositories.withTransaction(async (transaction) => { + await transaction.mintRepository.addOrUpdateMint(createdMint); + await transaction.mintRepository.setMintTrusted(updatedMint.mintUrl, true); + await transaction.authSessionRepository.deleteSession(deletedSession.mintUrl); + return 'committed'; + }); + + expect(result).toBe('committed'); + expect(repositories.mintRepository).toBe(mintRepository); + expect(repositories.authSessionRepository).toBe(authSessionRepository); + expect(await mintRepository.getAllMints()).toHaveLength(2); + expect((await mintRepository.getMintByUrl(updatedMint.mintUrl)).trusted).toBe(true); + expect(await mintRepository.getMintByUrl(createdMint.mintUrl)).toEqual(createdMint); + expect(await authSessionRepository.getSession(deletedSession.mintUrl)).toBeNull(); + }); + + it('rolls back staged creates, updates, and deletes across repositories', async () => { + const repositories = new MemoryRepositories(); + const mintRepository = repositories.mintRepository; + const authSessionRepository = repositories.authSessionRepository; + const updatedMint = createMint('https://updated-mint.test', false); + const deletedSession = { + mintUrl: 'https://session-mint.test', + accessToken: 'access-token', + expiresAt: 1_730_000_000, + }; + + await mintRepository.addOrUpdateMint(updatedMint); + await authSessionRepository.saveSession(deletedSession); + + await expect( + repositories.withTransaction(async (transaction) => { + await transaction.mintRepository.addOrUpdateMint(createMint('https://created-mint.test')); + await transaction.mintRepository.setMintTrusted(updatedMint.mintUrl, true); + await transaction.authSessionRepository.deleteSession(deletedSession.mintUrl); + throw new Error('rollback transaction'); + }), + ).rejects.toThrow('rollback transaction'); + + expect(repositories.mintRepository).toBe(mintRepository); + expect(repositories.authSessionRepository).toBe(authSessionRepository); + expect(await mintRepository.getAllMints()).toHaveLength(1); + expect((await mintRepository.getMintByUrl(updatedMint.mintUrl)).trusted).toBe(false); + expect(await authSessionRepository.getSession(deletedSession.mintUrl)).toEqual(deletedSession); + }); + + it('does not include concurrent root writes in a transaction rollback', async () => { + const repositories = new MemoryRepositories(); + const transactionEntered = createDeferred(); + const releaseTransaction = createDeferred(); + const transactionMint = createMint('https://transaction-mint.test'); + const outsideMint = createMint('https://outside-mint.test'); + + const transactionResult = repositories + .withTransaction(async (transaction) => { + await transaction.mintRepository.addOrUpdateMint(transactionMint); + transactionEntered.resolve(); + await releaseTransaction.promise; + throw new Error('rollback transaction'); + }) + .catch((error: unknown) => error); + + await transactionEntered.promise; + + let outsideWriteResolved = false; + const outsideWrite = repositories.mintRepository.addOrUpdateMint(outsideMint).then(() => { + outsideWriteResolved = true; + }); + await Promise.resolve(); + + expect(outsideWriteResolved).toBe(false); + + releaseTransaction.resolve(); + expect(await transactionResult).toBeInstanceOf(Error); + await outsideWrite; + + expect(await repositories.mintRepository.getAllMints()).toEqual([outsideMint]); + }); +});