engine

Documentation

Index

• Constants
• func ChannelConsensusCluster(clusterID flow.ChainID) string
• func ChannelIDs() []string
• func ChannelIDsByRole(role flow.Role) []string
• func ChannelSyncCluster(clusterID flow.ChainID) string
• func FullyQualifiedChannelName(channelID string, rootBlockID string) string
• func IsClusterChannelID(channelID string) (string, bool)
• func IsInvalidInputError(err error) bool
• func IsOutdatedInputError(err error) bool
• func LogError(log zerolog.Logger, err error)
• func NewInvalidInputError(msg string) error
• func NewInvalidInputErrorf(msg string, args ...interface{}) error
• func NewOutdatedInputErrorf(msg string, args ...interface{}) error
• func RolesByChannelID(channelID string) (flow.RoleList, bool)
• type InvalidInputError
  • func (e InvalidInputError) Error() string
  • func (e InvalidInputError) Unwrap() error
• type OutdatedInputError
  • func (e OutdatedInputError) Error() string
  • func (e OutdatedInputError) Unwrap() error
• type Unit
  • func NewUnit() *Unit
  • func (u *Unit) Ctx() context.Context
  • func (u *Unit) Do(f func() error) error
  • func (u *Unit) Done(actions ...func()) <-chan struct{}
  • func (u *Unit) Launch(f func())
  • func (u *Unit) LaunchAfter(delay time.Duration, f func())
  • func (u *Unit) LaunchPeriodically(f func(), interval time.Duration, delay time.Duration)
  • func (u *Unit) Quit() <-chan struct{}
  • func (u *Unit) Ready(checks ...func()) <-chan struct{}

Constants

const (

	// Channels used for testing
	TestNetwork = "test-network"
	TestMetrics = "test-metrics"

	// Channels for consensus protocols
	ConsensusCommittee = "consensus-committee"

	// Channels for protocols actively synchronizing state across nodes
	SyncCommittee = "sync-committee"

	SyncExecution = "sync-execution"

	// Channels for actively pushing entities to subscribers
	PushTransactions = "push-transactions"
	PushGuarantees   = "push-guarantees"
	PushBlocks       = "push-blocks"
	PushReceipts     = "push-receipts"
	PushApprovals    = "push-approvals"

	// Channels for actively requesting missing entities
	RequestCollections       = "request-collections"
	RequestChunks            = "request-chunks"
	RequestReceiptsByBlockID = "request-receipts-by-block-id"

	// Channel aliases to make the code more readable / more robust to errors
	ReceiveTransactions = PushTransactions
	ReceiveGuarantees   = PushGuarantees
	ReceiveBlocks       = PushBlocks
	ReceiveReceipts     = PushReceipts
	ReceiveApprovals    = PushApprovals

	ProvideCollections       = RequestCollections
	ProvideChunks            = RequestChunks
	ProvideReceiptsByBlockID = RequestReceiptsByBlockID
)

channel IDs

Functions

func ChannelConsensusCluster

func ChannelConsensusCluster(clusterID flow.ChainID) string

ChannelConsensusCluster returns a dynamic cluster consensus channel based on the chain ID of the cluster in question.

func ChannelIDs

func ChannelIDs() []string

ChannelIDs returns all channelIDs nodes of any role have subscribed to.

func ChannelIDsByRole

func ChannelIDsByRole(role flow.Role) []string

ChannelIDsByRole returns a list of all channel IDs the role subscribes to.

func ChannelSyncCluster

func ChannelSyncCluster(clusterID flow.ChainID) string

ChannelSyncCluster returns a dynamic cluster sync channel based on the chain ID of the cluster in question.

func FullyQualifiedChannelName

func FullyQualifiedChannelName(channelID string, rootBlockID string) string

FullyQualifiedChannelName returns the unique channel name made up of channel name string suffixed with root block id The root block id is used to prevent cross talks between nodes on different sporks

func IsClusterChannelID

