cometdump

package module
v0.2.0-alpha Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 13, 2025 License: MIT Imports: 22 Imported by: 0

README

CometDump

PkgGoDev

A Go library and CLI tool for efficiently storing and retrieving CometBFT blockchain data. CometDump downloads blocks from CometBFT-based chains and stores them in compressed chunks for fast local access.

Features

  • 🚀 Fast Block Retrieval: Efficiently fetch blocks from multiple CometBFT nodes with concurrent workers
  • 💾 Compressed Storage: Blocks are stored in compressed msgpack format using Brotli compression
  • 🔍 Smart Node Discovery: Automatically discover and select the best nodes from the network
  • Iterator Interface: Stream through blocks with Go's iterator pattern
  • 🌐 RPC Mirror: Serve stored blockchain data via CometBFT-compatible JSON-RPC endpoints
  • 🔧 Configurable Sync: Flexible sync configuration with version constraints and chunk sizing

Installation

go get github.com/ehsanranjbar/cometdump

Usage

Example

This example demonstrates syncing blocks from a CometBFT node, iterating through stored blocks, and accessing specific blocks:

package main

import (
    "context"
    "fmt"
    "log"
    "log/slog"
    "os"

    "github.com/ehsanranjbar/cometdump"
)

func main() {
    // Open or create a store directory
    store, err := cometdump.Open("./blockchain-data")
    if err != nil {
        log.Fatal(err)
    }

    // Create a logger
    logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
        Level: slog.LevelInfo,
    }))
    // Configure sync options and sync blocks from CometBFT nodes
    config := cometdump.DefaultSyncConfig("https://cosmos-rpc.publicnode.com:443/").
        WithExpandRemotes(true).                    // Discover additional nodes
        WithUseLatestVersion(true).                 // Use nodes with latest version
        WithHeight(1000).                           // Sync up to block 1,000
        WithLogger(logger.With("module", "sync"))

    ctx := context.Background()
    err = store.Sync(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("Sync completed!")

    // Iterate through all stored blocks using the Blocks method
    fmt.Println("\nIterating through blocks:")
    for block, err := range store.Blocks() {
        if err != nil {
            log.Printf("Error reading block: %v", err)
            continue
        }

        fmt.Printf("Block Height: %d, Hash: %s, Txs: %d\n",
            block.Block.Height,
            block.Block.Hash(),
            len(block.Block.Data.Txs))

        // Process first 10 blocks only for this example
        if block.Block.Height >= 10 {
            break
        }
    }
}
RPC Server Example

CometDump includes an RPC server that provides CometBFT-compatible JSON-RPC endpoints for accessing stored blockchain data:

package main

import (
    "log"
    "net/http"
    "os"

    cometlog "github.com/cometbft/cometbft/v2/libs/log"
    rpcserver "github.com/cometbft/cometbft/v2/rpc/jsonrpc/server"
    "github.com/ehsanranjbar/cometdump"
)

func main() {
    // Open the store
    store, err := cometdump.Open("./blockchain-data")
    if err != nil {
        log.Fatal(err)
    }

    // Create RPC server and register routes
    mux := http.NewServeMux()
    server := cometdump.NewRPCServer(store)
    logger := cometlog.NewLogger(os.Stdout)
    rpcserver.RegisterRPCFuncs(mux, server.GetRoutes(), logger)

    // Start the RPC server
    listener, err := rpcserver.Listen("tcp://localhost:8080", 0)
    if err != nil {
        log.Fatal(err)
    }

    log.Println("Starting RPC server", "address", "http://localhost:8080")

    if err := rpcserver.Serve(listener, mux, logger, rpcserver.DefaultConfig()); err != nil {
        log.Fatal(err)
    }
}

The RPC server provides the following CometBFT-compatible endpoints:

  • GET /block?height=N - Get block at specific height
  • GET /block_results?height=N - Get block results (transaction results, events)
  • GET /blockchain?minHeight=N&maxHeight=M - Get block headers in range
  • GET /header?height=N - Get block header at specific height

