orchestrator

Documentation

Index

• Constants
• Variables
• type BlockNumberWithError
• type ChainTracker
  • func NewChainTracker(rpc rpc.IRPCClient) *ChainTracker
  • func (ct *ChainTracker) Start(ctx context.Context)
• type Committer
  • func NewCommitter(rpc rpc.IRPCClient, storage storage.IStorage, poller *Poller, ...) *Committer
  • func (c *Committer) Start(ctx context.Context)
• type CommitterOption
  • func WithValidator(validator *Validator) CommitterOption
• type Orchestrator
  • func NewOrchestrator(rpc rpc.IRPCClient) (*Orchestrator, error)
  • func (o *Orchestrator) Start()
• type Poller
  • func NewPoller(rpc rpc.IRPCClient, storage storage.IStorage, opts ...PollerOption) *Poller
  • func (p *Poller) Request(ctx context.Context, blockNumbers []*big.Int) []common.BlockData
  • func (p *Poller) Start(ctx context.Context)
• type PollerOption
  • func WithPollerWorker(cfg *worker.Worker) PollerOption
• type ReorgHandler
  • func NewReorgHandler(rpc rpc.IRPCClient, storage storage.IStorage) *ReorgHandler
  • func (rh *ReorgHandler) RunFromBlock(ctx context.Context, latestCheckedBlock *big.Int) (lastCheckedBlock *big.Int, err error)
  • func (rh *ReorgHandler) Start(ctx context.Context)
• type Validator
  • func NewValidator(rpcClient rpc.IRPCClient, s storage.IStorage, w *worker.Worker) *Validator
  • func (v *Validator) FindAndFixGaps(startBlock *big.Int, endBlock *big.Int) error
  • func (v *Validator) FixBlocks(invalidBlocks []*big.Int, fixBatchSize int) error
  • func (v *Validator) ValidateBlock(blockData common.BlockData) (valid bool, err error)
  • func (v *Validator) ValidateBlockRange(startBlock *big.Int, endBlock *big.Int) (validBlocks []common.BlockData, invalidBlocks []common.BlockData, err error)
  • func (v *Validator) ValidateBlocks(blocks []common.BlockData) (validBlocks []common.BlockData, invalidBlocks []common.BlockData, err error)
• type WorkMode
• type WorkModeMonitor
  • func NewWorkModeMonitor(rpc rpc.IRPCClient, storage storage.IStorage) *WorkModeMonitor
  • func (m *WorkModeMonitor) RegisterChannel(ch chan WorkMode)
  • func (m *WorkModeMonitor) Start(ctx context.Context)
  • func (m *WorkModeMonitor) UnregisterChannel(ch chan WorkMode)

Constants

const (
	DEFAULT_PARALLEL_POLLERS  = 5
	DEFAULT_LOOKAHEAD_BATCHES = 5
)

const (
	DEFAULT_WORK_MODE_CHECK_INTERVAL          = 10
	DEFAULT_LIVE_MODE_THRESHOLD               = 500
	WorkModeLive                     WorkMode = "live"
	WorkModeBackfill                 WorkMode = "backfill"
)

const DEFAULT_BLOCKS_PER_COMMIT = 1000

const DEFAULT_CHAIN_TRACKER_POLL_INTERVAL = 60 * 1000 // 1 minutes

const DEFAULT_COMMITTER_TRIGGER_INTERVAL = 2000

const DEFAULT_REORG_HANDLER_BLOCKS_PER_SCAN = 100

const DEFAULT_REORG_HANDLER_INTERVAL = 1000

Variables

var ErrBlocksProcessed = fmt.Errorf("blocks are being processed")

var ErrNoNewBlocks = fmt.Errorf("no new blocks to poll")

Types

type BlockNumberWithError

type BlockNumberWithError struct {
	BlockNumber *big.Int
	Error       error
}

type ChainTracker

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

func NewChainTracker

func NewChainTracker(rpc rpc.IRPCClient) *ChainTracker

func (*ChainTracker) Start

func (ct *ChainTracker) Start(ctx context.Context)

type Committer

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

func NewCommitter

func NewCommitter(rpc rpc.IRPCClient, storage storage.IStorage, poller *Poller, opts ...CommitterOption) *Committer

func (*Committer) Start

func (c *Committer) Start(ctx context.Context)

type CommitterOption

type CommitterOption func(*Committer)

func WithValidator

func WithValidator(validator *Validator) CommitterOption

type Orchestrator

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

func NewOrchestrator

func NewOrchestrator(rpc rpc.IRPCClient) (*Orchestrator, error)

func (*Orchestrator) Start

func (o *Orchestrator) Start()

type Poller

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

func NewPoller

func NewPoller(rpc rpc.IRPCClient, storage storage.IStorage, opts ...PollerOption) *Poller

func (*Poller) Request

func (p *Poller) Request(ctx context.Context, blockNumbers []*big.Int) []common.BlockData

func (*Poller) Start

func (p *Poller) Start(ctx context.Context)

type PollerOption

type PollerOption func(*Poller)

func WithPollerWorker

func WithPollerWorker(cfg *worker.Worker) PollerOption

type ReorgHandler

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

func NewReorgHandler

func NewReorgHandler(rpc rpc.IRPCClient, storage storage.IStorage) *ReorgHandler

func (*ReorgHandler) RunFromBlock

func (rh *ReorgHandler) RunFromBlock(ctx context.Context, latestCheckedBlock *big.Int) (lastCheckedBlock *big.Int, err error)

func (*ReorgHandler) Start

func (rh *ReorgHandler) Start(ctx context.Context)

type Validator

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

func NewValidator

func NewValidator(rpcClient rpc.IRPCClient, s storage.IStorage, w *worker.Worker) *Validator

func (*Validator) FindAndFixGaps

func (v *Validator) FindAndFixGaps(startBlock *big.Int, endBlock *big.Int) error

func (*Validator) FixBlocks

func (v *Validator) FixBlocks(invalidBlocks []*big.Int, fixBatchSize int) error

func (*Validator) ValidateBlock

func (v *Validator) ValidateBlock(blockData common.BlockData) (valid bool, err error)

func (*Validator) ValidateBlockRange

func (v *Validator) ValidateBlockRange(startBlock *big.Int, endBlock *big.Int) (validBlocks []common.BlockData, invalidBlocks []common.BlockData, err error)

*

• Validate blocks in the range of startBlock to endBlock
• @param startBlock - The start block number (inclusive)
• @param endBlock - The end block number (inclusive)
• @return error - An error if the validation fails

func (*Validator) ValidateBlocks

func (v *Validator) ValidateBlocks(blocks []common.BlockData) (validBlocks []common.BlockData, invalidBlocks []common.BlockData, err error)

type WorkMode

type WorkMode string

type WorkModeMonitor

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

func NewWorkModeMonitor

func NewWorkModeMonitor(rpc rpc.IRPCClient, storage storage.IStorage) *WorkModeMonitor

func (*WorkModeMonitor) RegisterChannel

func (m *WorkModeMonitor) RegisterChannel(ch chan WorkMode)

RegisterChannel adds a new channel to receive work mode updates

func (*WorkModeMonitor) Start

func (m *WorkModeMonitor) Start(ctx context.Context)

func (*WorkModeMonitor) UnregisterChannel

func (m *WorkModeMonitor) UnregisterChannel(ch chan WorkMode)

UnregisterChannel removes a channel from receiving work mode updates