das

Documentation

Overview

Package das contains the most important functionality provided by celestia-node. It contains logic for running data availability sampling (DAS) routines on block headers in the network. DAS is the process of verifying the availability of block data by sampling chunks or shares of those blocks.

Package das can confirm the availability of block data in the network via the Availability interface which is implemented both in `full` and `light` mode. `Full` availability ensures the full reparation of a block's data square (meaning the instance will sample for enough shares to be able to fully repair the block's data square) while `light` availability samples for shares randomly until it is sufficiently likely that all block data is available as it is assumed that there are enough `light` availability instances active on the network doing sampling over the same block to collectively verify its availability.

The central component of this package is the `samplingCoordinator`. It launches parallel workers that perform DAS on new ExtendedHeaders in the network. The DASer kicks off this loop by loading its last DASed headers snapshot (`checkpoint`) and kicking off worker pool to DAS all headers between the checkpoint and the current network head. It subscribes to notifications about to new ExtendedHeaders, received via gossipsub. Newly found headers are being put into higher priority queue and will be sampled by the next available worker.

Index

• Variables
• type DASer
  • func NewDASer(da share.Availability, hsub libhead.Subscriber[*header.ExtendedHeader], ...) (*DASer, error)
  • func (d *DASer) InitMetrics() error
  • func (d *DASer) SamplingStats(ctx context.Context) (SamplingStats, error)
  • func (d *DASer) Start(ctx context.Context) error
  • func (d *DASer) Stop(ctx context.Context) error
  • func (d *DASer) WaitCatchUp(ctx context.Context) error
• type Option
  • func WithBackgroundStoreInterval(backgroundStoreInterval time.Duration) Option
  • func WithConcurrencyLimit(concurrencyLimit int) Option
  • func WithPriorityQueueSize(priorityQueueSize int) Option
  • func WithSampleFrom(sampleFrom uint64) Option
  • func WithSampleTimeout(sampleTimeout time.Duration) Option
  • func WithSamplingRange(samplingRange uint64) Option
• type Parameters
  • func DefaultParameters() Parameters
  • func (p *Parameters) Validate() error
• type SamplingStats
• type WorkerStats

Variables

var ErrInvalidOption = fmt.Errorf("das: invalid option")

ErrInvalidOption is an error that is returned by Parameters.Validate when supplied with invalid values. This error will also be returned by NewDASer if supplied with an invalid option

Types

type DASer

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

DASer continuously validates availability of data committed to headers.

func NewDASer

func NewDASer(
	da share.Availability,
	hsub libhead.Subscriber[*header.ExtendedHeader],
	getter libhead.Getter[*header.ExtendedHeader],
	dstore datastore.Datastore,
	bcast fraud.Broadcaster,
	options ...Option,
) (*DASer, error)

NewDASer creates a new DASer.

func (*DASer) InitMetrics

func (d *DASer) InitMetrics() error

func (*DASer) SamplingStats

func (d *DASer) SamplingStats(ctx context.Context) (SamplingStats, error)

SamplingStats returns the current statistics over the DA sampling process.

func (*DASer) Start

func (d *DASer) Start(ctx context.Context) error

Start initiates subscription for new ExtendedHeaders and spawns a sampling routine.

func (*DASer) Stop

func (d *DASer) Stop(ctx context.Context) error

Stop stops sampling.

func (*DASer) WaitCatchUp

func (d *DASer) WaitCatchUp(ctx context.Context) error

WaitCatchUp waits for DASer to indicate catchup is done

type Option

type Option func(*DASer)

Option is the functional option that is applied to the daser instance to configure DASing parameters (the Parameters struct)

func WithBackgroundStoreInterval

func WithBackgroundStoreInterval(backgroundStoreInterval time.Duration) Option

WithBackgroundStoreInterval is a functional option to configure the daser's `backgroundStoreInterval` parameter Refer to WithSamplingRange documentation to see an example of how to use this

