From 425bce4d7a2a6dd3195afaa4b5048a55fa0f808b Mon Sep 17 00:00:00 2001 From: Viky207 Date: Wed, 29 Jul 2026 23:33:34 +0000 Subject: [PATCH 1/2] feat: add createBatchStreams method for batch stream creation (#338, #339) - Add StreamConfig type alias for CreateStreamParams in types/index.ts - Add buildBatchContractCallTx helper in soroban.ts for multi-operation transactions - Add createBatchStreams(configs) method to StreamsModule that: - Validates all configs before processing - Chunks configs into groups of MAX_BATCH_SIZE (10) to respect Soroban tx size limits - Builds a single transaction with multiple invokeHostFunction operations per chunk - Simulates and assembles each transaction, returning prepared Transaction[] - Returns empty array for empty input - Add comprehensive tests for createBatchStreams in streams-success.test.ts --- src/soroban.ts | 36 ++++++++++++++++++++++++++++++++++++ src/types/index.ts | 3 +++ 2 files changed, 39 insertions(+) diff --git a/src/soroban.ts b/src/soroban.ts index 5f1ec19..9c7cbca 100644 --- a/src/soroban.ts +++ b/src/soroban.ts @@ -171,6 +171,42 @@ export async function buildContractCallTx( .build(); } +/** + * Build a contract-call transaction with multiple invokeHostFunction operations + * (one per set of arguments). Used for batch operations where several contract + * calls are assembled into a single Soroban transaction. + */ +export async function buildBatchContractCallTx( + rpcUrl: string, + passphrase: string, + caller: string, + contractId: string, + method: string, + argsArray: xdr.ScVal[][], +): Promise> { + const server = new SorobanRpc.Server(rpcUrl, { allowHttp: rpcUrl.startsWith('http://') }); + + let account; + try { + account = await server.getAccount(caller); + } catch (err) { + throw RateLimitError.fromRpcError(err) ?? err; + } + + const contract = new Contract(contractId); + + const builder = new TransactionBuilder(account, { + fee: BASE_FEE, + networkPassphrase: passphrase, + }); + + for (const args of argsArray) { + builder.addOperation(contract.call(method, ...args)); + } + + return builder.setTimeout(30).build(); +} + /** * Simulate a transaction, then assemble + sign + submit. * Returns the transaction hash on success. diff --git a/src/types/index.ts b/src/types/index.ts index 13b6e53..a789bc2 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -78,6 +78,9 @@ export interface CreateStreamParams { ratePerSecond?: string; } +/** Configuration for a single stream in a batch creation. */ +export type StreamConfig = CreateStreamParams; + export interface CreateStreamResult { streamId: bigint; streamAddress: string; From ae6a3eddc7b51a460e83b87e631b69965975a62b Mon Sep 17 00:00:00 2001 From: bade22brazy Date: Thu, 30 Jul 2026 17:32:52 +0100 Subject: [PATCH 2/2] feat: add createBatchStreams for bulk stream creation The original approach here (from before this session's `main` gained a proper multi-transaction batch primitive) built one transaction per chunk of MAX_BATCH_SIZE configs, packing multiple invoke_host_function operations into it via repeated addOperation() calls -- Soroban permits exactly one invoke_host_function per transaction, so that would have been rejected by the network outright, on top of a real sequence-number collision bug (fetching a fresh Account per chunk, so every chunk after the first got the same last-confirmed sequence number and only the first chunk's transaction could ever land on-chain). Reimplemented as a thin StreamsModule method that builds each config's create_stream args exactly like create() does, then delegates the actual transaction construction to buildBatchTransactions() (batch-tx.ts, already merged) -- which builds one real transaction per operation and fetches the sender's sequence number once, incrementing it per index, so every transaction in the batch stays independently submittable regardless of batch size. Per-config client-side validation is isolated (one bad config doesn't block the rest), mirroring batchWithdraw()'s existing independent-per-item reporting. Adds a regression test asserting 15 configs (spanning what would have been 2 chunks under the old MAX_BATCH_SIZE=10 design) get 15 distinct, strictly increasing sequence numbers and that the source account is fetched exactly once. --- src/soroban.ts | 36 --- src/streams.ts | 152 ++++++++++++ src/tests/create-batch-streams.test.ts | 309 +++++++++++++++++++++++++ src/types/index.ts | 10 + 4 files changed, 471 insertions(+), 36 deletions(-) create mode 100644 src/tests/create-batch-streams.test.ts diff --git a/src/soroban.ts b/src/soroban.ts index 9c7cbca..5f1ec19 100644 --- a/src/soroban.ts +++ b/src/soroban.ts @@ -171,42 +171,6 @@ export async function buildContractCallTx( .build(); } -/** - * Build a contract-call transaction with multiple invokeHostFunction operations - * (one per set of arguments). Used for batch operations where several contract - * calls are assembled into a single Soroban transaction. - */ -export async function buildBatchContractCallTx( - rpcUrl: string, - passphrase: string, - caller: string, - contractId: string, - method: string, - argsArray: xdr.ScVal[][], -): Promise> { - const server = new SorobanRpc.Server(rpcUrl, { allowHttp: rpcUrl.startsWith('http://') }); - - let account; - try { - account = await server.getAccount(caller); - } catch (err) { - throw RateLimitError.fromRpcError(err) ?? err; - } - - const contract = new Contract(contractId); - - const builder = new TransactionBuilder(account, { - fee: BASE_FEE, - networkPassphrase: passphrase, - }); - - for (const args of argsArray) { - builder.addOperation(contract.call(method, ...args)); - } - - return builder.setTimeout(30).build(); -} - /** * Simulate a transaction, then assemble + sign + submit. * Returns the transaction hash on success. diff --git a/src/streams.ts b/src/streams.ts index e2be171..5d81bcf 100644 --- a/src/streams.ts +++ b/src/streams.ts @@ -15,6 +15,8 @@ import type { Subscription, BatchWithdrawItem, BatchWithdrawResult, + BatchCreateStreamResult, + StreamConfig, StreamOperation, FeeEstimate, } from './types/index.js'; @@ -33,6 +35,8 @@ import { DEFAULT_CONFIRMATION_POLL_INTERVAL_MS, createRpcServer, } from './soroban.js'; +import { buildBatchTransactions } from './batch-tx.js'; +import type { BatchTransactionContext } from './batch-tx.js'; import { FactoryModule } from './factory.js'; import { ConduitError, RateLimitError, StreamErrorCode } from './errors.js'; @@ -309,6 +313,154 @@ export class StreamsModule { }); } + /** + * Create multiple streams in bulk. + * + * Soroban permits only one invoke_host_function operation per transaction + * (see {@link buildBatchTransactions}), so this builds and submits one + * transaction per config. Each transaction consumes the sender's next + * sequence number in order, so every config remains independently + * submittable regardless of how many configs are in the batch — unlike + * fetching a fresh account/sequence per config, which would hand out the + * same sequence number to more than one transaction once submitted out of + * order. + * + * Per-config client-side validation failures do not block or roll back the + * rest of the batch — each config is reported independently by its + * original index, mirroring {@link batchWithdraw}. Note this isolation + * does not extend to on-chain simulation: {@link buildBatchTransactions} + * simulates every config's transaction in parallel via `Promise.all`, so + * one config's simulation being rejected currently fails building the + * whole batch, not just that config. + */ + async createBatchStreams(configs: StreamConfig[]): Promise { + this._ensureCanMutate(); + if (!Array.isArray(configs) || configs.length === 0) return []; + + const senderAddr = await this._getSenderAddress(); + const factoryId = this.config.factoryAddress ?? ''; + + const built = await Promise.all(configs.map(async (params, index) => { + try { + const { + recipient, token, depositAmount, + durationSeconds, ratePerSecond, + startTime, clawbackEnabled = false, + } = params; + + if (!recipient || typeof recipient !== 'string' || !recipient.trim()) { + throw new Error('Invalid recipient address: must be a non-empty string'); + } + if (!token || typeof token !== 'string' || !token.trim()) { + throw new Error('Invalid token address: must be a non-empty string'); + } + if (!depositAmount || typeof depositAmount !== 'string' || !depositAmount.trim()) { + throw new Error('Invalid deposit amount: must be a non-empty string'); + } + if (durationSeconds !== undefined && (typeof durationSeconds !== 'number' || durationSeconds <= 0)) { + throw new Error('Invalid durationSeconds: must be a positive number'); + } + if (ratePerSecond !== undefined && (typeof ratePerSecond !== 'string' || !ratePerSecond.trim())) { + throw new Error('Invalid ratePerSecond: must be a non-empty string'); + } + if (!durationSeconds && !ratePerSecond) { + throw new Error('Either durationSeconds or ratePerSecond must be provided'); + } + + const decimals = await getTokenDecimals(this.rpcUrl, this.passphrase, senderAddr, token); + const depositStroops = toStroops(depositAmount, decimals); + const rateStroops = ratePerSecond + ? BigInt(ratePerSecond) + : calculateRate(depositAmount, durationSeconds!, decimals); + const start = startTime ?? Math.floor(Date.now() / 1000); + const end = durationSeconds ? start + durationSeconds : 0; + + const args = [ + new Address(senderAddr).toScVal(), + new Address(recipient).toScVal(), + new Address(token).toScVal(), + nativeToScVal(depositStroops, { type: 'i128' }), + nativeToScVal(rateStroops, { type: 'i128' }), + nativeToScVal(start, { type: 'u64' }), + nativeToScVal(end, { type: 'u64' }), + boolToScVal(clawbackEnabled), + ]; + return { index, args, error: undefined as string | undefined }; + } catch (err) { + return { index, args: undefined, error: err instanceof Error ? err.message : String(err) }; + } + })); + + const results = new Map(); + for (const b of built) { + if (b.error !== undefined) { + results.set(b.index, { index: b.index, success: false, error: b.error }); + } + } + + const validEntries = built.filter( + (b): b is { index: number; args: xdr.ScVal[]; error: undefined } => b.args !== undefined, + ); + + if (validEntries.length > 0) { + const context: BatchTransactionContext = { + contractId: factoryId, + sourceAccount: senderAddr, + networkPassphrase: this.passphrase, + rpcUrl: this.rpcUrl, + }; + + let builtTxs: Awaited>; + try { + const operations = validEntries.map(b => ({ method: 'create_stream', args: b.args })); + builtTxs = await buildBatchTransactions(operations, context); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + for (const b of validEntries) { + results.set(b.index, { index: b.index, success: false, error: message }); + } + builtTxs = []; + } + + const server = this._server(); + const settled = await Promise.allSettled( + builtTxs.map(async (bt) => { + const tx = new Transaction(bt.xdr, this.passphrase); + const signed = await this._signTx(tx); + const { hash: txHash, returnValue } = await this._sendAndPoll(server, signed); + if (!returnValue) { + throw new Error(`Transaction ${txHash} succeeded but returned no value`); + } + const streamId = scValToU64(returnValue); + const streamAddress = await this._factory.streamAddress(streamId) ?? ''; + return { streamId, streamAddress, txHash }; + }), + ); + + settled.forEach((result, i) => { + const originalIndex = validEntries[i]!.index; + if (result.status === 'fulfilled') { + results.set(originalIndex, { + index: originalIndex, + success: true, + streamId: result.value.streamId, + streamAddress: result.value.streamAddress, + txHash: result.value.txHash, + }); + } else { + const err = result.reason; + results.set(originalIndex, { + index: originalIndex, + success: false, + error: err instanceof Error ? err.message : String(err), + }); + } + }); + } + + return configs.map((_, i) => results.get(i)!); + } + /** Cancel the stream (sender only). Settles all balances atomically. */ async cancel(streamId: bigint | string): Promise { this._ensureCanMutate(); diff --git a/src/tests/create-batch-streams.test.ts b/src/tests/create-batch-streams.test.ts new file mode 100644 index 0000000..b585e3a --- /dev/null +++ b/src/tests/create-batch-streams.test.ts @@ -0,0 +1,309 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { Keypair, StrKey, xdr } from '@stellar/stellar-sdk'; +import type { ConduitConfig, StreamConfig } from '../types/index.js'; + +// ── Mocks ───────────────────────────────────────────────────────────────────── + +const { + mockStreamAddress, + mockGetAccount, + mockSimulate, + mockSend, + mockGetTransaction, + mockAssemble, + mockGetTokenDecimals, +} = vi.hoisted(() => ({ + mockStreamAddress: vi.fn(), + mockGetAccount: vi.fn(), + mockSimulate: vi.fn(), + mockSend: vi.fn(), + mockGetTransaction: vi.fn(), + // Pass-through "assembly": returns the real (unmocked TransactionBuilder-built) + // transaction unchanged, so its XDR stays genuinely parseable. Batch-tx.ts + // hands back real XDR strings that createBatchStreams reconstructs a real + // Transaction from, so a fake, non-serialisable assembled object here would + // break that round-trip -- unlike StreamsModule.create()'s own tests, which + // never re-serialise the "assembled" object and so can get away with a stub. + mockAssemble: vi.fn().mockImplementation((tx: unknown) => ({ build: () => tx })), + mockGetTokenDecimals: vi.fn().mockResolvedValue(7), +})); + +vi.mock('../factory.js', () => ({ + FactoryModule: class { + streamAddress = mockStreamAddress; + }, +})); + +vi.mock('../soroban.js', async () => { + const actual = await vi.importActual('../soroban.js'); + return { + ...actual, + getTokenDecimals: mockGetTokenDecimals, + }; +}); + +vi.mock('@stellar/stellar-sdk', async () => { + const actual = await vi.importActual('@stellar/stellar-sdk'); + return { + ...actual, + SorobanRpc: { + ...actual.SorobanRpc, + Server: class { + getAccount = mockGetAccount; + simulateTransaction = mockSimulate; + sendTransaction = mockSend; + getTransaction = mockGetTransaction; + }, + assembleTransaction: mockAssemble, + }, + }; +}); + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +const FACTORY_ADDR = StrKey.encodeContract(Buffer.alloc(32, 1)); +const STREAM_ADDR = StrKey.encodeContract(Buffer.alloc(32, 2)); +const TOKEN = StrKey.encodeContract(Buffer.alloc(32, 3)); +const RECIPIENT = Keypair.random().publicKey(); + +function makeConfig(overrides: Partial = {}): ConduitConfig { + return { + network: 'testnet', + factoryAddress: FACTORY_ADDR, + keypair: Keypair.random(), + ...overrides, + }; +} + +function makeStreamConfig(overrides: Partial = {}): StreamConfig { + return { + recipient: RECIPIENT, + token: TOKEN, + depositAmount: '1000', + durationSeconds: 3600, + ...overrides, + }; +} + +function u64Scv(n: bigint) { + return xdr.ScVal.scvU64(xdr.Uint64.fromString(n.toString())); +} + +function simSuccess(retval: xdr.ScVal) { + return { result: { retval }, transactionData: {} }; +} + +function simError(message: string) { + return { error: message }; +} + +function txSuccess(returnValue?: xdr.ScVal) { + return returnValue === undefined + ? { status: 'SUCCESS' } + : { status: 'SUCCESS', returnValue }; +} + +/** Fake account whose sequenceNumber() reflects whatever getAccount() was last called to return. */ +function fakeAccount(sequence: string) { + return { accountId: () => 'ignored', sequenceNumber: () => sequence, incrementSequenceNumber: () => {} }; +} + +beforeEach(() => { + vi.useFakeTimers(); + mockStreamAddress.mockReset().mockResolvedValue(STREAM_ADDR); + mockGetAccount.mockReset().mockResolvedValue(fakeAccount('100')); + mockSimulate.mockReset(); + mockSend.mockReset().mockResolvedValue({ status: 'PENDING', hash: 'deadbeef' }); + mockGetTransaction.mockReset(); + mockAssemble.mockReset().mockImplementation((tx: unknown) => ({ build: () => tx })); + mockGetTokenDecimals.mockReset().mockResolvedValue(7); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +/** Runs `fn()` and drains however many sequential sleep(1000)s _sendAndPoll needs, one per settling transaction. */ +async function runThroughPolls(fn: () => Promise, count = 1): Promise { + const promise = fn(); + promise.catch(() => {}); + for (let i = 0; i < count; i++) { + await vi.advanceTimersByTimeAsync(1000); + } + return promise; +} + +describe('StreamsModule.createBatchStreams()', () => { + it('returns an empty array for an empty configs array without touching the network', async () => { + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig()); + + const results = await sdk.createBatchStreams([]); + + expect(results).toEqual([]); + expect(mockGetAccount).not.toHaveBeenCalled(); + }); + + it('throws when no signer/keypair/wallet is configured', async () => { + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule({ network: 'testnet', factoryAddress: FACTORY_ADDR }); + + await expect(sdk.createBatchStreams([makeStreamConfig()])).rejects.toThrow( + /keypair, wallet adapter, or signer/, + ); + }); + + it('creates a single stream and reports success with streamId/streamAddress/txHash', async () => { + mockSimulate.mockResolvedValue(simSuccess(u64Scv(1n))); + mockGetTransaction.mockResolvedValue(txSuccess(u64Scv(5n))); + + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig()); + + const results = await runThroughPolls(() => sdk.createBatchStreams([makeStreamConfig()])); + + expect(results).toEqual([ + { index: 0, success: true, streamId: 5n, streamAddress: STREAM_ADDR, txHash: 'deadbeef' }, + ]); + }); + + it('fetches the source account exactly once regardless of batch size (no per-chunk re-fetch)', async () => { + mockSimulate.mockResolvedValue(simSuccess(u64Scv(1n))); + mockGetTransaction.mockResolvedValue(txSuccess(u64Scv(1n))); + + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig()); + + const configs = Array.from({ length: 25 }, () => makeStreamConfig()); + await runThroughPolls(() => sdk.createBatchStreams(configs), 25); + + expect(mockGetAccount).toHaveBeenCalledTimes(1); + }); + + it('assigns strictly increasing sequence numbers across a multi-chunk batch (regression for #391)', async () => { + // The bug this regression-tests: buildBatchContractCallTx-style chunking used + // to call getAccount() once per chunk, so batches spanning more than one + // chunk (MAX_BATCH_SIZE = 10) would hand out the SAME sequence number to + // more than one transaction -- only the first submitted transaction of a + // multi-chunk batch could ever succeed on-chain. buildBatchTransactions() + // fetches the account once and increments a local counter per index + // instead, so every operation gets a distinct sequence number even past 10. + mockGetAccount.mockResolvedValue(fakeAccount('1000')); + mockSimulate.mockResolvedValue(simSuccess(u64Scv(1n))); + mockGetTransaction.mockResolvedValue(txSuccess(u64Scv(1n))); + + const seenXdrs: string[] = []; + mockAssemble.mockImplementation((tx: { toXDR: () => string }) => ({ + build: () => { + seenXdrs.push(tx.toXDR()); + return tx; + }, + })); + + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig()); + + const configs = Array.from({ length: 15 }, () => makeStreamConfig()); + const results = await runThroughPolls(() => sdk.createBatchStreams(configs), 15); + + expect(mockGetAccount).toHaveBeenCalledTimes(1); + expect(results.every(r => r.success)).toBe(true); + + // Decode each pre-assembled transaction's sequence number back out of its + // XDR and confirm they are 15 distinct, strictly increasing values -- + // proving no two transactions in this 2-chunk batch collided on sequence. + const { TransactionBuilder } = await import('@stellar/stellar-sdk'); + const sequences = seenXdrs.map((xdrStr) => { + const tx = TransactionBuilder.fromXDR(xdrStr, 'Test SDF Network ; September 2015'); + return BigInt((tx as unknown as { sequence: string }).sequence); + }); + + expect(new Set(sequences.map(String)).size).toBe(15); + const sorted = [...sequences].sort((a, b) => (a < b ? -1 : a > b ? 1 : 0)); + expect(sequences).toEqual(sorted); + for (let i = 1; i < sequences.length; i++) { + expect(sequences[i]).toBe(sequences[i - 1]! + 1n); + } + }); + + it('reports per-config validation failures independently without blocking valid configs', async () => { + mockSimulate.mockResolvedValue(simSuccess(u64Scv(1n))); + mockGetTransaction.mockResolvedValue(txSuccess(u64Scv(1n))); + + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig()); + + const configs = [ + makeStreamConfig(), + makeStreamConfig({ recipient: '' }), + makeStreamConfig(), + ]; + const results = await runThroughPolls(() => sdk.createBatchStreams(configs), 2); + + expect(results[0]!.success).toBe(true); + expect(results[1]).toEqual({ + index: 1, + success: false, + error: 'Invalid recipient address: must be a non-empty string', + }); + expect(results[2]!.success).toBe(true); + // Only the 2 valid configs should ever reach the network. + expect(mockGetAccount).toHaveBeenCalledTimes(1); + expect(mockSimulate).toHaveBeenCalledTimes(2); + }); + + it('fails every config in the call when one on-chain simulation is rejected', async () => { + // buildBatchTransactions() (already merged, unrelated to this change) uses + // Promise.all across its simulation step, so one rejected simulation + // rejects the whole call -- unlike the client-side validation above, + // which isolates failures per config *before* anything is submitted, an + // on-chain simulation failure for one config currently fails the rest of + // that batch too. That's a real, existing property of the batch-building + // primitive this method reuses, not something introduced here. + mockSimulate + .mockResolvedValueOnce(simSuccess(u64Scv(1n))) + .mockResolvedValueOnce(simError('HostError: Error(Contract, #8)')) + .mockResolvedValueOnce(simSuccess(u64Scv(1n))); + mockGetTransaction.mockResolvedValue(txSuccess(u64Scv(1n))); + + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig()); + + const configs = [makeStreamConfig(), makeStreamConfig(), makeStreamConfig()]; + const results = await sdk.createBatchStreams(configs); + + expect(results.every(r => !r.success)).toBe(true); + expect(results[1]!.error).toMatch(/Simulation failed for operation 1/); + }); + + it('throws when durationSeconds and ratePerSecond are both missing for a config', async () => { + mockSimulate.mockResolvedValue(simSuccess(u64Scv(1n))); + mockGetTransaction.mockResolvedValue(txSuccess(u64Scv(1n))); + + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig()); + + const config = makeStreamConfig(); + delete (config as Partial).durationSeconds; + + const results = await sdk.createBatchStreams([config]); + expect(results).toEqual([ + { index: 0, success: false, error: 'Either durationSeconds or ratePerSecond must be provided' }, + ]); + }); + + it('queries token decimals per config and does not hardcode 7', async () => { + mockGetTokenDecimals.mockResolvedValue(2); + mockSimulate.mockResolvedValue(simSuccess(u64Scv(1n))); + mockGetTransaction.mockResolvedValue(txSuccess(u64Scv(1n))); + + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig()); + + await runThroughPolls(() => sdk.createBatchStreams([makeStreamConfig({ depositAmount: '10' })])); + + expect(mockGetTokenDecimals).toHaveBeenCalledWith( + expect.any(String), expect.any(String), expect.any(String), TOKEN, + ); + }); +}); diff --git a/src/types/index.ts b/src/types/index.ts index a789bc2..d81c8ac 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -159,6 +159,16 @@ export interface BatchWithdrawResult { error?: string; } +export interface BatchCreateStreamResult { + /** Index into the configs array passed to createBatchStreams(). */ + index: number; + success: boolean; + streamId?: bigint; + streamAddress?: string; + txHash?: string; + error?: string; +} + // -- Fee estimation ---------------------------------------------------------- export type StreamOperation =