Keep your application in sync

An application that keeps a processor’s output in its own database owns that copy, and reconciling it is application logic. Leani makes the loop short: it tells the application what it covers and which blocks the live lane skipped. This guide applies to processors with history_control = "application_subscriptions" and history_mode = "on_demand", whose history only the application requests. A node-owned on-demand processor announces skipped blocks the same way, but its operator requests them as POST /admin/v1/materialization-jobs.

Where blocks go missing

After downtime or a long outage, the live lane resumes at the newest verified finalized anchor. It skips the blocks between the processor’s cursor and that anchor. Automatic history refills them; on-demand history leaves them to the application. With [processors.live_gap_fill] (see Live gap fill) the lane fills a short range itself before it applies the next block, and the live stream delivers those blocks as live_recovery batches. It announces only what it did not fill: a longer range, a range history cannot serve yet, and the rest of a fill that failed or ran out of time.
At startup, retained unfinalized blocks are reverted as a reorg only when they are off the finalized chain, or when the node cannot prove that they are on it: without execution peers within a minute, or after an outage longer than about a day. The application receives their undo changes, and the blocks become part of the skipped range. After a restart that follows a long outage, and on reorgs, several blocks can be reverted in a row: one undo batch per block, newest first. Rebuild derived data once per such run, as Undo runs describes.
On the application’s side, a crash between its commit and its acknowledgement redelivers a batch, and a failed backfill subscription stops until acted upon.

The loop

Diff coverage on every live session. The live stream’s hello carries the processor’s coverage as the session starts. Holes in coverage.available, from the first block the application keeps through processedThrough, are blocks the processor never processed. Before its first block, everything from the processor’s start up to the finalized head is history to request. streamLive() exposes each session’s hello but does not reconnect: after a failure the SDK marks retryable, open a new session with backoff, which checks coverage again. subscribe() reconnects on its own and does not expose the hello. Keep this diff when the processor fills live gaps: it still finds the holes below the processor’s cursor, which no fill reaches, and the range of a fill that failed.
Keep announced gaps with the batch. Before the live lane applies a block past its cursor’s successor, the live stream delivers a finalized change of kind system.live_gap.put with data: { fromBlock, toBlock, reason: "not_filled" }, in a batch of its own. On a stream that keeps records until they are acknowledged, it survives reconnects. Store the range together with the batch’s rows, then acknowledge the batch. In the SDK, isLiveGapChange narrows the change, and applyEntityChange ignores it. A processor that fills live gaps announces only what it did not fill, so keep handling notices.
Request the kept ranges. liveGapBackfill(range, request) from @smart-byte/leani-sdk/backfill builds the fill_missing subscription with the idempotency key live-gap:{consumer}:{fromBlock}-{toBlock}. Creating it again for the same range returns the same subscription; a hole that history has partly filled meanwhile is a new range, so remember what you already requested. Drop a range once its subscription exists. When a request fails, keep the range and try again later: a failing request must never hold the live lane back.
Consume the history lanes beside the live lane, as Backfill while following live describes; each has its own lease and acknowledgement.
Act on failed subscriptions. A failed subscription stays failed: inspect its lastError and correct the cause. Then retry it with POST /admin/v1/backfill-subscriptions/{id}/retry (SDK retry(id)), which resumes from its committed progress. To abandon it instead, delete it, adding ?discardUnacknowledged=true (SDK delete(id, { discardUnacknowledged: true })) when the consumer will not read the rest.
Write idempotently. Live, history, and redelivered batches can overlap and arrive out of block order. Write each block’s rows as an upsert keyed by block, so that a block arriving twice, or from history after live, changes nothing. Deduplicate batches by cursor, not by block: block n can come as a notice batch and then as its own batch. A live_recovery batch can repeat a block that a history subscription delivers, such as one the application requested from its coverage diff while the lane filled the same range. A notice can also be followed by live_recovery batches for the blocks it announced, when the lane retries a fill after it held the block that follows, or after a crash or a restart of its live lane between the notice and that block.

import {
  applyEntityChange,
  isLiveGapChange,
  LeaniError,
  type EntityChangeTarget,
  type ProcessorCoverage,
} from "@smart-byte/leani-sdk";
import {
  createBackfillSubscriptionClient,
  liveGapBackfill,
  type BackfillRange,
  type BackfillStatus,
} from "@smart-byte/leani-sdk/backfill";

/**
 * Blocks from `from` that the processor lacks, through the last block it
 * processed. Before its first block, that is everything up to the finalized
 * head.
 */
