ingest

Documentation

Overview

Package ingest implements an engine for receiving transactions that need to be packaged into a collection.

Index

• type Config
  • func DefaultConfig() Config
• type Engine
  • func New(log zerolog.Logger, net network.Network, state protocol.State, ...) (*Engine, error)
  • func (e *Engine) Done() <-chan struct{}
  • 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) Start(ctx irrecoverable.SignalerContext)
  • func (e *Engine) Submit(channel network.Channel, originID flow.Identifier, event interface{})
  • func (e *Engine) SubmitLocal(event interface{})

Types

type Config

type Config struct {
	// how much buffer time there is between a transaction being ingested by a
	// collection node and being included in a collection and block
	ExpiryBuffer uint
	// the maximum transaction gas limit
	MaxGasLimit uint64
	// whether or not we check that transaction scripts are parse-able
	CheckScriptsParse bool
	// the maximum address index we accept
	MaxAddressIndex uint64
	// how many extra nodes in the responsible cluster we propagate transactions to
	// (we always send to at least one)
	PropagationRedundancy uint
	// the maximum transaction byte size limit
	MaxTransactionByteSize uint64
	// maximum collection byte size, it acts as hard limit max for the tx size.
	MaxCollectionByteSize uint64
	// maximum number of un-processed transaction messages to hold in the queue.
	MaxMessageQueueSize uint
}

Config defines configuration for the transaction ingest engine.

func DefaultConfig

func DefaultConfig() Config

type Engine

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

Engine is the transaction ingestion engine, which ensures that new transactions are delegated to the correct collection cluster, and prepared to be included in a collection.

func New

func New(
	log zerolog.Logger,
	net network.Network,
	state protocol.State,
	engMetrics module.EngineMetrics,
	mempoolMetrics module.MempoolMetrics,
	colMetrics module.CollectionMetrics,
	me module.Local,
	chain flow.Chain,
	pools *epochs.TransactionPools,
	config Config,
) (*Engine, error)

New creates a new collection ingest engine.

func (*Engine) Done

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

Done returns a done channel that is closed once the engine has fully stopped.

func (*Engine) Process

func (e *Engine) Process(channel network.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 the potential processing error when done.

func (*Engine) ProcessLocal

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

ProcessLocal processes an event originating on the local node. For local messages, we skip the message queue and immediately process the transaction, blocking the caller until we are finished. This way, we can provide context to the user about whether their transaction was accepted.

func (*Engine) Ready

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

Ready returns a ready channel that is closed once the engine has fully started.

func (*Engine) Start

func (e *Engine) Start(ctx irrecoverable.SignalerContext)

Start starts the engine by starting the message processing loop.

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.