> **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. # Transactional Streams Status: Unstable. A transactional stream keeps a destination, like a database, in step with a [streaming query](/sdks/typescript/streaming), even when your program crashes or the chain reorgs. Each event it emits has a transaction ID. When rows you processed must go, because a reorg replaced their blocks or because your program stopped before it recorded its progress, it tells you which transactions to undo. If you record which transaction wrote what, each row ends up in your destination exactly once. The `TransactionalStream` service builds on [`ProtocolStream`](/sdks/typescript/streaming). It saves its progress in a state store, which you provide. :::warning Transactional streams inherit the [stream's known issues](/sdks/typescript/streaming): they fail after 30 seconds, which the example below works around, and they lose the rows that arrive with a reorg. ::: ## 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. ## How it works The stream emits three kinds of events, each with an `id`, its transaction ID, which only ever increases: * **`Data`**: new rows, in `data`, from the block ranges in `ranges`. Save them with their transaction ID. * **`Watermark`**: the rows for `ranges` have all arrived. The stream can resume from here. * **`Undo`**: the transactions from `invalidate.start` to `invalidate.end`, inclusive, are void: delete everything they wrote. `cause` says why: a `Reorg`, or a `Rewind` after a restart. After you process an event, you commit it, and the stream saves its progress to the state store. When your program starts again, the stream loads it, and resumes after the last committed `Watermark`. If it had emitted events after that watermark, it first emits an `Undo` with the cause `Rewind` for them: you may have processed them, or not, and their rows will arrive again. ## Process a stream This program copies the dataset's blocks into a JSON file, `blocks.json`, and records which transaction wrote each row, so that an `Undo` can delete them. Its state store is a second file, `stream-state.json`, so that it survives restarts: ```ts [transactional.ts] import { ArrowFlight, ProtocolStream, TransactionalStream } from "@edgeandnode/amp" import { layerTransportGrpc } from "@edgeandnode/amp/arrow-flight/node" import { Effect, Layer } from "effect" import { existsSync, readFileSync, renameSync, writeFileSync } from "node:fs" import { layerFileStateStore } from "./file-state-store.ts" // The destination: block rows, each with the ID of the transaction that wrote it type Row = { block_num: unknown; gas_used: unknown; transaction: number } const load = (): Row[] => (existsSync("blocks.json") ? JSON.parse(readFileSync("blocks.json", "utf8")) : []) // Writes a temporary file, then renames it over the destination, so a crash never leaves it half-written const save = (rows: Row[]) => { writeFileSync("blocks.json.tmp", JSON.stringify(rows), { flush: true }) renameSync("blocks.json.tmp", "blocks.json") } const handle = (event: TransactionalStream.TransactionEvent) => Effect.sync(() => { switch (event._tag) { case "Data": { const rows = event.data.map((row) => ({ block_num: row.block_num, gas_used: row.gas_used, transaction: event.id, })) save([...load(), ...rows]) return console.log(`#${event.id} Data: saved ${rows.length} rows`) } case "Undo": { const { start, end } = event.invalidate save(load().filter((row) => row.transaction < start || row.transaction > end)) return console.log(`#${event.id} Undo (${event.cause._tag}): removed transactions ${start} to ${end}`) } case "Watermark": return console.log(`#${event.id} Watermark: blocks up to ${event.ranges[0]?.numbers.end} are complete`) } }) const program = Effect.gen(function* () { const stream = yield* TransactionalStream.TransactionalStream yield* stream.forEach(`SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks`, {}, handle).pipe( // Works around the keep-alive issue (see Streaming): start again from the saved state Effect.catchDefect((defect) => (defect instanceof RangeError ? Effect.void : Effect.die(defect))), Effect.forever, ) }) const FlightLayer = ArrowFlight.layer.pipe(Layer.provide(layerTransportGrpc({ baseUrl: "http://localhost:1602" }))) const AmpLayer = TransactionalStream.layer.pipe( Layer.provide(layerFileStateStore("stream-state.json")), Layer.provide(ProtocolStream.layer), Layer.provide(FlightLayer), ) await Effect.runPromise(program.pipe(Effect.provide(AmpLayer))) ``` `forEach` runs `handle` on each event, then commits it. `layerFileStateStore` is in [Save the state](/sdks/typescript/transactional-streams#save-the-state): save it beside this file as `file-state-store.ts`. Then run the program: :::terminal ```bash node transactional.ts ``` ``` #0 Data: saved 100 rows #1 Watermark: blocks up to 24000099 are complete ``` ::: The program keeps running, waiting for new blocks. Stop it with Ctrl+C, then run it again: it prints nothing, because it resumes after block 24,000,099, which it already processed. ### Recover from a crash If the program stops after it saves the rows of transaction 0, but before the watermark, the next run starts with an `Undo`: ``` #1 Undo (Rewind): removed transactions 0 to 0 #2 Data: saved 100 rows #3 Watermark: blocks up to 24000099 are complete ``` The stream didn't know whether transaction 0's rows were saved, so it voided them, and sent them again. `blocks.json` ends up with each block once, saved by transaction 2. This needs each write to the destination to be all or nothing. `save` writes a temporary file, then renames it over `blocks.json`: if the program stops while it writes, `blocks.json` keeps the rows it had, and the next run's `Undo` removes the ones it must. Writing to `blocks.json` directly would leave it half-written, and lose the rows of the transactions committed before. ### Handle reorgs When a [reorg](/sdks/typescript/streaming#reorgs) replaces blocks, the stream emits an `Undo` with the cause `Reorg`, which voids the transactions that carried those blocks, and `cause.invalidation` says which blocks they were. The example deletes their rows, like for a rewind. To recover, the stream needs the last watermark from before the reorg: it keeps the watermarks of the last `retention` blocks (128 by default). It fails when it can't: * [`UnrecoverableReorgError`](/sdks/typescript/errors#unrecoverablereorgerror): the reorg goes back further than the watermarks it kept. Raise `retention` to handle deeper reorgs. * [`PartialReorgError`](/sdks/typescript/errors#partialreorgerror): the reorg starts in the middle of a watermark's block range, so part of that range is still valid. This can happen when the stream started with a range that covered many blocks, like a first batch of history. Either way, start over: delete the destination's rows and the state, and run the stream again. :::warning **Known issue:** the rows Amp sends for the replacement blocks [are lost](/sdks/typescript/streaming#reorgs). After an `Undo` with the cause `Reorg`, query the blocks in `cause.invalidation` again with [`flight.query`](/sdks/typescript/querying), once the next `Watermark` covers them. ::: [Test reorgs on a local chain](/sdks/typescript/streaming#test-reorgs-on-a-local-chain) shows how to cause a reorg, to see your code handle it. ## Commit Committing tells the stream you've processed an event, and everything before it. The stream then saves the watermarks up to that event in the state store. Commits are cheap to repeat: committing an event that's already committed does nothing. `forEach` commits each event after its handler succeeds. If the handler fails, the stream stops without committing it, and `forEach` fails with the handler's error: when it runs again, it rewinds to the last committed watermark. To decide when to commit, use `streamTransactional`. It emits each event with a `CommitHandle`, whose `commit` Effect commits it. Commit after your destination has durably saved what the event changed, like after your database's own transaction commits. You don't have to commit every event: committing one commits all the events before it too. ```ts yield* stream.streamTransactional(sql).pipe( Stream.runForEach(([event, handle]) => Effect.gen(function* () { yield* processEvent(event) if (event._tag === "Watermark") yield* handle.commit }), ), ) ``` ## Save the state A state store keeps the stream's progress: the next transaction ID, and the watermarks it can resume or recover from. The SDK has one, `InMemoryStateStore`, which keeps them in memory: when your program exits, they're lost, and the next run starts from the beginning. Use it for tests and experiments. To resume after a restart, provide a `StateStore` that saves to durable storage, like a file or your destination's database. It has four operations: | Operation | What it does | | ---------------- | --------------------------------------------------------------------------------------------------------------------------- | | `load` | Returns the saved state: the watermark buffer, and the next transaction ID. `emptySnapshot` when there's none yet. | | `advance(next)` | Saves the next transaction ID. The stream calls it before it emits each event, so IDs are never reused, even after a crash. | | `commit(commit)` | Adds `commit.insert`'s watermarks to the buffer, after removing those up to `commit.prune`'s ID, if set. | | `truncate(from)` | Removes the watermarks with IDs from `from` on, after a reorg. | Each must be durable when it succeeds, and `commit` must be idempotent: committing the same watermarks again changes nothing. When one fails, fail with a `StateStoreError`: the stream stops. This one saves the state as JSON in a file: ```ts [file-state-store.ts] import { TransactionalStream } from "@edgeandnode/amp" import { Effect, Layer } from "effect" import { existsSync, readFileSync, renameSync, writeFileSync } from "node:fs" type Snapshot = TransactionalStream.StateSnapshot type Operation = "advance" | "commit" | "truncate" | "load" /** A StateStore that keeps a transactional stream's state in a JSON file. */ export function layerFileStateStore(path: string): Layer.Layer { const attempt = (operation: Operation, run: () => A) => Effect.try({ try: run, catch: (cause) => new TransactionalStream.StateStoreError({ reason: String(cause), operation, cause }), }) const read = (): Snapshot => existsSync(path) ? JSON.parse(readFileSync(path, "utf8")) : TransactionalStream.emptySnapshot // Writes a temporary file, then renames it over the state, so a crash never leaves it half-written const update = (operation: Operation, change: (state: Snapshot) => Snapshot) => attempt(operation, () => { writeFileSync(`${path}.tmp`, JSON.stringify(change(read())), { flush: true }) renameSync(`${path}.tmp`, path) }) return Layer.succeed(TransactionalStream.StateStore, { load: attempt("load", read), advance: (next) => update("advance", (state) => ({ ...state, next })), commit: ({ insert, prune }) => update("commit", (state) => { // Skips the watermarks it already saved, so committing twice changes nothing const newest = state.buffer.at(-1)?.[0] ?? -1 return { ...state, buffer: [ ...state.buffer.filter(([id]) => prune === undefined || id > prune), ...insert.filter(([id]) => id > newest), ], } }), truncate: (from) => update("truncate", (state) => ({ ...state, buffer: state.buffer.filter(([id]) => id < from) })), }) } ``` When your destination is a database, store the state in it too, and update both in the same database transaction: then a crash can't leave one ahead of the other. ## Reference Import these from `@edgeandnode/amp`'s `TransactionalStream` namespace, or from `@edgeandnode/amp/transactional-stream`. ### `TransactionalStream` ```ts class TransactionalStream extends Context.Service ``` The service that runs transactional streams. Get it with `yield* TransactionalStream.TransactionalStream`. ### `TransactionalStreamService` ```ts interface TransactionalStreamService { readonly streamTransactional: ( sql: string, options?: TransactionalStreamOptions, ) => Stream readonly forEach: ( sql: string, options: TransactionalStreamOptions, handler: (event: TransactionEvent) => Effect, ) => Effect } ``` #### `streamTransactional` Runs `sql` as a streaming query, resuming from the state store's last watermark, and emits each event with its [`CommitHandle`](/sdks/typescript/transactional-streams#commithandle). Starts with an `Undo` with the cause `Rewind` when events after that watermark were emitted before. * **Parameters**: `sql`, the query; `options`, the [`TransactionalStreamOptions`](/sdks/typescript/transactional-streams#transactionalstreamoptions). * **Returns**: `Stream`. * **Fails with**: [`TransactionalStreamError`](/sdks/typescript/errors#transactional-stream-errors). #### `forEach` Runs `streamTransactional`, calls `handler` on each event, and commits the event once `handler` succeeds. * **Parameters**: `sql`, the query; `options`, the [`TransactionalStreamOptions`](/sdks/typescript/transactional-streams#transactionalstreamoptions), `{}` for the defaults; `handler`, a function from an event to an Effect. * **Returns**: `Effect`, where `E` and `R` are `handler`'s. Never succeeds: the stream doesn't end. * **Fails with**: a [`TransactionalStreamError`](/sdks/typescript/errors#transactional-stream-errors), or `handler`'s error. The failed event isn't committed. ### `TransactionalStreamOptions` ```ts interface TransactionalStreamOptions { readonly retention?: number readonly schema?: QueryOptions["schema"] } ``` | Option | Type | Default | Description | | ----------- | ---------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `retention` | `number` | `128` | How many blocks of watermarks to keep, to recover from reorgs. A reorg deeper than this fails with `UnrecoverableReorgError`. Like `{ retention: 512 }` for a chain with deep reorgs. | | `schema` | A `Schema` | None | Checks each row, as in [`query`](/sdks/typescript/querying#give-rows-a-type). | ### `layer` ```ts const layer: Layer ``` Builds the `TransactionalStream` service. Needs [`ProtocolStream`](/sdks/typescript/streaming#protocolstream) and a [`StateStore`](/sdks/typescript/transactional-streams#statestore). ### `TransactionEvent` ```ts type TransactionEvent = TransactionEventData | TransactionEventUndo | TransactionEventWatermark ``` An event of the stream. Tell them apart by `_tag`: | Type | `_tag` | Properties | | --------------------------- | ------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `TransactionEventData` | `"Data"` | `id`: `TransactionId`. `data`: the rows. `ranges`: the [block ranges](/sdks/typescript/domain#blockrange) they're from. | | `TransactionEventWatermark` | `"Watermark"` | `id`. `ranges`: the block ranges whose rows have all arrived. `prune`: an `Option` of the last transaction whose watermark this one let the stream forget, after `retention`. | | `TransactionEventUndo` | `"Undo"` | `id`. `cause`: an [`UndoCause`](/sdks/typescript/transactional-streams#undocause). `invalidate`: a [`TransactionIdRange`](/sdks/typescript/transactional-streams#transactionidrange), the transactions to undo. | Their schemas are `TransactionEventSchema`, `TransactionEventDataSchema`, `TransactionEventUndoSchema`, and `TransactionEventWatermarkSchema`. ### `TransactionId` ```ts type TransactionId = number & Brand<"Amp/TransactionalStream/TransactionId"> ``` An event's ID: a non-negative integer, one more than the last event's, never reused. Its schema is `TransactionIdSchema`: create one with `TransactionIdSchema.make(5)`. ### `TransactionIdRange` ```ts type TransactionIdRange = { readonly start: TransactionId; readonly end: TransactionId } ``` The transactions from `start` to `end`, both included. Its schema is `TransactionIdRangeSchema`. ### `UndoCause` ```ts type UndoCause = UndoCauseReorg | UndoCauseRewind ``` Why an `Undo` happened: * `UndoCauseReorg`, `{ _tag: "Reorg", invalidation }`: a reorg replaced the blocks in `invalidation`, an array of [`InvalidationRange`](/sdks/typescript/streaming#invalidationrange)s. * `UndoCauseRewind`, `{ _tag: "Rewind" }`: the stream restarted, and the events after its last committed watermark will arrive again. Their schemas are `UndoCauseSchema`, `UndoCauseReorgSchema`, and `UndoCauseRewindSchema`. ### Event constructors ```ts function dataEvent( id: TransactionId, data: ReadonlyArray>, ranges: ReadonlyArray, ): TransactionEventData function undoEvent(id: TransactionId, cause: UndoCause, invalidate: TransactionIdRange): TransactionEventUndo function watermarkEvent( id: TransactionId, ranges: ReadonlyArray, prune: TransactionId | null, ): TransactionEventWatermark function reorgCause(invalidation: ReadonlyArray): UndoCauseReorg function rewindCause(): UndoCauseRewind ``` Create events and causes, for tests of code that handles them. ### `CommitHandle` ```ts interface CommitHandle { readonly id: TransactionId readonly commit: Effect } ``` Commits the event with ID `id`, and every event before it, when you run `commit`. Running it again does nothing. ### `makeCommitHandle` ```ts function makeCommitHandle( id: TransactionId, commitFn: (id: TransactionId) => Effect, ): CommitHandle ``` Creates a `CommitHandle` whose `commit` runs `commitFn(id)`. The stream uses it; you may in tests. ### `StateStore` ```ts class StateStore extends Context.Service ``` The service that saves a transactional stream's state. Provide it with `InMemoryStateStore.layer`, or with your own, as in [Save the state](/sdks/typescript/transactional-streams#save-the-state). ### `StateStoreService` ```ts interface StateStoreService { readonly advance: (next: TransactionId) => Effect readonly commit: (commit: Commit) => Effect readonly truncate: (from: TransactionId) => Effect readonly load: Effect } ``` The operations a state store implements, described in [Save the state](/sdks/typescript/transactional-streams#save-the-state). ### `StateSnapshot` ```ts interface StateSnapshot { readonly buffer: ReadonlyArray]> readonly next: TransactionId } ``` A saved state: `buffer`, the watermarks it can resume or recover from, oldest first, each with its transaction ID; and `next`, the next transaction ID. ### `Commit` ```ts interface Commit { readonly insert: ReadonlyArray]> readonly prune: TransactionId | undefined } ``` What `StateStore.commit` saves: the watermarks to add, oldest first, and the ID of the last watermark to remove, if any. Remove before adding. ### `emptySnapshot`, `emptyCommit`, and `makeCommit` ```ts const emptySnapshot: StateSnapshot // { buffer: [], next: 0 } const emptyCommit: Commit // { insert: [], prune: undefined } function makeCommit( insert: ReadonlyArray]>, prune?: TransactionId, ): Commit ``` The state of a stream that hasn't started, a commit that changes nothing, and a function that creates a `Commit`. ### `InMemoryStateStore` A namespace with state stores that keep the state in memory, lost when your program exits: | Export | Type | Description | | ------------------------- | -------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- | | `layer` | `Layer` | A store that starts empty. | | `layerWithState(initial)` | `Layer` | A store that starts with `initial`, a `StateSnapshot`, to test a restart. | | `layerTest` | `Layer` | A store that starts empty, and a `TestState` service to read it. | | `TestState` | Service | `{ get: Effect }`, the store's current state, for tests: `const snapshot = yield* (yield* InMemoryStateStore.TestState).get`. | ### Recovery functions The functions the stream uses to manage its watermarks, for tests and for state stores that need them: | Function | Returns | | ------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `findRecoveryPoint(buffer, invalidation)` | The last watermark in `buffer` that a reorg's `invalidation` doesn't affect, as `[id, ranges]`, or `undefined` when they're all affected. | | `findPruningPoint(buffer, retention)` | The ID of the last watermark that can be forgotten: one whose blocks all end more than `retention` blocks before the newest watermark's start. `undefined` when there's none. | | `checkPartialReorg(recoveryRanges, invalidation)` | The network whose reorg starts inside the recovery watermark's ranges, after their start, or `undefined`. | | `compressCommits(pendingCommits)` | A single `{ insert, prune }` that applies `pendingCommits`, an array of `[id, PendingCommit]`, in order. | `PendingCommit`, `{ ranges, prune }`, is a watermark waiting to be committed. Import it from `@edgeandnode/amp/transactional-stream/algorithms`. ### `makeStateActor` ```ts import { makeStateActor } from "@edgeandnode/amp/transactional-stream/state-actor" const makeStateActor: (store: StateStoreService, retention: number) => Effect ``` Creates the object that assigns transaction IDs, applies events to the state, and commits. `streamTransactional` creates one per stream. A `StateActor` has `watermark`, the last watermark; `peek`, the next ID; `execute(action)`, which turns an `Action` (a protocol `Message`, or a `Rewind`) into an event and its `CommitHandle`; and `commit(id)`. You don't need it unless you build your own stream. ### Errors `TransactionalStreamError`, `StateStoreError`, `UnrecoverableReorgError`, and `PartialReorgError` are in [Errors](/sdks/typescript/errors#transactional-stream-errors).