> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need. # Streaming Status: Unstable. A [streaming query](/querying/streaming-queries) doesn't end: Amp sends the rows for the blocks it has, then new rows as its datasets sync new blocks. Use one to keep an app, a cache, or a database up to date with a chain. The `ProtocolStream` service runs streaming queries, and tells you what each part of the stream means: new rows, a range of blocks that's complete, or a [reorg](/get-started/concepts/reorgs) that replaced blocks you already received. To process each part exactly once, even across restarts, use a [transactional stream](/sdks/typescript/transactional-streams) on top of it. :::warning Streams have known issues: they [fail after 30 seconds](/sdks/typescript/streaming#keep-a-stream-running), and [lose rows after a reorg](/sdks/typescript/streaming#reorgs). Read this page's workarounds before you rely on them. ::: ## Prerequisites * The SDK, [installed](/sdks/typescript#install-the-sdk). * A query server to connect to (see [Connecting](/sdks/typescript/connecting)). The examples use the [Quickstart](/get-started/quickstart)'s Amp and its `acme/eth_mainnet` dataset. ## Start a stream `protocol.stream(sql)` runs `sql` as a streaming query, and returns a [Stream](/sdks/typescript/effect#streams) of messages. `ProtocolStream.layer` builds the service from `ArrowFlight`'s: ```ts [stream.ts] import { ArrowFlight, ProtocolStream } from "@edgeandnode/amp" import { layerTransportGrpc } from "@edgeandnode/amp/arrow-flight/node" import { Effect, Layer, Stream } from "effect" const program = Effect.gen(function* () { const protocol = yield* ProtocolStream.ProtocolStream yield* protocol.stream(`SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks`).pipe( Stream.runForEach((message) => Effect.sync(() => { switch (message._tag) { case "Data": { const [range] = message.ranges return console.log( `Data: ${message.data.length} rows, blocks ${range?.numbers.start} to ${range?.numbers.end}`, ) } case "Watermark": return console.log(`Watermark: blocks up to ${message.ranges[0]?.numbers.end} are complete`) case "Reorg": return console.log(`Reorg: blocks from ${message.invalidation[0]?.start} were replaced`) } }), ), ) }) const FlightLayer = ArrowFlight.layer.pipe(Layer.provide(layerTransportGrpc({ baseUrl: "http://localhost:1602" }))) const AmpLayer = ProtocolStream.layer.pipe(Layer.provide(FlightLayer)) await Effect.runPromise(program.pipe(Effect.provide(AmpLayer))) ``` :::terminal ```bash node stream.ts ``` ``` Data: 100 rows, blocks 24000000 to 24000099 Watermark: blocks up to 24000099 are complete ``` ::: The stream then waits for new blocks. The Quickstart's dataset stopped at block 24,000,099, so none come. After 30 seconds, the stream fails with `RangeError: Offset is outside the bounds of the DataView`, a known issue: [Keep a stream running](/sdks/typescript/streaming#keep-a-stream-running) shows how to resume it. ## Messages Each message has a `_tag`, which says what it is: * **`Data`**: new rows, in `data`, decoded as [`query` decodes them](/sdks/typescript/querying#how-values-are-decoded). `ranges` has the [block ranges](/sdks/typescript/domain#blockrange) they come from, for each chain the query reads. A range's rows can arrive in several `Data` messages. * **`Watermark`**: all the rows for `ranges` have arrived. Save `ranges` to [resume](/sdks/typescript/streaming#resume-a-stream) from there later. * **`Reorg`**: blocks you received were replaced. `invalidation` lists, for each chain, the first and last block replaced: discard what you derived from them. See [Reorgs](/sdks/typescript/streaming#reorgs). Amp streams in microbatches: it runs the query over a range of blocks, sends the rows, then a `Watermark`, then moves on to the next range once its datasets have synced it. Each `Data` message is one of Amp's record batches. [Streaming Queries](/querying/streaming-queries) explains how Amp sizes its microbatches, and which SQL works in a stream. ## Reorgs A [reorg](/get-started/concepts/reorgs) happens when a chain replaces its most recent blocks with others. Amp re-extracts the new blocks, then re-sends their rows, with block ranges that start before the end of the last ones it sent. `ProtocolStream` compares each message's ranges with the previous ones: when a range jumps back, it emits a `Reorg` message instead. `invalidation` has an [`InvalidationRange`](/sdks/typescript/streaming#invalidationrange) for each chain the reorg affects: its `network`, and the `start` and `end` blocks, inclusive. Delete or undo everything you derived from those blocks: the rows Amp sends next replace them. :::warning **Known issue: rows are lost after a reorg.** Amp sends the replacement blocks' rows in the same record batch that reveals the reorg, but `ProtocolStream` emits the `Reorg` message in place of that batch's `Data`: those rows never reach you. The next messages carry the blocks after them. Until this is fixed in the SDK, after a `Reorg`, query the replaced blocks again, with [`flight.query`](/sdks/typescript/querying), once the next `Watermark` covers them. ::: ### Test reorgs on a local chain To see how your code handles reorgs, stream from a local chain that reorgs when you ask it to. [Anvil](https://getfoundry.sh/anvil/overview), from Foundry, runs a local Ethereum chain: :::steps ### Start a local chain [Install Foundry](https://getfoundry.sh/introduction/installation), then start Anvil, with a block every 2 seconds. It serves its RPC API on port 8545: ```bash anvil --block-time 2 ``` ### Extract it with Amp In another terminal, register Anvil as a [provider](/providers/evm-rpc), and deploy a dataset that follows it, as in the [Quickstart](/get-started/quickstart): ```bash cat > anvil.toml << 'EOF' kind = "evm-rpc" network = "anvil" url = "http://localhost:8545" EOF ampctl provider register anvil_rpc ./anvil.toml ampctl manifest generate --kind evm-rpc --network anvil --start-block 0 --out anvil.json ampctl dataset register acme/anvil ./anvil.json --tag 1.0.0 ampctl dataset deploy acme/anvil@1.0.0 ``` ### Stream from it Change the query in `stream.ts` to `SELECT block_num, gas_used FROM "acme/anvil".blocks`, and run it. A `Data` and a `Watermark` message arrive every 2 seconds, with each new block. ### Reorg the chain In a third terminal, ask Anvil to replace its last 2 blocks: ```bash curl -s localhost:8545 -H 'content-type: application/json' \ -d '{"jsonrpc":"2.0","id":1,"method":"anvil_reorg","params":[2,[]]}' ``` The stream reports the reorg: ``` Data: 1 rows, blocks 587 to 587 Watermark: blocks up to 587 are complete Data: 1 rows, blocks 588 to 588 Watermark: blocks up to 588 are complete Reorg: blocks from 587 were replaced Watermark: blocks up to 589 are complete Data: 1 rows, blocks 590 to 590 Watermark: blocks up to 590 are complete ``` The rows of blocks 587 to 589 are missing after the `Reorg`: that's the [known issue](/sdks/typescript/streaming#reorgs) above. ::: ## Resume a stream To pick up where a stream left off, after a restart or a failure, pass the last `Watermark`'s ranges as `resumeWatermark`. Amp then sends only the rows after them: ```ts const resumed = protocol.stream(sql, { resumeWatermark: lastWatermark.ranges }) ``` Save the ranges somewhere that survives a restart, like a file or a database, each time a `Watermark` arrives. Rows that arrived after the last `Watermark` arrive again. A [transactional stream](/sdks/typescript/transactional-streams) does this for you, and tells you which rows to discard. If the chain reorged the blocks you resume after, Amp notices, and re-sends them. :::warning **Known issue: a reorg while a stream is stopped goes unnoticed.** `ProtocolStream` only detects reorgs by comparing messages within one stream. When it resumes after blocks that were replaced meanwhile, it emits the re-sent rows as `Data`, without a `Reorg`: you have both the old and the new rows for those blocks. ::: ## Keep a stream running While a stream waits for new blocks, Amp sends an empty message every 30 seconds (its query server's [`keep_alive_interval`](/running-amp/configuration-reference#serverkeep_alive_interval)), to keep the connection open. The SDK can't read it: the stream fails with a [defect](/sdks/typescript/errors#defects), `RangeError: Offset is outside the bounds of the DataView`. Until that's fixed in the SDK, catch it, and resume from the last watermark: ```ts [follow.ts] import { ArrowFlight, type Domain, ProtocolStream } from "@edgeandnode/amp" import { layerTransportGrpc } from "@edgeandnode/amp/arrow-flight/node" import { Effect, Layer, Stream } from "effect" const sql = `SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks` const handle = (message: ProtocolStream.ProtocolMessage) => Effect.sync(() => { const ranges = (ranges: ReadonlyArray) => ranges.map((range) => `${range.network} ${range.numbers.start}-${range.numbers.end}`).join(", ") switch (message._tag) { case "Data": return console.log(`Data: ${message.data.length} rows, ${ranges(message.ranges)}`) case "Watermark": return console.log(`Watermark: ${ranges(message.ranges)}`) case "Reorg": return console.log(`Reorg: ${message.invalidation.map((r) => `${r.network} ${r.start}-${r.end}`).join(", ")}`) } }) const program = Effect.gen(function* () { const protocol = yield* ProtocolStream.ProtocolStream // The last watermark, to resume from after a failure let resumeWatermark: ReadonlyArray | undefined const follow = Effect.suspend(() => protocol.stream(sql, resumeWatermark ? { resumeWatermark } : {}).pipe( Stream.tap((message) => Effect.sync(() => message._tag === "Watermark" && (resumeWatermark = message.ranges))), Stream.runForEach(handle), ), ) yield* follow.pipe( // Works around the keep-alive issue: resume when the stream fails with its RangeError Effect.catchDefect((defect) => defect instanceof RangeError ? Effect.logWarning(`Stream failed, resuming: ${defect}`) : Effect.die(defect), ), Effect.forever, ) }) const FlightLayer = ArrowFlight.layer.pipe(Layer.provide(layerTransportGrpc({ baseUrl: "http://localhost:1602" }))) const AmpLayer = ProtocolStream.layer.pipe(Layer.provide(FlightLayer)) await Effect.runPromise(program.pipe(Effect.provide(AmpLayer))) ``` :::terminal ```bash node follow.ts ``` ``` Data: 100 rows, mainnet 24000000-24000099 Watermark: mainnet 24000000-24000099 [17:14:59.663] WARN (#1): Stream failed, resuming: RangeError: Offset is outside the bounds of the DataView [17:15:29.697] WARN (#1): Stream failed, resuming: RangeError: Offset is outside the bounds of the DataView ``` ::: `Effect.suspend` builds the stream again each time `follow` runs, with the latest watermark, and `Effect.forever` runs it again each time it stops. Other defects still stop the program. Here, nothing is repeated, because the stream fails while it waits for new blocks, after a `Watermark`. But the empty messages come every 30 seconds whether or not blocks arrive, so on a busy chain, one can come between a `Data` message and its `Watermark`. The stream then resumes from the `Watermark` before, and that `Data` arrives again, as after any [resume](/sdks/typescript/streaming#resume-a-stream). Process rows so that getting them twice is harmless, like upserts by block number, or use a [transactional stream](/sdks/typescript/transactional-streams), which tells you which rows to undo. ## Raw streams `flight.streamQuery(sql, { stream: true })` also runs a streaming query, and gives you Amp's record batches as they arrive, with their `metadata`, without checking them or detecting reorgs. `ProtocolStream` is built on it. Use it only to handle Amp's [streaming protocol](/querying/arrow-flight) yourself. ## Validation `ProtocolStream` checks that each message's block ranges follow from the previous ones, as Amp guarantees: * Each block range names its parent block's hash (`prevHash`), except a chain's first block, and it matches the hash of the previous range's last block. * The chains don't change during a stream, and each appears once per message. * No blocks are skipped. A message that breaks one of these fails the stream with a [`ProtocolValidationError`](/sdks/typescript/errors#protocolvalidationerror), whose `cause` says which. It means Amp sent something unexpected: [resume](/sdks/typescript/streaming#resume-a-stream) from the last watermark. ## Reference Import these from `@edgeandnode/amp`'s `ProtocolStream` namespace, or from `@edgeandnode/amp/protocol-stream`. ### `ProtocolStream` ```ts class ProtocolStream extends Context.Service ``` The service that runs streaming queries. Get it with `yield* ProtocolStream.ProtocolStream`. ### `ProtocolStreamService` ```ts interface ProtocolStreamService { readonly stream: (sql: string, options?: ProtocolStreamOptions) => Stream } ``` #### `stream` Runs `sql` as a streaming query, and emits a [`ProtocolMessage`](/sdks/typescript/streaming#protocolmessage) for each record batch Amp sends. Never ends on its own. * **Parameters**: `sql`, the query; `options`, the [`ProtocolStreamOptions`](/sdks/typescript/streaming#protocolstreamoptions). * **Returns**: `Stream`. * **Fails with**: [`ProtocolStreamError`](/sdks/typescript/errors#protocol-stream-errors): a `ProtocolArrowFlightError` when the query fails, whose `cause` is the [`ArrowFlightError`](/sdks/typescript/querying#arrowflighterror), or a `ProtocolValidationError`. ### `ProtocolStreamOptions` ```ts interface ProtocolStreamOptions { readonly schema?: QueryOptions["schema"] readonly resumeWatermark?: ReadonlyArray } ``` | Option | Type | Default | Description | | ----------------- | --------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `schema` | A `Schema` | None | Checks each row, as in [`query`](/sdks/typescript/querying#give-rows-a-type). The messages' `data` stays typed `Record`: cast it to your schema's type. | | `resumeWatermark` | `ReadonlyArray` | None | Starts after these block ranges: a previous `Watermark`'s `ranges`. See [Resume a stream](/sdks/typescript/streaming#resume-a-stream). | The stream's values use the default decoding: 64-bit integers as strings, and so on. ### `layer` ```ts const layer: Layer ``` Builds the `ProtocolStream` service. Needs [`ArrowFlight`](/sdks/typescript/querying#arrowflight). ### `ProtocolMessage` ```ts type ProtocolMessage = ProtocolMessageData | ProtocolMessageReorg | ProtocolMessageWatermark ``` A message of the stream. `ProtocolMessage` is also its schema. Tell them apart by `_tag`: | Type | `_tag` | Properties | | -------------------------- | ------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | | `ProtocolMessageData` | `"Data"` | `data`: `ReadonlyArray>`, the rows. `ranges`: `ReadonlyArray`, the block ranges they're from. | | `ProtocolMessageWatermark` | `"Watermark"` | `ranges`: `ReadonlyArray`, the block ranges whose rows have all arrived. | | `ProtocolMessageReorg` | `"Reorg"` | `previous`: the ranges before the reorg. `incoming`: the ranges that revealed it. `invalidation`: `ReadonlyArray`, the blocks replaced. | Each type is also its schema, with a `make` function, like `ProtocolMessageWatermark.make({ ranges })`. ### `InvalidationRange` ```ts const InvalidationRange: Schema.Struct<{ network: Network; start: BlockNumber; end: BlockNumber }> ``` Blocks a reorg replaced on one chain, `network`, from `start` to `end`, both included. ### `makeInvalidationRange` ```ts function makeInvalidationRange(network: string, start: number, end: number): InvalidationRange ``` Creates an `InvalidationRange`. Throws when the values aren't valid: `network` must be lowercase, and the blocks non-negative integers. ### `invalidates` ```ts function invalidates(invalidation: InvalidationRange, range: BlockRange): boolean ``` Whether a reorg's `invalidation` affects `range`: `true` when they're on the same chain and share at least one block. Use it to find which of the ranges you saved a reorg affects: ```ts const affected = saved.filter((range) => message.invalidation.some((inv) => ProtocolStream.invalidates(inv, range))) ``` ### `data`, `reorg`, and `watermark` ```ts function data(records: ReadonlyArray>, ranges: ReadonlyArray): ProtocolMessageData function reorg( previous: ReadonlyArray, incoming: ReadonlyArray, invalidation: ReadonlyArray, ): ProtocolMessageReorg function watermark(ranges: ReadonlyArray): ProtocolMessageWatermark ``` Create messages, for tests of code that handles them. ### Validation functions The checks [Validation](/sdks/typescript/streaming#validation) describes, which `stream` runs on each message. Each succeeds with nothing, or fails with a [`ValidationError`](/sdks/typescript/errors#protocolvalidationerror): * `validatePrevHash(range)`: checks that a range starting at block 0 has no parent hash, or an all-zero one, and that any other range has one. Fails with `MissingPrevHashError`, `InvalidPrevHashError`. * `validateNetworks(previous, incoming)`: checks that each chain appears once in `incoming`, and the same chains as in `previous`, unless `previous` is empty. Fails with `DuplicateNetworkError`, `NetworkCountChangedError`, `UnexpectedNetworkError`. * `validateConsecutiveness(previous, incoming)`: checks that each incoming range follows the previous one: it starts at the next block with the previous one's hash as its parent, jumps back with a different hash (a reorg), or repeats it. Fails with `HashMismatchOnConsecutiveBlocksError`, `InvalidReorgError`, `GapError`. * `validateAll(previous, incoming)`: runs all of the above, for each incoming range. Fails with any `ValidationError`. They take [`BlockRange`](/sdks/typescript/domain#blockrange)s, and return `Effect`s. ### `blockRangeEquals` ```ts import { blockRangeEquals } from "@edgeandnode/amp/protocol-stream/validation" function blockRangeEquals(a: BlockRange, b: BlockRange): boolean ``` Whether two block ranges have the same chain, first and last blocks, hash, and parent hash. ### Errors `ProtocolStreamError`, `ProtocolArrowFlightError`, `ProtocolValidationError`, `ValidationError`, and the validation errors are in [Errors](/sdks/typescript/errors#protocol-stream-errors).