Requirements

  • Go 1.23.5 or later
  • Compatible with CometBFT v2.x nodes

License

This project is licensed under the MIT License - see the LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit issues, feature requests, or pull requests.

  • CometBFT - The underlying consensus engine
  • Cosmos SDK - Framework for building blockchain applications

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type BlockRecord

type BlockRecord struct {
	BlockID               cmtypes.BlockID
	Block                 *cmtypes.Block
	TxResults             []*abcitypes.ExecTxResult
	FinalizeBlockEvents   []abcitypes.Event
	ValidatorUpdates      []abcitypes.ValidatorUpdate
	ConsensusParamUpdates *cmtproto.ConsensusParams
	BlockResultsAppHash   []byte
}

BlockRecord is a structure that holds the details of a block, including its ID, transactions, events, and other related data. It is used to encode and decode block data in a msgpack format.

func BlockRecordFromRPCResults

func BlockRecordFromRPCResults(block *coretypes.ResultBlock, blockResults *coretypes.ResultBlockResults) *BlockRecord

BlockRecordFromRPCResults creates a BlockRecord from the given ResultBlock and ResultBlockResults.

func (*BlockRecord) DecodeMsgpack

func (b *BlockRecord) DecodeMsgpack(dec *msgpack.Decoder) error

DecodeMsgpack decodes the Block from a msgpack format.

func (*BlockRecord) EncodeMsgpack

func (b *BlockRecord) EncodeMsgpack(enc *msgpack.Encoder) error

EncodeMsgpack encodes the Block into a msgpack format.

func (*BlockRecord) ToResultBlock

func (b *BlockRecord) ToResultBlock() *coretypes.ResultBlock

ToResultBlock converts the resultBlockProto back to a ResultBlock.

func (*BlockRecord) ToResultBlockResults

func (b *BlockRecord) ToResultBlockResults() *coretypes.ResultBlockResults

ToResultBlockResults converts the Block to a ResultBlockResults.

type NoNodesAvailableError

type NoNodesAvailableError struct {
	// contains filtered or unexported fields
}

NoNodesAvailableError is returned when no nodes are available for the given height range.

func (*NoNodesAvailableError) Error

func (e *NoNodesAvailableError) Error() string

func (*NoNodesAvailableError) Range

func (e *NoNodesAvailableError) Range() (int64, int64)

Range returns the height range for which no nodes are available.

type RPCServer

type RPCServer struct {
	// contains filtered or unexported fields
}

RPCServer is a struct that provides methods to handle RPC requests for the cometdump store. It contains a reference to the Store, which is used to access blockchain data.

func NewRPCServer

func NewRPCServer(store *Store) *RPCServer

NewRPCServer creates a new RPC server for the cometdump store.

func (*RPCServer) Block

func (server *RPCServer) Block(_ *rpctypes.Context, heightPtr *int64) (*coretypes.ResultBlock, error)

Block gets block at a given height. If no height is provided, it will fetch the latest block. More: https://docs.cometbft.com/main/rpc/#/Info/block

func (*RPCServer) BlockResults

func (server *RPCServer) BlockResults(_ *rpctypes.Context, heightPtr *int64) (*coretypes.ResultBlockResults, error)

BlockResults gets ABCIResults at a given height. If no height is provided, it will fetch results for the latest block.

Results are for the height of the block containing the txs. Thus response.results.deliver_tx[5] is the results of executing getBlock(h).Txs[5] More: https://docs.cometbft.com/main/rpc/#/Info/block_results

func (*RPCServer) BlockchainInfo

func (server *RPCServer) BlockchainInfo(_ *rpctypes.Context, minHeight, maxHeight int64) (*coretypes.ResultBlockchainInfo, error)

BlockchainInfo gets block headers for minHeight <= height <= maxHeight. More: https://docs.cometbft.com/main/rpc/#/Info/blockchain

