substreams-sink

name	substreams-sink
description	Expert knowledge for consuming Substreams data in applications. Use when building sinks, real-time data pipelines, or integrating Substreams outputs into Go, JavaScript, Python, or Rust applications.
license	Apache-2.0
compatibility	{"platforms":["claude-code","cursor","vscode","windsurf"]}
metadata	{"version":"1.4.0","author":"StreamingFast","documentation":"https://substreams.streamingfast.io"}

Substreams Sink Development Expert

Expert assistant for consuming Substreams data - building production-grade sinks and data pipelines.

Skill routing

User goal	Skill
Custom app sink (Go / JS / Python / Rust SDK consumer)	this skill (substreams-sink)
SQL module (db_out / DatabaseChanges / From-proto)	substreams-sql
Run the sink yourself (CLI, schema, ops on your infra)	substreams-sink-deploy-local
StreamingFast hosts the sink (Portal HostedService)	substreams-hosted-sink (+ portal-api)
graph_out / EntityChanges module (produce entities)	this skill (inline proto) + substreams-dev

Prefer existing sink CLIs (substreams-sink-sql, files, kv) over a hand-rolled consumer when they already cover the destination.

Core Concepts

What is a Substreams Sink?

A Substreams sink is an application that:

• Connects to a Substreams endpoint via gRPC
• Streams processed blockchain data from a Substreams package (.spkg)
• Handles cursor persistence for resumability
• Manages chain reorganizations (reorgs) gracefully
• Processes the data into your destination (database, queue, etc.)

Note: Before building a custom sink, consider using existing solutions:

• substreams-sink-sql - For PostgreSQL and ClickHouse. Handles cursor management, reorgs, batching, and schema management out of the box. Install via brew install streamingfast/tap/substreams-sink-sql or download binaries. For full CLI ops, load substreams-sink-deploy-local.
• substreams-sink-kv - For key-value stores.
• substreams-sink-files - For file-based outputs (JSON, CSV, Parquet).

Important: Do NOT use substreams-sink-postgres - this is a deprecated name. Use substreams-sink-sql which supports both PostgreSQL and ClickHouse.

The examples in this guide use database code for illustration purposes. For production SQL database sinks, substreams-sink-sql is highly recommended as it solves cursor persistence, reorg handling, batching, and many edge cases already.

Key Components

1. Endpoint: gRPC server providing Substreams data (e.g., mainnet.eth.streamingfast.io:443)
2. Package (.spkg): Compiled Substreams with modules and protobuf schemas
3. Module: The specific output module to stream from
4. Cursor: Opaque string for resuming streams at exact position
5. Block Range: Start and stop blocks for the stream

Authentication

All Substreams endpoints require authentication. Get your API key from The Graph Market - sign up at thegraph.market/auth/signup.

CLI Authentication (Recommended):

substreams auth  # Interactive authentication, stores token locally

Quick Token Generation: Visit thegraph.market/auth/substreams-devenv to generate a JWT token from your API key directly in the browser.

Environment Variables (Alternative):

# Set API key (recommended)
export SUBSTREAMS_API_KEY=<your-api-key>

# Or set bearer token directly
export SUBSTREAMS_API_TOKEN=<your-jwt-token>

The substreams auth command handles token exchange and local storage automatically.

Note — this is data-plane auth, not Portal auth. SUBSTREAMS_API_KEY / SUBSTREAMS_API_TOKEN authenticate the sink against the streaming data endpoints. They are separate from the StreamingFast Portal admin API (billing, usage, hosted deployments), which the portal-api and portal-api-jwt skills cover. A Portal device-code/Bearer token from portal-api-jwt is scoped to Portal routes only and will not authenticate a sink — use the substreams auth flow here.

Sink Output Types — Pick the Right Proto FIRST

Before writing a graph_out or db_out module, choose the correct output proto. These are different sink contracts; mixing them up produces a buildable-but-wrong pipeline.

