store

Documentation

Index

• Constants
• type Cache
  • func (c *Cache[K, V]) Get(r storage.Reader, key K) (V, error)
  • func (c *Cache[K, V]) Insert(key K, resource V)
  • func (c *Cache[K, V]) IsCached(key K) bool
  • func (c *Cache[K, V]) PutTx(rw storage.ReaderBatchWriter, key K, resource V) error
  • func (c *Cache[K, V]) Remove(key K)
• type ChunkDataPacks
  • func NewChunkDataPacks(collector module.CacheMetrics, db storage.DB, collections storage.Collections, ...) *ChunkDataPacks
  • func (ch *ChunkDataPacks) BatchRemove(chunkID flow.Identifier, rw storage.ReaderBatchWriter) error
  • func (ch *ChunkDataPacks) BatchStore(c *flow.ChunkDataPack, rw storage.ReaderBatchWriter) error
  • func (ch *ChunkDataPacks) ByChunkID(chunkID flow.Identifier) (*flow.ChunkDataPack, error)
  • func (ch *ChunkDataPacks) Remove(chunkIDs []flow.Identifier) error
  • func (ch *ChunkDataPacks) Store(cs []*flow.ChunkDataPack) error
• type ChunksQueue
  • func NewChunkQueue(collector module.CacheMetrics, db storage.DB) *ChunksQueue
  • func (q *ChunksQueue) AtIndex(index uint64) (*chunks.Locator, error)
  • func (q *ChunksQueue) Init(defaultIndex uint64) (bool, error)
  • func (q *ChunksQueue) LatestIndex() (uint64, error)
  • func (q *ChunksQueue) StoreChunkLocator(locator *chunks.Locator) (bool, error)
• type ConsumerProgressInitializer
  • func NewConsumerProgress(db storage.DB, consumer string) *ConsumerProgressInitializer
  • func (cpi *ConsumerProgressInitializer) Initialize(defaultIndex uint64) (storage.ConsumerProgress, error)
• type ResultApprovals
  • func NewResultApprovals(collector module.CacheMetrics, db storage.DB) *ResultApprovals
  • func (r *ResultApprovals) ByChunk(resultID flow.Identifier, chunkIndex uint64) (*flow.ResultApproval, error)
  • func (r *ResultApprovals) ByID(approvalID flow.Identifier) (*flow.ResultApproval, error)
  • func (r *ResultApprovals) Index(resultID flow.Identifier, chunkIndex uint64, approvalID flow.Identifier) error
  • func (r *ResultApprovals) Store(approval *flow.ResultApproval) error

Constants

const DefaultCacheSize = uint(1000)

const DefaultChunkQueuesCacheSize = uint(1000)

const JobQueueChunksQueue = "JobQueueChunksQueue"

Types

type Cache

type Cache[K comparable, V any] struct {
	// contains filtered or unexported fields
}

func (*Cache[K, V]) Get

func (c *Cache[K, V]) Get(r storage.Reader, key K) (V, error)

Get will try to retrieve the resource from cache first, and then from the injected. During normal operations, the following error returns are expected:

• `storage.ErrNotFound` if key is unknown.

func (*Cache[K, V]) Insert

func (c *Cache[K, V]) Insert(key K, resource V)

Insert will add a resource directly to the cache with the given ID

func (*Cache[K, V]) IsCached

func (c *Cache[K, V]) IsCached(key K) bool

IsCached returns true if the key exists in the cache. It DOES NOT check whether the key exists in the underlying data store.

func (*Cache[K, V]) PutTx

func (c *Cache[K, V]) PutTx(rw storage.ReaderBatchWriter, key K, resource V) error

PutTx will return tx which adds a resource to the cache with the given ID.

func (*Cache[K, V]) Remove

func (c *Cache[K, V]) Remove(key K)

type ChunkDataPacks

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

func NewChunkDataPacks

func NewChunkDataPacks(collector module.CacheMetrics, db storage.DB, collections storage.Collections, byChunkIDCacheSize uint) *ChunkDataPacks

func (*ChunkDataPacks) BatchRemove

func (ch *ChunkDataPacks) BatchRemove(chunkID flow.Identifier, rw storage.ReaderBatchWriter) error

BatchRemove removes ChunkDataPack c keyed by its ChunkID in provided batch No errors are expected during normal operation, even if no entries are matched. If Badger unexpectedly fails to process the request, the error is wrapped in a generic error and returned.

func (*ChunkDataPacks) BatchStore

func (ch *ChunkDataPacks) BatchStore(c *flow.ChunkDataPack, rw storage.ReaderBatchWriter) error

