ingestion

Documentation

Index

• type Deltas
  • func NewDeltas(limit uint, opts ...stdmap.OptionFunc) (*Deltas, error)
  • func (s *Deltas) Add(delta *messages.ExecutionStateDelta) bool
  • func (s *Deltas) All() []*messages.ExecutionStateDelta
  • func (s *Deltas) ByBlockID(blockID flow.Identifier) (*messages.ExecutionStateDelta, bool)
  • func (s *Deltas) Rem(blockID flow.Identifier) bool
• type Engine
  • func New(logger zerolog.Logger, net network.Network, me module.Local, ...) (*Engine, error)
  • func (e *Engine) BlockProcessable(b *flow.Header)
  • func (e *Engine) Done() <-chan struct{}
  • func (e *Engine) ExecuteScriptAtBlockID(ctx context.Context, script []byte, arguments [][]byte, ...) ([]byte, error)
  • func (e *Engine) GetAccount(ctx context.Context, addr flow.Address, blockID flow.Identifier) (*flow.Account, error)
  • func (e *Engine) GetRegisterAtBlockID(ctx context.Context, owner, controller, key []byte, blockID flow.Identifier) ([]byte, error)
  • func (e *Engine) OnCollection(originID flow.Identifier, entity flow.Entity)
  • func (e *Engine) Process(channel network.Channel, originID flow.Identifier, event interface{}) error
  • func (e *Engine) ProcessLocal(event interface{}) error
  • func (e *Engine) Ready() <-chan struct{}
  • func (e *Engine) Submit(channel network.Channel, originID flow.Identifier, event interface{})
  • func (e *Engine) SubmitLocal(event interface{})
• type IngestRPC
• type Mempool
  • func (m *Mempool) Run(f func(blockByCollection *stdmap.BlockByCollectionBackdata, ...) error) error

Types

type Deltas

type Deltas struct {
	*stdmap.Backend
}

func NewDeltas

func NewDeltas(limit uint, opts ...stdmap.OptionFunc) (*Deltas, error)

NewDeltas creates a new memory pool for state deltas

func (*Deltas) Add

func (s *Deltas) Add(delta *messages.ExecutionStateDelta) bool

Add adds an state deltas to the mempool.

func (*Deltas) All

func (s *Deltas) All() []*messages.ExecutionStateDelta

All returns all block Deltass from the pool.

func (*Deltas) ByBlockID

func (s *Deltas) ByBlockID(blockID flow.Identifier) (*messages.ExecutionStateDelta, bool)

ByBlockID returns the state deltas for a block from the mempool.

func (*Deltas) Rem

func (s *Deltas) Rem(blockID flow.Identifier) bool

Rem will remove a deltas by block ID.

type Engine

type Engine struct {
	psEvents.Noop // satisfy protocol events consumer interface
	// contains filtered or unexported fields
}

An Engine receives and saves incoming blocks.

func New

func New(
	logger zerolog.Logger,
	net network.Network,
	me module.Local,
	request module.Requester,
	state protocol.State,
	blocks storage.Blocks,
	collections storage.Collections,
	events storage.Events,
	serviceEvents storage.ServiceEvents,
	transactionResults storage.TransactionResults,
	executionEngine computation.ComputationManager,
	providerEngine provider.ProviderEngine,
	execState state.ExecutionState,
	metrics module.ExecutionMetrics,
	tracer module.Tracer,
	extLog bool,
	syncFilter flow.IdentityFilter,
	syncDeltas mempool.Deltas,
	syncThreshold int,
	syncFast bool,
	checkStakedAtBlock func(blockID flow.Identifier) (bool, error),
	pauseExecution bool,
) (*Engine, error)

func (*Engine) BlockProcessable

func (e *Engine) BlockProcessable(b *flow.Header)

BlockProcessable handles the new verified blocks (blocks that have passed consensus validation) received from the consensus nodes Note: BlockProcessable might be called multiple times for the same block.

func (*Engine) Done

func (e *Engine) Done() <-chan struct{}

Done returns a channel that will close when the engine has successfully stopped.

func (*Engine) ExecuteScriptAtBlockID

func (e *Engine) ExecuteScriptAtBlockID(ctx context.Context, script []byte, arguments [][]byte, blockID flow.Identifier) ([]byte, error)

func (*Engine) GetAccount

func (e *Engine) GetAccount(ctx context.Context, addr flow.Address, blockID flow.Identifier) (*flow.Account, error)

func (*Engine) GetRegisterAtBlockID

func (e *Engine) GetRegisterAtBlockID(ctx context.Context, owner, controller, key []byte, blockID flow.Identifier) ([]byte, error)

func (*Engine) OnCollection

func (e *Engine) OnCollection(originID flow.Identifier, entity flow.Entity)

OnCollection is a callback for handling the collections requested by the collection requester.

func (*Engine) Process

func (e *Engine) Process(channel network.Channel, originID flow.Identifier, event interface{}) error

func (*Engine) ProcessLocal

func (e *Engine) ProcessLocal(event interface{}) error

ProcessLocal processes an event originating on the local node.

func (*Engine) Ready

func (e *Engine) Ready() <-chan struct{}

Ready returns a channel that will close when the engine has successfully started.

func (*Engine) Submit

func (e *Engine) Submit(channel network.Channel, originID flow.Identifier, event interface{})

Submit submits the given event from the node with the given origin ID for processing in a non-blocking manner. It returns instantly and logs a potential processing error internally when done.

func (*Engine) SubmitLocal

func (e *Engine) SubmitLocal(event interface{})

SubmitLocal submits an event originating on the local node.

type IngestRPC

type IngestRPC interface {

	// ExecuteScriptAtBlockID executes a script at the given Block id
	ExecuteScriptAtBlockID(ctx context.Context, script []byte, arguments [][]byte, blockID flow.Identifier) ([]byte, error)

	// GetAccount returns the Account details at the given Block id
	GetAccount(ctx context.Context, address flow.Address, blockID flow.Identifier) (*flow.Account, error)

	// GetRegisterAtBlockID returns the value of a register at the given Block id (if available)
	GetRegisterAtBlockID(ctx context.Context, owner, controller, key []byte, blockID flow.Identifier) ([]byte, error)
}

IngestRPC represents the RPC calls that the execution ingest engine exposes to support the Access Node API calls

type Mempool

type Mempool struct {
	ExecutionQueue    *stdmap.Queues
	BlockByCollection *stdmap.BlockByCollections
}

func (*Mempool) Run

func (m *Mempool) Run(f func(blockByCollection *stdmap.BlockByCollectionBackdata, executionQueue *stdmap.QueuesBackdata) error) error