You want to write to...	Output proto type	Crate / proto package	When to use
The Graph (subgraph entities)	sf.substreams.sink.entity.v1.EntityChanges	substreams-entity-change (BROKEN — see below)	graph_out modules feeding a Graph Node, hosted subgraph, or substreams-sink-subgraph
Postgres / ClickHouse / SQL DB	sf.substreams.sink.database.v1.DatabaseChanges	substreams-database-change = "4"	db_out modules feeding substreams-sink-sql or hosted SQL sink
Custom sink (Go/Rust consumer)	Your own proto type	n/a	Bespoke consumers reading raw module output

Rule: never use DatabaseChanges for graph-out, never use EntityChanges for SQL sinks. They are not interchangeable. Acceptance tests in eval corpus auto-zero on wrong proto type.

Graph Node / The Graph Output (Entity Changes)

Blocker — read before adding substreams-entity-change.

Do not rely on the substreams-entity-change crate for modern substreams = "0.7" pipelines:

• v1 pins prost = "0.11" / substreams = "0.5" → prost trait conflicts with the current toolchain.
• v2.0.0 has prost ^0.13 but still depends on substreams ^0.6, so it does not drop cleanly into a 0.7 tree.

No version constraint fully fixes this for 0.7 today. Prefer inlining the proto (below). Re-check crates.io before changing this advice.

New projects: prefer SQL (db_out + substreams-sql / substreams-sink-sql) over graph_out unless you specifically need Graph Node / EntityChanges.

Workaround: inline the entity-change proto

Define sf.substreams.sink.entity.v1 types inline. This proto is for graph-out (The Graph) only. Do NOT use it for SQL sinks — that's a different proto package (sf.substreams.sink.database.v1).

1. Add the proto definition (proto/entity.proto):

syntax = "proto3";
package sf.substreams.sink.entity.v1;

message EntityChanges {
  repeated EntityChange entity_changes = 5;  // NOT 1 — consumers read field 5
}

message EntityChange {
  string entity = 1;
  string id     = 2;
  // Deprecated, this is not used within `graph-node`.
  uint64 ordinal = 3;
  enum Operation {
    OPERATION_UNSPECIFIED = 0;
    OPERATION_CREATE      = 1;
    OPERATION_UPDATE      = 2;
    OPERATION_DELETE      = 3;
    OPERATION_FINAL       = 4;
  }
  Operation operation   = 4;
  repeated Field fields = 5;
}

message Value {
  oneof typed {
    int32  int32      = 1;
    string bigdecimal = 2;
    string bigint     = 3;
    string string     = 4;
    string bytes      = 5;  // hex string, NOT proto `bytes`
    bool   bool       = 6;
    int64  timestamp  = 7;
    Array  array      = 10;
  }
}

message Array {
  repeated Value value = 1;
}

message Field {
  string name = 1;
  optional Value new_value = 3;
  // Deprecated, this is not used within `graph-node`.
  optional Value old_value = 5;
}

Field numbers are load-bearing. entity_changes is 5, not 1; old_value is 5, not 2; Value.bytes is a string (hex), not proto bytes. Getting any of these wrong produces a module that builds, streams, and silently writes zero entities — the consumer decodes a field number you never wrote. substreams run will still look correct, because it decodes with your own package's descriptor; the mismatch only surfaces against a real Graph Node.