func (*RPCServer) GetRoutes

func (server *RPCServer) GetRoutes() rpccore.RoutesMap

GetRoutes returns a map of RPC function names to their corresponding RPC functions. Each function is registered with its expected parameters and caching behavior. The functions handle requests for block data, block results, blockchain information, and headers.

func (*RPCServer) Header

func (server *RPCServer) Header(_ *rpctypes.Context, heightPtr *int64) (*coretypes.ResultHeader, error)

Header gets block header at a given height. If no height is provided, it will fetch the latest header. More: https://docs.cometbft.com/main/rpc/#/Info/header

type Store

type Store struct {
	// contains filtered or unexported fields
}

Store represents a directory where blocks are stored as chunks of data.

func Open

func Open(path string) (*Store, error)

Open initializes a new Store at the specified path or returns an existing one. If the directory does not exist, it will be created. It also acquires a lock on the directory to prevent concurrent access.

func (*Store) BlockAt

func (s *Store) BlockAt(height int64) (*BlockRecord, error)

BlockAt retrieves a block record by its height from the store.

func (*Store) Blocks

func (s *Store) Blocks() iter.Seq2[*BlockRecord, error]

Blocks returns an iterator which iterates over all BlockRecords in the store with order

func (*Store) BlocksInRange

func (s *Store) BlocksInRange(startHeight, endHeight int64) ([]*BlockRecord, error)

BlocksInRange returns a list of BlockRecords in the specified height range.

func (*Store) Sync

func (s *Store) Sync(ctx context.Context, conf SyncConfig) error

Sync fetches blocks up until the latest block height (or a specific height if provided) and stores them in the store directory.

type SyncConfig

type SyncConfig struct {
	// Remotes is a list of node RPC endpoints to connect to.
	Remotes []string
	// ExpandRemotes indicates whether to expand the remotes by querying the chain network info.
	ExpandRemotes bool
	// VersionConstraint is a semantic version constraint for app versions of the nodes.
	VersionConstraint *semver.Constraints
	// UseLatestVersion indicates whether to use the nodes with the latest application version.
	UseLatestVersion bool
	// ChunkSize is the number of blocks to put in each file/chunk.
	ChunkSize int
	// Height is the height up to which store should be synced.
	// If 0, it will fetch up to the latest block height.
	Height int64
	// FetchSize is the number of blocks to fetch in each RPC call.
	FetchSize int
	// NumWorkers is the number of concurrent workers to fetch blocks.
	NumWorkers int
	// OutputChan is a channel that can be optionally used to receive the BlockRecords as they are stored.
	OutputChan chan *BlockRecord
	// Logger is the logger to use for logging during the sync process.
	Logger *slog.Logger
}

SyncConfig defines config for the Sync method.

func DefaultSyncConfig

func DefaultSyncConfig(remotes ...string) SyncConfig

DefaultSyncConfig provides default options for the Sync method.

func (SyncConfig) WithExpandRemotes

func (c SyncConfig) WithExpandRemotes(expand bool) SyncConfig

WithExpandRemotes sets whether to expand the remotes by querying the chain network info.

func (SyncConfig) WithHeight

func (c SyncConfig) WithHeight(height int64) SyncConfig

WithHeight sets the height up to which blocks should be fetched.

func (SyncConfig) WithLogger

func (conf SyncConfig) WithLogger(logger *slog.Logger) SyncConfig

WithLogger sets the logger for the sync operation.

func (SyncConfig) WithUseLatestVersion

func (c SyncConfig) WithUseLatestVersion(useLatest bool) SyncConfig

WithUseLatestVersion indicates whether to use the latest version of the remote.

func (SyncConfig) WithVersionConstraint

func (c SyncConfig) WithVersionConstraint(constraint string) SyncConfig

WithVersionConstraint sets a version constraint for the remotes.

Directories

Path Synopsis
internal

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL