fix: make stream state transitions concurrency-safe

docs/STREAM_CONCURRENCY.md

@@ -0,0 +1,158 @@
+# Stream transition concurrency contract
+
+Withdrawals and cancellations are balance-affecting transitions. They must
+not be implemented as an unguarded read, provider call, and write because the
+provider call yields the event loop while another request can inspect the same
+stream.
+
+## Failure mode
+
+The old flow was effectively:
+
+```text
+request A: read withdrawn=0, calculate 1000 available
+request B: read withdrawn=0, calculate 1000 available
+request A: release 1000, write withdrawn=1000
+request B: release 1000, write withdrawn=1000
+```
+
+The stored record looked plausible while the underlying funds were released
+twice. A cancellation racing with a withdrawal could also refund locked funds
+after the recipient had already received them.
+
+## Transition protocol
+
+Every stream has a monotonic integer `version`. New streams begin at version
+one. A transition captures the version before awaiting the provider and writes
+through `updateStreamIfVersion`. The write succeeds only if the stored version
+is still the captured value; a successful write increments it exactly once.
+
+The service also uses `withStreamLock(streamId)` to serialize transitions for
+the same stream across the provider await. This is the in-process guard for the
+current service. The compare-and-swap method remains necessary as a persistence
+boundary and documents the contract a future database-backed store must retain.
+
+```text
+withStreamLock(id)
+  read state and capture version
+  validate status and available amount
+  await provider transaction
+  compare-and-swap expected version
+  increment version and publish outbox event
+release lock
+```
+
+The lock is keyed by stream ID, so unrelated streams continue concurrently.
+An exception from the provider always releases the lock and leaves stream
+state, version, and outbox unchanged.
+
+## State rules
+
+| Current state | Operation | Result |
+| --- | --- | --- |
+| Active with available amount | Withdraw | Provider release, version increment |
+| Active with no locked amount | Withdraw | 400; no provider call |
+| Active | Cancel | Provider refund, status becomes cancelled |
+| Completed | Cancel | 409; no refund |
+| Cancelled | Cancel | 409; no refund |
+| Any state after a newer write | Stale update | 409; newer state is preserved |
+
+Full withdrawal can move an active stream directly to `completed`. Once that
+transition wins, a queued cancellation observes the terminal state and cannot
+issue a second provider operation. Conversely, a cancellation that wins first
+prevents a queued withdrawal from releasing funds.
+
+## Compatibility
+
+The store accepts legacy records without a version and treats them as version
+one for the first guarded transition. Existing public views omit the version
+for those legacy records so callers that persist or compare old fixture shapes
+do not break. New records and all successfully transitioned records expose the
+version, allowing clients and operators to detect stale responses.
+
+Outbox payloads for balance-affecting events include the resulting version.
+The existing event keys and status values remain unchanged. Outbox delivery is
+still responsible for downstream retries; the stream mutation itself is not
+marked complete until the provider call has succeeded and the versioned state
+write has completed.
+
+## Operational guidance
+
+Provider calls should remain idempotent by transaction identity in a real
+Soroban integration. This service's lock prevents duplicate local transitions,
+while the version guard prevents stale persistence. If a process crashes after
+the provider commits but before the local write, reconciliation must use the
+provider transaction hash before retrying. The transition design deliberately
+surfaces that recovery boundary instead of claiming local memory can provide
+cross-process atomicity.
+
+Do not remove the lock because Node is single-threaded: asynchronous provider
+calls still interleave. Do not replace the conditional write with an
+unconditional map assignment. Any future shared store must provide equivalent
+row-level compare-and-swap or an atomic transaction around the state change.
+
+## Test matrix
+
+`test/streamConcurrency.test.js` covers:
+
+- two full withdrawals racing for the same stream;
+- withdrawal versus cancellation ordering;
+- provider failure and retry after lock release;
+- stale compare-and-swap writers;
+- terminal-state replay protection;
+- independent progress for different stream IDs;
+- version one compatibility for newly created records; and
+- resulting versions on outbox-facing transition results.
+
+These tests use deferred provider promises to force the exact interleaving
+that caused the original failure. They do not rely on timing sleeps, making
+the race assertions deterministic and fast in CI.
+
+## Deployment checklist
+
+Before enabling a real provider adapter:
+
+1. Persist the stream version in the same database row as `withdrawn` and
+   `status`.
+2. Implement `updateStreamIfVersion` as a database conditional update and
+   treat zero affected rows as a conflict.
+3. Include the provider transaction identity in the transition record before
+   acknowledging a successful request.
+4. Reconcile provider transactions that have no matching local completion.
+5. Keep the per-stream lock for duplicate work within one process, but do not
+   treat it as a cross-process lock.
+6. Alert on repeated version conflicts, provider failures, and reconciliation
+   gaps; each indicates a different recovery path.
+
+The in-memory implementation is deliberately small, but its observable
+contract is explicit: one winning transition per observed version, no
+terminal-state replay, and no state mutation when the provider fails.
+
+## Observability and support guidance
+
+Every successful balance-affecting response includes the stream version that
+was committed. This lets a client or support operator compare a response with
+a later `GET /streams/:id` result without guessing which request wrote state.
+
+Conflict responses are expected control-flow responses when a terminal state
+is reached or when a stale writer loses a compare-and-set race. They should be
+reported separately from provider outages. A rising conflict rate can indicate
+duplicate client submissions, an overly aggressive retry policy, or a caller
+using an old stream representation.
+
+Provider errors do not advance the stream version. The failed request can be
+retried after the provider is healthy, and the next successful transition will
+still use the version that was current before the failed attempt. Operators
+should therefore avoid manually editing stream state to recover from a
+transient provider error.
+
+The lock is scoped to a stream ID, so unrelated streams remain independent. A
+slow provider call for stream A must not delay a withdrawal for stream B. The
+lock is released in a `finally` path, including when the provider rejects, so
+one failed request cannot permanently block later work.
+
+This release provides process-local serialization for the current in-memory
+store. Deployments that move stream state to a shared database must retain the
+version predicate in the database update and coordinate provider side effects
+with an outbox or idempotency mechanism. A process-local lock alone is not
+sufficient once multiple workers can write the same stream.