func IsClusterChannelID(channelID string) (string, bool)

IsClusterChannelID returns true if channel ID is a cluster-related channel ID. At the current implementation, only collection nodes are involved in a cluster-related channel ID. If the channel ID is a cluster-related one, this method also strips off the channel prefix and returns it.

func IsInvalidInputError

func IsInvalidInputError(err error) bool

IsInvalidInputError returns whether the given error is an InvalidInputError error

func IsOutdatedInputError

func IsOutdatedInputError(err error) bool

func LogError

func LogError(log zerolog.Logger, err error)

LogError logs the engine processing error

func NewInvalidInputError

func NewInvalidInputError(msg string) error

func NewInvalidInputErrorf

func NewInvalidInputErrorf(msg string, args ...interface{}) error

func NewOutdatedInputErrorf

func NewOutdatedInputErrorf(msg string, args ...interface{}) error

func RolesByChannelID

func RolesByChannelID(channelID string) (flow.RoleList, bool)

RolesByChannelID returns list of flow roles involved in the channelID.

Types

type InvalidInputError

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

InvalidInputError are errors for caused by invalid inputs. It's useful to distinguish these known errors from exceptions. By distinguishing errors from exceptions, we can log them differently. For instance, log InvalidInputError error as a warn log, and log other error as an error log.

func (InvalidInputError) Error

func (e InvalidInputError) Error() string

func (InvalidInputError) Unwrap

func (e InvalidInputError) Unwrap() error

type OutdatedInputError

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

OutdatedInputError are for inputs that are outdated. An outdated input doesn't mean whether the input was invalid or not, knowing that would take more computation that isn't necessary. An outdated input could also for a duplicated input: the duplication is outdated.

func (OutdatedInputError) Error

func (e OutdatedInputError) Error() string

func (OutdatedInputError) Unwrap

func (e OutdatedInputError) Unwrap() error

type Unit

type Unit struct {
	sync.Mutex // can be used to synchronize the engine
	// contains filtered or unexported fields
}

Unit handles synchronization management, startup, and shutdown for engines.

func NewUnit

func NewUnit() *Unit

NewUnit returns a new unit.

func (*Unit) Ctx

func (u *Unit) Ctx() context.Context

Ctx returns a context with the same lifecycle scope as the unit. In particular, it is cancelled when Done is called, so it can be used as the parent context for processes spawned by any engine whose lifecycle is managed by a unit.

func (*Unit) Do

func (u *Unit) Do(f func() error) error

Do synchronously executes the input function f unless the unit has shut down. It returns the result of f. If f is executed, the unit will not shut down until after f returns.

func (*Unit) Done

func (u *Unit) Done(actions ...func()) <-chan struct{}

Done returns a channel that is closed when the unit is done. A unit is done when (i) the series of "action" functions are executed and (ii) all pending functions invoked with `Do` or `Launch` have completed.

The engine using the unit is responsible for defining these action functions as required.

func (*Unit) Launch

func (u *Unit) Launch(f func())

Launch asynchronously executes the input function unless the unit has shut down. If f is executed, the unit will not shut down until after f returns.

func (*Unit) LaunchAfter

func (u *Unit) LaunchAfter(delay time.Duration, f func())

LaunchAfter asynchronously executes the input function after a certain delay unless the unit has shut down.

func (*Unit) LaunchPeriodically

func (u *Unit) LaunchPeriodically(f func(), interval time.Duration, delay time.Duration)

LaunchPeriodically asynchronously executes the input function on `interval` periods unless the unit has shut down. If f is executed, the unit will not shut down until after f returns.

func (*Unit) Quit

func (u *Unit) Quit() <-chan struct{}

Quit returns a channel that is closed when the unit begins to shut down.

func (*Unit) Ready

func (u *Unit) Ready(checks ...func()) <-chan struct{}

Ready returns a channel that is closed when the unit is ready. A unit is ready when the series of "check" functions are executed.

The engine using the unit is responsible for defining these check functions as required.