func WithConcurrencyLimit

func WithConcurrencyLimit(concurrencyLimit int) Option

WithConcurrencyLimit is a functional option to configure the daser's `ConcurrencyLimit` parameter Refer to WithSamplingRange documentation to see an example of how to use this

func WithPriorityQueueSize

func WithPriorityQueueSize(priorityQueueSize int) Option

WithPriorityQueueSize is a functional option to configure the daser's `priorityQueuSize` parameter Refer to WithSamplingRange documentation to see an example of how to use this

func WithSampleFrom

func WithSampleFrom(sampleFrom uint64) Option

WithSampleFrom is a functional option to configure the daser's `SampleFrom` parameter Refer to WithSamplingRange documentation to see an example of how to use this

func WithSampleTimeout

func WithSampleTimeout(sampleTimeout time.Duration) Option

WithSampleFrom is a functional option to configure the daser's `SampleTimeout` parameter Refer to WithSamplingRange documentation to see an example of how to use this

func WithSamplingRange

func WithSamplingRange(samplingRange uint64) Option

WithSamplingRange is a functional option to configure the daser's `SamplingRange` parameter

Usage:
```
	WithSamplingRange(10)(daser)
```

or

```
	option := WithSamplingRange(10)
	// shenanigans to create daser
	option(daser)

```

type Parameters

type Parameters struct {
	//  SamplingRange is the maximum amount of headers processed in one job.
	SamplingRange uint64

	// ConcurrencyLimit defines the maximum amount of sampling workers running in parallel.
	ConcurrencyLimit int

	// BackgroundStoreInterval is the period of time for background checkpointStore to perform a
	// checkpoint backup.
	BackgroundStoreInterval time.Duration

	// PriorityQueueSize defines the size limit of the priority queue
	PriorityQueueSize int

	// SampleFrom is the height sampling will start from if no previous checkpoint was saved
	SampleFrom uint64

	// SampleTimeout is a maximum amount time sampling of single block may take until it will be
	// canceled
	SampleTimeout time.Duration
}

Parameters is the set of parameters that must be configured for the daser

func DefaultParameters

func DefaultParameters() Parameters

DefaultParameters returns the default configuration values for the daser parameters

func (*Parameters) Validate

func (p *Parameters) Validate() error

Validate validates the values in Parameters

All parameters must be positive and non-zero, except:
	BackgroundStoreInterval = 0 disables background storer,
	PriorityQueueSize = 0 disables prioritization of recently produced blocks for sampling

type SamplingStats

type SamplingStats struct {
	// all headers before SampledChainHead were successfully sampled
	SampledChainHead uint64 `json:"head_of_sampled_chain"`
	// all headers before CatchupHead were submitted to sampling workers
	CatchupHead uint64 `json:"head_of_catchup"`
	// NetworkHead is the height of the most recent header in the network
	NetworkHead uint64 `json:"network_head_height"`
	// Failed contains all skipped header's heights with corresponding try count
	Failed map[uint64]int `json:"failed,omitempty"`
	// Workers has information about each currently running worker stats
	Workers []WorkerStats `json:"workers,omitempty"`
	// Concurrency currently running parallel workers
	Concurrency int `json:"concurrency"`
	// CatchUpDone indicates whether all known headers are sampled
	CatchUpDone bool `json:"catch_up_done"`
	// IsRunning tracks whether the DASer service is running
	IsRunning bool `json:"is_running"`
}

SamplingStats collects information about the DASer process. Currently, there are only two sampling routines: the main sampling routine which performs sampling over current network headers, and the `catchUp` routine which performs sampling over past headers from the last sampled checkpoint.

type WorkerStats

type WorkerStats struct {
	Curr uint64 `json:"current"`
	From uint64 `json:"from"`
	To   uint64 `json:"to"`

	ErrMsg string `json:"error,omitempty"`
}