Documentation / @etherfold/core / StreamBuilder

Class: StreamBuilder<ABI, ProcessResultType>

Defined in: packages/core/src/streamBuilder.ts:278

The STREAM-BUILDER of ADR-0003, and the receiving half of the wire contract of ADR-0004.

It takes contiguous ranges of raw logs from a stateless log-fetcher, derives every retraction itself, drives an EventProcessor, and is authoritative about where the next range must start. It makes NO chain calls, which is the whole reason it is a separate object from IndexerGeneration: that class opens load() with eth_chainId, so the half of a split deployment that hosts the processor cannot use it (recorded in indexer-load-needs-a-chain-so-the-server-half-cannot-call-it).

Why the cursor is read on every call and never cached

The intended host is serverless: several isolates may serve the same database, and an instance can be created and destroyed between two batches. An in-memory cursor would be one instance's private opinion of a value the DATABASE owns, and two instances would disagree the moment either of them moved it. So expectedFromBlock() and receive() both read the persisted cursor through processor.load() first. That costs a read per batch and buys the property the design is built on: the cursor a sender is told is the cursor the state actually has.

The three identities, and who checks which

• source and config are asserted by the SENDER, on every batch, and checked here against what this receiver indexes (WireContextMismatchError).
• processor cannot be asserted by the sender at all, so it is checked against the PERSISTED cursor instead: a cursor written by another processor version, or for another source, is state that means something else, and it is not resumed on top of. WHAT HAPPENS TO IT depends on whether there is a container above this receiver: with one, the running fold is a GENERATION beside the stored one and nothing is discarded; without one, the state is discarded (processor.clear()), which is the hole docs/reviews/todo-triage.md found in every persistence layer, closed the only way a receiver holding a single store can close it. See currentLastSync and receivingContainer.ts.

What it deliberately does not do

It does not batch a large stream into several process() calls the way IndexerGeneration.feed does. A batch is one HTTP request and the sender chose its size; splitting it here would add a second place where a partially-applied range is possible, and the processor already applies block by block atomically.

The two things it hands OUT, and the one it must not lose

It knows no database, so both durable facts a fold concludes leave through a PORT the store's owner supplies: the reorg COUNT (recordReorg, ADR-0050) and the emission STREAM (appendEmissions, ADR-0052). Neither is a fact about the transport, so neither may live on an entrance: a combined process folds through createDirectIngestion and reaches no route, and putting either write at both entrances would double it on the shape that both concludes and receives.

They are ORDERED DIFFERENTLY on purpose, and that is the only asymmetry between them. The count is taken AFTER the batch was applied and its failure is swallowed; the stream is written BEFORE, and its failure refuses the batch. See noteReorg and receive.

Type Parameters

ABI

ABI extends Abi

ProcessResultType

ProcessResultType = unknown

Implements

• LogIngestion

Constructors

Constructor

new StreamBuilder<ABI, ProcessResultType>(processor, source, config): StreamBuilder<ABI, ProcessResultType>

Defined in: packages/core/src/streamBuilder.ts:322

Parameters

processor

EventProcessor<ABI, ProcessResultType>

source

IndexingSource<ABI>

config

StreamBuilderOptions<ABI>

Returns

StreamBuilder<ABI, ProcessResultType>

Properties

context

readonly context: WireContext

Defined in: packages/core/src/streamBuilder.ts:284

The {source, config} a sender must assert to be talking to this receiver.

Implementation of

LogIngestion.context

---

defaultFromBlock

readonly defaultFromBlock: number

Defined in: packages/core/src/streamBuilder.ts:280

The earliest block this source can have anything to say about.

---

streamConfig

readonly streamConfig: UsedStreamConfig

Defined in: packages/core/src/streamBuilder.ts:282

The resolved stream config, which is what config in the context hashes.

---

streamDigest

readonly streamDigest: string

Defined in: packages/core/src/streamBuilder.ts:286

WHICH stream this folds, as everything that stores its emissions keys them. See LogIngestion.

Implementation of

LogIngestion.streamDigest

Accessors

generation

Get Signature

get generation(): GenerationId

Defined in: packages/core/src/streamBuilder.ts:296

WHICH generation this is: the stream it folds, and the fold itself.

A GETTER rather than a field only because the stream half is one: the fold half is the identity this receiver's ARRIVAL derived (ADR-0086), which is a constant, since the config a bundle was built with is IN the bundle. See LogIngestion.

Returns

GenerationId

Implementation of

LogIngestion.generation

Methods

expectedFromBlock()

expectedFromBlock(): Promise<number>

Defined in: packages/core/src/streamBuilder.ts:365

Where the next batch must start.

It is NOT lastToBlock + 1: it reaches back to latestBlock - finality so the unconfirmed window is re-delivered every round, which is the only way a reorg is ever detected. A sender that computed this itself would be holding the state that makes it stateless.

Reading this can WRITE, and WHICH write depends on whether a container is attached (see currentLastSync). WITHOUT one, a persisted cursor belonging to another source, config or processor version is cleared here rather than merely ignored: the alternative is answering with a number derived from state this server is about to wipe, so the read and the following write would disagree. Nothing is lost that was not already invalid, and the same reconciliation happens on load() in the single-process shape. WITH one, the write is a REGISTRY write instead -- the fold this receiver runs is resolved-or-created as a generation -- and no state is discarded at all.

Returns

Promise<number>

Implementation of

LogIngestion.expectedFromBlock

---

receive()

receive(batch): Promise<IngestionResult<ABI>>

Defined in: packages/core/src/streamBuilder.ts:378

Apply one batch, or refuse it having applied nothing.

The order of the three refusals is deliberate. Identity first, because a batch for another source must never be told to resume from a block number that means nothing to it. Then the envelope, which is cheap and local. Then the cursor, which is generateStreamToAppend's own check and therefore the engine's, not a second one that could disagree with it.

Parameters

batch

WireBatch<ABI>

Returns

Promise<IngestionResult<ABI>>

Implementation of

LogIngestion.receive