export function coverageHoles(coverage: ProcessorCoverage, from: number): BackfillRange[] {
  const through = coverage.processedThrough ?? coverage.chainFinalizedHead?.number;
  if (through === undefined) return [];
  const holes: BackfillRange[] = [];
  let next = Math.max(from, coverage.configuredStartBlock);
  for (const { fromBlock, toBlock } of coverage.available) {
    if (fromBlock > next) holes.push({ fromBlock: next, toBlock: Math.min(fromBlock - 1, through) });
    next = Math.max(next, toBlock + 1);
  }
  if (next <= through) holes.push({ fromBlock: next, toBlock: through });
  return holes.filter((hole) => hole.fromBlock <= hole.toBlock);
}

/** An idempotent destination that also keeps the ranges it still has to request. */
export interface SyncedDestination<T> extends EntityChangeTarget<T> {
  /** Keep a range to request, together with the rows of the batch that announced it. */
  rememberGap(range: BackfillRange): Promise<void>;
  /** The ranges kept and not yet requested. */
  gaps(): Promise<BackfillRange[]>;
  /** Drop a range once a subscription delivers it. */
  forgetGap(range: BackfillRange): Promise<void>;
}

export interface SyncOptions {
  baseUrl: string;
  processor: string;
  consumer: string;
  credential: string;
  /** The first block the destination keeps. */
  fromBlock: number;
  /**
   * Consume each new subscription's history lane, beside the live lane:
   * start `runBackfill()` from `@smart-byte/leani-sdk/backfill` here with
   * `liveGapBackfill(subscription, …)` and the same processor and consumer,
   * lease TTL and credential included. Do not await it, which would hold the
   * live lane back, but catch its rejection. `keepInSync` forgets the range
   * once this returns, so until a run resolves `completed` the range is the
   * application's: keep its request and run it again after a rejection or a
   * restart, once the cause is fixed. An abort or a failure keeps the
   * subscription.
   */
  onBackfill(subscription: BackfillStatus): void | Promise<void>;
  signal?: AbortSignal;
}

/**
 * Follow a processor's live lane into a destination, and request as history
 * every block the processor lacks: the holes in the coverage each session
 * starts with, and the blocks the live lane announces it skipped.
 */
export async function keepInSync<T>(
  destination: SyncedDestination<T>,
  options: SyncOptions,
): Promise<void> {
  const leani = createBackfillSubscriptionClient({ baseUrl: options.baseUrl });
  const requested: BackfillRange[] = [];
  const request = async () => {
    for (const range of await destination.gaps()) {
      const inFlight = requested.some(
        (earlier) => earlier.fromBlock <= range.fromBlock && range.toBlock <= earlier.toBlock,
      );
      if (!inFlight) {
        try {
          await options.onBackfill(await leani.create(liveGapBackfill(range, {
            processor: options.processor,
            consumer: {
              id: options.consumer,
              role: "required",
              leaseTtlSeconds: 300,
              credential: options.credential,
            },
            signal: options.signal,
          })));
        } catch (error) {
          // Keep the range and try again after the next batch: a request
          // that fails must not hold the live lane back.
          console.warn(`request for blocks ${range.fromBlock}-${range.toBlock} failed`, error);
          return;
        }
        requested.push(range);
      }
      await destination.forgetGap(range);
    }
  };
  let delayMs = 1_000;
  while (!options.signal?.aborted) {
    try {
      const session = await leani.streamLive<T>(options.processor, options.consumer, {
        credential: options.credential,
        signal: options.signal,
      });
      delayMs = 1_000;
      try {
        for (const hole of coverageHoles(session.hello.coverage, options.fromBlock)) {
          await destination.rememberGap(hole);
        }
        await request();
        for await (const record of session) {
          if (record.type !== "batch") continue;
          for (const change of record.changes) {
            if (isLiveGapChange(change)) await destination.rememberGap(change.data);
            else await applyEntityChange(destination, change);
          }
          await session.acknowledge(record.acknowledgeableCursor);
          await request();
        }
      } finally {
        await session.close();
      }
    } catch (error) {
      if (options.signal?.aborted) return;
      // An outage ends only this session: reconnect, which redelivers what
      // was not acknowledged and checks coverage again. A replaced session
      // can still hold the lease for a moment. Anything else, such as a
      // rejected credential, is the caller's.
      const transient =
        error instanceof LeaniError &&
        (error.retryable ||
          error.code === "consumer_session_active" ||
          error.code === "consumer_session_lost");
      if (!transient) throw error;
      console.warn(`live session failed; reconnecting in ${delayMs} ms`, error);
      await new Promise((resolve) => setTimeout(resolve, delayMs));
      delayMs = Math.min(delayMs * 2, 30_000);
    }
  }
}

Follow the live lane and request every block the processor lacks as history.Verified source

Built by Smart-Byte · Legal notice (Impressum) · Privacy