[#1499] Make milestone release and redirect operations atomic

db/migrations/20260830000000_create_settlement_operations.cjs

@@ -0,0 +1,54 @@
+/**
+ * Durable state for milestone release/redirect operations.
+ *
+ * The operation identity is supplied by the caller and is unique per
+ * milestone. This lets a retried request recover the original operation
+ * instead of creating a second payout attempt.
+ */
+exports.up = async function up(knex) {
+  await knex.schema.createTable('settlement_operations', (table) => {
+    table.uuid('id').primary().defaultTo(knex.raw('gen_random_uuid()'))
+    table.string('milestone_id', 64).notNullable()
+      .references('id').inTable('milestones').onDelete('CASCADE')
+    table.string('operation_key', 255).notNullable()
+    table
+      .enu('operation_type', ['release', 'redirect'], {
+        useNative: true,
+        enumName: 'settlement_operation_type',
+      })
+      .notNullable()
+    table
+      .enu('status', ['pending', 'submitted', 'confirmed', 'failed'], {
+        useNative: true,
+        enumName: 'settlement_operation_status',
+      })
+      .notNullable()
+      .defaultTo('pending')
+    table.integer('attempt_count').notNullable().defaultTo(0)
+    table.string('transaction_hash', 128).nullable()
+    table.string('failure_code', 64).nullable()
+    table.text('failure_message').nullable()
+    table.string('requested_by', 255).notNullable()
+    table.jsonb('request_fingerprint').notNullable()
+    table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+    table.timestamp('updated_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+    table.timestamp('submitted_at', { useTz: true }).nullable()
+    table.timestamp('confirmed_at', { useTz: true }).nullable()
+
+    table.unique(['milestone_id', 'operation_key'], {
+      indexName: 'idx_settlement_operations_identity',
+    })
+    table.unique(['transaction_hash'], {
+      indexName: 'idx_settlement_operations_transaction_hash',
+    })
+    table.index(['milestone_id', 'status'], 'idx_settlement_operations_milestone_status')
+    table.index(['status', 'updated_at'], 'idx_settlement_operations_recovery')
+  })
+}
+
+exports.down = async function down(knex) {
+  await knex.schema.dropTableIfExists('settlement_operations')
+  await knex.raw('DROP TYPE IF EXISTS settlement_operation_status')
+  await knex.raw('DROP TYPE IF EXISTS settlement_operation_type')
+}
+

docs/settlement-operations.md

@@ -0,0 +1,88 @@
+# Atomic milestone settlement operations
+
+Milestone releases and redirects are represented by a durable settlement
+operation. The operation is the unit of retry, not the HTTP request and not a
+new blockchain transaction on every retry.
+
+## State machine
+
+```text
+pending ──submit──> submitted ──confirm──> confirmed
+   ▲                    │
+   │                    └──fail──> failed ──submit──> submitted
+   └──────────────────────────────────────────────────
+```
+
+`confirmed` is terminal. A release or redirect cannot become complete merely
+because a worker constructed a transaction or because the submission endpoint
+returned. Only a callback carrying the exact submitted transaction hash can
+advance `submitted` to `confirmed`.
+
+## Operation identity
+
+Each operation has:
+
+- `milestone_id` and caller-provided `operation_key` as a unique logical
+  identity;
+- `operation_type`, either `release` or `redirect`;
+- a SHA-256 request fingerprint covering the milestone, type, destination,
+  amount, and asset;
+- `attempt_count`, submitted/confirmed timestamps, and operator-visible
+  failure details.
+
+Reusing an operation key with the same request returns the existing operation.
+Reusing it with a changed payout destination, amount, or type returns an
+identity conflict. This prevents an accidental retry from becoming a second
+settlement or from changing the beneficiary under an old key.
+
+## Atomic database behavior
+
+`settlement_operations` has a unique `(milestone_id, operation_key)` index.
+Submission, failure recording, and confirmation each run in a transaction,
+lock the operation row, and use a conditional status update. Concurrent
+workers therefore observe one of these safe outcomes:
+
+| Concurrent action | Result |
+| --- | --- |
+| same pending operation submitted twice | one attempt, same submitted record returned |
+| same submitted operation confirmed twice | one confirmation, same record returned |
+| different transaction hash | conflict; original hash is preserved |
+| timeout after broadcast | `failed`, then retry with the same operation id |
+| process restart | row is recovered by operation id and key |
+
+The final confirmation update is guarded by both operation id, status, and
+transaction hash. A late callback for a previous attempt cannot confirm a new
+attempt accidentally.
+
+## Recovery procedure
+
+1. Find `pending`, `submitted`, or `failed` rows for the milestone.
+2. For `confirmed`, do not submit or retry; the settlement is complete.
+3. For `submitted`, reconcile the stored transaction hash with the chain
+   before deciding whether to record confirmation or failure.
+4. For `failed`, retry `submit` using the existing operation id and logical
+   key. Do not create a new key to bypass an unknown outcome.
+5. Record the operator-visible failure code and a redacted message when the
+   chain or RPC is unavailable.
+
+The `SettlementOperationLedger` mirrors the production transitions and can
+snapshot/restore records for deterministic worker and restart tests. The
+database adapter is the source of truth in deployed environments.
+
+## Failure and observability trade-offs
+
+The design deliberately preserves failed rows rather than deleting them. This
+uses a small amount of storage but gives operators a complete attempt history,
+supports safe recovery after a timeout, and prevents duplicate payouts. Error
+messages are bounded and must not contain secrets, signing material, or full
+request payloads. A monitoring job should alert on old `submitted` operations
+and high retry counts; it must not auto-create replacement operations.
+
+## Test coverage
+
+The state-machine tests cover malformed requests, release/redirect identity,
+same-key replay, changed-payload conflicts, duplicate submission, transaction
+hash conflicts, timeout/rejected-transaction recovery, confirmation gating,
+duplicate callbacks, terminal behavior, and restart restoration for every
+non-terminal state.
+

src/services/settlementOperations.test.ts

@@ -0,0 +1,230 @@
+import {
+  assertSettlementConfirmed,
+  SettlementOperationError,
+  SettlementOperationLedger,
+} from './settlementOperations.js'
+import { describe, expect, it } from '@jest/globals'
+
+const request = (overrides: Record<string, unknown> = {}) => ({
+  milestoneId: 'ms-1499-test',
+  operationKey: 'release-2026-08-30',
+  operationType: 'release' as const,
+  requestedBy: 'creator-1',
+  destination: 'GDESTINATION',
+  amount: '100.0000000',
+  assetCode: 'USDC',
+  ...overrides,
+})
+
+const tx = 'tx-hash-0001'
+
+describe('settlement operation identity', () => {
+  it('creates a pending operation with no side effect status', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+
+    expect(operation.id).toBeTruthy()
+    expect(operation.status).toBe('pending')
+    expect(operation.attemptCount).toBe(0)
+    expect(operation.operationType).toBe('release')
+    expect(operation.transactionHash).toBeNull()
+    expect(operation.submittedAt).toBeNull()
+    expect(operation.confirmedAt).toBeNull()
+  })
+
+  it('returns the same operation for an identical retry', () => {
+    const ledger = new SettlementOperationLedger()
+    const first = ledger.create(request())
+    const retry = ledger.create(request())
+
+    expect(retry).toEqual(first)
+    expect(ledger.snapshot()).toHaveLength(1)
+  })
+
+  it.each([
+    ['operationType', { operationType: 'unknown' }],
+    ['operationKey', { operationKey: 'contains spaces' }],
+    ['amount', { amount: '0' }],
+    ['destination', { destination: '' }],
+    ['requestedBy', { requestedBy: '' }],
+  ])('rejects malformed %s before allocation', (_field, overrides) => {
+    const ledger = new SettlementOperationLedger()
+    expect(() => ledger.create(request(overrides))).toThrow(SettlementOperationError)
+    expect(ledger.snapshot()).toEqual([])
+  })
+
+  it('rejects an idempotency-key reuse with a changed destination', () => {
+    const ledger = new SettlementOperationLedger()
+    ledger.create(request())
+
+    expect(() => ledger.create(request({ destination: 'GOTHERDESTINATION' }))).toThrow(/different settlement request/i)
+    expect(ledger.snapshot()).toHaveLength(1)
+  })
+
+  it('allows release and redirect to share no identity', () => {
+    const ledger = new SettlementOperationLedger()
+    const release = ledger.create(request())
+    const redirect = ledger.create(request({ operationType: 'redirect', operationKey: 'redirect-2026-08-30' }))
+
+    expect(redirect.id).not.toBe(release.id)
+    expect(ledger.snapshot()).toHaveLength(2)
+  })
+})
+
+describe('submission and retry state machine', () => {
+  it('moves pending to submitted exactly once', () => {
+    const ledger = new SettlementOperationLedger()
+    const pending = ledger.create(request())
+    const submitted = ledger.submit(pending.id, tx)
+    const retry = ledger.submit(pending.id, tx)
+
+    expect(submitted.status).toBe('submitted')
+    expect(submitted.attemptCount).toBe(1)
+    expect(retry).toEqual(submitted)
+    expect(ledger.snapshot()[0]?.attemptCount).toBe(1)
+  })
+
+  it('rejects a second transaction hash for a submitted operation', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+
+    expect(() => ledger.submit(operation.id, 'tx-hash-0002')).toThrow(/another transaction/i)
+    expect(ledger.get(operation.id)?.attemptCount).toBe(1)
+  })
+
+  it.each(['short', '', 'contains spaces'])('rejects malformed transaction hashes (%s)', (hash) => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    expect(() => ledger.submit(operation.id, hash)).toThrow(/transactionHash/i)
+    expect(ledger.get(operation.id)?.status).toBe('pending')
+  })
+
+  it('records a failure as recoverable and increments attempts on retry', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+    const failed = ledger.fail(operation.id, 'RPC_TIMEOUT', 'Soroban RPC timed out')
+    const retry = ledger.submit(operation.id, 'tx-hash-0002')
+
+    expect(failed.status).toBe('failed')
+    expect(failed.failureCode).toBe('RPC_TIMEOUT')
+    expect(failed.failureMessage).toBe('Soroban RPC timed out')
+    expect(retry.status).toBe('submitted')
+    expect(retry.attemptCount).toBe(2)
+    expect(retry.failureCode).toBeNull()
+  })
+
+  it('does not allow failed or pending operations to be confirmed', () => {
+    const ledger = new SettlementOperationLedger()
+    const pending = ledger.create(request())
+    expect(() => ledger.confirm(pending.id, tx)).toThrow(/submitted/i)
+    ledger.submit(pending.id, tx)
+    ledger.fail(pending.id, 'REJECTED_TX', 'Transaction rejected')
+    expect(() => ledger.confirm(pending.id, tx)).toThrow(/submitted/i)
+  })
+})
+
+describe('confirmation gate', () => {
+  it('exposes a completion guard that rejects every non-confirmed state', () => {
+    for (const status of ['pending', 'submitted', 'failed'] as const) {
+      expect(() => assertSettlementConfirmed({ status })).toThrow(/not confirmed/i)
+    }
+    expect(() => assertSettlementConfirmed({ status: 'confirmed' })).not.toThrow()
+  })
+
+  it('confirms only the exact submitted transaction', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+
+    expect(() => ledger.confirm(operation.id, 'tx-hash-0002')).toThrow(/does not match/i)
+    expect(ledger.get(operation.id)?.status).toBe('submitted')
+
+    const confirmed = ledger.confirm(operation.id, tx)
+    expect(confirmed.status).toBe('confirmed')
+    expect(confirmed.confirmedAt).toBeTruthy()
+  })
+
+  it('makes the confirmation callback idempotent', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+    const first = ledger.confirm(operation.id, tx)
+    const replay = ledger.confirm(operation.id, tx)
+
+    expect(replay).toEqual(first)
+    expect(ledger.snapshot()).toHaveLength(1)
+  })
+
+  it('rejects a different callback after confirmation', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+    ledger.confirm(operation.id, tx)
+
+    expect(() => ledger.confirm(operation.id, 'tx-hash-0002')).toThrow(/another transaction/i)
+  })
+
+  it('does not permit a confirmed operation to be resubmitted', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+    ledger.confirm(operation.id, tx)
+
+    const result = ledger.submit(operation.id, 'tx-hash-0002')
+    expect(result.status).toBe('confirmed')
+    expect(result.transactionHash).toBe(tx)
+    expect(result.attemptCount).toBe(1)
+  })
+})
+
+describe('restart recovery', () => {
+  it('restores pending, submitted, failed, and confirmed operations', () => {
+    const firstProcess = new SettlementOperationLedger()
+    const pending = firstProcess.create(request({ operationKey: 'pending' }))
+    const submitted = firstProcess.create(request({ operationKey: 'submitted' }))
+    firstProcess.submit(submitted.id, 'tx-hash-0003')
+    const failed = firstProcess.create(request({ operationKey: 'failed' }))
+    firstProcess.submit(failed.id, 'tx-hash-0004')
+    firstProcess.fail(failed.id, 'NETWORK_ERROR', 'Network unavailable')
+    const confirmed = firstProcess.create(request({ operationKey: 'confirmed' }))
+    firstProcess.submit(confirmed.id, 'tx-hash-0005')
+    firstProcess.confirm(confirmed.id, 'tx-hash-0005')
+
+    const secondProcess = new SettlementOperationLedger()
+    secondProcess.restore(firstProcess.snapshot())
+
+    expect(secondProcess.get(pending.id)?.status).toBe('pending')
+    expect(secondProcess.get(submitted.id)?.status).toBe('submitted')
+    expect(secondProcess.get(failed.id)?.status).toBe('failed')
+    expect(secondProcess.get(confirmed.id)?.status).toBe('confirmed')
+    expect(secondProcess.create(request({ operationKey: 'pending' })).id).toBe(pending.id)
+  })
+
+  it('retries the same failed logical operation after restoration', () => {
+    const firstProcess = new SettlementOperationLedger()
+    const operation = firstProcess.create(request())
+    firstProcess.submit(operation.id, tx)
+    firstProcess.fail(operation.id, 'TIMEOUT', 'Submission timed out')
+
+    const secondProcess = new SettlementOperationLedger()
+    secondProcess.restore(firstProcess.snapshot())
+    const retry = secondProcess.submit(operation.id, 'tx-hash-0006')
+
+    expect(retry.id).toBe(operation.id)
+    expect(retry.attemptCount).toBe(2)
+    expect(retry.status).toBe('submitted')
+  })
+
+  it('does not silently replace restored records with a new operation', () => {
+    const source = new SettlementOperationLedger()
+    const original = source.create(request())
+    const restored = new SettlementOperationLedger()
+    restored.restore(source.snapshot())
+
+    const replay = restored.create(request())
+    expect(replay.id).toBe(original.id)
+    expect(restored.snapshot()).toHaveLength(1)
+  })
+})