From 382efc10730948762fae288be3ccf4f47c8d0adc Mon Sep 17 00:00:00 2001 From: Emelie-Dev Date: Fri, 28 Aug 2026 10:38:56 +0100 Subject: [PATCH] Make transaction sequence allocation restart-safe --- .../0023_transaction_sequences.down.sql | 1 + migrations/0023_transaction_sequences.sql | 9 +++ src/services/postgresSequenceStore.test.ts | 35 +++++++++++ src/services/postgresSequenceStore.ts | 32 ++++++++++ src/services/sequenceManager.ts | 60 ++++++++++++++++++- src/services/transactionBuilder.test.ts | 32 ++++++++++ src/services/transactionBuilder.ts | 52 ++++++++++++++++ 7 files changed, 219 insertions(+), 2 deletions(-) create mode 100644 migrations/0023_transaction_sequences.down.sql create mode 100644 migrations/0023_transaction_sequences.sql create mode 100644 src/services/postgresSequenceStore.test.ts create mode 100644 src/services/postgresSequenceStore.ts diff --git a/migrations/0023_transaction_sequences.down.sql b/migrations/0023_transaction_sequences.down.sql new file mode 100644 index 00000000..92618627 --- /dev/null +++ b/migrations/0023_transaction_sequences.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS transaction_sequences; diff --git a/migrations/0023_transaction_sequences.sql b/migrations/0023_transaction_sequences.sql new file mode 100644 index 00000000..2848fe01 --- /dev/null +++ b/migrations/0023_transaction_sequences.sql @@ -0,0 +1,9 @@ +CREATE TABLE IF NOT EXISTS transaction_sequences ( + account_id TEXT PRIMARY KEY, + next_sequence NUMERIC(38, 0) NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_transaction_sequences_updated_at + ON transaction_sequences (updated_at); diff --git a/src/services/postgresSequenceStore.test.ts b/src/services/postgresSequenceStore.test.ts new file mode 100644 index 00000000..e6ec8f7a --- /dev/null +++ b/src/services/postgresSequenceStore.test.ts @@ -0,0 +1,35 @@ +import assert from 'node:assert/strict'; +import { PostgresSequenceStore } from './postgresSequenceStore.js'; +import type { Queryable } from '../db.js'; + +describe('PostgresSequenceStore', () => { + test('uses an atomic upsert that advances from the greater of durable and ledger state', async () => { + const query = jest.fn().mockResolvedValue({ rows: [{ allocated_sequence: '43' }] }); + const store = new PostgresSequenceStore({ query } as unknown as Queryable); + + const allocated = await store.allocate('GACCOUNT', 42n); + + assert.equal(allocated, 43n); + assert.match(query.mock.calls[0]?.[0], /ON CONFLICT \(account_id\) DO UPDATE/); + assert.match(query.mock.calls[0]?.[0], /GREATEST\(transaction_sequences\.next_sequence, \$2::numeric\) \+ 1/); + assert.deepEqual(query.mock.calls[0]?.[1], ['GACCOUNT', '42']); + }); + + test('surfaces partial database failures as retryable allocation failures to callers', async () => { + const store = new PostgresSequenceStore({ + query: jest.fn().mockRejectedValue(new Error('database unavailable')), + } as unknown as Queryable); + + await expect(store.allocate('GACCOUNT', 42n)).rejects.toThrow('database unavailable'); + }); + + test('fails explicitly when the database write commits no returned sequence', async () => { + const store = new PostgresSequenceStore({ + query: jest.fn().mockResolvedValue({ rows: [] }), + } as unknown as Queryable); + + await expect(store.allocate('GACCOUNT', 42n)).rejects.toThrow( + 'Sequence allocation did not return a reserved value' + ); + }); +}); diff --git a/src/services/postgresSequenceStore.ts b/src/services/postgresSequenceStore.ts new file mode 100644 index 00000000..cff5dc12 --- /dev/null +++ b/src/services/postgresSequenceStore.ts @@ -0,0 +1,32 @@ +import { writeQuery, type Queryable } from '../db.js'; +import type { SequenceStore } from './sequenceManager.js'; + +interface SequenceRow { + allocated_sequence: string; +} + +export class PostgresSequenceStore implements SequenceStore { + constructor(private readonly db: Queryable = { query: writeQuery }) {} + + async allocate(accountId: string, ledgerNextSequence: bigint): Promise { + const { rows } = await this.db.query( + ` + INSERT INTO transaction_sequences (account_id, next_sequence) + VALUES ($1, ($2::numeric + 1)) + ON CONFLICT (account_id) DO UPDATE + SET + next_sequence = GREATEST(transaction_sequences.next_sequence, $2::numeric) + 1, + updated_at = NOW() + RETURNING (next_sequence - 1)::text AS allocated_sequence + `, + [accountId, ledgerNextSequence.toString()] + ); + + const allocatedSequence = rows[0]?.allocated_sequence; + if (!allocatedSequence) { + throw new Error('Sequence allocation did not return a reserved value'); + } + + return BigInt(allocatedSequence); + } +} diff --git a/src/services/sequenceManager.ts b/src/services/sequenceManager.ts index 3ecc967a..998da8cd 100644 --- a/src/services/sequenceManager.ts +++ b/src/services/sequenceManager.ts @@ -58,6 +58,21 @@ export interface HorizonAccountLoader { export interface SequenceManagerOptions { /** Horizon loader used to fetch account objects. */ loader: HorizonAccountLoader; + /** Optional durable allocator used to make allocations survive restarts. */ + store?: SequenceStore; + /** Timeout for the durable allocation step. Defaults to 5 seconds. */ + allocationTimeoutMs?: number; +} + +export interface SequenceStore { + allocate(accountId: string, ledgerNextSequence: bigint): Promise; +} + +export class SequenceAllocationError extends Error { + constructor(message: string) { + super(message); + this.name = 'SequenceAllocationError'; + } } // ── SequenceManager ──────────────────────────────────────────────────────────── @@ -88,9 +103,13 @@ export class SequenceManager { /** Horizon loader injected at construction time. */ private readonly loader: HorizonAccountLoader; + private readonly store?: SequenceStore; + private readonly allocationTimeoutMs: number; constructor(options: SequenceManagerOptions) { this.loader = options.loader; + this.store = options.store; + this.allocationTimeoutMs = options.allocationTimeoutMs ?? 5_000; } /** @@ -135,8 +154,14 @@ export class SequenceManager { // Horizon returns sequence as a decimal string; parse to bigint for // exact arithmetic (sequence numbers can exceed Number.MAX_SAFE_INTEGER // on very active accounts). - const sequence = BigInt(account.sequence) + 1n; - return sequence; + const ledgerNextSequence = BigInt(account.sequence) + 1n; + if (!this.store) { + return ledgerNextSequence; + } + + return await this.withAllocationTimeout( + this.store.allocate(accountId, ledgerNextSequence) + ); } finally { // Always release the lock, even if loadAccount() threw. resolveSlot(); @@ -161,4 +186,35 @@ export class SequenceManager { hasLock(accountId: string): boolean { return this.locks.has(accountId); } + + private async withAllocationTimeout(allocation: Promise): Promise { + let timeoutId: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + allocation, + new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject( + new SequenceAllocationError( + 'Timed out while reserving a durable Stellar sequence number' + ) + ); + }, this.allocationTimeoutMs); + }), + ]); + } catch (error) { + if (error instanceof SequenceAllocationError) { + throw error; + } + throw new SequenceAllocationError( + `Failed to reserve a durable Stellar sequence number: ${ + error instanceof Error && error.message.trim() ? error.message : 'Unknown error' + }` + ); + } finally { + if (timeoutId) { + clearTimeout(timeoutId); + } + } + } } diff --git a/src/services/transactionBuilder.test.ts b/src/services/transactionBuilder.test.ts index b5024f1e..a59e4230 100644 --- a/src/services/transactionBuilder.test.ts +++ b/src/services/transactionBuilder.test.ts @@ -174,6 +174,38 @@ describe('TransactionBuilderService', () => { assert.equal(mockAddMemo.mock.calls.length, 0); }); + test('uses a durable reserved sequence when a sequence store is configured', async () => { + const sequenceStore = { + allocate: jest.fn().mockResolvedValue(9n), + }; + const service = new TransactionBuilderService({ sequenceStore }); + + await service.buildDepositTransaction({ + userPublicKey: 'GUSERPUBLICKEY123', + vaultContractId: 'CVAULTTEST', + amountUsdc: '12.3456789', + }); + + assert.deepEqual(sequenceStore.allocate.mock.calls[0], ['GUSERPUBLICKEY123', 2n]); + assert.equal(mockBuild.mock.calls[0]?.[0].sourceAccount.sequence, '8'); + }); + + test('returns an explicit recoverable network error when durable sequence allocation fails', async () => { + const service = new TransactionBuilderService({ + sequenceStore: { + allocate: jest.fn().mockRejectedValue(new Error('database unavailable')), + }, + }); + + await expect( + service.buildDepositTransaction({ + userPublicKey: 'GUSERPUBLICKEY123', + vaultContractId: 'CVAULTTEST', + amountUsdc: '12.3456789', + }) + ).rejects.toThrow(NetworkError); + }); + test('uses the correct network passphrase from configuration', async () => { const service = new TransactionBuilderService(); diff --git a/src/services/transactionBuilder.ts b/src/services/transactionBuilder.ts index 4cd69326..9a977fa2 100644 --- a/src/services/transactionBuilder.ts +++ b/src/services/transactionBuilder.ts @@ -12,6 +12,8 @@ import { extractSimulationDetails, type SimulationDetails, } from '../lib/simulationDiagnostics.js'; +import { PostgresSequenceStore } from './postgresSequenceStore.js'; +import { SequenceAllocationError, type SequenceStore } from './sequenceManager.js'; export type StellarNetwork = 'testnet' | 'mainnet'; @@ -137,6 +139,8 @@ interface NormalizedMemo { export interface TransactionBuilderServiceOptions { createServer?: (horizonUrl: string) => HorizonAccountLoader; + sequenceStore?: SequenceStore | false; + sequenceAllocationTimeoutMs?: number; baseFee?: string | number; timeoutSeconds?: number; /** Maximum number of retries for transient Horizon errors. Default: 3. */ @@ -210,6 +214,21 @@ export class TransactionBuilderService { shouldRetry: isHorizonTransientError, } ); + const sequenceStore = + this.options.sequenceStore === undefined + ? process.env.NODE_ENV === 'test' + ? undefined + : new PostgresSequenceStore() + : this.options.sequenceStore || undefined; + + if (sequenceStore) { + const accountSequence = (sourceAccount as { sequence?: unknown }).sequence; + const ledgerNextSequence = BigInt(String(accountSequence)) + 1n; + const reservedSequence = await this.withSequenceAllocationTimeout( + sequenceStore.allocate(sourceKey, ledgerNextSequence) + ); + (sourceAccount as { sequence: string }).sequence = (reservedSequence - 1n).toString(); + } } catch (error) { throw this.mapLoadAccountError(sourceKey, error); } @@ -408,6 +427,10 @@ export class TransactionBuilderService { } private mapLoadAccountError(accountId: string, error: unknown): Error { + if (error instanceof SequenceAllocationError) { + return new NetworkError(error.message); + } + const message = this.getErrorMessage(error).toLowerCase(); if ( @@ -434,4 +457,33 @@ export class TransactionBuilderService { return 'Unknown error'; } + + private async withSequenceAllocationTimeout(allocation: Promise): Promise { + let timeoutId: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + allocation, + new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject( + new SequenceAllocationError( + 'Timed out while reserving a durable Stellar sequence number' + ) + ); + }, this.options.sequenceAllocationTimeoutMs ?? 5_000); + }), + ]); + } catch (error) { + if (error instanceof SequenceAllocationError) { + throw error; + } + throw new SequenceAllocationError( + `Failed to reserve a durable Stellar sequence number: ${this.getErrorMessage(error)}` + ); + } finally { + if (timeoutId) { + clearTimeout(timeoutId); + } + } + } }