sealing

Documentation

Index

• Constants
• type Core
  • func NewCore(log zerolog.Logger, coreMetrics module.EngineMetrics, tracer module.Tracer, ...) (*Core, error)
  • func (c *Core) CheckSealing() error
  • func (c *Core) OnApproval(originID flow.Identifier, approval *flow.ResultApproval) error
  • func (c *Core) OnReceipt(originID flow.Identifier, receipt *flow.ExecutionReceipt) error
• type Engine
  • func NewEngine(log zerolog.Logger, engineMetrics module.EngineMetrics, tracer module.Tracer, ...) (*Engine, error)
  • func (e *Engine) Done() <-chan struct{}
  • func (e *Engine) HandleReceipt(originID flow.Identifier, receipt flow.Entity)
  • func (e *Engine) Process(originID flow.Identifier, event interface{}) error
  • func (e *Engine) ProcessLocal(event interface{}) error
  • func (e *Engine) Ready() <-chan struct{}
  • func (e *Engine) Submit(originID flow.Identifier, event interface{})
  • func (e *Engine) SubmitLocal(event interface{})
• type Event
• type EventSink
• type RequestTracker
  • func NewRequestTracker(blackoutPeriodMin, blackoutPeriodMax int) *RequestTracker
  • func (rt *RequestTracker) Get(resultID flow.Identifier, chunkIndex uint64) *RequestTrackerItem
  • func (rt *RequestTracker) GetAll() map[flow.Identifier]map[uint64]*RequestTrackerItem
  • func (rt *RequestTracker) Remove(resultID flow.Identifier)
  • func (rt *RequestTracker) Set(resultID flow.Identifier, chunkIndex uint64, item *RequestTrackerItem)
• type RequestTrackerItem
  • func NewRequestTrackerItem(blackoutPeriodMin, blackoutPeriodMax int) *RequestTrackerItem
  • func (i *RequestTrackerItem) IsBlackout() bool
  • func (i *RequestTrackerItem) Update()

Constants

const DefaultEmergencySealingActive = false

DefaultEmergencySealingActive is a flag which indicates when emergency sealing is active, this is a temporary measure to make fire fighting easier while seal & verification is under development.

const DefaultEmergencySealingThreshold = 400

DefaultEmergencySealingThreshold is the default number of blocks which indicates that ER should be sealed using emergency sealing.

const DefaultRequiredApprovalsForSealConstruction = 0

DefaultRequiredApprovalsForSealConstruction is the default number of approvals required to construct a candidate seal for subsequent inclusion in block.

Types

type Core

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

Core implements the core algorithms of the sealing protocol, i.e. determining, which Execution Result has accumulated sufficient approvals for it to be sealable. Specifically:

• Core tracks which execution Results (from ExecutionReceipts) were incorporated in the blocks.
• It processes the ResultApprovals and matches them to execution results.
• When an incorporated Result has collected sufficient approvals, a candidate Seal is generated and stored in the IncorporatedResultSeals mempool. Spwecifically, we require that each chunk must have a minimal number of approvals, `requiredApprovalsForSealConstruction`, from assigned Verifiers.

NOTE: Core is designed to be non-thread safe and cannot be used in concurrent environment user of this object needs to ensure single thread access.

func NewCore

func NewCore(
	log zerolog.Logger,
	coreMetrics module.EngineMetrics,
	tracer module.Tracer,
	mempool module.MempoolMetrics,
	conMetrics module.ConsensusMetrics,
	state protocol.State,
	me module.Local,
	receiptRequester module.Requester,
	receiptsDB storage.ExecutionReceipts,
	headersDB storage.Headers,
	indexDB storage.Index,
	incorporatedResults mempool.IncorporatedResults,
	receipts mempool.ExecutionTree,
	approvals mempool.Approvals,
	seals mempool.IncorporatedResultSeals,
	pendingReceipts mempool.PendingReceipts,
	assigner module.ChunkAssigner,
	receiptValidator module.ReceiptValidator,
	approvalValidator module.ApprovalValidator,
	requiredApprovalsForSealConstruction uint,
	emergencySealingActive bool,
	approvalConduit network.Conduit,
) (*Core, error)

func (*Core) CheckSealing

func (c *Core) CheckSealing() error

CheckSealing checks if there is anything worth sealing at the moment.

func (*Core) OnApproval

func (c *Core) OnApproval(originID flow.Identifier, approval *flow.ResultApproval) error