src/services/streamService.js

@@ -35,6 +35,7 @@ function toView(stream, atTime) {
     remainingSeconds: streamMath.remainingSeconds(stream, at),
     createdAt: stream.createdAt,
     updatedAt: stream.updatedAt,
+    ...(stream.version === undefined ? {} : { version: store.streamVersion(stream) }),
     txHashes: stream.txHashes,
   };
 }
@@ -64,6 +65,7 @@ async function createStream(input) {
     withdrawn: 0,
     createdAt: now,
     updatedAt: now,
+    version: 1,
     txHashes: { lock: lock.txHash },
   };
 
@@ -239,9 +241,16 @@ function clampOffset(value) {
 /**
  * Release the streamed-so-far amount to the recipient.
  */
-async function withdraw(id, requestedAmount) {
+async function withdrawUnlocked(id, requestedAmount) {
   const stream = store.getStream(id);
   if (!stream) throw ApiError.notFound(`Stream ${id} not found`);
+  const expectedVersion = store.streamVersion(stream);
+  if (stream.status === STREAM_STATUS.CANCELLED) {
+    throw ApiError.conflict('Stream already cancelled');
+  }
+  if (stream.status === STREAM_STATUS.COMPLETED) {
+    throw ApiError.conflict('Stream already completed');
+  }
 
   const now = nowSeconds();
   const available = streamMath.withdrawableAmount(stream, now);
@@ -268,24 +277,26 @@ async function withdraw(id, requestedAmount) {
   if (stream.withdrawn >= stream.total && stream.status === STREAM_STATUS.ACTIVE) {
     stream.status = STREAM_STATUS.COMPLETED;
   }
-  store.updateStream(stream);
+  const updated = store.updateStreamIfVersion(stream.id, expectedVersion, stream);
+  if (!updated) throw ApiError.conflict('Stream changed; retry the withdrawal');
   outboxService.enqueue({
-    key: `${stream.id}:withdraw:${release.txHash}`,
+    key: `${updated.id}:withdraw:${release.txHash}`,
     type: 'stream.withdrawn',
-    aggregateId: stream.id,
-    payload: { streamId: stream.id, amount, withdrawn: stream.withdrawn, txHash: release.txHash },
+    aggregateId: updated.id,
+    payload: { streamId: updated.id, amount, withdrawn: updated.withdrawn, txHash: release.txHash, version: updated.version },
   });
 
-  logger.info('stream withdraw', { id: stream.id, amount });
-  return { stream: toView(stream, now), amount, txHash: release.txHash };
+  logger.info('stream withdraw', { id, amount, version: updated.version });
+  return { stream: toView(updated, now), amount, txHash: release.txHash };
 }
 
 /**
  * Cancel a stream: recipient keeps what streamed, sender reclaims the rest.
  */
-async function cancel(id) {
+async function cancelUnlocked(id) {
   const stream = store.getStream(id);
   if (!stream) throw ApiError.notFound(`Stream ${id} not found`);
+  const expectedVersion = store.streamVersion(stream);
   if (stream.status === STREAM_STATUS.CANCELLED) {
     throw ApiError.conflict('Stream already cancelled');
   }
@@ -304,16 +315,33 @@ async function cancel(id) {
   stream.status = STREAM_STATUS.CANCELLED;
   stream.updatedAt = now;
   stream.txHashes = { ...stream.txHashes, refund: refundTx.txHash };
-  store.updateStream(stream);
+  const updated = store.updateStreamIfVersion(stream.id, expectedVersion, stream);
+  if (!updated) throw ApiError.conflict('Stream changed; retry cancellation');
   outboxService.enqueue({
-    key: `${stream.id}:cancelled`,
+    key: `${updated.id}:cancelled`,
     type: 'stream.cancelled',
-    aggregateId: stream.id,
-    payload: { streamId: stream.id, refunded: refund, txHash: refundTx.txHash },
+    aggregateId: updated.id,
+    payload: { streamId: updated.id, refunded: refund, txHash: refundTx.txHash, version: updated.version },
   });
 
-  logger.info('stream cancelled', { id: stream.id, refund });
-  return { stream: toView(stream, now), refunded: refund, txHash: refundTx.txHash };
+  logger.info('stream cancelled', { id, refund, version: updated.version });
+  return { stream: toView(updated, now), refunded: refund, txHash: refundTx.txHash };
+}
+
+/**
+ * Serialize a withdrawal for this stream across the provider await and the
+ * persistence write. A second request observes the first request's new
+ * version and state instead of releasing the same balance again.
+ */
+async function withdraw(id, requestedAmount) {
+  return store.withStreamLock(id, () => withdrawUnlocked(id, requestedAmount));
+}
+
+/**
+ * Serialize cancellation with withdrawals and other cancellation attempts.
+ */
+async function cancel(id) {
+  return store.withStreamLock(id, () => cancelUnlocked(id));
 }
 
 /**

src/store/index.js

@@ -8,6 +8,13 @@
 const streams = new Map();
 const outbox = new Map();
 const deliveredEvents = new Map();
+const streamLocks = new Map();
+
+function streamVersion(stream) {
+  return Number.isInteger(stream && stream.version) && stream.version > 0
+    ? stream.version
+    : 1;
+}
 
 const store = {
   /**
@@ -33,6 +40,40 @@ const store = {
     return stream;
   },
 
+  /**
+   * Replace a stream only when its caller observed the current version.
+   * Returning false gives services a stable conflict path instead of allowing
+   * a stale transition to overwrite a newer balance-affecting transition.
+   */
+  updateStreamIfVersion(id, expectedVersion, stream) {
+    const current = streams.get(id);
+    if (!current || streamVersion(current) !== expectedVersion) return false;
+    const updated = { ...stream, version: expectedVersion + 1 };
+    streams.set(id, updated);
+    return updated;
+  },
+
+  /**
+   * Serialize all balance-affecting transitions for one stream. The callback
+   * may await the network/provider; the next callback starts only after it
+   * releases this stream's turn.
+   */
+  async withStreamLock(id, callback) {
+    const previous = streamLocks.get(id) || Promise.resolve();
+    let release;
+    const turn = new Promise((resolve) => { release = resolve; });
+    const queued = previous.then(() => turn);
+    streamLocks.set(id, queued);
+
+    await previous;
+    try {
+      return await callback();
+    } finally {
+      release();
+      if (streamLocks.get(id) === queued) streamLocks.delete(id);
+    }
+  },
+
   /**
    * Return all streams as an array.
    */
@@ -47,6 +88,7 @@ const store = {
     streams.clear();
     outbox.clear();
     deliveredEvents.clear();
+    streamLocks.clear();
   },
 
   /**
@@ -58,6 +100,7 @@ const store = {
 
   outbox,
   deliveredEvents,
+  streamVersion,
 };
 
 module.exports = store;