BatchStore stores ChunkDataPack c keyed by its ChunkID in provided batch. No errors are expected during normal operation, but it may return generic error if entity is not serializable or Badger unexpectedly fails to process request

func (*ChunkDataPacks) ByChunkID

func (ch *ChunkDataPacks) ByChunkID(chunkID flow.Identifier) (*flow.ChunkDataPack, error)

func (*ChunkDataPacks) Remove

func (ch *ChunkDataPacks) Remove(chunkIDs []flow.Identifier) error

Remove removes multiple ChunkDataPacks cs keyed by their ChunkIDs in a batch. No errors are expected during normal operation, even if no entries are matched.

func (*ChunkDataPacks) Store

func (ch *ChunkDataPacks) Store(cs []*flow.ChunkDataPack) error

Store stores multiple ChunkDataPacks cs keyed by their ChunkIDs in a batch. No errors are expected during normal operation, but it may return generic error

type ChunksQueue

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

ChunksQueue stores a queue of chunk locators that assigned to me to verify. Job consumers can read the locators as job from the queue by index. Chunk locators stored in this queue are unique.

func NewChunkQueue

func NewChunkQueue(collector module.CacheMetrics, db storage.DB) *ChunksQueue

NewChunkQueue will initialize the underlying badger database of chunk locator queue.

func (*ChunksQueue) AtIndex

func (q *ChunksQueue) AtIndex(index uint64) (*chunks.Locator, error)

AtIndex returns the chunk locator stored at the given index in the queue.

func (*ChunksQueue) Init

func (q *ChunksQueue) Init(defaultIndex uint64) (bool, error)

Init initializes chunk queue's latest index with the given default index. It returns (false, nil) if the chunk queue is already initialized. It returns (true, nil) if the chunk queue is successfully initialized.

func (*ChunksQueue) LatestIndex

func (q *ChunksQueue) LatestIndex() (uint64, error)

LatestIndex returns the index of the latest chunk locator stored in the queue.

func (*ChunksQueue) StoreChunkLocator

func (q *ChunksQueue) StoreChunkLocator(locator *chunks.Locator) (bool, error)

StoreChunkLocator stores a new chunk locator that assigned to me to the job queue. A true will be returned, if the locator was new. A false will be returned, if the locator was duplicate.

type ConsumerProgressInitializer

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

ConsumerProgressInitializer is a helper to initialize the consumer progress index in storage It prevents the consumer from being used before initialization

func NewConsumerProgress

func NewConsumerProgress(db storage.DB, consumer string) *ConsumerProgressInitializer

func (*ConsumerProgressInitializer) Initialize

func (cpi *ConsumerProgressInitializer) Initialize(defaultIndex uint64) (storage.ConsumerProgress, error)

type ResultApprovals

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

ResultApprovals implements persistent storage for result approvals.

CAUTION suitable only for _Verification Nodes_ for persisting their _own_ approvals!

• In general, the Flow protocol requires multiple approvals for the same chunk from different verification nodes. In other words, there are multiple different approvals for the same chunk.
• Internally, ResultApprovals populates an index from Executed Chunk ➜ ResultApproval. This is *only safe* for Verification Nodes when tracking their own approvals (for the same ExecutionResult, a Verifier will always produce the same approval)

func NewResultApprovals

func NewResultApprovals(collector module.CacheMetrics, db storage.DB) *ResultApprovals

func (*ResultApprovals) ByChunk

func (r *ResultApprovals) ByChunk(resultID flow.Identifier, chunkIndex uint64) (*flow.ResultApproval, error)

ByChunk retrieves a ResultApproval by result ID and chunk index. The ResultApprovals store is only used within a verification node, where it is assumed that there is never more than one approval per chunk.

func (*ResultApprovals) ByID

func (r *ResultApprovals) ByID(approvalID flow.Identifier) (*flow.ResultApproval, error)

ByID retrieves a ResultApproval by its ID

func (*ResultApprovals) Index

func (r *ResultApprovals) Index(resultID flow.Identifier, chunkIndex uint64, approvalID flow.Identifier) error

Index indexes a ResultApproval by chunk (ResultID + chunk index). This operation is idempotent (repeated calls with the same value are equivalent to just calling the method once; still the method succeeds on each call).

CAUTION: the Flow protocol requires multiple approvals for the same chunk from different verification nodes. In other words, there are multiple different approvals for the same chunk. Therefore, the index Executed Chunk ➜ ResultApproval ID (populated here) is *only safe* to be used by Verification Nodes for tracking their own approvals.

func (*ResultApprovals) Store

func (r *ResultApprovals) Store(approval *flow.ResultApproval) error

Store stores a ResultApproval