OnApproval processes a new result approval.

func (*Core) OnReceipt

func (c *Core) OnReceipt(originID flow.Identifier, receipt *flow.ExecutionReceipt) error

OnReceipt processes a new execution receipt. Any error indicates an unexpected problem in the protocol logic. The node's internal state might be corrupted. Hence, returned errors should be treated as fatal.

type Engine

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

Engine is a wrapper for sealing `Core` which implements logic for queuing and filtering network messages which later will be processed by sealing engine. Purpose of this struct is to provide an efficient way how to consume messages from network layer and pass them to `Core`. Engine runs 2 separate gorourtines that perform pre-processing and consuming messages by Core.

func NewEngine

func NewEngine(log zerolog.Logger,
	engineMetrics module.EngineMetrics,
	tracer module.Tracer,
	mempool module.MempoolMetrics,
	conMetrics module.ConsensusMetrics,
	net module.Network,
	state protocol.State,
	me module.Local,
	receiptRequester module.Requester,
	receiptsDB storage.ExecutionReceipts,
	headersDB storage.Headers,
	indexDB storage.Index,
	incorporatedResults mempool.IncorporatedResults,
	receipts mempool.ExecutionTree,
	approvals mempool.Approvals,
	seals mempool.IncorporatedResultSeals,
	pendingReceipts mempool.PendingReceipts,
	assigner module.ChunkAssigner,
	receiptValidator module.ReceiptValidator,
	approvalValidator module.ApprovalValidator,
	requiredApprovalsForSealConstruction uint,
	emergencySealingActive bool) (*Engine, error)

NewEngine constructs new `EngineEngine` which runs on it's own unit.

func (*Engine) Done

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

func (*Engine) HandleReceipt

func (e *Engine) HandleReceipt(originID flow.Identifier, receipt flow.Entity)

HandleReceipt pipes explicitly requested receipts to the process function. Receipts can come from this function or the receipt provider setup in the engine constructor.

func (*Engine) Process

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

Process sends event into channel with pending events. Generally speaking shouldn't lock for too long.

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 ready channel that is closed once the engine has fully started. For the propagation engine, we consider the engine up and running upon initialization.

func (*Engine) Submit

func (e *Engine) Submit(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 Event

type Event struct {
	OriginID flow.Identifier
	Msg      interface{}
}

type EventSink

type EventSink chan *Event // Channel to push pending events

type RequestTracker

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

RequestTracker is an index of RequestTrackerItems indexed by execution result ID and chunk index. It is not concurrency-safe.

func NewRequestTracker

func NewRequestTracker(blackoutPeriodMin, blackoutPeriodMax int) *RequestTracker

NewRequestTracker instantiates a new RequestTracker with blackout periods between min and max seconds.

func (*RequestTracker) Get

func (rt *RequestTracker) Get(resultID flow.Identifier, chunkIndex uint64) *RequestTrackerItem

Get returns the tracker item for a specific chunk, and creates a new one if it doesn't exist.

func (*RequestTracker) GetAll

func (rt *RequestTracker) GetAll() map[flow.Identifier]map[uint64]*RequestTrackerItem

GetAll returns a map of all the items in the tracker indexed by execution result ID and chunk index.

func (*RequestTracker) Remove

func (rt *RequestTracker) Remove(resultID flow.Identifier)

Remove removes all entries pertaining to an execution result

func (*RequestTracker) Set

func (rt *RequestTracker) Set(resultID flow.Identifier, chunkIndex uint64, item *RequestTrackerItem)

Set inserts or updates the tracker item for a specific chunk.

type RequestTrackerItem

type RequestTrackerItem struct {
	Requests    uint
	NextTimeout time.Time
	// contains filtered or unexported fields
}

RequestTrackerItem is an object that keeps track of how many times a request has been made, as well as the time until a new request can be made. It is not concurrency-safe.

func NewRequestTrackerItem

func NewRequestTrackerItem(blackoutPeriodMin, blackoutPeriodMax int) *RequestTrackerItem

NewRequestTrackerItem instantiates a new RequestTrackerItem where the NextTimeout is evaluated to the current time plus a random blackout period contained between min and max.

func (*RequestTrackerItem) IsBlackout

func (i *RequestTrackerItem) IsBlackout() bool

func (*RequestTrackerItem) Update

func (i *RequestTrackerItem) Update()

Update increments the number of requests and recomputes the NextTimeout.