ingestion

Documentation

Index

• type Core
  • func NewCore(log zerolog.Logger, tracer module.Tracer, mempool module.MempoolMetrics, ...) *Core
  • func (e *Core) OnGuarantee(originID flow.Identifier, guarantee *flow.CollectionGuarantee) error
• type Engine
  • func New(log zerolog.Logger, engineMetrics module.EngineMetrics, net network.Network, ...) (*Engine, error)
  • func (e *Engine) Process(channel channels.Channel, originID flow.Identifier, event interface{}) error
  • func (e *Engine) ProcessLocal(event interface{}) error
  • func (e *Engine) Submit(channel channels.Channel, originID flow.Identifier, event interface{})
  • func (e *Engine) SubmitLocal(event interface{})

Types

type Core

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

Core represents core logic of the ingestion engine. It contains logic for handling single collection which are channeled from engine in concurrent way.

func NewCore

func NewCore(
	log zerolog.Logger,
	tracer module.Tracer,
	mempool module.MempoolMetrics,
	state protocol.State,
	headers storage.Headers,
	pool mempool.Guarantees,
) *Core

func (*Core) OnGuarantee

func (e *Core) OnGuarantee(originID flow.Identifier, guarantee *flow.CollectionGuarantee) error

OnGuarantee is used to process collection guarantees received from nodes that are not consensus nodes (notably collection nodes). Returns expected errors:

• engine.InvalidInputError if the collection violates protocol rules
• engine.UnverifiableInputError if the reference block of the collection is unknown
• engine.OutdatedInputError if the collection is already expired

All other errors are unexpected and potential symptoms of internal state corruption.

type Engine

type Engine struct {
	component.Component
	// contains filtered or unexported fields
}

Engine represents the ingestion engine, used to funnel collections from a cluster of collection nodes to the set of consensus nodes. It represents the link between collection nodes and consensus nodes and has a counterpart with the same engine ID in the collection node.

func New

func New(
	log zerolog.Logger,
	engineMetrics module.EngineMetrics,
	net network.Network,
	me module.Local,
	core *Core,
) (*Engine, error)

New creates a new collection propagation engine.

func (*Engine) Process

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

Process processes the given event from the node with the given origin ID in a blocking manner. It returns error only in unexpected scenario.

func (*Engine) ProcessLocal

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

ProcessLocal processes an event originating on the local node.

func (*Engine) Submit

func (e *Engine) Submit(channel channels.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.