complete

Documentation

Index

• Constants
• type Compactor
  • func NewCompactor(l *Ledger, w realWAL.LedgerWAL, logger zerolog.Logger, checkpointCapacity uint, ...) (*Compactor, error)
  • func (c *Compactor) Done() <-chan struct{}
  • func (c *Compactor) Ready() <-chan struct{}
  • func (c *Compactor) Subscribe(observer observable.Observer)
  • func (c *Compactor) Unsubscribe(observer observable.Observer)
• type Ledger
  • func NewLedger(wal realWAL.LedgerWAL, capacity int, metrics module.LedgerMetrics, ...) (*Ledger, error)
  • func (l *Ledger) Checkpointer() (*realWAL.Checkpointer, error)
  • func (l *Ledger) CollectStats(payloadCallBack func(payload *ledger.Payload)) (*LedgerStats, error)
  • func (l *Ledger) Done() <-chan struct{}
  • func (l *Ledger) DumpTrieAsJSON(state ledger.State, writer io.Writer) error
  • func (l *Ledger) FindTrieByStateCommit(commitment flow.StateCommitment) (*trie.MTrie, error)
  • func (l *Ledger) ForestSize() int
  • func (l *Ledger) Get(query *ledger.Query) (values []ledger.Value, err error)
  • func (l *Ledger) GetSingleValue(query *ledger.QuerySingleValue) (value ledger.Value, err error)
  • func (l *Ledger) HasState(state ledger.State) bool
  • func (l *Ledger) InitialState() ledger.State
  • func (l *Ledger) MemSize() (int64, error)
  • func (l *Ledger) MigrateAt(state ledger.State, migrations []ledger.Migration, ...) (*trie.MTrie, error)
  • func (l *Ledger) MostRecentTouchedState() (ledger.State, error)
  • func (l *Ledger) Prove(query *ledger.Query) (proof ledger.Proof, err error)
  • func (l *Ledger) Ready() <-chan struct{}
  • func (l *Ledger) Set(update *ledger.Update) (newState ledger.State, trieUpdate *ledger.TrieUpdate, err error)
  • func (l *Ledger) TrieUpdateChan() <-chan *WALTrieUpdate
  • func (l *Ledger) Tries() ([]*trie.MTrie, error)
  • func (l *Ledger) ValueSizes(query *ledger.Query) (valueSizes []int, err error)
• type LedgerStats
• type WALTrieUpdate

Constants

const (
	DefaultCacheSize         = 1000
	DefaultPathFinderVersion = 1
)

Types

type Compactor

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

Compactor is a long-running goroutine responsible for: - writing WAL record from trie update, - starting checkpointing async when enough segments are finalized.

Compactor communicates with Ledger through channels to ensure that by the end of any trie update processing, update is written to WAL and new trie is pushed to trie queue.

Compactor stores pointers to tries in ledger state in a fix-sized checkpointing queue (FIFO). Checkpointing queue is decoupled from main ledger state to allow separate optimization and looser coupling, etc. CAUTION: If the forest LRU Cache is used for main state, then ledger state and checkpointing queue may contain different tries. This will be resolved automaticaly after the forest LRU Cache (code outside checkpointing) is replaced by something like a FIFO queue.

func NewCompactor

func NewCompactor(
	l *Ledger,
	w realWAL.LedgerWAL,
	logger zerolog.Logger,
	checkpointCapacity uint,
	checkpointDistance uint,
	checkpointsToKeep uint,
	triggerCheckpointOnNextSegmentFinish *atomic.Bool,
) (*Compactor, error)

NewCompactor creates new Compactor which writes WAL record and triggers checkpointing asynchronously when enough segments are finalized. The checkpointDistance is a flag that specifies how many segments need to be finalized to trigger checkpointing. However, if a prior checkpointing is already running and not finished, then more segments than specified could be accumulated for the new checkpointing (to reduce memory). All returned errors indicate that Compactor can't be created. Since failure to create Compactor will end up blocking ledger updates, the caller should handle all returned errors as unrecoverable.

func (*Compactor) Done

func (c *Compactor) Done() <-chan struct{}

Done returns channel which would be closed when Compactor goroutine exits.

func (*Compactor) Ready

func (c *Compactor) Ready() <-chan struct{}

Ready returns channel which would be closed when Compactor goroutine starts.

func (*Compactor) Subscribe

func (c *Compactor) Subscribe(observer observable.Observer)

Subscribe subscribes observer to Compactor.

func (*Compactor) Unsubscribe

func (c *Compactor) Unsubscribe(observer observable.Observer)

Unsubscribe unsubscribes observer to Compactor.

type Ledger

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

Ledger (complete) is a fast memory-efficient fork-aware thread-safe trie-based key/value storage. Ledger holds an array of registers (key-value pairs) and keeps tracks of changes over a limited time. Each register is referenced by an ID (key) and holds a value (byte slice). Ledger provides atomic batched updates and read (with or without proofs) operation given a list of keys. Every update to the Ledger creates a new state which captures the state of the storage. Under the hood, it uses binary Merkle tries to generate inclusion and non-inclusion proofs. Ledger is fork-aware which means any update can be applied at any previous state which forms a tree of tries (forest). The forest is in memory but all changes (e.g. register updates) are captured inside write-ahead-logs for crash recovery reasons. In order to limit the memory usage and maintain the performance storage only keeps a limited number of tries and purge the old ones (FIFO-based); in other words, Ledger is not designed to be used for archival usage but make it possible for other software components to reconstruct very old tries using write-ahead logs.

