Barrow

Barrow is a framework for building blockchain indexing tools. It provides a simple API for defining and running indexing jobs on the Cardano blockchain.

Installation

npm i @ada-anvil/barrow

Usage

Controller

The Controller class is the main entry point for defining and running indexing jobs.

Constructor:

new Controller(config, startOpts?)

Parameters:

1. config (required): Configuration object with the following properties:

   • syncClient: An instance of SyncClient that provides a generator for sync events.
   • errorHandler (optional): An instance of ErrorHandler that handles errors during sync events.
   • logger (optional): A function that handles log events.
   • tracingConfig (optional): An object that configures tracing for the controller.

2. startOpts (optional): Default options to use for all start() calls. These will be merged with options passed to start(), with start() options taking precedence. See Sync Job Configuration for available options (excluding point).

SyncClient

The SyncClient interface defines a method for generating sync events.

Currently, the only implementation of SyncClient is OgmiosSyncClient, which uses the Ogmios API to sync with the blockchain.

Future support is planned for other sync clients such as Dolos.

ErrorHandler

The ErrorHandler class is responsible for handling errors during sync events.

Methods:

• register(filter, handler): Registers an error handler or retry policy for a specific error type or class.
• handle(error): Processes an error and returns the handling result.
• reset(): Resets the error handler to its initial state.

Retry policies

Built-in retry handlers:

• ErrorHandler.retry(options): Retries the sync event after a specified delay.
• ErrorHandler.retryWithBackoff(options): Retries the sync event with exponential backoff.

Error filters

A filter can be:

• An Error class (only instances of that class will be handled)
• A function that takes an error and returns a boolean indicating whether to handle it

Retry options

Options for ErrorHandler.retry and ErrorHandler.retryWithBackoff:

• maxRetries (optional): Maximum number of retries (default: 3)
• baseDelay (optional): Base delay in milliseconds between retries (default: 1000)
• backoff (optional): Use exponential backoff (default: false)
• persistent (optional): Preserve error handler state between retries (default: false)

Getting Started

Step 1: Install Dependencies

Install the Ogmios client:

npm i @cardano-ogmios/client

Step 2: Create a Sync Client

Create an instance of OgmiosSyncClient:

import { OgmiosSyncClient } from "@ada-anvil/barrow/ogmios";

const syncClient = new OgmiosSyncClient({
  host: "localhost",
  port: 1337,
  tls: false,
});

Configuration options:

• host: Ogmios node hostname
• port: Ogmios node port
• tls: Enable TLS connection

Step 3: Create a Controller

Create a Controller instance with your sync client:

import { Controller, ErrorHandler } from "@ada-anvil/barrow";

const controller = new Controller({
  syncClient,
  errorHandler: new ErrorHandler(),
});

You can optionally provide default start options as a second parameter. These defaults will be merged with options passed to start():

const controller = new Controller(
  {
    syncClient,
    errorHandler: new ErrorHandler(),
  },
  {
    // Default options for all start() calls
    throttle: [100, "milliseconds"],
    fn: (event) => {
      console.log(event);
    },
  },
);

Step 4: Start Syncing

Start the controller with a sync job configuration:

await controller.start({
  fn: (event) => {
    console.log(event);
  },
  point: {
    slot: 101163751,
    id: "fa5a6a51632b90557665fcb33970f4fb372dff6ad0191e083ff3b6b221f2b87e",
  },
  throttle: [100, "milliseconds"],
});

// Wait for sync completion
await controller.waitForCompletion();

Controlling Sync Jobs

Pause and Resume:

await controller.pause(); // Preserves state
await controller.resume(); // Resumes from paused point

Restart:

Calling start() on a paused job resets the state and starts from scratch.

Job Completion

A sync job can complete in two ways:

1. Using takeUntil: The takeUntil function returns true

   await controller.start({
     fn: (event) => {
       /* process event */
     },
     point: startPoint,
     takeUntil: ({ state }) => state.meta.syncTip?.slot >= targetSlot,
   });

   Note: takeUntil runs on ALL events, including filtered ones. This allows you to stop syncing based on conditions that don't depend on event processing:

   await controller.start({
     fn: (event) => {
       /* process event */
     },
     filter: (event) => event.type === "apply", // Only process apply events
     point: startPoint,
     takeUntil: ({ state }) => state.counters.filterCount >= 100, // Stop after 100 filtered events
   });

   To run takeUntil only on processed events, you can use the isFilteredOut property:

   await controller.start({
     fn: (event) => {
       /* process event */
     },
     filter: (event) => event.type === "apply",
     point: startPoint,
     takeUntil: ({ lastEvent, state }) => {
       // Return early if the event was filtered out
       if (lastEvent.isFilteredOut) return false;
   
       // Only count processed events
       return state.counters.applyCount >= 10;
     },
   });

