fix(core): make memory transactions rollback-safe

.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.

packages/core/repositories/memory/MemoryRepositories.ts

@@ -37,6 +37,62 @@ import {
   MemoryPaymentRequestReceiveOperationRepository,
 } from './MemoryPaymentRequestReceiveRepository';
 
+type MutableContainer = Map<unknown, unknown> | unknown[];
+
+interface MutableContainerSnapshot {
+  repository: object;
+  property: string;
+  value: MutableContainer;
+}
+
+function cloneTransactionValue<T>(value: T, seen = new Map<object, unknown>()): 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<string, unknown>;
+  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<void> = Promise.resolve();
+  private activeOperations = 0;
+  private operationsDrained: Promise<void> = Promise.resolve();
+  private releaseOperationsDrained?: () => void;
 
   constructor() {
     this.mintRepository = new MemoryMintRepository();
@@ -85,13 +146,111 @@ 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<void> {
     // No-op: Memory repositories don't require initialization
   }
 
   async withTransaction<T>(fn: (repos: RepositoryTransactionScope) => Promise<T>): Promise<T> {
-    return fn(this);
+    const previousTransaction = this.transactionTail;
+    let releaseTransaction!: () => void;
+    const currentTransaction = new Promise<void>((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<T extends object>(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<T>(fn: () => T | Promise<T>): Promise<T> {
+    while (true) {
+      const transaction = this.transactionTail;
+      await transaction;
+      if (transaction === this.transactionTail) break;
+    }
+
+    if (this.activeOperations === 0) {
+      this.operationsDrained = new Promise<void>((resolve) => {
+        this.releaseOperationsDrained = resolve;
+      });
+    }
+    this.activeOperations += 1;
+    try {
+      return await fn();
+    } finally {
+      this.activeOperations -= 1;
+      if (this.activeOperations === 0) {
+        this.releaseOperationsDrained?.();
+        this.releaseOperationsDrained = undefined;
+      }
+    }
   }
 }

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<void>((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]);
+  });
+});