func NewLedger

func NewLedger(
	wal realWAL.LedgerWAL,
	capacity int,
	metrics module.LedgerMetrics,
	log zerolog.Logger,
	pathFinderVer uint8) (*Ledger, error)

NewLedger creates a new in-memory trie-backed ledger storage with persistence.

func (*Ledger) Checkpointer

func (l *Ledger) Checkpointer() (*realWAL.Checkpointer, error)

Checkpointer returns a checkpointer instance

func (*Ledger) CollectStats

func (l *Ledger) CollectStats(payloadCallBack func(payload *ledger.Payload)) (*LedgerStats, error)

func (*Ledger) Done

func (l *Ledger) Done() <-chan struct{}

Done implements interface module.ReadyDoneAware

func (*Ledger) DumpTrieAsJSON

func (l *Ledger) DumpTrieAsJSON(state ledger.State, writer io.Writer) error

DumpTrieAsJSON export trie at specific state as JSONL (each line is JSON encoding of a payload)

func (*Ledger) FindTrieByStateCommit

func (l *Ledger) FindTrieByStateCommit(commitment flow.StateCommitment) (*trie.MTrie, error)

FindTrieByStateCommit iterates over the ledger tries and compares the root hash to the state commitment if a match is found it is returned, otherwise a nil value is returned indicating no match was found

func (*Ledger) ForestSize

func (l *Ledger) ForestSize() int

ForestSize returns the number of tries stored in the forest

func (*Ledger) Get

func (l *Ledger) Get(query *ledger.Query) (values []ledger.Value, err error)

Get read the values of the given keys at the given state it returns the values in the same order as given registerIDs and errors (if any)

func (*Ledger) GetSingleValue

func (l *Ledger) GetSingleValue(query *ledger.QuerySingleValue) (value ledger.Value, err error)

GetSingleValue reads value of a single given key at the given state.

func (*Ledger) HasState

func (l *Ledger) HasState(state ledger.State) bool

HasState returns true if the given state exists inside the ledger

func (*Ledger) InitialState

func (l *Ledger) InitialState() ledger.State

InitialState returns the state of an empty ledger

func (*Ledger) MemSize

func (l *Ledger) MemSize() (int64, error)

MemSize return the amount of memory used by ledger TODO implement an approximate MemSize method

func (*Ledger) MigrateAt

func (l *Ledger) MigrateAt(
	state ledger.State,
	migrations []ledger.Migration,
	targetPathFinderVersion uint8,
) (*trie.MTrie, error)

func (*Ledger) MostRecentTouchedState

func (l *Ledger) MostRecentTouchedState() (ledger.State, error)

MostRecentTouchedState returns a state which is most recently touched.

func (*Ledger) Prove

func (l *Ledger) Prove(query *ledger.Query) (proof ledger.Proof, err error)

Prove provides proofs for a ledger query and errors (if any).

Proves are generally _not_ provided in the register order of the query. In the current implementation, proofs are sorted in a deterministic order specified by the forest and mtrie implementation.

func (*Ledger) Ready

func (l *Ledger) Ready() <-chan struct{}

Ready implements interface module.ReadyDoneAware it starts the EventLoop's internal processing loop.

func (*Ledger) Set

func (l *Ledger) Set(update *ledger.Update) (newState ledger.State, trieUpdate *ledger.TrieUpdate, err error)

Set updates the ledger given an update. It returns the state after update and errors (if any)

func (*Ledger) TrieUpdateChan

func (l *Ledger) TrieUpdateChan() <-chan *WALTrieUpdate

TrieUpdateChan returns a channel which is used to receive trie updates that needs to be logged in WALs. This channel is closed when ledger component shutdowns down.

func (*Ledger) Tries

func (l *Ledger) Tries() ([]*trie.MTrie, error)

Tries returns the tries stored in the forest

func (*Ledger) ValueSizes

func (l *Ledger) ValueSizes(query *ledger.Query) (valueSizes []int, err error)

ValueSizes read the values of the given keys at the given state. It returns value sizes in the same order as given registerIDs and errors (if any)

type LedgerStats

type LedgerStats struct {
	TrieCount        uint64 `json:"tries_count"`
	NodeCount        uint64 `json:"node_count"`
	InterimNodeCount uint64 `json:"interim_node_count"`
	LeafNodeCount    uint64 `json:"leaf_node_count"`
}

type WALTrieUpdate

type WALTrieUpdate struct {
	Update   *ledger.TrieUpdate // Update data needs to be encoded and saved in WAL.
	ResultCh chan<- error       // ResultCh channel is used to send WAL update result from Compactor to Ledger.
	TrieCh   <-chan *trie.MTrie // TrieCh channel is used to send new trie from Ledger to Compactor.
}

WALTrieUpdate is a message communicated through channel between Ledger and Compactor.