diff --git a/README.md b/README.md index 1c93ea6..3dbe6d6 100644 --- a/README.md +++ b/README.md @@ -95,23 +95,34 @@ request: `amount` is optional on `withdraw` (omitting it withdraws the full available balance, same as the single-stream endpoint) and not allowed on `cancel`. -Each item is applied independently and best-effort: one item failing (stream -not found, already cancelled, nothing withdrawable, etc.) does not stop the -rest of the batch. The response reports a per-item outcome rather than a -single pass/fail for the whole request: +Each item is applied independently in input order under an explicit partial +commit contract: one item failing (stream not found, already cancelled, +nothing withdrawable, etc.) does not stop the rest of the batch, and successful +items are not rolled back when a later item fails. Send an `Idempotency-Key` +header to make retries safe; the response stores successful item outcomes and +retries only failed items for the same normalized request. The response +contains operation and per-item correlation IDs: ```json { + "operationId": "batch_...", + "correlationId": "batch_...", + "atomicity": "partial", + "replayed": false, "results": [ - { "id": "stream_abc", "action": "withdraw", "ok": true, "stream": { "...": "..." }, "amount": 100, "txHash": "tx_..." }, - { "id": "stream_def", "action": "cancel", "ok": false, "error": { "message": "Stream stream_def not found", "code": "NOT_FOUND", "statusCode": 404 } } + { "index": 0, "itemCorrelationId": "batch_...:item:1", "id": "stream_abc", "action": "withdraw", "ok": true, "stream": { "...": "..." }, "amount": 100, "txHash": "tx_..." }, + { "index": 1, "itemCorrelationId": "batch_...:item:2", "id": "stream_def", "action": "cancel", "ok": false, "error": { "message": "Stream stream_def not found", "code": "NOT_FOUND", "statusCode": 404 } } ], "count": 2, "succeeded": 1, - "failed": 1 + "failed": 1, + "retryableFailures": 1 } ``` +See [`docs/BATCH_SEMANTICS.md`](docs/BATCH_SEMANTICS.md) for the complete +retry, ordering, and failure contract. + Validation happens up front and rejects the whole request if malformed: an empty or oversized batch, an unknown `id`/`action` shape, or the same `id` appearing twice in one batch (which would otherwise let a single request diff --git a/docs/BATCH_SEMANTICS.md b/docs/BATCH_SEMANTICS.md new file mode 100644 index 0000000..cea217f --- /dev/null +++ b/docs/BATCH_SEMANTICS.md @@ -0,0 +1,164 @@ +# Batch stream mutation semantics + +`POST /api/streams/batch` is a partial-commit endpoint. It deliberately does +not pretend that several independent provider transactions form one atomic +ledger transaction. The API makes that boundary visible and gives callers the +information needed to reconcile a mixed result. + +## Contract at a glance + +| Property | Contract | +| --- | --- | +| Execution order | Items execute sequentially in the order received. | +| Atomicity | `partial`: a successful item is committed even if a later item fails. | +| Rollback | There is no cross-item rollback. Each item owns its own mutation. | +| Item identity | `index` and `itemCorrelationId` remain stable for a request shape. | +| Batch identity | `operationId` and `correlationId` identify the whole operation. | +| Retry key | `Idempotency-Key` binds retries to one normalized request. | +| Success retry | A previously successful item is returned from the operation record and is not submitted again. | +| Failed retry | A failed item may be attempted again; previously successful items remain untouched. | +| Error shape | Every failed item includes `message`, machine-readable `code`, and `statusCode`. | +| Duplicate IDs | Rejected during validation before any provider call. | + +## Why partial commit is explicit + +Withdraw and cancel actions are separate provider operations. The backend can +serialize them and report their local state transitions, but it cannot turn a +set of separate external transactions into one all-or-nothing transaction. +Calling the endpoint atomic would make a failed later item look like an +instruction to reverse earlier ledger work, which is unsafe. + +The partial contract gives a caller two useful facts: + +1. Every `ok: true` item has a committed local state transition and an event + payload associated with that transition. +2. Every `ok: false` item has an actionable error and can be retried without + replaying the successful items in the same request. + +The top-level `succeeded`, `failed`, and `retryableFailures` counts are derived +from the returned item list. The list is always in input order, including when +some items fail. + +## Correlation IDs + +The server generates an `operationId` such as `batch_550e8400-e29b-41d4-a716- +446655440000`. `correlationId` is an alias for this value at the response +level, so clients can use either conventional name in logs and tracing. + +Each item receives `batch-id:item:N`, where `N` is one-based. The item ID does +not depend on a stream's mutable state or on a transaction hash. It therefore +remains useful when comparing the original response with a replay response. + +For example, a three-item request always has item IDs ending in `:item:1`, +`:item:2`, and `:item:3`, even if the second item fails. The `index` field uses +zero-based indexing for direct array correlation. + +## Idempotency-Key behavior + +Clients should generate one stable key for one logical batch and send it with +every retry. Keys are trimmed and limited to 128 characters. An empty key is +treated as absent; an overlong key is rejected before provider work begins. + +The in-memory store records the key, the normalized request fingerprint, the +operation ID, and each item outcome. Reusing a key with a different item list, +action, order, or amount returns `409 CONFLICT`. This prevents a caller from +accidentally attaching a new business operation to an old retry record. + +The first request processes every item in order. It saves each outcome after +the item finishes, rather than waiting for the whole batch. If the request is +repeated after a response was returned, successful outcomes are replayed from +the record and no provider call is made for those items. + +Failed outcomes are retained for diagnostics but are eligible for another +attempt. This is important for transient provider failures: a retry can make +progress on the failed item while still protecting successful items from +duplicate release or refund calls. A retry response sets `replayed` to false +when it had to execute at least one failed item. + +Concurrent requests using the same key are serialized by the store. The first +request creates the operation record and the second request waits. After the +first request commits its success, the second request reads and replays that +success instead of entering the provider again. + +This implementation's record is process-local because the repository uses an +in-memory store. A production deployment must persist the operation record and +item outcomes in a shared database or idempotency service before relying on +the contract across workers or restarts. + +## Error handling + +Errors remain item-scoped and use the same stable codes as direct mutation +endpoints: + +| Situation | Code | Retry guidance | +| --- | --- | --- | +| Stream does not exist | `NOT_FOUND` | Correct the item; retrying unchanged will fail again. | +| Stream is already terminal | `CONFLICT` | Refresh state; do not retry as a new mutation. | +| Nothing is withdrawable | `BAD_REQUEST` | Wait for vesting or adjust the request. | +| Provider or unexpected service failure | `INTERNAL_ERROR` or service code | Retry with the same key after checking provider status. | +| Key reused for another request | `CONFLICT` | Generate a new key only for the genuinely new request. | + +The server always continues to the next item after an item-scoped failure. +Validation errors are different: malformed batches, unsupported actions, +duplicate stream IDs, and invalid amounts reject the entire request before any +item is executed. This prevents a request from having a partially applied +interpretation of malformed input. + +## Ordering and duplicate protection + +The validator trims stream IDs before producing the cleaned request. It rejects +the same cleaned ID more than once, even when the actions differ. For example, +one request cannot withdraw and then cancel the same stream in a chosen order. +That restriction removes an otherwise confusing dependency on which action +happened to run first. + +Distinct streams execute in array order. This makes provider call order, +outbox order, response order, and retry reconciliation deterministic. The +service does not use `Promise.all` for item execution because parallel calls +would make partial failure ordering and accounting harder to explain. + +## Rollback boundary + +An individual stream transition follows the existing provider-then-local-write +sequence. If the provider rejects before returning a transaction, the stream +record and its outbox event remain unchanged. The next batch item can still +run. If the provider commits but the process fails before the local write, the +provider transaction identity must be reconciled by the production adapter; +the in-memory mock cannot provide crash recovery. + +Consequently, a caller should reconcile the `results` array, not infer state +from the top-level HTTP status alone. For every success, retain the returned +transaction hash and item correlation ID. For every failure, retain the error +code and retry the same key only when the error is operationally retryable. + +## Client algorithm + +1. Build and validate the complete ordered update list. +2. Generate one idempotency key for the logical batch. +3. Submit the list with the `Idempotency-Key` header. +4. Persist `operationId`, every item correlation ID, transaction hash, and + outcome. +5. On a timeout or transient failure, repeat the identical request and key. +6. Treat successful replayed items as already committed. +7. Investigate non-retryable item codes instead of changing the request under + the old key. + +Do not generate a new key merely because the HTTP response was lost. A new key +would intentionally create a new operation and could submit the same provider +mutation again. + +## Test coverage + +`test/batchSemantics.test.js` covers: + +- mixed success and failure with stable order and correlation IDs; +- duplicate-ID rejection before item execution; +- no provider repetition for a fully successful replay; +- retrying a failed item while preserving an earlier success; +- concurrent same-key requests and operation-level serialization; +- key fingerprint conflicts; +- provider failure without rolling back a prior successful item; and +- idempotency-key length validation before provider work. + +The existing batch service and validator tests remain in place. Together they +cover the prior per-item behavior as well as the new explicit contract. diff --git a/src/constants/batch.js b/src/constants/batch.js index d0921ae..ac029a7 100644 --- a/src/constants/batch.js +++ b/src/constants/batch.js @@ -7,6 +7,7 @@ */ const BATCH = Object.freeze({ MAX_ITEMS: 25, + MAX_IDEMPOTENCY_KEY_LENGTH: 128, ACTIONS: Object.freeze(['withdraw', 'cancel']), }); diff --git a/src/controllers/streamController.js b/src/controllers/streamController.js index d407782..e3739bd 100644 --- a/src/controllers/streamController.js +++ b/src/controllers/streamController.js @@ -97,7 +97,9 @@ async function cancel(req, res) { * own ok/error outcome so partial application is visible to the caller. */ async function batchUpdate(req, res) { - const result = await streamService.batchUpdate(req.validated.updates); + const result = await streamService.batchUpdate(req.validated.updates, { + idempotencyKey: req.get('Idempotency-Key'), + }); res.json(result); } diff --git a/src/services/streamService.js b/src/services/streamService.js index 60cf010..440fe10 100644 --- a/src/services/streamService.js +++ b/src/services/streamService.js @@ -5,12 +5,13 @@ const stellarService = require('./stellarService'); const streamMath = require('./streamMath'); const ApiError = require('../utils/ApiError'); const logger = require('../utils/logger'); -const { newStreamId } = require('../utils/ids'); +const { newStreamId, newBatchOperationId } = require('../utils/ids'); const { nowSeconds } = require('../utils/time'); const money = require('../utils/money'); const { STREAM_STATUS } = require('../constants/streamStatus'); const { PAGINATION } = require('../constants/pagination'); const outboxService = require('./outboxService'); +const { BATCH } = require('../constants/batch'); /** * Build the public-facing view of a stream, enriching the stored record with @@ -317,44 +318,133 @@ async function cancel(id) { } /** - * Apply a batch of withdraw/cancel actions in a single request. Each item is - * applied independently and best-effort: one item failing (e.g. a stream not - * found, or nothing withdrawable) does not stop the rest of the batch from - * being applied. The per-item outcome is reported back so callers can tell - * exactly which updates succeeded and which didn't, rather than getting a - * single pass/fail for the whole request. + * Helpers for the batch contract. Batches are partial-commit operations: items + * run in request order, each gets a stable correlation id, and one item error + * does not roll back earlier successful items. */ -async function batchUpdate(updates) { +function clone(value) { + return JSON.parse(JSON.stringify(value)); +} + +function normalizeIdempotencyKey(value) { + if (value === undefined || value === null || value === '') return null; + if (typeof value !== 'string') { + throw ApiError.badRequest('Idempotency-Key must be a string'); + } + const key = value.trim(); + if (!key) return null; + if (key.length > BATCH.MAX_IDEMPOTENCY_KEY_LENGTH) { + throw ApiError.badRequest( + `Idempotency-Key must not exceed ${BATCH.MAX_IDEMPOTENCY_KEY_LENGTH} characters` + ); + } + return key; +} + +function batchFingerprint(updates) { + return JSON.stringify(updates); +} + +function itemCorrelationId(operation, index) { + return `${operation.operationId}:item:${index + 1}`; +} + +async function executeBatch(operation, updates, replayable) { const results = []; + let replayed = replayable; + + for (const [index, item] of updates.entries()) { + const previous = operation.outcomes.get(index); + if (previous && previous.ok) { + results.push(clone(previous)); + continue; + } + replayed = false; - for (const item of updates) { + let outcome; try { - const outcome = + const transition = item.action === 'withdraw' ? await withdraw(item.id, item.amount) : await cancel(item.id); - results.push({ id: item.id, action: item.action, ok: true, ...outcome }); + outcome = { + index, + itemCorrelationId: itemCorrelationId(operation, index), + id: item.id, + action: item.action, + ok: true, + ...transition, + }; } catch (err) { const statusCode = err instanceof ApiError ? err.statusCode : 500; const code = err instanceof ApiError ? err.code : ApiError.codeFor(statusCode); - results.push({ + outcome = { + index, + itemCorrelationId: itemCorrelationId(operation, index), id: item.id, action: item.action, ok: false, error: { message: err.message, code, statusCode }, - }); + }; + } + if (operation.key) { + store.saveBatchOutcome(operation.key, index, outcome); + } else { + operation.outcomes.set(index, clone(outcome)); } + results.push(outcome); } + const failed = results.filter((r) => !r.ok).length; + operation.completed = failed === 0; return { + operationId: operation.operationId, + correlationId: operation.operationId, + atomicity: 'partial', + replayed, results, count: results.length, succeeded: results.filter((r) => r.ok).length, - failed: results.filter((r) => !r.ok).length, + failed, + retryableFailures: failed, }; } +/** + * Apply a batch in input order under an explicit partial-commit contract. + * Successful outcomes are cached by Idempotency-Key; a retry resumes only + * failed items, so an already committed item is never submitted twice. + */ +async function batchUpdate(updates, { idempotencyKey } = {}) { + const key = normalizeIdempotencyKey(idempotencyKey); + const fingerprint = batchFingerprint(updates); + + if (!key) { + const operation = { + key: null, + fingerprint, + operationId: newBatchOperationId(), + outcomes: new Map(), + completed: false, + }; + return executeBatch(operation, updates, false); + } + + return store.withBatchLock(key, async () => { + const existing = store.getBatchOperation(key); + if (existing && existing.fingerprint !== fingerprint) { + throw ApiError.conflict('Idempotency-Key was reused with a different batch'); + } + const operation = existing || store.createBatchOperation({ + key, + fingerprint, + operationId: newBatchOperationId(), + }); + return executeBatch(operation, updates, Boolean(existing)); + }); +} + module.exports = { toView, createStream, @@ -366,4 +456,5 @@ module.exports = { withdraw, cancel, batchUpdate, + normalizeIdempotencyKey, }; diff --git a/src/store/index.js b/src/store/index.js index adfb224..2b0d508 100644 --- a/src/store/index.js +++ b/src/store/index.js @@ -8,6 +8,8 @@ const streams = new Map(); const outbox = new Map(); const deliveredEvents = new Map(); +const batchOperations = new Map(); +const batchLocks = new Map(); const store = { /** @@ -33,6 +35,49 @@ const store = { return stream; }, + /** Return the idempotency record for a batch key, if one exists. */ + getBatchOperation(key) { + return batchOperations.get(key); + }, + + /** Create an idempotency record. Callers serialize creation with a lock. */ + createBatchOperation({ key, fingerprint, operationId }) { + const operation = { + key, + fingerprint, + operationId, + outcomes: new Map(), + completed: false, + }; + batchOperations.set(key, operation); + return operation; + }, + + /** Store one immutable item outcome so successful work is not repeated. */ + saveBatchOutcome(key, index, outcome) { + const operation = batchOperations.get(key); + if (!operation) return false; + operation.outcomes.set(index, JSON.parse(JSON.stringify(outcome))); + return true; + }, + + /** Serialize concurrent requests that present the same idempotency key. */ + async withBatchLock(key, callback) { + const previous = batchLocks.get(key) || Promise.resolve(); + let release; + const turn = new Promise((resolve) => { release = resolve; }); + const queued = previous.then(() => turn); + batchLocks.set(key, queued); + + await previous; + try { + return await callback(); + } finally { + release(); + if (batchLocks.get(key) === queued) batchLocks.delete(key); + } + }, + /** * Return all streams as an array. */ @@ -47,6 +92,8 @@ const store = { streams.clear(); outbox.clear(); deliveredEvents.clear(); + batchOperations.clear(); + batchLocks.clear(); }, /** @@ -58,6 +105,7 @@ const store = { outbox, deliveredEvents, + batchOperations, }; module.exports = store; diff --git a/src/utils/ids.js b/src/utils/ids.js index d5f8f7b..00ea835 100644 --- a/src/utils/ids.js +++ b/src/utils/ids.js @@ -18,4 +18,12 @@ function newTxHash() { return `tx_${uuidv4().replace(/-/g, '')}`; } -module.exports = { newStreamId, newTxHash }; +/** + * Generate a correlation id for one batch request. It is separate from a + * stream id so logs and retries can group several item outcomes safely. + */ +function newBatchOperationId() { + return `batch_${uuidv4()}`; +} + +module.exports = { newStreamId, newTxHash, newBatchOperationId }; diff --git a/test/batchSemantics.test.js b/test/batchSemantics.test.js new file mode 100644 index 0000000..89c8a28 --- /dev/null +++ b/test/batchSemantics.test.js @@ -0,0 +1,265 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); + +const store = require('../src/store'); +const streamService = require('../src/services/streamService'); +const stellarService = require('../src/services/stellarService'); +const { validateBatchUpdate } = require('../src/validators/streamValidators'); +const { STREAM_STATUS } = require('../src/constants/streamStatus'); + +const SENDER = 'GALICE0000000000000000000000000000000000000000000000'; +const RECIPIENT_PREFIX = 'GBOB000000000000000000000000000000000000000000000'; + +function seedStream(overrides = {}) { + const now = Math.floor(Date.now() / 1000); + const stream = { + id: `stream_${Math.random().toString(16).slice(2)}`, + sender: SENDER, + recipient: `${RECIPIENT_PREFIX}${Math.random().toString(16).slice(2, 7)}`, + total: 1000, + asset: 'XLM', + startTime: now - 200, + endTime: now - 100, + status: STREAM_STATUS.ACTIVE, + withdrawn: 0, + createdAt: now - 200, + updatedAt: now - 200, + txHashes: { lock: 'tx_seed' }, + ...overrides, + }; + store.insertStream(stream); + return stream; +} + +function installProvider({ releaseFunds, refundFunds } = {}) { + const originals = { + releaseFunds: stellarService.releaseFunds, + refundFunds: stellarService.refundFunds, + }; + if (releaseFunds) stellarService.releaseFunds = releaseFunds; + if (refundFunds) stellarService.refundFunds = refundFunds; + return () => { + stellarService.releaseFunds = originals.releaseFunds; + stellarService.refundFunds = originals.refundFunds; + }; +} + +function tx(name) { + return { txHash: `tx_${name}`, network: 'test', asset: 'XLM' }; +} + +test('returns a deterministic partial contract and preserves input order', async (t) => { + t.after(() => store.clear()); + const restore = installProvider({ + releaseFunds: async ({ recipient }) => tx(recipient.slice(-3)), + refundFunds: async () => tx('refund'), + }); + t.after(restore); + + const first = seedStream(); + const second = seedStream(); + const result = await streamService.batchUpdate([ + { id: first.id, action: 'withdraw', amount: 125 }, + { id: 'stream_missing', action: 'cancel' }, + { id: second.id, action: 'cancel' }, + ], { idempotencyKey: 'batch-order-1' }); + + assert.equal(result.atomicity, 'partial'); + assert.match(result.operationId, /^batch_[0-9a-f-]+$/); + assert.equal(result.correlationId, result.operationId); + assert.equal(result.results.length, 3); + assert.deepEqual(result.results.map((item) => item.index), [0, 1, 2]); + assert.deepEqual(result.results.map((item) => item.itemCorrelationId), [ + `${result.operationId}:item:1`, + `${result.operationId}:item:2`, + `${result.operationId}:item:3`, + ]); + assert.equal(result.results[0].ok, true); + assert.equal(result.results[1].error.code, 'NOT_FOUND'); + assert.equal(result.results[2].ok, true); + assert.equal(result.succeeded, 2); + assert.equal(result.failed, 1); + assert.equal(result.retryableFailures, 1); +}); + +test('does not repeat successful items when the same key is replayed', async (t) => { + t.after(() => store.clear()); + let releases = 0; + const restore = installProvider({ + releaseFunds: async () => { + releases += 1; + return tx(`release-${releases}`); + }, + }); + t.after(restore); + + const first = seedStream(); + const second = seedStream(); + const updates = [ + { id: first.id, action: 'withdraw' }, + { id: second.id, action: 'withdraw' }, + ]; + const initial = await streamService.batchUpdate(updates, { idempotencyKey: 'batch-replay-1' }); + const replay = await streamService.batchUpdate(updates, { idempotencyKey: 'batch-replay-1' }); + + assert.equal(releases, 2); + assert.equal(initial.replayed, false); + assert.equal(replay.replayed, true); + assert.equal(replay.operationId, initial.operationId); + assert.deepEqual(replay.results, initial.results); + assert.equal(store.getStream(first.id).withdrawn, 1000); + assert.equal(store.getStream(second.id).withdrawn, 1000); +}); + +test('retries failed items while retaining successful item outcomes', async (t) => { + t.after(() => store.clear()); + const successful = seedStream(); + const retryable = seedStream(); + let successfulCalls = 0; + let retryableCalls = 0; + const restore = installProvider({ + releaseFunds: async ({ recipient }) => { + if (recipient === retryable.recipient) { + retryableCalls += 1; + if (retryableCalls === 1) throw new Error('provider temporarily unavailable'); + } else { + successfulCalls += 1; + } + return tx(`release-${successfulCalls}-${retryableCalls}`); + }, + }); + t.after(restore); + + const updates = [ + { id: successful.id, action: 'withdraw' }, + { id: retryable.id, action: 'withdraw' }, + ]; + const first = await streamService.batchUpdate(updates, { idempotencyKey: 'batch-resume-1' }); + assert.equal(first.succeeded, 1); + assert.equal(first.failed, 1); + assert.equal(first.results[1].error.code, 'INTERNAL_ERROR'); + assert.equal(store.getStream(successful.id).withdrawn, 1000); + assert.equal(store.getStream(retryable.id).withdrawn, 0); + + const retry = await streamService.batchUpdate(updates, { idempotencyKey: 'batch-resume-1' }); + assert.equal(retry.succeeded, 2); + assert.equal(retry.failed, 0); + assert.equal(retry.replayed, false); + assert.equal(successfulCalls, 1, 'the previously successful item is not submitted again'); + assert.equal(retryableCalls, 2, 'the failed item receives one retry'); + assert.equal(store.getStream(retryable.id).withdrawn, 1000); +}); + +test('rejects reuse of a key with a different normalized request', async (t) => { + t.after(() => store.clear()); + const restore = installProvider({ releaseFunds: async () => tx('release') }); + t.after(restore); + const stream = seedStream(); + const key = 'batch-fingerprint-1'; + + await streamService.batchUpdate([{ id: stream.id, action: 'withdraw', amount: 100 }], { + idempotencyKey: key, + }); + await assert.rejects( + () => streamService.batchUpdate([{ id: stream.id, action: 'withdraw', amount: 200 }], { + idempotencyKey: key, + }), + (error) => error.statusCode === 409 && error.code === 'CONFLICT' + ); + assert.equal(store.getStream(stream.id).withdrawn, 100); +}); + +test('serializes concurrent requests with the same idempotency key', async (t) => { + t.after(() => store.clear()); + const stream = seedStream(); + let calls = 0; + let releaseProvider; + const providerStarted = new Promise((resolve) => { releaseProvider = resolve; }); + let continueProvider; + const providerGate = new Promise((resolve) => { continueProvider = resolve; }); + const restore = installProvider({ + releaseFunds: async () => { + calls += 1; + releaseProvider(); + await providerGate; + return tx('concurrent'); + }, + }); + t.after(restore); + + const updates = [{ id: stream.id, action: 'withdraw' }]; + const firstPromise = streamService.batchUpdate(updates, { idempotencyKey: 'batch-concurrent-1' }); + await providerStarted; + const secondPromise = streamService.batchUpdate(updates, { idempotencyKey: 'batch-concurrent-1' }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(calls, 1, 'the second request waits for the first operation record'); + continueProvider(); + + const [first, second] = await Promise.all([firstPromise, secondPromise]); + assert.equal(calls, 1); + assert.equal(first.operationId, second.operationId); + assert.equal(second.replayed, true); + assert.equal(store.getStream(stream.id).withdrawn, 1000); +}); + +test('keeps prior successes when a later item fails; no batch rollback is implied', async (t) => { + t.after(() => store.clear()); + const first = seedStream(); + const second = seedStream(); + let refundCalls = 0; + const restore = installProvider({ + releaseFunds: async () => tx('release'), + refundFunds: async () => { + refundCalls += 1; + if (refundCalls === 2) throw new Error('refund unavailable'); + return tx('refund'); + }, + }); + t.after(restore); + + const result = await streamService.batchUpdate([ + { id: first.id, action: 'cancel' }, + { id: second.id, action: 'cancel' }, + ]); + + assert.equal(result.results[0].ok, true); + assert.equal(result.results[1].ok, false); + assert.equal(store.getStream(first.id).status, STREAM_STATUS.CANCELLED); + assert.equal(store.getStream(second.id).status, STREAM_STATUS.ACTIVE); + assert.equal(result.results[1].error.code, 'INTERNAL_ERROR'); +}); + +test('validator rejects duplicate IDs before any item can be applied', () => { + const result = validateBatchUpdate({ + updates: [ + { id: 'stream_same', action: 'cancel' }, + { id: 'stream_same', action: 'withdraw' }, + ], + }); + assert.deepEqual(result.value, undefined); + assert.equal(result.error.length, 1); + assert.match(result.error[0], /duplicated in this batch/); +}); + +test('rejects overlong idempotency keys before provider work begins', async (t) => { + t.after(() => store.clear()); + let calls = 0; + const restore = installProvider({ + releaseFunds: async () => { + calls += 1; + return tx('release'); + }, + }); + t.after(restore); + const stream = seedStream(); + + await assert.rejects( + () => streamService.batchUpdate([{ id: stream.id, action: 'withdraw' }], { + idempotencyKey: 'x'.repeat(129), + }), + (error) => error.statusCode === 400 && error.code === 'BAD_REQUEST' + ); + assert.equal(calls, 0); +});