From e714dfb3fb76357bdbd013ee63ec84cae636dcd3 Mon Sep 17 00:00:00 2001 From: SarahDoma <108196951+SarahDoma@users.noreply.github.com> Date: Sun, 30 Aug 2026 13:38:19 +0100 Subject: [PATCH 1/7] test(streams,events,soroban): add failing tests for #504, #506, #509 - batchWithdraw() must submit sequentially, not concurrently, so it cannot be verified against the current Promise.allSettled(...map) implementation. - dispatchEvent() must decode created/force_cxl/xfer_rec/set_op/rm_op, which are not yet in TOPIC or the switch statement. - buildContractCallTx()/resolveFee() must honour a configurable inclusion fee, which does not exist yet (fee is hardcoded to BASE_FEE). All 13 new assertions fail against the current implementation. --- src/tests/batch-withdraw.test.ts | 26 +++++++ src/tests/events.test.ts | 104 +++++++++++++++++++++++++++ src/tests/soroban-fee-config.test.ts | 70 ++++++++++++++++++ src/tests/streams-success.test.ts | 49 +++++++++++++ 4 files changed, 249 insertions(+) create mode 100644 src/tests/soroban-fee-config.test.ts diff --git a/src/tests/batch-withdraw.test.ts b/src/tests/batch-withdraw.test.ts index baed9ca..66aa0ac 100644 --- a/src/tests/batch-withdraw.test.ts +++ b/src/tests/batch-withdraw.test.ts @@ -65,4 +65,30 @@ describe('StreamsModule.batchWithdraw()', () => { /keypair, wallet adapter, or signer/, ); }); + + it('submits withdrawals one at a time, not concurrently (#504)', async () => { + // A concurrent Promise.allSettled(...map) would call withdraw() for every + // item before any of them resolve. Serialised submission must instead + // wait for withdraw #1 to fully settle before withdraw #2 is even + // invoked — otherwise every withdrawal would read the same account + // sequence number and only one could ever land on-chain. + const sdk = new StreamsModule(makeConfig()); + let inFlight = 0; + let maxConcurrent = 0; + const callOrder: bigint[] = []; + + vi.spyOn(sdk, 'withdraw').mockImplementation(async (streamId) => { + callOrder.push(BigInt(streamId)); + inFlight++; + maxConcurrent = Math.max(maxConcurrent, inFlight); + await new Promise(resolve => setTimeout(resolve, 5)); + inFlight--; + return `hash-${streamId}`; + }); + + await sdk.batchWithdraw([{ streamId: 1n }, { streamId: 2n }, { streamId: 3n }]); + + expect(maxConcurrent).toBe(1); + expect(callOrder).toEqual([1n, 2n, 3n]); + }); }); \ No newline at end of file diff --git a/src/tests/events.test.ts b/src/tests/events.test.ts index 93cac08..6ae27fa 100644 --- a/src/tests/events.test.ts +++ b/src/tests/events.test.ts @@ -9,16 +9,26 @@ import type { ResumeEvent, TopUpEvent, ClawbackEvent, + CreatedEvent, + ForceCancelEvent, + RecipientTransferEvent, + OperatorSetEvent, + OperatorRevokedEvent, } from '../types/index.js'; // ── Fixture helpers ─────────────────────────────────────────────────────────── const actorAddress = Keypair.random().publicKey(); +const otherAddress = Keypair.random().publicKey(); function topic(name: string): xdr.ScVal { return xdr.ScVal.scvSymbol(name); } +function address(addr: string): xdr.ScVal { + return new Address(addr).toScVal(); +} + function actorTopic(): xdr.ScVal { return new Address(actorAddress).toScVal(); } @@ -150,6 +160,100 @@ describe('dispatchEvent — clawback', () => { }); }); +// ── created: (recipient, token, deposit_amount, rate_per_second, start_time, end_time) ── + +describe('dispatchEvent — created', () => { + it('decodes the address and numeric tuple fields (#506)', () => { + const event = makeEvent( + 'created', + tuple(address(otherAddress), address(otherAddress), i128(1_000_000n), i128(10n), u64(1_700_000_000), u64(1_700_100_000)), + ); + let received: CreatedEvent | undefined; + const handlers: StreamEventHandlers = { onCreated: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ + sender: actorAddress, + recipient: otherAddress, + token: otherAddress, + depositAmount: 1_000_000n, + ratePerSecond: 10n, + startTime: 1_700_000_000, + endTime: 1_700_100_000, + sequence: 0n, + }); + }); + + it('does not call onCreated when the handler is not registered', () => { + const event = makeEvent('created', tuple(address(otherAddress), address(otherAddress), i128(1n), i128(1n), u64(1), u64(2))); + expect(() => dispatchEvent(event, {})).not.toThrow(); + }); +}); + +// ── force_cxl: (payout_amount, refund_amount) ──────────────────────────────── + +describe('dispatchEvent — force_cxl', () => { + it('decodes both i128 tuple fields, attributing the actor as the recipient (#506)', () => { + const event = makeEvent('force_cxl', tuple(i128(50_000n), i128(25_000n))); + let received: ForceCancelEvent | undefined; + const handlers: StreamEventHandlers = { onForceCancel: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ + recipient: actorAddress, + payoutAmount: 50_000n, + refundAmount: 25_000n, + sequence: 0n, + }); + }); +}); + +// ── xfer_rec: new_recipient: address (bare scalar) ─────────────────────────── + +describe('dispatchEvent — xfer_rec', () => { + it('decodes the new recipient address, attributing the actor as the previous recipient (#506)', () => { + const event = makeEvent('xfer_rec', address(otherAddress)); + let received: RecipientTransferEvent | undefined; + const handlers: StreamEventHandlers = { onRecipientTransfer: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ + previousRecipient: actorAddress, + newRecipient: otherAddress, + sequence: 0n, + }); + }); +}); + +// ── set_op / rm_op: operator: address (bare scalar) ────────────────────────── + +describe('dispatchEvent — set_op', () => { + it('decodes the delegated operator address (#506)', () => { + const event = makeEvent('set_op', address(otherAddress)); + let received: OperatorSetEvent | undefined; + const handlers: StreamEventHandlers = { onOperatorSet: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ sender: actorAddress, operator: otherAddress, sequence: 0n }); + }); +}); + +describe('dispatchEvent — rm_op', () => { + it('decodes the revoked operator address (#506)', () => { + const event = makeEvent('rm_op', address(otherAddress)); + let received: OperatorRevokedEvent | undefined; + const handlers: StreamEventHandlers = { onOperatorRevoke: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ sender: actorAddress, operator: otherAddress, sequence: 0n }); + }); +}); + // ── Robustness ───────────────────────────────────────────────────────────── describe('dispatchEvent — sequence (topics[2])', () => { diff --git a/src/tests/soroban-fee-config.test.ts b/src/tests/soroban-fee-config.test.ts new file mode 100644 index 0000000..87591d0 --- /dev/null +++ b/src/tests/soroban-fee-config.test.ts @@ -0,0 +1,70 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { BASE_FEE, StrKey } from '@stellar/stellar-sdk'; + +const CONTRACT_ID = StrKey.encodeContract(Buffer.alloc(32, 1)); + +const { mockGetAccount, mockTransactionBuilder } = vi.hoisted(() => ({ + mockGetAccount: vi.fn(), + mockTransactionBuilder: vi.fn(), +})); + +vi.mock('@stellar/stellar-sdk', async () => { + const actual = await vi.importActual('@stellar/stellar-sdk'); + return { + ...actual, + SorobanRpc: { + ...actual.SorobanRpc, + Server: class { + getAccount = mockGetAccount; + }, + }, + TransactionBuilder: mockTransactionBuilder, + }; +}); + +import { resolveFee, buildContractCallTx } from '../soroban.js'; + +beforeEach(() => { + mockGetAccount.mockReset().mockResolvedValue({ accountId: () => 'GACCOUNT', sequenceNumber: () => '1' }); + mockTransactionBuilder.mockReset().mockImplementation(function MockTransactionBuilder() { + return { + addOperation: vi.fn().mockReturnThis(), + setTimeout: vi.fn().mockReturnThis(), + build: vi.fn().mockReturnValue({ _stub: 'tx' }), + }; + }); +}); + +describe('resolveFee() (#509)', () => { + it('defaults to BASE_FEE when neither fee nor feeMultiplier is set', () => { + expect(resolveFee({})).toBe(BASE_FEE); + }); + + it('returns the explicit fee, ignoring feeMultiplier', () => { + expect(resolveFee({ fee: '5000', feeMultiplier: 10 })).toBe('5000'); + }); + + it('scales BASE_FEE by feeMultiplier', () => { + expect(resolveFee({ feeMultiplier: 10 })).toBe((BigInt(BASE_FEE) * 10n).toString()); + }); +}); + +describe('buildContractCallTx() fee parameter (#509)', () => { + it('defaults the TransactionBuilder fee to BASE_FEE when no fee is passed', async () => { + await buildContractCallTx('https://rpc', 'passphrase', 'GCALLER', CONTRACT_ID, 'withdraw', []); + + expect(mockTransactionBuilder).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ fee: BASE_FEE }), + ); + }); + + it('passes an explicit fee through to the TransactionBuilder', async () => { + await buildContractCallTx('https://rpc', 'passphrase', 'GCALLER', CONTRACT_ID, 'withdraw', [], '10000'); + + expect(mockTransactionBuilder).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ fee: '10000' }), + ); + }); +}); diff --git a/src/tests/streams-success.test.ts b/src/tests/streams-success.test.ts index c90d9b9..bad7c82 100644 --- a/src/tests/streams-success.test.ts +++ b/src/tests/streams-success.test.ts @@ -495,3 +495,52 @@ describe('StreamsModule.transferRecipient() — success path', () => { ); }); }); + +describe('StreamsModule — configurable inclusion fee (#509)', () => { + beforeEach(() => { + mockStreamAddress.mockResolvedValue(STREAM_ADDR); + mockSimulate.mockResolvedValue(simSuccess(xdr.ScVal.scvVoid())); + mockGetTransaction.mockResolvedValue(txSuccess()); + }); + + it('passes BASE_FEE by default to buildContractCallTx for a submitted operation', async () => { + const { buildContractCallTx } = await import('../soroban.js'); + const { StreamsModule } = await import('../streams.js'); + const { BASE_FEE } = await import('@stellar/stellar-sdk'); + const sdk = new StreamsModule(makeConfig()); + + await runThroughFirstPoll(() => sdk.pause(1n)); + + expect(buildContractCallTx).toHaveBeenLastCalledWith( + expect.any(String), expect.any(String), expect.any(String), + STREAM_ADDR, 'pause', [], BASE_FEE, + ); + }); + + it('honours an explicit config.fee override', async () => { + const { buildContractCallTx } = await import('../soroban.js'); + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig({ fee: '5000' })); + + await runThroughFirstPoll(() => sdk.pause(1n)); + + expect(buildContractCallTx).toHaveBeenLastCalledWith( + expect.any(String), expect.any(String), expect.any(String), + STREAM_ADDR, 'pause', [], '5000', + ); + }); + + it('honours config.feeMultiplier scaled off BASE_FEE', async () => { + const { buildContractCallTx } = await import('../soroban.js'); + const { StreamsModule } = await import('../streams.js'); + const { BASE_FEE } = await import('@stellar/stellar-sdk'); + const sdk = new StreamsModule(makeConfig({ feeMultiplier: 10 })); + + await runThroughFirstPoll(() => sdk.pause(1n)); + + expect(buildContractCallTx).toHaveBeenLastCalledWith( + expect.any(String), expect.any(String), expect.any(String), + STREAM_ADDR, 'pause', [], (BigInt(BASE_FEE) * 10n).toString(), + ); + }); +}); From 0f98fd858ed0ca35391cc9875b50843def5b5032 Mon Sep 17 00:00:00 2001 From: SarahDoma <108196951+SarahDoma@users.noreply.github.com> Date: Sun, 30 Aug 2026 13:38:49 +0100 Subject: [PATCH 2/7] fix(streams): serialise batchWithdraw submissions to avoid sequence collisions (#504) batchWithdraw() fired all N withdraw() calls concurrently via Promise.allSettled(...map). Each withdraw() -> _invoke() -> buildContractCallTx() reads the caller account's sequence number via getAccount() and builds a transaction on top of it, so concurrent calls all read the same sequence number. With a single keypair/wallet (the normal case), at most one of the N transactions could ever land; the rest failed with txBAD_SEQ, defeating batchWithdraw's entire purpose. Submitting withdrawals one at a time (await each before starting the next) guarantees getAccount() only observes the sequence after the previous transaction has landed, so every submission gets a distinct, ordered sequence number. Closes #504 --- src/streams.ts | 50 ++++++++++++++++++++++++++++++-------------------- 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/src/streams.ts b/src/streams.ts index 0671684..b8cb0a8 100644 --- a/src/streams.ts +++ b/src/streams.ts @@ -337,35 +337,45 @@ export class StreamsModule { } /** - * Withdraw from multiple streams concurrently. + * Withdraw from multiple streams. * * Note: Soroban currently permits only one invoke_host_function * operation per transaction, so this cannot be assembled into a single * atomic transaction the way classic Stellar payment operations can. - * Each withdrawal is submitted as its own transaction; they run - * concurrently and are reported independently so a failure on one - * streamId (e.g. StreamNotFound, insufficient balance) does not block - * or roll back the others. + * Each withdrawal is submitted as its own transaction; they are reported + * independently so a failure on one streamId (e.g. StreamNotFound, + * insufficient balance) does not block or fail the others. + * + * Withdrawals are submitted one at a time, not concurrently. Each + * withdraw() -> _invoke() -> buildContractCallTx() reads the caller + * account's current sequence number via getAccount() and builds a + * transaction on top of it. Firing all N withdrawals at once means every + * one of them reads the same sequence number and builds a transaction + * with the same value, so with a single keypair/wallet at most one + * submission can ever land — the rest fail with txBAD_SEQ (see #504). + * Awaiting each withdrawal (submit + confirm) before starting the next + * guarantees getAccount() only observes the sequence after the previous + * transaction has landed, so every submission gets a distinct, ordered + * sequence number. */ async batchWithdraw(withdrawals: BatchWithdrawItem[]): Promise { this._ensureCanMutate(); - const settled = await Promise.allSettled( - withdrawals.map(w => this.withdraw(w.streamId, w.amount)), - ); - - return settled.map((result, i) => { - const streamId = BigInt(withdrawals[i]!.streamId); - if (result.status === 'fulfilled') { - return { streamId, success: true, txHash: result.value }; + const results: BatchWithdrawResult[] = []; + for (const w of withdrawals) { + const streamId = BigInt(w.streamId); + try { + const txHash = await this.withdraw(w.streamId, w.amount); + results.push({ streamId, success: true, txHash }); + } catch (err) { + results.push({ + streamId, + success: false, + error: err instanceof Error ? err.message : String(err), + }); } - const err = result.reason; - return { - streamId, - success: false, - error: err instanceof Error ? err.message : String(err), - }; - }); + } + return results; } /** Cancel the stream (sender only). Settles all balances atomically. */ From c1af0646a50760f4dd2e5aaff25ab9534adfde7f Mon Sep 17 00:00:00 2001 From: SarahDoma <108196951+SarahDoma@users.noreply.github.com> Date: Sun, 30 Aug 2026 13:40:04 +0100 Subject: [PATCH 3/7] fix(events,types): decode created/force_cxl/xfer_rec/set_op/rm_op events (#506) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TOPIC and dispatchEvent() only handled withdrawn, cancelled, paused, resumed, topped_up, and clawback, even though the stream contract also emits created, force_cxl (force-cancel by recipient), xfer_rec (recipient transfer), set_op (operator delegated), and rm_op (operator revoked). A subscriber was silently never notified of any of these — notably a recipient transfer, where the current subscriber may have just lost the stream. Adds the five missing topic constants, their StreamEventHandlers callbacks (onCreated, onForceCancel, onRecipientTransfer, onOperatorSet, onOperatorRevoke), and typed event payloads (CreatedEvent, ForceCancelEvent, RecipientTransferEvent, OperatorSetEvent, OperatorRevokedEvent) with tuple/scalar decoders matching the existing pattern for the six handled topics. Closes #506 --- src/events.ts | 89 ++++++++++++++++++++++++++++++++++++++++++++++ src/types/index.ts | 62 ++++++++++++++++++++++++++++++++ 2 files changed, 151 insertions(+) diff --git a/src/events.ts b/src/events.ts index 8a94690..703b6c0 100644 --- a/src/events.ts +++ b/src/events.ts @@ -15,10 +15,21 @@ import type { ResumeEvent, TopUpEvent, ClawbackEvent, + CreatedEvent, + ForceCancelEvent, + RecipientTransferEvent, + OperatorSetEvent, + OperatorRevokedEvent, } from './types/index.js'; import { scValToI128, scValToU64, createRpcServer } from './soroban.js'; // ── Event topic names (match symbol_short!() values in Rust) ───────────────── +// +// `created`, `force_cxl`, `xfer_rec`, `set_op` and `rm_op` are emitted by +// contracts/stream/src/events.rs alongside the six handled below, but were +// never wired into TOPIC/dispatchEvent — a subscriber was silently never +// notified of a recipient transfer, a recipient-initiated force-cancel, or +// an operator delegation/revocation (see #506). const TOPIC = { WITHDRAWN: 'withdrawn', @@ -27,6 +38,11 @@ const TOPIC = { RESUMED: 'resumed', TOPPED_UP: 'topped_up', CLAWBACK: 'clawback', + CREATED: 'created', + FORCE_CANCELLED: 'force_cxl', + RECIPIENT_TRANSFERRED: 'xfer_rec', + OPERATOR_SET: 'set_op', + OPERATOR_REVOKED: 'rm_op', } as const; // ── Parser helpers ──────────────────────────────────────────────────────────── @@ -51,6 +67,10 @@ function u64Field(fields: xdr.ScVal[], index: number): number { return field ? Number(scValToU64(field)) : 0; } +function addressFieldAt(fields: xdr.ScVal[], index: number): string { + return addressField(fields[index]); +} + /** * Decodes an address topic to its G.../C... string. `ScVal.address()?.accountId()` * returns the raw XDR PublicKey object, not a string — calling `.toString()` @@ -302,6 +322,75 @@ export function dispatchEvent( handlers.onClawback(data); break; } + + case TOPIC.CREATED: { + if (!handlers.onCreated) break; + // data: (recipient, token, deposit_amount: i128, rate_per_second: i128, start_time: u64, end_time: u64) + const fields = tupleFields(event.value); + const data: CreatedEvent = { + sender: actor, + recipient: addressFieldAt(fields, 0), + token: addressFieldAt(fields, 1), + depositAmount: i128Field(fields, 2), + ratePerSecond: i128Field(fields, 3), + startTime: u64Field(fields, 4), + endTime: u64Field(fields, 5), + sequence, + }; + handlers.onCreated(data); + break; + } + + case TOPIC.FORCE_CANCELLED: { + if (!handlers.onForceCancel) break; + // data: (payout_amount: i128, refund_amount: i128) — mirrors cancel()'s + // (refund_amount, withdrawn_so_far) shape but recipient-initiated. + const fields = tupleFields(event.value); + const data: ForceCancelEvent = { + recipient: actor, + payoutAmount: i128Field(fields, 0), + refundAmount: i128Field(fields, 1), + sequence, + }; + handlers.onForceCancel(data); + break; + } + + case TOPIC.RECIPIENT_TRANSFERRED: { + if (!handlers.onRecipientTransfer) break; + // data: new_recipient: address (bare scalar, not a tuple) + const data: RecipientTransferEvent = { + previousRecipient: actor, + newRecipient: addressField(event.value), + sequence, + }; + handlers.onRecipientTransfer(data); + break; + } + + case TOPIC.OPERATOR_SET: { + if (!handlers.onOperatorSet) break; + // data: operator: address (bare scalar, not a tuple) + const data: OperatorSetEvent = { + sender: actor, + operator: addressField(event.value), + sequence, + }; + handlers.onOperatorSet(data); + break; + } + + case TOPIC.OPERATOR_REVOKED: { + if (!handlers.onOperatorRevoke) break; + // data: operator: address (bare scalar, not a tuple) + const data: OperatorRevokedEvent = { + sender: actor, + operator: addressField(event.value), + sequence, + }; + handlers.onOperatorRevoke(data); + break; + } } return sequence; diff --git a/src/types/index.ts b/src/types/index.ts index 6983423..57067cb 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -125,6 +125,58 @@ export interface ResumeEvent { resumedAt: number; sender: string; sequence: b export interface TopUpEvent { amount: bigint; newBalance: bigint; sender: string; sequence: bigint; } export interface ClawbackEvent { amount: bigint; sender: string; sequence: bigint; } +/** Published once when the factory deploys a new DripStream (`created` topic). */ +export interface CreatedEvent { + /** The stream's sender (topics[1] actor). */ + sender: string; + recipient: string; + token: string; + depositAmount: bigint; + ratePerSecond: bigint; + startTime: number; + endTime: number; + sequence: bigint; +} + +/** + * Published when the recipient force-cancels a paused stream after the + * pause threshold has elapsed (`force_cxl` topic). Settles atomically like + * `cancelled`, but is recipient-initiated rather than sender-initiated. + */ +export interface ForceCancelEvent { + /** The recipient who force-cancelled (topics[1] actor). */ + recipient: string; + /** Earned-but-unwithdrawn amount paid out to the recipient. */ + payoutAmount: bigint; + /** Unstreamed remainder refunded to the sender. */ + refundAmount: bigint; + sequence: bigint; +} + +/** Published when the recipient role is transferred to a new address (`xfer_rec` topic). */ +export interface RecipientTransferEvent { + /** The outgoing recipient who initiated the transfer (topics[1] actor). */ + previousRecipient: string; + newRecipient: string; + sequence: bigint; +} + +/** Published when an operator is delegated on the stream (`set_op` topic). */ +export interface OperatorSetEvent { + /** The address that granted the operator role (topics[1] actor). */ + sender: string; + operator: string; + sequence: bigint; +} + +/** Published when a previously-delegated operator is revoked (`rm_op` topic). */ +export interface OperatorRevokedEvent { + /** The address that revoked the operator role (topics[1] actor). */ + sender: string; + operator: string; + sequence: bigint; +} + /** A gap detected in the per-contract event sequence — see `DataKey::EventSequence` in contracts/stream/src/events.rs. */ export interface EventGap { /** The sequence number that should have come next. */ @@ -140,6 +192,16 @@ export interface StreamEventHandlers { onResume?: (e: ResumeEvent) => void; onTopUp?: (e: TopUpEvent) => void; onClawback?: (e: ClawbackEvent) => void; + /** Called when the factory deploys a new stream. Optional — most subscribers attach after a stream already exists. */ + onCreated?: (e: CreatedEvent) => void; + /** Called when the recipient force-cancels a paused stream. */ + onForceCancel?: (e: ForceCancelEvent) => void; + /** Called when the recipient role is transferred to a new address — the current subscriber may have just lost the stream. */ + onRecipientTransfer?: (e: RecipientTransferEvent) => void; + /** Called when an operator is delegated on the stream. */ + onOperatorSet?: (e: OperatorSetEvent) => void; + /** Called when a delegated operator is revoked. */ + onOperatorRevoke?: (e: OperatorRevokedEvent) => void; /** Called when an event polling request fails. Polling continues afterward. */ onError?: (error: Error) => void; /** Called when a non-contiguous event sequence is observed (missed events across a poll gap or reconnect). */ From 3ea0cc81a6fa59c9ee5d1f71143689cf99d084a2 Mon Sep 17 00:00:00 2001 From: SarahDoma <108196951+SarahDoma@users.noreply.github.com> Date: Sun, 30 Aug 2026 13:42:09 +0100 Subject: [PATCH 4/7] fix(soroban,streams,types): add configurable inclusion fee (#509) buildContractCallTx() and StreamsModule always submitted transactions with fee: BASE_FEE (100 stroops) and there was no ConduitConfig knob to raise it. Under inclusion-fee pressure (surge pricing, congested ledgers) a 100-stroop bid is not selected; _sendAndPoll then exhausts its poll attempts and throws a misleading "Transaction timed out" instead of a "fee too low" error. Adds ConduitConfig.fee (explicit stroops amount) and ConduitConfig.feeMultiplier (multiple of BASE_FEE), resolved once via the new exported resolveFee() and threaded through buildContractCallTx()'s new optional fee parameter (default BASE_FEE, so existing callers are unaffected) for every submitted (non-read-only) StreamsModule operation: create(), the shared _invoke() path (withdraw/cancel/pause/resume/topUp/transferRecipient/forceCancel), and clawback(). Also fixes a pre-existing, unrelated build break on main: streams.ts used the Signer type without importing it, so `npm run build` and `npm run typecheck` failed unconditionally (`Cannot find name 'Signer'`) before this PR's changes could even be verified by CI. Updates two existing tests' soroban.js mocks/assertions for the new buildContractCallTx() signature (network-switcher-validation.test.ts, streams-success.test.ts). Closes #509 --- src/soroban.ts | 25 ++++++++++++++++++- src/streams.ts | 15 ++++++++--- src/tests/network-switcher-validation.test.ts | 1 + src/tests/streams-success.test.ts | 2 +- src/types/index.ts | 14 +++++++++++ 5 files changed, 52 insertions(+), 5 deletions(-) diff --git a/src/soroban.ts b/src/soroban.ts index 0a931b5..bc32736 100644 --- a/src/soroban.ts +++ b/src/soroban.ts @@ -145,11 +145,33 @@ export function createRpcServer(rpcUrl: string): SorobanRpc.Server { return proxied; } +/** + * Resolve the inclusion (bid) fee, in stroops, for a submitted transaction. + * + * An explicit `fee` always wins. Otherwise `feeMultiplier` scales + * `BASE_FEE`. With neither set, this returns `BASE_FEE` (the network + * minimum) unchanged — the previous, always-hardcoded behaviour. `BASE_FEE` + * alone is not competitive under inclusion-fee pressure (surge pricing, + * congested ledgers): the bid goes unselected, `_sendAndPoll` exhausts its + * poll attempts, and the caller sees a misleading "Transaction timed out" + * instead of "fee too low" (see #509). + */ +export function resolveFee(config: { fee?: string; feeMultiplier?: number }): string { + if (config.fee !== undefined) return config.fee; + if (config.feeMultiplier !== undefined) { + return (BigInt(BASE_FEE) * BigInt(config.feeMultiplier)).toString(); + } + return BASE_FEE; +} + /** * Build a contract-call transaction for simulate or submit. * * Fetches the caller's account from the RPC to get the current sequence * number, then wraps the call in a TransactionBuilder. + * + * @param fee - Inclusion fee in stroops. Defaults to `BASE_FEE`; pass the + * result of {@link resolveFee} to honour `ConduitConfig.fee`/`feeMultiplier`. */ export async function buildContractCallTx( rpcUrl: string, @@ -158,6 +180,7 @@ export async function buildContractCallTx( contractId: string, method: string, args: xdr.ScVal[], + fee: string = BASE_FEE, ): Promise> { const server = createRpcServer(rpcUrl); @@ -171,7 +194,7 @@ export async function buildContractCallTx( const contract = new Contract(contractId); return new TransactionBuilder(account, { - fee: BASE_FEE, + fee, networkPassphrase: passphrase, }) .addOperation(contract.call(method, ...args)) diff --git a/src/streams.ts b/src/streams.ts index b8cb0a8..bc1b947 100644 --- a/src/streams.ts +++ b/src/streams.ts @@ -18,6 +18,7 @@ import type { FeeEstimate, } from './types/index.js'; import type { WalletAdapter } from './adapters/types.js'; +import type { Signer } from './signer.js'; import { KeypairWalletAdapter } from './adapters/keypair.js'; import { toStroops, calculateRate, bigintSafeStringify } from './utils.js'; import { @@ -35,6 +36,7 @@ import { DEFAULT_CONFIRMATION_MAX_ATTEMPTS, DEFAULT_CONFIRMATION_POLL_INTERVAL_MS, createRpcServer, + resolveFee, } from './soroban.js'; import { STREAM_FLAG_PAUSED, @@ -82,6 +84,12 @@ export class StreamsModule { private readonly _factory: FactoryModule; private activeWallet?: WalletAdapter; + /** + * Inclusion (bid) fee, in stroops, for transactions this module submits. + * Resolved once from `config.fee` / `config.feeMultiplier` (see #509). + */ + private readonly _fee: string; + /** * Session-scoped cache of stream ID → contract address resolutions. * Avoids a redundant factory RPC on every get/withdraw/cancel/pause/resume/topUp/clawback call @@ -110,6 +118,7 @@ export class StreamsModule { this.rpcUrl = config.rpcUrl ?? DEFAULT_RPC[config.network]; this.passphrase = NETWORK_PASSPHRASE[config.network]; this._factory = new FactoryModule(config); + this._fee = resolveFee(config); if (config.wallet) { this.activeWallet = config.wallet; @@ -250,7 +259,7 @@ export class StreamsModule { boolToScVal(clawbackEnabled), ]; - const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, factoryId, 'create_stream', args); + const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, factoryId, 'create_stream', args, this._fee); const server = this._server(); const sim = await catchNetworkError('simulateTransaction (create)', server.simulateTransaction(tx)); @@ -450,7 +459,7 @@ export class StreamsModule { this._ensureCanMutate(); const addr = await this._resolveAddr(BigInt(streamId)); const caller = await this._getSenderAddress(); - const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, caller, addr, 'clawback', []); + const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, caller, addr, 'clawback', [], this._fee); const server = this._server(); const sim = await catchNetworkError('simulateTransaction (clawback)', server.simulateTransaction(tx)); @@ -790,7 +799,7 @@ export class StreamsModule { /** Simulate -> assemble -> sign -> submit -> poll. Returns txHash. */ private async _invoke(contractId: string, method: string, args: xdr.ScVal[]): Promise { const senderAddr = await this._getSenderAddress(); - const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, contractId, method, args); + const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, contractId, method, args, this._fee); const server = this._server(); const sim = await catchNetworkError('simulateTransaction (invoke)', server.simulateTransaction(tx)); if (SorobanRpc.Api.isSimulationError(sim)) { diff --git a/src/tests/network-switcher-validation.test.ts b/src/tests/network-switcher-validation.test.ts index 0d50eef..dd47fe4 100644 --- a/src/tests/network-switcher-validation.test.ts +++ b/src/tests/network-switcher-validation.test.ts @@ -31,6 +31,7 @@ import { WalletConnectAdapter } from '../adapters/walletconnect.js'; vi.mock('../soroban.js', () => ({ buildContractCallTx: vi.fn().mockResolvedValue({ _stub: 'tx' }), simulateReadOnly: vi.fn(), + resolveFee: () => '100', scValToU64: (v: { u64: () => { toString: () => string } }) => BigInt(v.u64().toString()), scValToI128: () => 0n, NETWORK_PASSPHRASE: { diff --git a/src/tests/streams-success.test.ts b/src/tests/streams-success.test.ts index bad7c82..d0d81d9 100644 --- a/src/tests/streams-success.test.ts +++ b/src/tests/streams-success.test.ts @@ -491,7 +491,7 @@ describe('StreamsModule.transferRecipient() — success path', () => { await runThroughFirstPoll(() => sdk.transferRecipient(1n, newRecipient)); expect(buildContractCallTx).toHaveBeenCalledWith( expect.any(String), expect.any(String), expect.any(String), - STREAM_ADDR, 'transfer_recipient', [expect.anything()], + STREAM_ADDR, 'transfer_recipient', [expect.anything()], expect.any(String), ); }); }); diff --git a/src/types/index.ts b/src/types/index.ts index 57067cb..4124b28 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -36,6 +36,20 @@ export interface ConduitConfig { confirmationPollIntervalMs?: number; /** Override Soroban confirmation polling attempts; default 30 */ confirmationMaxAttempts?: number; + /** + * Explicit inclusion (bid) fee in stroops for submitted transactions. + * Takes precedence over `feeMultiplier`. Defaults to `BASE_FEE` (100 + * stroops) when neither is set, which is the network minimum and is not + * competitive under inclusion-fee pressure (surge pricing, congested + * ledgers) — see #509. + */ + fee?: string; + /** + * Multiplier applied to `BASE_FEE` to compute the inclusion fee for + * submitted transactions, e.g. `10` bids 10x the network minimum. + * Ignored when `fee` is set. Defaults to `1` (BASE_FEE, unchanged). + */ + feeMultiplier?: number; } export interface StreamInfo { From 40e9b7baee44c640fb372ec0e7120abd1a3acaad Mon Sep 17 00:00:00 2001 From: SarahDoma <108196951+SarahDoma@users.noreply.github.com> Date: Sun, 30 Aug 2026 13:43:15 +0100 Subject: [PATCH 5/7] docs(streams,events,soroban): update api.md and CHANGELOG.md for #504, #506, #509 - Document batchWithdraw() (previously undocumented) and its sequential-submission behaviour. - Document ConduitConfig.fee/feeMultiplier and add an inclusion-fee callout under ConduitConfig. - Document the five new StreamEventHandlers callbacks and their decoded payload shapes in the subscribe() example and event-payload note. - Add CHANGELOG.md entries under [Unreleased] for all three fixes plus the incidental Signer-import build-break fix. --- CHANGELOG.md | 4 ++++ docs/api.md | 63 ++++++++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 63 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a1aa851..141c765 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,8 @@ All notable changes are documented here. Format based on [Keep a Changelog](http - Direct unit tests for `resolvePassphrase()` covering explicit passphrase present/blank, named network known/unknown, and neither-provided branches (#462) - `StreamsModule.forceCancel()` — wraps the contract's `force_cancel()` so a recipient can force-cancel a stream paused beyond the 30-day threshold (previously only a prose TODO; `StreamErrorCode.PauseThresholdNotMet` is now reachable through the SDK) (#453) - `StreamsModule.transferRecipient()` — wraps the contract's `transfer_recipient()` so the current recipient can reassign the recipient role (previously only a prose TODO) (#454) +- `ConduitConfig.fee` / `ConduitConfig.feeMultiplier` — configure the inclusion (bid) fee for submitted transactions, via `resolveFee()` (exported from `src/soroban.ts`) and a new `fee` parameter on `buildContractCallTx()`. Previously every transaction hardcoded `fee: BASE_FEE` with no override, so under inclusion-fee pressure a 100-stroop bid would never be selected (#509) +- `StreamEventHandlers.onCreated`, `.onForceCancel`, `.onRecipientTransfer`, `.onOperatorSet`, `.onOperatorRevoke` — `dispatchEvent()` previously silently ignored the `created`, `force_cxl`, `xfer_rec`, `set_op`, and `rm_op` contract topics, so a live subscriber was never notified when the recipient role was transferred, a paused stream was force-cancelled by the recipient, or an operator was delegated/revoked (#506) ### Performance - `FactoryModule.streamAddress()` now caches resolved stream→contract-address lookups in-memory, since the mapping is fixed at stream creation and never changes. Eliminates redundant RPC round trips on every `StreamsModule` read/write operation (`get`, `withdraw`, `cancel`, `pause`, `resume`, `topUp`, `clawback`) and on each page of `list()`, which previously re-resolved the same address for every stream on every call. @@ -56,6 +58,8 @@ All notable changes are documented here. Format based on [Keep a Changelog](http - `catchNetworkError()` no longer reclassifies *any* `TypeError` whose text happens to contain `fetch`/`connect`/`network`/etc. It now only reclassifies errors that are provably transport failures: the canonical fetch/axios network messages (`fetch failed`, `Failed to fetch`, `Network Error`, `Load failed`) or an error (or its nested `cause`) carrying a network errno code such as `ECONNREFUSED`/`ENOTFOUND`/`ERR_NETWORK`. A programming `TypeError` (e.g. `Cannot read properties of undefined (reading 'connect')`) is re-thrown as-is instead of being masked as a network outage (#457). - `NonceManager` now throws a descriptive error for an unparseable nonce string (e.g. `startNonce: 'not-a-number'`) instead of silently coercing it to `0n`, which masked caller bugs as an explicit zero (#458). - `StreamBuilder.build()` now stringifies a numeric `ratePerSecond` so the runtime value matches the declared `ratePerSecond?: string` return type; previously a `number` input passed through unchanged, so callers trusting the type (`.trim()`, string concatenation) hit runtime errors (#459). +- `StreamsModule.batchWithdraw()` now submits withdrawals one at a time instead of firing them all concurrently via `Promise.allSettled(...map)`. Concurrent submission meant every withdrawal read the same account sequence number from `getAccount()` and built a transaction with the same value — with a single keypair/wallet, at most one of the N withdrawals could ever land, the rest failing with `txBAD_SEQ`, defeating the whole purpose of a batch (#504) +- `src/streams.ts` was missing its `Signer` type import, so `npm run build` / `npm run typecheck` failed unconditionally on `main` (`Cannot find name 'Signer'`) — added the missing `import type { Signer } from './signer.js'`. - `StreamsModule.withdraw()` / `topUp()` now reject `amount <= 0n` client-side (before any RPC round-trip), matching `create()`'s fail-fast validation philosophy instead of relying on the contract's `InvalidAmount` simulate+reject cycle (#451) - `StreamsModule.list()` no longer silently drops `recipient` when both `sender` and `recipient` are provided — it now returns the de-duplicated union of both filters (#452) diff --git a/docs/api.md b/docs/api.md index 2a2463f..676ac30 100644 --- a/docs/api.md +++ b/docs/api.md @@ -20,6 +20,17 @@ new ConduitClient(config: ConduitConfig) | `factoryAddress` | `string` | | Deployed factory | | `governorAddress` | `string` | | Deployed governor | | `wallet` | `WalletAdapter` | | — | +| `fee` | `string` | | Explicit inclusion (bid) fee in stroops for submitted transactions. Takes precedence over `feeMultiplier`. Defaults to `BASE_FEE` (100 stroops) | +| `feeMultiplier` | `number` | | Multiplier applied to `BASE_FEE` to compute the inclusion fee, e.g. `10` bids 10x the network minimum. Ignored when `fee` is set | + +> **Inclusion fee.** Every submitted transaction previously used `BASE_FEE` +> (the network minimum, 100 stroops) unconditionally, with no way to raise +> it. Under inclusion-fee pressure (surge pricing, congested ledgers) a +> 100-stroop bid is not selected, and the SDK's confirmation polling +> eventually throws a misleading `Transaction timed out` instead of +> surfacing "fee too low". Set `fee` (an exact stroops amount) or +> `feeMultiplier` (a multiple of `BASE_FEE`) on `ConduitConfig` to bid +> higher; `fee` wins if both are set. ### Convenience methods @@ -86,6 +97,38 @@ const total = await client.streams.streamedTotal(streamId); --- +### `batchWithdraw(withdrawals) → Promise` + +| Param | Type | Notes | +|-------|------|-------| +| `withdrawals` | `{ streamId: bigint \| string; amount?: bigint }[]` | One entry per stream to withdraw from | + +```typescript +interface BatchWithdrawResult { + streamId: bigint; + success: boolean; + txHash?: string; // present when success is true + error?: string; // present when success is false +} +``` + +Withdraws from multiple streams. Soroban permits only one +`invoke_host_function` operation per transaction, so this cannot be a single +atomic transaction — each withdrawal is submitted as its own transaction and +reported independently, so a failure on one streamId (e.g. +`StreamNotFound`, insufficient balance) does not fail the others. + +Withdrawals are submitted **one at a time, not concurrently**. Each +transaction is built from the caller account's current sequence number; +submitting all N withdrawals at once means every one of them would read the +same sequence number and collide, so with a single keypair/wallet at most +one could ever land on-chain. Awaiting each withdrawal in turn guarantees +every submission gets a distinct, ordered sequence number. + +**Requires:** `keypair`, `wallet`, or `signer` set (recipient) + +--- + ### `cancel(streamId) → Promise` Atomically settles both parties (recipient gets owed amount, sender gets refund). @@ -203,6 +246,11 @@ const sub = client.streams.subscribe(streamId, { onResume: e => console.log('Resumed at:', e.resumedAt), onTopUp: e => console.log('Topped up:', e.amount), onClawback: e => console.log('Clawback:', e.amount), + onCreated: e => console.log('Created:', e.recipient), + onForceCancel: e => console.log('Force-cancelled by:', e.recipient), + onRecipientTransfer: e => console.log('Recipient transferred to:', e.newRecipient), + onOperatorSet: e => console.log('Operator set:', e.operator), + onOperatorRevoke: e => console.log('Operator revoked:', e.operator), onError: err => console.warn('poll failed:', err), pollInterval: 3000, // ms; default 5000 maxBackoffMs: 60_000, // cap on exponential backoff after failures; default 60_000 @@ -222,10 +270,17 @@ sub.unsubscribe(); > **All event payload fields are decoded.** `src/events.ts`'s `dispatchEvent()` parses every > multi-field event from its tuple `ScVal`s: `onWithdraw` → `{ recipient, amount, > totalWithdrawn, remaining }`, `onCancel` → `{ sender, refundAmount, withdrawnSoFar }`, -> `onPause` → `{ sender, pausedAt, withdrawable }`, `onTopUp` → `{ sender, amount, newBalance }`. -> Single-field events are decoded from their bare scalar: `onResume` → `{ sender, resumedAt }`, -> `onClawback` → `{ sender, amount }`. These fields are real values from the chain — no -> placeholder `0`/`0n` values remain. +> `onPause` → `{ sender, pausedAt, withdrawable }`, `onTopUp` → `{ sender, amount, newBalance }`, +> `onCreated` → `{ sender, recipient, token, depositAmount, ratePerSecond, startTime, endTime }`, +> `onForceCancel` → `{ recipient, payoutAmount, refundAmount }`. Single-field events are decoded +> from their bare scalar: `onResume` → `{ sender, resumedAt }`, `onClawback` → `{ sender, amount }`, +> `onRecipientTransfer` → `{ previousRecipient, newRecipient }`, `onOperatorSet` → `{ sender, +> operator }`, `onOperatorRevoke` → `{ sender, operator }`. These fields are real values from the +> chain — no placeholder `0`/`0n` values remain. +> +> `onRecipientTransfer`, `onForceCancel`, `onOperatorSet`, and `onOperatorRevoke` were previously +> silently ignored — a live subscriber was never notified when the recipient role moved to +> another address, a paused stream was force-cancelled, or an operator was delegated/revoked. --- From 96c777f456b80353f46075e203dfeb27b55b8681 Mon Sep 17 00:00:00 2001 From: SarahDoma <108196951+SarahDoma@users.noreply.github.com> Date: Sun, 30 Aug 2026 13:53:55 +0100 Subject: [PATCH 6/7] feat(module44): add stream liquidity-risk / runway calculator (#378) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements Feature #44 (issue #378), an SDK-side helper for flagging streams that are about to run out of scheduled balance: - assessSingleItem()/assessBatch() classify each stream's remaining runway (seconds until endTime) into 'inactive' | 'critical' | 'warning' | 'healthy', with configurable thresholds. - estimateTopUpNeeded() computes the stroops needed via top_up() to reach a target runway. - Follows the shared LruMemoCache pattern already used by Module26/Module36/Module48, and reports getPerformanceMetrics()'s measuredSpeedupPercent as an honest, workload-dependent measurement from this instance's own accumulated hit/miss timings — never a fixed assumed percentage (#378 asked for "improve performance by 20%", which this repo has already established is not a claim that can be made honestly without a specific, measured workload; see the same rationale applied to Module26/Module36 in CHANGELOG.md). - 26 unit tests covering constructor validation, all four risk classifications (including open-ended and already-ended streams), cache hit/miss/eviction/bypass behaviour, batch chunking, and estimateTopUpNeeded()'s edge cases — 98.5% statement / 100% function coverage on module44.ts. Closes #378 --- src/index.ts | 9 ++ src/module44.ts | 211 ++++++++++++++++++++++++++++++++ src/tests/module44.test.ts | 245 +++++++++++++++++++++++++++++++++++++ 3 files changed, 465 insertions(+) create mode 100644 src/module44.ts create mode 100644 src/tests/module44.test.ts diff --git a/src/index.ts b/src/index.ts index 475ef00..e0f62a9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -97,3 +97,12 @@ export type { Module49Metrics, } from './module49.js'; +export { Module44 } from './module44.js'; +export type { + Module44Config, + StreamRiskItem, + LiquidityRiskLevel, + StreamRiskAssessment, + Module44Metrics, +} from './module44.js'; + diff --git a/src/module44.ts b/src/module44.ts new file mode 100644 index 0000000..c282447 --- /dev/null +++ b/src/module44.ts @@ -0,0 +1,211 @@ +import type { StreamInfo } from './types/index.js'; +import { LruMemoCache } from './lru-memo-cache.js'; + +export interface Module44Config { + /** Maximum number of risk assessments retained in the LRU cache */ + cacheSize?: number; + /** Enable memoized assessment; actual speedup depends on hit rate — see `getPerformanceMetrics()` for a measured value */ + enableOptimization?: boolean; + /** Preferred chunk size for batch processing stream items */ + batchChunkSize?: number; + /** Runway (seconds) below which a stream is classified 'critical'. Default 86400 (1 day). */ + criticalThresholdSecs?: number; + /** Runway (seconds) below which a stream is classified 'warning'. Default 604800 (7 days). */ + warningThresholdSecs?: number; +} + +export interface StreamRiskItem { + id: string; + stream: StreamInfo; + /** Observation timestamp (unix seconds) */ + timestamp?: number; +} + +export type LiquidityRiskLevel = 'inactive' | 'critical' | 'warning' | 'healthy'; + +export interface StreamRiskAssessment { + id: string; + /** + * Seconds of streaming remaining before the stream reaches `endTime`. + * `null` for an open-ended stream (`endTime === 0`) — its runway is not + * bounded by a schedule, only by the sender keeping the balance topped up. + */ + runwaySecs: number | null; + riskLevel: LiquidityRiskLevel; + isCached: boolean; + computedAt: number; +} + +export interface Module44Metrics { + totalAssessed: number; + cacheHits: number; + cacheMisses: number; + /** + * Measured, not assumed: `(avgMissMs - avgHitMs) / avgMissMs * 100`, based + * on this instance's own accumulated timings. `null` until at least one + * hit and one miss have both been recorded (nothing to compare yet). + */ + measuredSpeedupPercent: number | null; + averageExecutionTimeMs: number; +} + +const DEFAULT_CRITICAL_THRESHOLD_SECS = 86_400; // 1 day +const DEFAULT_WARNING_THRESHOLD_SECS = 604_800; // 7 days + +/** + * Module 44: stream liquidity-risk / runway calculator. + * + * Implements Feature #44 with LRU-memoized runway and risk-level assessment + * per stream, so a dashboard can flag streams that are about to run out of + * scheduled balance. Speedup from caching is workload-dependent + * (proportional to cache hit rate); call `getPerformanceMetrics()` for this + * instance's own measured hit/miss timing rather than assuming a fixed + * percentage. + */ +export class Module44 { + private readonly enableOptimization: boolean; + private readonly batchChunkSize: number; + private readonly criticalThresholdSecs: number; + private readonly warningThresholdSecs: number; + + private readonly cache: LruMemoCache; + private totalAssessed = 0; + private totalExecutionTimeMs = 0; + + constructor(config: Module44Config = {}) { + this.cache = new LruMemoCache(config.cacheSize ?? 1000); + this.enableOptimization = config.enableOptimization ?? true; + this.batchChunkSize = config.batchChunkSize ?? 50; + this.criticalThresholdSecs = config.criticalThresholdSecs ?? DEFAULT_CRITICAL_THRESHOLD_SECS; + this.warningThresholdSecs = config.warningThresholdSecs ?? DEFAULT_WARNING_THRESHOLD_SECS; + + if (this.criticalThresholdSecs < 0) { + throw new Error('Module44: criticalThresholdSecs cannot be negative'); + } + if (this.warningThresholdSecs <= this.criticalThresholdSecs) { + throw new Error('Module44: warningThresholdSecs must be greater than criticalThresholdSecs'); + } + } + + /** + * Assess a batch of streams' liquidity risk with fast-path cache lookup. + */ + public assessBatch(items: StreamRiskItem[]): StreamRiskAssessment[] { + const results: StreamRiskAssessment[] = new Array(items.length); + + for (let i = 0; i < items.length; i += this.batchChunkSize) { + const chunkEnd = Math.min(i + this.batchChunkSize, items.length); + for (let j = i; j < chunkEnd; j++) { + const item = items[j]; + if (!item) continue; + + results[j] = this.assessSingleItem(item); + } + } + + return results; + } + + /** + * Assess a single stream's runway and liquidity risk level. + */ + public assessSingleItem(item: StreamRiskItem): StreamRiskAssessment { + const start = performance.now(); + const nowSec = item.timestamp ?? Math.floor(Date.now() / 1000); + const cacheKey = `${item.id}_${item.stream.paused ? 1 : 0}_${item.stream.cancelled ? 1 : 0}_${item.stream.ratePerSecond.toString()}_${item.stream.endTime}_${nowSec}`; + + if (this.enableOptimization) { + const cached = this.cache.get(cacheKey); + if (cached) { + const elapsed = performance.now() - start; + this.cache.recordHit(elapsed); + this.totalAssessed++; + this.totalExecutionTimeMs += elapsed; + return { ...cached, isCached: true }; + } + } + + const result = this.computeAssessment(item.id, item.stream, nowSec); + + if (this.enableOptimization) { + this.cache.set(cacheKey, result); + } + + const elapsed = performance.now() - start; + this.cache.recordMiss(elapsed); + this.totalAssessed++; + this.totalExecutionTimeMs += elapsed; + return result; + } + + private computeAssessment(id: string, stream: StreamInfo, nowSec: number): StreamRiskAssessment { + if (stream.cancelled || stream.paused || stream.ratePerSecond <= 0n) { + return { id, runwaySecs: 0, riskLevel: 'inactive', isCached: false, computedAt: nowSec }; + } + + if (stream.endTime === 0) { + // Open-ended: not bounded by a schedule, so treat as healthy — the + // sender is expected to keep the balance topped up via top_up(). + return { id, runwaySecs: null, riskLevel: 'healthy', isCached: false, computedAt: nowSec }; + } + + const runwaySecs = Math.max(0, stream.endTime - nowSec); + let riskLevel: LiquidityRiskLevel; + if (runwaySecs <= 0) { + riskLevel = 'inactive'; + } else if (runwaySecs < this.criticalThresholdSecs) { + riskLevel = 'critical'; + } else if (runwaySecs < this.warningThresholdSecs) { + riskLevel = 'warning'; + } else { + riskLevel = 'healthy'; + } + + return { id, runwaySecs, riskLevel, isCached: false, computedAt: nowSec }; + } + + /** + * Amount (in stroops) that must be added via `top_up()` for the stream's + * runway to reach `targetRunwaySecs` from `nowSec`. Returns `0n` when the + * stream is already inactive/paused/cancelled or already meets the target. + */ + public estimateTopUpNeeded( + stream: StreamInfo, + targetRunwaySecs: number, + nowSec = Math.floor(Date.now() / 1000), + ): bigint { + if (targetRunwaySecs <= 0 || stream.ratePerSecond <= 0n || stream.cancelled || stream.paused) { + return 0n; + } + + const currentRunwaySecs = stream.endTime === 0 ? Infinity : Math.max(0, stream.endTime - nowSec); + if (currentRunwaySecs >= targetRunwaySecs) return 0n; + + const deficitSecs = Math.ceil(targetRunwaySecs - currentRunwaySecs); + return stream.ratePerSecond * BigInt(deficitSecs); + } + + /** + * Reset performance cache and internal state. + */ + public clearCache(): void { + this.cache.clear(); + this.totalAssessed = 0; + this.totalExecutionTimeMs = 0; + } + + /** + * Retrieve performance metrics, including a measured (not assumed) cache speedup. + */ + public getPerformanceMetrics(): Module44Metrics { + const { cacheHits, cacheMisses, measuredSpeedupPercent } = this.cache.metrics(); + + return { + totalAssessed: this.totalAssessed, + cacheHits, + cacheMisses, + measuredSpeedupPercent: this.enableOptimization ? measuredSpeedupPercent : null, + averageExecutionTimeMs: this.totalAssessed > 0 ? this.totalExecutionTimeMs / this.totalAssessed : 0, + }; + } +} diff --git a/src/tests/module44.test.ts b/src/tests/module44.test.ts new file mode 100644 index 0000000..92a080d --- /dev/null +++ b/src/tests/module44.test.ts @@ -0,0 +1,245 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import { Module44 } from '../module44.js'; +import type { StreamInfo } from '../types/index.js'; + +describe('Module44 (SDK Feature #44)', () => { + let module44: Module44; + const now = 1_000_000; + + const mockStream: StreamInfo = { + id: 1n, + address: 'CCSTREAM44ADDRESSXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX', + sender: 'GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFXYCZLYC3ZCHB2D4P3CF', + recipient: 'GA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVSGZ', + token: 'native', + ratePerSecond: 100n, + startTime: now - 500, + endTime: now + 500, + withdrawn: 0n, + paused: false, + pausedAt: 0, + cancelled: false, + clawbackEnabled: false, + }; + + beforeEach(() => { + module44 = new Module44({ cacheSize: 10, batchChunkSize: 5 }); + }); + + describe('Constructor & Configuration', () => { + it('initializes with default options and no speedup measurement yet', () => { + const defaultMod = new Module44(); + const metrics = defaultMod.getPerformanceMetrics(); + expect(metrics.totalAssessed).toBe(0); + expect(metrics.measuredSpeedupPercent).toBeNull(); + }); + + it('never reports a speedup when optimization is disabled', () => { + const customMod = new Module44({ enableOptimization: false }); + const item = { id: 'stream-1', stream: mockStream, timestamp: now }; + customMod.assessSingleItem(item); + customMod.assessSingleItem(item); + expect(customMod.getPerformanceMetrics().measuredSpeedupPercent).toBeNull(); + }); + + it('throws when criticalThresholdSecs is negative', () => { + expect(() => new Module44({ criticalThresholdSecs: -1 })).toThrow(/criticalThresholdSecs/); + }); + + it('throws when warningThresholdSecs does not exceed criticalThresholdSecs', () => { + expect(() => new Module44({ criticalThresholdSecs: 100, warningThresholdSecs: 100 })).toThrow(/warningThresholdSecs/); + expect(() => new Module44({ criticalThresholdSecs: 100, warningThresholdSecs: 50 })).toThrow(/warningThresholdSecs/); + }); + }); + + describe('assessSingleItem — risk classification', () => { + it('classifies a cancelled stream as inactive with zero runway', () => { + const item = { id: 's1', stream: { ...mockStream, cancelled: true }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result).toMatchObject({ id: 's1', runwaySecs: 0, riskLevel: 'inactive', isCached: false, computedAt: now }); + }); + + it('classifies a paused stream as inactive with zero runway', () => { + const item = { id: 's1', stream: { ...mockStream, paused: true }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.riskLevel).toBe('inactive'); + expect(result.runwaySecs).toBe(0); + }); + + it('classifies a zero-rate stream as inactive', () => { + const item = { id: 's1', stream: { ...mockStream, ratePerSecond: 0n }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.riskLevel).toBe('inactive'); + }); + + it('classifies an open-ended stream as healthy with null runway', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: 0 }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.runwaySecs).toBeNull(); + expect(result.riskLevel).toBe('healthy'); + }); + + it('classifies runway already past endTime as inactive', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: now - 1 }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.runwaySecs).toBe(0); + expect(result.riskLevel).toBe('inactive'); + }); + + it('classifies runway below the critical threshold as critical', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: now + 3600 }, timestamp: now }; // 1 hour + const result = module44.assessSingleItem(item); + expect(result.runwaySecs).toBe(3600); + expect(result.riskLevel).toBe('critical'); + }); + + it('classifies runway between critical and warning thresholds as warning', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: now + 172_800 }, timestamp: now }; // 2 days + const result = module44.assessSingleItem(item); + expect(result.riskLevel).toBe('warning'); + }); + + it('classifies runway above the warning threshold as healthy', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: now + 1_000_000 }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.riskLevel).toBe('healthy'); + }); + + it('honours custom critical/warning thresholds', () => { + const custom = new Module44({ criticalThresholdSecs: 10, warningThresholdSecs: 20 }); + const item = { id: 's1', stream: { ...mockStream, endTime: now + 15 }, timestamp: now }; + expect(custom.assessSingleItem(item).riskLevel).toBe('warning'); + }); + + it('uses system current time if timestamp is omitted', () => { + const item = { id: 's1', stream: mockStream }; + const result = module44.assessSingleItem(item); + expect(result.computedAt).toBeGreaterThan(0); + }); + }); + + describe('Optimization & Caching', () => { + it('serves cached results on duplicate evaluation requests', () => { + const item = { id: 's1', stream: mockStream, timestamp: now }; + + const firstPass = module44.assessSingleItem(item); + expect(firstPass.isCached).toBe(false); + + const secondPass = module44.assessSingleItem(item); + expect(secondPass.isCached).toBe(true); + expect(secondPass.riskLevel).toBe(firstPass.riskLevel); + + const metrics = module44.getPerformanceMetrics(); + expect(metrics.cacheHits).toBe(1); + expect(metrics.cacheMisses).toBe(1); + expect( + metrics.measuredSpeedupPercent === null || typeof metrics.measuredSpeedupPercent === 'number', + ).toBe(true); + }); + + it('evicts oldest cache item when cacheSize threshold is reached', () => { + const smallCacheMod = new Module44({ cacheSize: 2 }); + + smallCacheMod.assessSingleItem({ id: 'item-1', stream: mockStream, timestamp: 1000 }); + smallCacheMod.assessSingleItem({ id: 'item-2', stream: mockStream, timestamp: 1001 }); + smallCacheMod.assessSingleItem({ id: 'item-3', stream: mockStream, timestamp: 1002 }); + + const reQuery = smallCacheMod.assessSingleItem({ id: 'item-1', stream: mockStream, timestamp: 1000 }); + expect(reQuery.isCached).toBe(false); + }); + + it('bypasses cache when optimization is disabled', () => { + const unoptimizedMod = new Module44({ enableOptimization: false }); + const item = { id: 's1', stream: mockStream, timestamp: now }; + + unoptimizedMod.assessSingleItem(item); + const secondPass = unoptimizedMod.assessSingleItem(item); + + expect(secondPass.isCached).toBe(false); + expect(unoptimizedMod.getPerformanceMetrics().measuredSpeedupPercent).toBeNull(); + }); + }); + + describe('assessBatch', () => { + it('assesses a batch of stream items in chunked iterations', () => { + const batchItems = Array.from({ length: 12 }, (_, i) => ({ + id: `stream-${i}`, + stream: { ...mockStream, id: BigInt(i) }, + timestamp: now, + })); + + const results = module44.assessBatch(batchItems); + + expect(results).toHaveLength(12); + expect(results[0]?.id).toBe('stream-0'); + expect(results[11]?.id).toBe('stream-11'); + + expect(module44.getPerformanceMetrics().totalAssessed).toBe(12); + }); + + it('reports the same totalAssessed whether items go through assessBatch or assessSingleItem directly', () => { + const viaSingle = new Module44({ cacheSize: 10 }); + viaSingle.assessSingleItem({ id: 'a', stream: mockStream, timestamp: now }); + viaSingle.assessSingleItem({ id: 'b', stream: { ...mockStream, id: 2n }, timestamp: now }); + expect(viaSingle.getPerformanceMetrics().totalAssessed).toBe(2); + + const viaBatch = new Module44({ cacheSize: 10 }); + viaBatch.assessBatch([ + { id: 'a', stream: mockStream, timestamp: now }, + { id: 'b', stream: { ...mockStream, id: 2n }, timestamp: now }, + ]); + expect(viaBatch.getPerformanceMetrics().totalAssessed).toBe(2); + }); + }); + + describe('estimateTopUpNeeded', () => { + it('returns 0n when the target runway is already met', () => { + const stream = { ...mockStream, endTime: now + 10_000 }; + expect(module44.estimateTopUpNeeded(stream, 100, now)).toBe(0n); + }); + + it('computes the exact deficit amount in stroops for a bounded stream', () => { + const stream = { ...mockStream, endTime: now + 100, ratePerSecond: 50n }; + // Needs runway of 200s but only has 100s -> 100s deficit * 50/s = 5000n + expect(module44.estimateTopUpNeeded(stream, 200, now)).toBe(5000n); + }); + + it('returns 0n for an open-ended stream (runway already infinite)', () => { + const stream = { ...mockStream, endTime: 0 }; + expect(module44.estimateTopUpNeeded(stream, 1_000_000, now)).toBe(0n); + }); + + it('returns 0n for non-positive target runway', () => { + expect(module44.estimateTopUpNeeded(mockStream, 0, now)).toBe(0n); + expect(module44.estimateTopUpNeeded(mockStream, -10, now)).toBe(0n); + }); + + it('returns 0n for a zero-rate, paused, or cancelled stream', () => { + expect(module44.estimateTopUpNeeded({ ...mockStream, ratePerSecond: 0n }, 1000, now)).toBe(0n); + expect(module44.estimateTopUpNeeded({ ...mockStream, paused: true }, 1000, now)).toBe(0n); + expect(module44.estimateTopUpNeeded({ ...mockStream, cancelled: true }, 1000, now)).toBe(0n); + }); + + it('defaults nowSec to the current time when omitted', () => { + const stream = { ...mockStream, endTime: 0 }; + expect(() => module44.estimateTopUpNeeded(stream, 1000)).not.toThrow(); + }); + }); + + describe('clearCache & Metrics', () => { + it('resets cache state and performance counters', () => { + const item = { id: 's1', stream: mockStream, timestamp: now }; + module44.assessSingleItem(item); + module44.assessSingleItem(item); + + expect(module44.getPerformanceMetrics().cacheHits).toBe(1); + + module44.clearCache(); + + const freshMetrics = module44.getPerformanceMetrics(); + expect(freshMetrics.cacheHits).toBe(0); + expect(freshMetrics.cacheMisses).toBe(0); + expect(freshMetrics.totalAssessed).toBe(0); + }); + }); +}); From 4a8facf075ba4110b90b1d0241756217616295d7 Mon Sep 17 00:00:00 2001 From: SarahDoma <108196951+SarahDoma@users.noreply.github.com> Date: Sun, 30 Aug 2026 13:54:01 +0100 Subject: [PATCH 7/7] docs(module44): document Module44 API and add CHANGELOG entry (#378) Adds a "Module44 (Feature #44)" section to docs/api.md mirroring the existing Module26/Module36/Module48/Module49 sections, and a CHANGELOG.md [Unreleased] entry under Added. --- CHANGELOG.md | 1 + docs/api.md | 28 ++++++++++++++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 141c765..4244663 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ All notable changes are documented here. Format based on [Keep a Changelog](http - `StreamsModule.transferRecipient()` — wraps the contract's `transfer_recipient()` so the current recipient can reassign the recipient role (previously only a prose TODO) (#454) - `ConduitConfig.fee` / `ConduitConfig.feeMultiplier` — configure the inclusion (bid) fee for submitted transactions, via `resolveFee()` (exported from `src/soroban.ts`) and a new `fee` parameter on `buildContractCallTx()`. Previously every transaction hardcoded `fee: BASE_FEE` with no override, so under inclusion-fee pressure a 100-stroop bid would never be selected (#509) - `StreamEventHandlers.onCreated`, `.onForceCancel`, `.onRecipientTransfer`, `.onOperatorSet`, `.onOperatorRevoke` — `dispatchEvent()` previously silently ignored the `created`, `force_cxl`, `xfer_rec`, `set_op`, and `rm_op` contract topics, so a live subscriber was never notified when the recipient role was transferred, a paused stream was force-cancelled by the recipient, or an operator was delegated/revoked (#506) +- `Module44` stream liquidity-risk / runway calculator for Feature #44 (#378); follows `Module26`/`Module36`/`Module48`'s shared `LruMemoCache` pattern and reports an honest, workload-dependent `measuredSpeedupPercent` rather than a fixed percentage. Classifies each stream's remaining runway into `'inactive' | 'critical' | 'warning' | 'healthy'` and exposes `estimateTopUpNeeded()` to compute the stroops needed to reach a target runway ### Performance - `FactoryModule.streamAddress()` now caches resolved stream→contract-address lookups in-memory, since the mapping is fixed at stream creation and never changes. Eliminates redundant RPC round trips on every `StreamsModule` read/write operation (`get`, `withdraw`, `cancel`, `pause`, `resume`, `topUp`, `clawback`) and on each page of `list()`, which previously re-resolved the same address for every stream on every call. diff --git a/docs/api.md b/docs/api.md index 676ac30..4525ae4 100644 --- a/docs/api.md +++ b/docs/api.md @@ -838,4 +838,32 @@ new Module49(config?: Module49Config) * `clearCache(): void` - Clears the internal lookup cache and metrics. * `getPerformanceMetrics(): Module49Metrics` - Returns real-time metrics (`totalProcessed`, `cacheHits`, `cacheMisses`, `hitRate`, `averageExecutionTimeMs`). +--- + +## `Module44` (Feature #44) + +Stream liquidity-risk / runway calculator implementing Feature #44. For each stream, computes the remaining `runwaySecs` until `endTime` and classifies it into a `LiquidityRiskLevel` (`'inactive' | 'critical' | 'warning' | 'healthy'`), so a dashboard can flag streams about to run out of scheduled balance. Uses the same LRU-memoized lookup pattern as `Module26`/`Module36`/`Module48`; actual speedup is workload-dependent (proportional to cache hit rate) — see `getPerformanceMetrics()` for a measured value, never a fixed assumed percentage. + +### Constructor + +```typescript +new Module44(config?: Module44Config) +``` + +| Option | Type | Default | Notes | +|--------|------|---------|-------| +| `cacheSize` | `number` | `1000` | Max entries in memoization cache | +| `enableOptimization` | `boolean` | `true` | Enables memoized lookup caching | +| `batchChunkSize` | `number` | `50` | Stream chunk size for batch processing | +| `criticalThresholdSecs` | `number` | `86400` (1 day) | Runway below this is `'critical'`. Throws if negative | +| `warningThresholdSecs` | `number` | `604800` (7 days) | Runway below this (and at/above `criticalThresholdSecs`) is `'warning'`. Throws if not greater than `criticalThresholdSecs` | + +### Methods + +* `assessSingleItem(item: StreamRiskItem): StreamRiskAssessment` — Computes one stream's `runwaySecs` and `riskLevel`. A cancelled, paused, or zero-rate stream is `'inactive'` with `runwaySecs: 0`. An open-ended stream (`endTime === 0`) is `'healthy'` with `runwaySecs: null` — its runway isn't bounded by a schedule, only by the sender keeping the balance topped up. +* `assessBatch(items: StreamRiskItem[]): StreamRiskAssessment[]` — Assesses an array of streams in `batchChunkSize` chunks. +* `estimateTopUpNeeded(stream: StreamInfo, targetRunwaySecs: number, nowSec?: number): bigint` — Stroops needed via `top_up()` for the stream's runway to reach `targetRunwaySecs`. Returns `0n` if the stream is inactive/paused/cancelled, the target is non-positive, or the target is already met (including any open-ended stream, whose runway is treated as unbounded). +* `clearCache(): void` — Clears the internal lookup cache and metrics. +* `getPerformanceMetrics(): Module44Metrics` — Returns `totalAssessed`, `cacheHits`, `cacheMisses`, `averageExecutionTimeMs`, and `measuredSpeedupPercent` (a real measurement derived from this instance's own accumulated hit/miss timings, `null` until both have occurred at least once). +