Skip to content

feat(streams): read many streams in one RPC call (#904) - #1042

Open
arkhindi12 wants to merge 4 commits into
conduit-protocol:mainfrom
arkhindi12:feat/904-batch-stream-read
Open

arkhindi12 wants to merge 4 commits into
conduit-protocol:mainfrom
arkhindi12:feat/904-batch-stream-read

Conversation

@arkhindi12

@arkhindi12 arkhindi12 commented Nov 4, 2026 •

Copy link
Copy Markdown
Contributor

Closes #904.

Fetching N streams costs N simulateTransaction calls today. getStreamInfos resolves each id and simulates it, and getStreamsBatch — despite the name — only chunked that same per-stream loop. BatchGetStreamsOptions declared chunkSize/chunkDelayMs, but the concurrency knob never changed the call count, so the batching was cosmetic.

before:  20 streams  ->  20 simulations (+ 20 address resolutions)
after:   20 streams  ->   2 getLedgerEntries calls

API — implemented as the issue specifies

The issue names it exactly: getMultipleStreams(streamAddresses: string[]): Promise<StreamData[]> in src/factory.ts. That is what this adds, in that file, taking contract addresses.

const streams = await client.factory.getMultipleStreams(addresses);

Behaviour:

  • addresses with no ledger entry (expired, never initialised, not a stream) are omitted rather than throwing, so one bad address cannot fail a page
  • duplicate addresses collapse to one key
  • an empty list makes no network call
  • an aborted signal rejects before any RPC

Returns StreamInfo[] — id, address and the decoded state, the same shape the per-stream path produces.

The id→address batching landed as streams.getStreamInfosBatched and now also backs client.getStreamsBatch, which is what its name already claimed.

Why this works

Soroban's getLedgerEntries accepts many keys per request, and the protocol stores everything needed in directly addressable entries. From the contracts repo:

The address index (contracts/factory/src/storage.rs):

/// **Persistent storage.** Maps stream ID to its deployed contract address.
/// Key: `DataKey::StreamAddr(u64)` — the stream's unique ID
StreamAddr(u64),

The stream state (contracts/stream/src/state.rs):

pub fn try_load(env: &Env) -> Result<StreamInfo, Error> {
    let s = env.storage().instance();
    s.get::<_, StreamInfo>(&DataKey::Config)
}

The entire StreamInfo struct lives under DataKey::Config in instance storage — one ledger entry per contract. So N streams resolve from one batched call. No contract call executes and no signer is required: this is a pure ledger read.

Verified against testnet before writing code — two keys, one call, 1.4s:

entries: 1 | latestLedger: 5562305 | ms: 1408

Key encoding — the part that would fail silently

DataKey is a #[contracttype] enum, and a variant with a payload encodes as ScVal::Vec whose first element is the variant name as a Symbol, not a u32 index:

// soroban-sdk-macros, derive_enum.rs
let tup = (Symbol::try_from_val(env, &case_name_str_lit)?, #(#field_convs,)*);

So DataKey::StreamAddr(42) is scvVec([scvSymbol("StreamAddr"), scvU64(42)]). Encoding it as an index yields a key the RPC never matches — no error, just empty results forever. A round-trip test pins the exact encoding, including that the reader rejects a key whose variant name differs.

Two @stellar/stellar-base typing gaps

Both documented at the call site rather than papered over:

1. LedgerKeyContractData is not public. Declared only inside the private xdrHidden namespace and never re-exported, so the public constructor is untyped — and constructing it directly (via new or Reflect.construct) returns a bare ChildStruct without the union accessors the RPC needs. Keys therefore go through the typed static factory xdr.LedgerKey.contractData.

2. Typings and runtime disagree. curr.d.ts declares key, contract, durability and val as accessor methods:

key(value?: ScVal): ScVal;

At runtime they are plain properties — entry.key() throws key is not a function. A single field() accessor reads the property and keeps the cast in one documented place.

Also fixed

LedgerEntryResult.val is a LedgerEntryData, not an ScVal, so the stored address lives under its contractData arm. My first pass read it directly and the test caught it.

Tests

27 new/updated across four files. The two that assert the issue's premise directly:

// streams module
expect(simulateTransaction).not.toHaveBeenCalled();
expect(getLedgerEntries).toHaveBeenCalledTimes(2);
expect(getLedgerEntries.mock.calls[0]).toHaveLength(5);

// factory module — the API the issue names
expect(getLedgerEntries).toHaveBeenCalledTimes(1);
expect(getLedgerEntries.mock.calls[0]).toHaveLength(2);
getMultipleStreams      6   N addresses -> 1 call, 0 simulations, dedupe,
                             empty input, missing entries, abort
getStreamInfosBatched   7   call count, parse parity, missing ids, dedupe,
                             empty input, abort
ledger-keys            11   key encoding (name-not-index), round-trip readers,
                             foreign-key rejection, Config extraction
streams-batch           3   chunking via batched path, no per-stream fallback,
                             failures preserved

Fixtures build real XDR — ScContractInstance with a Config entry via contractExecutableWasm, wrapped in LedgerEntryData — so the tests exercise the actual decode path rather than a stub shape. RPC is mocked at the module level (createRpcServer), never at a private method.

Verification

new/updated tests        27/27 pass
full suite               28 pre-existing failures — identical to main
passing files            143 -> 146
typecheck                26 -> 23 pre-existing errors, none introduced

main is red until #1038 lands; the remaining errors are in client.ts / custom-fetch.ts / soroban.ts, which that PR fixes. Verified combined with #1038: Lint & typecheck, Test, Build, TypeDoc, size and bundle jobs all pass.

Note on the contracts build

I tried to deploy drip-stream to testnet for an end-to-end check, but streamFi-contracts HEAD does not compile — lib.rs references DataKey::InvoiceMetadataUri and DataKey::MilestoneMetadataUri, neither of which exists in storage.rs (5 errors). That is upstream of this change, so validation rests on the storage layout read from source plus real XDR fixtures.

Happy to add an integration test against a deployed stream contract if you can point me at a testnet factory address — none is checked in, and the app repo's are mocks.

zaguza added 2 commits November 4, 2026 20:55
Fetching N streams cost N `simulateTransaction` calls. The existing
`getStreamInfos` resolves each id and simulates it, and `getStreamsBatch` —
despite its name — only chunked that same per-stream loop, so "batching" was
still N+1. `BatchGetStreamsOptions` declared `chunkSize`/`chunkDelayMs` but the
concurrency knob did nothing for call count.

Soroban exposes `getLedgerEntries`, which takes many keys per request, and the
protocol stores everything needed in directly addressable entries:

* the factory maps `DataKey::StreamAddr(u64)` → stream contract address, in
  persistent storage (`contracts/factory/src/storage.rs`); and
* each stream keeps its entire `StreamInfo` under `DataKey::Config` in
  **instance** storage — `env.storage().instance().set(&DataKey::Config, info)`
  in `contracts/stream/src/state.rs` — and instance storage is one ledger entry
  per contract.

So `getStreamInfosBatched` issues **two** `getLedgerEntries` calls for any
number of streams: one for the address index, one for the instance state. No
contract call runs and no signer is needed. Verified against testnet: two keys,
one call, 1.4s.

Parsing is shared with the per-stream path — both decode the same on-chain
`StreamInfo` map through `parseStreamInfo`, so the returned shape is identical.

`getStreamsBatch` now drives the batched path, which is what its name and its
docstring already claimed.

## Key encoding

`DataKey` is a `#[contracttype]` enum, and a variant **with a payload** encodes
as `ScVal::Vec` whose first element is the variant *name* as a `Symbol`, not a
`u32` index:

```rust
// soroban-sdk-macros, derive_enum.rs
let tup = (Symbol::try_from_val(env, &case_name_str_lit)?, #(#field_convs,)*);
```

So `DataKey::StreamAddr(42)` is `scvVec([scvSymbol("StreamAddr"), scvU64(42)])`.
Encoding it as an index produces a key the RPC silently never matches, so this
is pinned by a round-trip test.

## `@stellar/stellar-base` typing gaps

Two things had to be worked around, both documented at the call site:

* `LedgerKeyContractData` is declared only inside the private `xdrHidden`
  namespace and never re-exported, so the public constructor is untyped and
  constructing it directly yields a bare `ChildStruct` without the union
  accessors. Keys therefore go through the typed static factory
  `xdr.LedgerKey.contractData`.
* The typings declare `key`, `contract`, `durability` and `val` as accessor
  *methods* (`key(): ScVal`), but at runtime they are plain *properties* —
  `entry.key()` throws `is not a function`. A single `field()` accessor reads
  the property and keeps the cast in one place.

Also fixed while here: `LedgerEntryResult.val` is a `LedgerEntryData`, so the
stored address lives under its `contractData` arm, not directly on `val`.

## Tests

21 new/updated tests. The central one asserts the point of the issue directly:
five streams produce **two** `getLedgerEntries` calls and **zero**
`simulateTransaction` calls.

```
getStreamInfosBatched  7 tests  (call count, parse parity, missing ids,
                                 dedupe, empty input, abort)
ledger-keys            11 tests (key encoding incl. name-not-index, round-trip
                                 readers, foreign-key rejection, Config
                                 extraction from instance storage)
streams-batch          3 tests  (chunking via the batched path, no fallback to
                                 per-stream, failures preserved)
```

Verified: 21/21 pass. Full suite unchanged from `main` — 28 pre-existing
failures in both, passing files 143 → 145 (the two new test files). Typecheck
27 → 25 pre-existing errors (none introduced; `main` is red until conduit-protocol#1038 lands).
`LedgerEntryDataContractData.val` is typed as an accessor method
(`val(): ScVal`) but is a plain property at runtime, the same typing gap as
`key` / `contract` / `durability` documented in ledger-keys.ts. The earlier
cast still called it as a method, which typechecked but would throw
`val is not a function` on a real RPC response.

Reads it through an explicit property cast, matching the `field()` accessor
used by the key helpers.

Typecheck: 26 pre-existing errors on main -> 24 with this branch.
@arkhindi12

Copy link
Copy Markdown
Contributor Author

CI status: red because main is red, not because of this branch

This PR's jobs fail on the same errors main already has. main has not typechecked since 2026-10-31 — the Lint & typecheck job fails there too, which skips Build and therefore the whole downstream chain.

Measured on the same checkout:

main                   26 type errors
this branch            24 type errors   (2 fewer: the two I fixed in streams.ts)

Errors remaining are all in files this PR does not touch:

src/custom-fetch.ts   6
src/client.ts         4
src/index.ts          3
src/soroban.ts        2
src/events.ts         1
src/streams.ts        2   (DurationBreakdown, an overload mismatch)

None of them come from ledger-keys.ts or getStreamInfosBatched — grepping the CI log for either name returns nothing.

#1038 fixes the SDK-wide typecheck; once it lands, this branch should go green without changes. Locally with #1038 applied:

typecheck        0 errors
new tests        21/21 pass
full suite       28 pre-existing failures (identical to main), 145 passing files

@arkhindi12

Copy link
Copy Markdown
Contributor Author

Verified green together with #1038

Ran the full CI matrix on #1038 + this branch combined (temporary draft #1043):

Lint & typecheck    pass   59s
Test                pass  1m40s
Build               pass   53s
Build TypeDoc       pass  1m11s
Size limit          pass   51s
Bundle size check   pass  1m06s
@streamfi/react     fail  1m06s   <- unrelated, see below

Local state on that combined branch:

typecheck            0 errors
full suite           152/152 files, 1663/1663 tests pass

The one failing job is @streamfi/react, and its errors are rpcHeaders / makeHeaderFetch not existing in packages/react/src/context/StreamFiProvider.tsx — the exact defect #1039 fixes. Nothing to do with this branch; #1039 makes that job pass.

So the dependency chain is:

  1. fix: restore green CI — 23 type errors and 28 failing tests #1038 — SDK typecheck (unblocks Build)
  2. fix(react): build @streamfi/react again — 20 type errors, 40 failing tests #1039 — packages/react (unblocks @streamfi/react)
  3. this PR — src/factory.ts: implement batch query to fetch multiple stream details in a single RPC round-trip #904 batched stream reads

Merging in that order leaves CI fully green.

The mock's parameters were untyped once the return value was cast, tripping
TS7006 under noImplicitAny. Typing them as bigint[] keeps the cast local to
the return value.
@arkhindi12

Copy link
Copy Markdown
Contributor Author

Final state — verified with #1038 applied

Last push adds one thing that would have broken CI after #1038 merged: the mock in streams-batch.test.ts needed both a return-type cast and explicit parameter types, otherwise noImplicitAny rejects it (TS7006). Caught it by running the combined branch rather than trusting the isolated one.

Combined with #1038, on the exact commit in this PR:

typecheck            0 errors
test suite           152/152 files, 1663/1663 tests pass

CI on this PR alone still reports 18 type errors — all inherited from main (26 there). Grepping the log for ledger-keys, getStreamInfosBatched, or streams-batch returns nothing. Verified again after the final push:

$ gh run view --job <lint> --log-failed | grep -E "ledger-keys|getStreamInfosBatched|streams-batch"
(no output)

Temporary verification branches deleted; only feat/904-batch-stream-read remains on the fork.

…nduit-protocol#904)

The issue names the API exactly: `getMultipleStreams(streamAddresses:
string[]): Promise<StreamData[]>` in `src/factory.ts`. The earlier commits on
this branch added the ledger-key machinery and a batched reader on the streams
module, but not this method — so the public surface the issue asks for was
still missing.

Implemented as specified, in the specified file. It takes contract addresses
(not ids), so it needs only one key per stream: each stream stores its whole
`StreamInfo` under `DataKey::Config` in instance storage, which is one ledger
entry per contract. One `getLedgerEntries` call covers every address.

Behaviour:

* addresses with no ledger entry (expired, never initialised, not a stream)
  are omitted rather than throwing, so one bad address cannot fail a page;
* duplicate addresses collapse to a single key;
* an empty list makes no network call;
* an aborted signal rejects before any RPC.

Returns `StreamInfo[]`, which carries `id`, `address` and the decoded state —
the same shape the per-stream path produces.

Tests: 6, asserting the issue's premise directly (N addresses → 1 call, 0
simulations) plus dedupe, empty input, missing entries and abort.

The fixture builds a real `ScContractInstance` with a `Config` entry using
`contractExecutableWasm`, so the decode path is exercised rather than stubbed.

Typecheck 26 -> 23 pre-existing errors. Full suite: 146 passing files (was
143 on main), same 28 pre-existing failures.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

src/factory.ts: implement batch query to fetch multiple stream details in a single RPC round-trip

1 participant