Query once, then follow live

A normal query followed by a separate stream connection has a race: a change can commit after the query but before the stream starts. queryAndFollow() returns the snapshot and its opaque change boundary atomically. Passing that boundary to subscribe({ after }) closes the gap.

Included-and-finalized processors only

query-and-follow answers 409 finalized_only_snapshot for a processor with publish = "finalized_only". Its retained state already holds included blocks whose changes are unpublished, and the stream never emits an undo for them, so a snapshot could seed rows no later event corrects. Follow such a processor’s change stream from its head instead.

First follow the source installation, including the SDK setup. Stop any other node using API port 8080 before this example. The supplied profile retains the latest USDC/WETH pool state and a bounded change stream; its public Xatu history needs no provider key:

LEANI_SDK_DIR="$(mktemp -d)"
cp "$LEANI_SOURCE/examples/sdk-stream/node.toml" "$LEANI_SDK_DIR/leani.toml"
cd "$LEANI_SDK_DIR"
leani doctor --json
leani backfill --processor usdc-weth-latest --from 17000000 --to 17000999
leani serve

In another terminal, from the repository root:

cd "$LEANI_SOURCE"
bun examples/sdk-stream/index.ts

Expect Loaded 1 pools. The process then waits for more changes; this historical-only profile has no live source, so silence after the seed is expected. Press Ctrl-C to stop. The example keeps the price map in memory; for a durable destination, continue with PostgreSQL.

This historical-only profile lets you inspect the seed and resume boundary. For ongoing P2P updates, enable live execution and verified finality as described in the Uniswap subscription guide.

Query once, then follow without a race

import {
  applyEntityChange, createLeaniClient, LeaniError,
  type GenericSnapshotPage, type LeaniClient, type UniswapPoolPrice,
} from "@leani/sdk";

// Run examples/sdk-stream/node.toml first (see the guide).
export async function followPrices(
  leani: LeaniClient = createLeaniClient({ baseUrl: "http://127.0.0.1:8080" }),
  prices = new Map<string, UniswapPoolPrice>(),
  signal?: AbortSignal,
): Promise<void> {
  const processor = "usdc-weth-latest";
  const collection = "uniswap.pools.current";
  const target = {
    put: (key: string, value: UniswapPoolPrice) => { prices.set(key, value); },
    delete: (key: string) => { prices.delete(key); },
  };
  while (!signal?.aborted) {
    try {
      const snapshot = await leani.processors.queryAndFollow<UniswapPoolPrice>(
        processor, collection, { signal },
      );
      const seed = new Map<string, UniswapPoolPrice>();
      try {
        let page: GenericSnapshotPage<UniswapPoolPrice> = snapshot;
        while (true) {
          for (const entity of page.data) seed.set(entity.key, entity.data);
          if (!page.nextCursor) break;
          page = await leani.processors.queryEntities<UniswapPoolPrice>(
            processor, collection, { cursor: page.nextCursor, signal },
          );
        }
      } finally {
        // Cleanup gets its own deadline even if the subscription was cancelled.
        await leani.processors.releaseSnapshot(processor, snapshot.snapshotId)
          .catch((error: unknown) => {
            if (!(error instanceof LeaniError && error.status === 404)) throw error;
          });
      }
      prices.clear();
      for (const [key, value] of seed) prices.set(key, value);
      console.log(`Loaded ${prices.size} pools`);
      for await (const change of leani.processors.subscribe<UniswapPoolPrice>(
        processor, { after: snapshot.boundaryCursor, signal },
      )) {
        // Both apply and undo carry the entity mutation to perform.
        await applyEntityChange(target, change);
        if (change.operation === "finalized") {
          console.log("finalized through", change.data.throughBlock);
        }
      }
    } catch (error) {
      if (signal?.aborted) return;
      // ResetRequiredError from SSE is also a LeaniError with cursor_expired.
      if (!(error instanceof LeaniError) ||
          !["cursor_expired", "query_snapshot_expired"].includes(error.code)) throw error;
      console.log("Retention moved; rebuilding the snapshot");
    }
  }
}

if (import.meta.main) await followPrices();

Seed local state at an exact boundary, resume SSE there, and handle apply, undo, and finalized events.Verified source

1. Follow every nextCursor to materialize the complete stable snapshot.
2. Release the snapshot when the local seed is committed.
3. Start the resumable stream at boundaryCursor.
4. Apply both apply and undo; an undo envelope carries the inverse entity mutation.
5. Treat finalized as an explicit transition and reset_required as a rebuild instruction. The SDK throws ResetRequiredError on an SSE reset; expired HTTP cursors throw LeaniError with code: "cursor_expired". Never infer either transition from elapsed time.

SSE resume is not a durable acknowledgement

The stream cursor closes an in-process query/follow race and reconnects the live feed. For a destination that must fence Leani’s retention across process restarts, use a named durable consumer.

The page limit only controls transport size. Creating the first page copies the entire matching collection under the store writer lock. Defaults cap each snapshot at 100,000 rows / 64 MiB and all outstanding snapshots together at 32 snapshots / 128 MiB (including key and row overhead). Physical store admission also applies. Release snapshots promptly; the node expires them after five minutes and cleans expired rows every 30 seconds. query_snapshot_capacity returns retryable HTTP 503 when aggregate admission is full. Very slow readers can outlive the retained stream and must rebuild, as the example does.