Wire compatibility: Both the package name AND the exact message/field definitions (numbers + types) must match. package sf.substreams.sink.entity.v1; is required — do NOT change it to sf.substreams.sink.database.v1 (that's the SQL sink). The block above is a verbatim copy of the canonical source; re-copy from there rather than retyping, and do not simplify field types or renumber fields.

2. Reference in manifest (substreams.yaml):

protobuf:
  files:
    - proto/entity.proto
    # ... other protos

modules:
  - name: graph_out
    kind: map
    inputs:
      - map: map_events
    output:
      type: proto:sf.substreams.sink.entity.v1.EntityChanges  # ← graph-out
      # NOT: proto:sf.substreams.sink.database.v1.DatabaseChanges (that's SQL)

3. Use the generated type in Rust:

use crate::pb::sf::substreams::sink::entity::v1::{EntityChange, EntityChanges, Field, Value};
use crate::pb::sf::substreams::sink::entity::v1::entity_change::Operation;
use crate::pb::sf::substreams::sink::entity::v1::value::Typed;

// `Field.new_value` and `old_value` are `Value` messages, so the prost-generated
// Rust type is `Option<Value>` — you cannot assign a raw String/Vec<u8>/u64 directly.
// Wrap each scalar in the appropriate `Typed::*` oneof variant. Helpers below
// cover the common cases; add `Typed::Bool`, `Typed::Int32`, `Typed::Bigdecimal`,
// `Typed::Timestamp` (i64), `Typed::Array` as needed for your schema.
// Note `Typed::Bytes` takes a `String` (hex), not a `Vec<u8>` — see the proto above.
fn val_string(s: impl Into<String>) -> Option<Value> {
    Some(Value { typed: Some(Typed::String(s.into())) })
}
fn val_bigint(decimal_str: impl Into<String>) -> Option<Value> {
    // BigInt is wire-encoded as its decimal string, e.g. "12345"
    Some(Value { typed: Some(Typed::Bigint(decimal_str.into())) })
}

#[substreams::handlers::map]
pub fn graph_out(events: Events) -> Result<EntityChanges, substreams::errors::Error> {
    let mut changes = EntityChanges::default();

    for mint in events.mints {
        changes.entity_changes.push(EntityChange {
            entity: "NftMint".to_string(),
            id: format!("{}-{}", mint.tx_hash, mint.log_index),
            ordinal: mint.ordinal,
            operation: Operation::Create as i32,
            fields: vec![
                // tokenId: uint256 → BigInt-as-decimal-string
                Field { name: "tokenId".to_string(), old_value: None, new_value: val_bigint(mint.token_id.clone()) },
                // address / hash fields: keep as hex strings for Graph Node consumption
                Field { name: "to".to_string(),      old_value: None, new_value: val_string(mint.to.clone()) },
                Field { name: "txHash".to_string(),  old_value: None, new_value: val_string(mint.tx_hash.clone()) },
            ],
        });
    }

    Ok(changes)
}

Do NOT add substreams-entity-change to Cargo.toml when using this workaround. The generated proto code is sufficient; adding the crate triggers the prost conflict.

For SQL sinks (Postgres / ClickHouse / db_out), use substreams-database-change = "4" — see substreams-sql/SKILL.md. Those crates ARE compatible with the current toolchain. Do NOT inline DatabaseChanges proto and call your module graph_out — that mixes sink types and the run will fail (or worse, succeed silently with garbage data).

Language Recommendations

Language	Recommendation	Best For
Go	Official SDK (Recommended)	Production sinks, StreamingFast sinks
JavaScript	Official SDK	Web apps, Node.js services
Python	Reference implementation	Prototyping, data analysis
Rust	Reference implementation	High-performance custom sinks

Quick Start by Language

Go (Recommended)

Import path is github.com/streamingfast/substreams/sink (package lives in the main Substreams module; the old standalone github.com/streamingfast/substreams-sink module is deprecated).

package main

import (
    "context"
    "fmt"

    "github.com/spf13/cobra"
    "github.com/spf13/pflag"
    . "github.com/streamingfast/cli"
    "github.com/streamingfast/logging"
    pbsubstreamsrpc "github.com/streamingfast/substreams/pb/sf/substreams/rpc/v2"
    "github.com/streamingfast/substreams/sink"
)

var expectedOutputModuleType = "sf.substreams.sink.database.v1.DatabaseChanges" // adjust to your module
var zlog, tracer = logging.RootLogger("my-sink", "github.com/you/my-sink")

func main() {
    logging.InstantiateLoggers()
    Run("my-sink", "Custom Substreams sink",
        Command(sinkRunE,
            "sink [<manifest> [<output_module>]]",
            "Run the sink",
            RangeArgs(0, 2),
            Flags(func(flags *pflag.FlagSet) { sink.AddFlagsToSet(flags) }),
        ),
        OnCommandErrorLogAndExit(zlog),
    )
}

func sinkRunE(cmd *cobra.Command, args []string) error {
    manifestPath := "substreams.yaml"
    outputModule := sink.InferOutputModuleFromPackage
    if len(args) > 0 {
        manifestPath = args[0]
    }
    if len(args) > 1 {
        outputModule = args[1]
    }

    sinker, err := sink.NewFromViper(
        cmd,
        expectedOutputModuleType,
        manifestPath,
        outputModule,
        "my-sink/1.0.0", // userAgent
        zlog,
        tracer,
    )
    if err != nil {
        return fmt.Errorf("create sinker: %w", err)
    }

    // Load persisted cursor on restart; blank = start from module initialBlock / flags
    cursor := sink.NewBlankCursor()
    sinker.Run(context.Background(), cursor, sink.NewSinkerHandlers(
        handleBlockScopedData,
        handleBlockUndoSignal,
    ))
    return nil
}

func handleBlockScopedData(ctx context.Context, data *pbsubstreamsrpc.BlockScopedData, isLive *bool, cursor *sink.Cursor) error {
    // Process data.Output.MapOutput, then persist cursor AFTER success
    _ = cursor
    return nil
}

func handleBlockUndoSignal(ctx context.Context, undoSignal *pbsubstreamsrpc.BlockUndoSignal, cursor *sink.Cursor) error {
    // Rewind data to undoSignal.LastValidBlock, persist cursor
    _ = cursor
    return nil
}

CLI flags (via AddFlagsToSet): -e/--endpoint, -s/--start-block, -t/--stop-block, --final-blocks-only, -p/--params, auth env vars, etc.

See references/go-sink.md for complete guide.

JavaScript (Node.js)

Prefer createGrpcTransport (faster for large streams) over Connect HTTP. Requires "type": "module" in package.json. Pin your deps — the unpinned install fails to resolve:

npm install @substreams/core@0.16.0 @connectrpc/connect@1.3.0 @connectrpc/connect-node@1.3.0

@connectrpc/connect 2.x peers @bufbuild/protobuf ^2, but @substreams/core still peers ^1 — installing both unpinned gives ERESOLVE.

Official pattern:

import {
    createRequest,
    streamBlocks,
    createAuthInterceptor,
    createRegistry,
    fetchSubstream,
} from "@substreams/core";
import { createGrpcTransport } from "@connectrpc/connect-node";

const TOKEN = process.env.SUBSTREAMS_API_TOKEN;
const ENDPOINT = "https://mainnet.eth.streamingfast.io:443";
const SPKG = "https://spkg.io/streamingfast/substreams-eth-block-meta-v0.4.3.spkg";
const MODULE = "db_out";

const pkg = await fetchSubstream(SPKG);
const registry = createRegistry(pkg);
const transport = createGrpcTransport({
    baseUrl: ENDPOINT,
    interceptors: [createAuthInterceptor(TOKEN)],
    jsonOptions: { typeRegistry: registry },
});

const request = createRequest({
    substreamPackage: pkg,
    outputModule: MODULE,
    productionMode: true,
    startBlockNum: 17000000n, // number | bigint — a string is a TS type error
    stopBlockNum: "+1000",
    startCursor: (await getCursor()) ?? undefined,
});

let blockCount = 0;
for await (const response of streamBlocks(transport, request)) {
    const msg = response.message;
    if (msg.case === "blockScopedData") {
        // Process msg.value.output, then persist msg.value.cursor AFTER success
        await writeCursor(msg.value.cursor);
        blockCount++;
    } else if (msg.case === "blockUndoSignal") {
        // Rewind to msg.value.lastValidBlock.number, persist lastValidCursor
        await writeCursor(msg.value.lastValidCursor);
    } else if (msg.case === "fatalError") {
        // Server-side module failure — without this branch it looks like success
        throw new Error(`fatal error in module ${msg.value.module}: ${msg.value.reason}`);
    }
}

// A clean exit with zero blocks is NOT success — on createGrpcTransport a bad
// token ends the stream silently instead of throwing Unauthenticated.
if (blockCount === 0) throw new Error("stream ended with 0 blocks — check your token");

Wrap the stream loop in retry/backoff for normal disconnects. See references/javascript-sink.md for complete guide.

Python

import grpc
from sf.substreams.rpc.v2 import service_pb2, service_pb2_grpc

creds = grpc.ssl_channel_credentials()
with grpc.secure_channel("mainnet.eth.streamingfast.io:443", creds) as channel:
    stub = service_pb2_grpc.StreamStub(channel)
    metadata = [("authorization", f"Bearer {token}")]

    request = service_pb2.Request(
        start_block_num=17000000,
        stop_block_num=17001000,
        modules=package.modules,
        output_module="map_events",
        production_mode=True,
    )

    for response in stub.Blocks(request, metadata=metadata):
        if response.WhichOneof("message") == "block_scoped_data":
            # Process block data
            pass
        elif response.WhichOneof("message") == "block_undo_signal":
            # Handle reorg
            pass

See references/python-sink.md for complete guide.

Rust

Reference implementation (no official SDK). Stream wrapper shape from the examples:

use futures03::StreamExt; // brings `stream.next()` into scope
use substreams_stream::{BlockResponse, SubstreamsStream};

// `substreams_stream` / `substreams` here are LOCAL modules vendored from the
// examples repo (`mod substreams_stream;`) — there is no `substreams-stream`
// crate, and do NOT `cargo add substreams` (that's the WASM module crate).

// SubstreamsStream::new(endpoint, cursor, package, output_module, start_block, stop_block)
let mut stream = SubstreamsStream::new(
    endpoint,                 // Arc<SubstreamsEndpoint>
    cursor,                   // Option<String>
    Some(package),            // Option<Package>
    "map_events".to_string(), // output module name
    start_block,              // i64
    stop_block,               // u64 (0 = live)
);

while let Some(response) = stream.next().await {
    match response? {
        BlockResponse::New(data) => {
            // Process block data, then persist cursor AFTER success
            persist_cursor(&data.cursor)?;
        }
        BlockResponse::Undo(signal) => {
            // Handle reorg, then persist last_valid_cursor
            rewind_to_block(signal.last_valid_block.as_ref())?;
            persist_cursor(&signal.last_valid_cursor)?;
        }
    }
}

See references/rust-sink.md for complete guide.

Critical Concepts

Cursor Management

The cursor is the most critical piece of sink development.

RULE #1: Persist cursor AFTER successful processing, never before.
RULE #2: On restart, load persisted cursor and resume from there.
RULE #3: Blank/empty cursor means start from the beginning.

The cursor is an opaque string that encodes:

• Block number and hash
• Module execution state
• Position within the block

Cursor persistence patterns:

Storage	Use Case	Example
File	Development, single instance	cursor.txt
Database	Production, multi-instance	Cursors table with module key
Redis	High availability	Key-value with TTL

See references/cursor-reorg.md for detailed patterns.

Chain Reorganization (Reorg) Handling

When a chain reorganizes, you receive a BlockUndoSignal:

BlockUndoSignal {
    last_valid_block: BlockRef { number: 17000100, id: "0xabc..." }
    last_valid_cursor: "opaque-cursor-string"
}

Required actions:

1. Delete/revert all data for blocks > last_valid_block.number
2. Persist the last_valid_cursor
3. Continue streaming (new blocks will follow automatically)

Final blocks only mode (recommended for most sinks):

• Set final_blocks_only: true in request (--final-blocks-only on the CLI, finalBlocksOnly: true in JS)
• Only receive blocks that cannot be reorganized
• Eliminates need for undo handling
• Trade-off: you lag the chain tip by that chain's finality distance — ~13 min on Ethereum mainnet (2 epochs), ~seconds on Solana, varies elsewhere

Error Handling & Retry

Fatal errors (do not retry):

• Unauthenticated - Invalid or expired token
• InvalidArgument - Bad request parameters

Retryable errors (implement exponential backoff):

• Unavailable - Server temporarily unavailable
• ResourceExhausted - Rate limited
• Internal - usually a transient stream reset (RST_STREAM) on long-lived streams, not a server bug — retry it
• Connection timeouts

The official Rust reference implementation treats only Unauthenticated as fatal and reconnects on everything else. Do not classify Internal as fatal: it is the standard gRPC surfacing of a routine disconnect, and treating it as fatal kills sinks on exactly the reconnect they should ride out.

A clean stream end is not proof of success. Handle the fatal_error message of the response oneof explicitly — a server-side module panic arrives there, and a switch that only covers block_scoped_data / block_undo_signal will fall through and report success. Likewise, if you asked for a range and processed zero blocks, treat that as an error rather than a clean exit.

Exponential backoff pattern:

Base delay: 500ms
Max delay: 45-90 seconds
Jitter: Add random 0-100ms

Production Mode vs Development Mode

Feature	Production Mode	Development Mode
Parallel execution	Yes	No
Output	Single module only	All modules
Performance	Optimized	Debug-friendly
Use case	Sinks	Testing, debugging

Always use production mode for sinks:

Language	How
Python	production_mode=True — field on service_pb2.Request
JavaScript	productionMode: true — option to createRequest
Rust	production_mode: true — field on the v3 Request
Go	already the default — see below

In Go, production mode is already the default with sink.NewFromViper — just do not pass --development-mode. (There is no sink.WithProductionMode(); the sink.With* options do not include one, and all of them are deprecated in favour of configuring SinkerConfig via sink.NewFromConfig.)

Common Endpoints

Network	Endpoint
Ethereum Mainnet	mainnet.eth.streamingfast.io:443
Ethereum Sepolia	sepolia.eth.streamingfast.io:443
Polygon	polygon.streamingfast.io:443
Arbitrum One	arb-one.streamingfast.io:443
Optimism	mainnet.optimism.streamingfast.io:443
Base Mainnet	base-mainnet.streamingfast.io:443
BNB	bnb.streamingfast.io:443
Solana Mainnet-Beta	mainnet.sol.streamingfast.io:443
NEAR Mainnet	mainnet.near.streamingfast.io:443

Full list: docs.substreams.dev/reference-material/chains-and-endpoints (raw data: TheGraphNetworksRegistry.json)

Block Range Syntax

# Explicit range
--start-block 17000000 --stop-block 17001000

# From manifest initialBlock — omit --start-block entirely (empty = use initialBlock).
# There is no `:` syntax; `--start-block :` fails with "start block is invalid".
--stop-block 17001000

# Relative stop (process 1000 blocks)
--start-block 17000000 --stop-block +1000

# To chain head (live streaming)
--start-block 17000000 --stop-block 0

Module Parameters

Pass runtime parameters to modules:

# Single parameter
-p "map_events=0xa0b86a33e6..."

# Multiple parameters
-p "map_events=0xa0b86a33..." -p "filter_module=type:transfer"

# JSON parameters
-p 'map_events={"contracts":["0x123","0x456"],"min_value":1000}'

Protobuf Code Generation

Generate language bindings from .spkg files:

# Install buf
go install github.com/bufbuild/buf/cmd/buf@latest

# Generate from local .spkg
buf generate --exclude-path="google" ./my-substreams.spkg#format=bin

# Generate from URL
buf generate "https://spkg.io/streamingfast/substreams-eth-block-meta-v0.4.3.spkg#format=binpb"

# Generate from buf registry
buf generate buf.build/streamingfast/substreams --include-imports

Troubleshooting

Connection Issues

"Unauthenticated" error:

• Verify API key/token is set correctly
• Check token hasn't expired
• Ensure correct environment variable name

"Connection refused" error:

• Verify endpoint URL and port
• Check TLS is enabled for https://
• Test network connectivity

Empty Output

No data received:

• Verify initialBlock in manifest is before your start block
• Check the output module name is correct
• Ensure the block range contains relevant data
• Try a known-good block range first
• JS: a silent clean exit with zero blocks on createGrpcTransport usually means a bad token — the trailers-only Unauthenticated is swallowed. Re-run with createConnectTransport to surface the real ConnectError.
• graph_out writing nothing? Check your EntityChanges proto field numbers against the canonical proto — entity_changes must be 5. substreams run will look fine either way.

Performance Issues

Slow processing:

• Enable production mode
• Use gzip compression
• Increase connection timeout
• Consider final_blocks_only for non-realtime needs

Resources

• Official Documentation
• Go Sink Reference
• JavaScript Sink Reference
• Python Sink Reference
• Rust Sink Reference
• Cursor & Reorg Handling

Getting Help

• Discord Community
• GitHub Issues
• Example Repository