syncer

README

Syncer

The Syncer package provides support for syncing blocks from any Rosetta Data API implementation. If you want to see an example of how to use this package, take a look at rosetta-cli.

Features

• Automatic handling of block re-orgs
• Multi-threaded block fetching (using the fetcher package)
• Implementable Handler to define your own block processing logic (ex: store processed blocks to a db or print our balance changes)

Installation

go get github.com/coinbase/rosetta-sdk-go/syncer

Future Work

• Sync multiple shards in a sharded blockchain

Documentation

Index

• Constants
• Variables
• func Err(err error) bool
• type Handler
• type Helper
• type Option
  • func WithAdjustmentWindow(adjustmentWindow int64) Option
  • func WithCacheSize(cacheSize int) Option
  • func WithMaxConcurrency(concurrency int64) Option
  • func WithPastBlockLimit(blocks int) Option
  • func WithPastBlocks(blocks []*types.BlockIdentifier) Option
  • func WithSizeMultiplier(sizeMultiplier float64) Option
• type Syncer
  • func New(network *types.NetworkIdentifier, helper Helper, handler Handler, ...) *Syncer
  • func (s *Syncer) Sync(ctx context.Context, startIndex int64, endIndex int64) error
  • func (s *Syncer) Tip() *types.BlockIdentifier

Constants

const (
	// DefaultPastBlockLimit is the maximum number of previously
	// processed block headers we keep in the syncer to handle
	// reorgs correctly. If there is a reorg greater than
	// DefaultPastBlockLimit, it will not be handled correctly.
	DefaultPastBlockLimit = 100

	// DefaultConcurrency is the default number of
	// blocks the syncer will try to get concurrently.
	DefaultConcurrency = int64(4) // nolint:gomnd

	// DefaultCacheSize is the default size of the preprocess
	// cache for the syncer.
	DefaultCacheSize = 2000 << 20 // 2 GB

	// LargeCacheSize will aim to use 5 GB of memory.
	LargeCacheSize = 5000 << 20 // 5 GB

	// SmallCacheSize will aim to use 500 MB of memory.
	SmallCacheSize = 500 << 20 // 500 MB

	// TinyCacheSize will aim to use 200 MB of memory.
	TinyCacheSize = 200 << 20 // 200 MB

	// DefaultMaxConcurrency is the maximum concurrency we will
	// attempt to sync with.
	DefaultMaxConcurrency = int64(256) // nolint:gomnd

	// MinConcurrency is the minimum concurrency we will
	// attempt to sync with.
	MinConcurrency = int64(1) // nolint:gomnd

	// DefaultAdjustmentWindow is how frequently we will
	// consider increasing our concurrency.
	DefaultAdjustmentWindow = 5

	// DefaultSizeMultiplier is used to pad our average size adjustment.
	// This can be used to account for the overhead associated with processing
	// a particular block (i.e. balance adjustments, coins created, etc).
	DefaultSizeMultiplier = float64(10) // nolint:gomnd

)

Variables

var (
	// ErrCannotRemoveGenesisBlock is returned by the syncer when
	// a Rosetta implementation indicates that the
	// genesis block should be orphaned.
	ErrCannotRemoveGenesisBlock = errors.New("cannot remove genesis block")

	// ErrOrphanHead is returned by the Helper when
	// the current head should be orphaned. In some
	// cases, it may not be possible to populate a block
	// if the head of the canonical chain is not yet synced.
	ErrOrphanHead = errors.New("orphan head")

	// ErrBlockResultNil is returned by the syncer
	// when attempting to process a block and the block
	// result is nil.
	ErrBlockResultNil = errors.New("block result is nil")

	// ErrGetCurrentHeadBlockFailed is returned by the syncer when
	// the current head block index is not able to get
	ErrGetCurrentHeadBlockFailed = errors.New("unable to get current head block index")

	// ErrOutOfOrder is returned when the syncer examines
	// a block that is out of order. This typically
	// means the Helper has a bug.
	ErrOutOfOrder = errors.New("block processing is out of order")
)

Named error types for Syncer errors

Functions

func Err

func Err(err error) bool

Err takes an error as an argument and returns whether or not the error is one thrown by the syncer package

Types

type Handler

type Handler interface {
	// BlockSeen is invoked AT LEAST ONCE
	// by the syncer prior to calling BlockAdded
	// with the same arguments. This allows for
	// storing block data before it is sequenced.
	BlockSeen(
		ctx context.Context,
		block *types.Block,
	) error

	BlockAdded(
		ctx context.Context,
		block *types.Block,
	) error

	BlockRemoved(
		ctx context.Context,
		block *types.BlockIdentifier,
	) error
}

Handler is called at various times during the sync cycle to handle different events. It is common to write logs or perform reconciliation in the sync processor.

type Helper

type Helper interface {
	NetworkStatus(
		context.Context,
		*types.NetworkIdentifier,
	) (*types.NetworkStatusResponse, error)

	Block(
		context.Context,
		*types.NetworkIdentifier,
		*types.PartialBlockIdentifier,
	) (*types.Block, error)
}

Helper is called at various times during the sync cycle to get information about a blockchain network. It is common to implement this helper using the Fetcher package.

type Option

type Option func(s *Syncer)

Option is used to overwrite default values in Syncer construction. Any Option not provided falls back to the default value.

func WithAdjustmentWindow

func WithAdjustmentWindow(adjustmentWindow int64) Option

WithAdjustmentWindow overrides the default adjustment window.

func WithCacheSize

func WithCacheSize(cacheSize int) Option

WithCacheSize overrides the default cache size.

func WithMaxConcurrency

func WithMaxConcurrency(concurrency int64) Option

WithMaxConcurrency overrides the default max concurrency.

func WithPastBlockLimit

func WithPastBlockLimit(blocks int) Option

WithPastBlockLimit overrides the default past block limit

func WithPastBlocks

func WithPastBlocks(blocks []*types.BlockIdentifier) Option

WithPastBlocks provides the syncer with a cache of previously processed blocks to handle reorgs.

func WithSizeMultiplier

func WithSizeMultiplier(sizeMultiplier float64) Option

WithSizeMultiplier overrides the default size multiplier.

type Syncer

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

Syncer coordinates blockchain syncing without relying on a storage interface. Instead, it calls a provided Handler whenever a block is added or removed. This provides the client the opportunity to define the logic used to handle each new block. In the rosetta-cli, we handle reconciliation, state storage, and logging in the handler.

func New

func New(
	network *types.NetworkIdentifier,
	helper Helper,
	handler Handler,
	cancel context.CancelFunc,
	options ...Option,
) *Syncer

New creates a new Syncer. If pastBlocks is left nil, it will be set to an empty slice.

func (*Syncer) Sync

func (s *Syncer) Sync(
	ctx context.Context,
	startIndex int64,
	endIndex int64,
) error

Sync cycles endlessly until there is an error or the requested range is synced. When the requested range is synced, context is canceled.

func (*Syncer) Tip

func (s *Syncer) Tip() *types.BlockIdentifier

Tip returns the last observed tip. The tip is recorded at the start of each sync range and should only be thought of as a best effort approximation of tip.

This can be very helpful to callers who want to know an approximation of tip very frequently (~every second) but don't want to implement their own caching logic.