2. Using handler return value: The fn handler returns { done: true }

   await controller.start({
     fn: (event) => {
       // Process event
       if (someCondition) {
         return { done: true };
       }
     },
     point: startPoint,
   });

Throttling

The throttle option allows you to control the rate of event processing by adding delays between events. Throttle is applied to ALL events, including filtered ones:

await controller.start({
  fn: (event) => {
    /* process event */
  },
  filter: (event) => event.type === "apply", // Only process apply events
  point: startPoint,
  throttle: [100, "milliseconds"], // Delays after ALL events (filtered and processed)
});

Sync Job Configuration

Configuration properties:

• point (required): Starting point for syncing (slot and block ID)
• fn (optional): Function that handles sync events
• throttle (optional): Throttle duration for sync events. Throttle applies to ALL events (including filtered events)
• filter (optional): Function to filter sync events
• takeUntil (optional): Function that returns true to stop syncing. This function runs on ALL events (including filtered events). Use lastEvent.isFilteredOut to handle filtered events differently

Data Structures

Sync Event:

• type: Event type
• block: Block that was synced
• tip: Current chain tip

Point:

• slot: Slot number
• id: Block hash

Tip:

• slot: Slot number
• id: Block hash
• height: Block height

Block:

• type: Block type
• era: Cardano era
• id: Block hash
• height: Block height
• slot (optional): Slot number

Logger

Barrow provides built-in logging support using Pino.

Setup

Install Pino:

npm i pino

Configure the logger:

import { pinoLogger } from "@ada-anvil/barrow/pino";
import { pino } from "pino";

const controller = new Controller({
  syncClient: new OgmiosSyncClient({
    host: "localhost",
    port: 1337,
    tls: false,
  }),
  logger: pinoLogger(pino()),
});

Tracing

Barrow supports OpenTelemetry for metrics and tracing.

Setup

Install OpenTelemetry:

npm i @opentelemetry/api

Configure tracing:

import { otelTracingConfig } from "@ada-anvil/barrow/otel";

const controller = new Controller({
  syncClient: new OgmiosSyncClient({
    host: "localhost",
    port: 1337,
    tls: false,
  }),
  tracingConfig: otelTracingConfig(),
});

The otelTracingConfig function accepts either:

• A Meter instance
• A configuration object with name, version (optional), and opts (optional)

Available Metrics

Metric Definitions

| Metric Key | Type | Name | Description | Value Type | Unit | | -------------- | --------- | ---------------- | --------------------------------- | ---------- | ------------ | | status | gauge | status | Controller status | int | - | | syncTipSlot | gauge | sync_tip_slot | Sync tip slot | int | - | | syncTipHeight | gauge | sync_tip_height | Sync tip height | int | - | | chainTipSlot | gauge | chain_tip_slot | Chain tip slot | int | - | | chainTipHeight | gauge | chain_tip_height | Chain tip height | int | - | | isSynced | gauge | is_synced | Is synced (1 = yes, 0 = no) | int | - | | processingTime | histogram | processing_time | Time it takes to process an event | - | milliseconds | | arrivalTime | histogram | arrival_time | Time it takes to receive an event | - | milliseconds | | applyCount | gauge | apply_count | Number of apply events | int | - | | resetCount | gauge | reset_count | Number of reset events | int | - | | filterCount | gauge | filter_count | Number of filtered events | int | - | | errorCount | gauge | error_count | Number of errors | int | - |

Examples

Example implementations are available in the src/examples directory.

Running Examples

1. Install dependencies:

   npm i

2. Create a .env file in the project root:

   OGMIOS_NODE_HOST=<ogmios-node-host>
   OGMIOS_NODE_PORT=<ogmios-node-port>
   OGMIOS_NODE_TLS=<ogmios-node-tls>

3. Run an example:

   npm run example kitchen-sink