pipeline

Documentation

Index

• Variables
• func BlockToClock(block *pbbstream.Block) *pbsubstreams.Clock
• func BuildRequestDetails(ctx context.Context, request *pbsubstreamsrpc.Request, ...) (req *reqctx.RequestDetails, undoSignal *pbsubstreamsrpc.BlockUndoSignal, ...)
• func BuildRequestDetailsFromSubrequest(request *pbssinternal.ProcessRangeRequest) (req *reqctx.RequestDetails)
• func NewStoreBoundary(interval uint64, requestStartBlockNum uint64, requestStopBlock uint64) *storeBoundary
• type CursorResolver
  • func NewCursorResolver(hub *hub.ForkableHub, mergedBlocksStore, forkedBlocksStore dstore.Store) CursorResolver
• type ForkHandler
  • func NewForkHandler() *ForkHandler
• type Option
  • func WithFinalBlocksOnly() Option
  • func WithHighestStage(stage uint32) Option
  • func WithPendingUndoMessage(msg *pbsubstreamsrpc.Response) Option
  • func WithPostBlockHook(f substreams.BlockHook) Option
  • func WithPostJobHook(f substreams.PostJobHook) Option
  • func WithPreBlockHook(f substreams.BlockHook) Option
• type Pipeline
  • func New(ctx context.Context, execGraph *exec.Graph, stores *Stores, ...) *Pipeline
  • func (p *Pipeline) BuildModuleExecutors(ctx context.Context) error
  • func (p *Pipeline) GetStoreMap() store.Map
  • func (p *Pipeline) Init(ctx context.Context) (err error)
  • func (p *Pipeline) InitTier1StoresAndBackprocess(ctx context.Context, reqPlan *plan.RequestPlan) (err error)
  • func (p *Pipeline) InitTier2Stores(ctx context.Context) (err error)
  • func (p *Pipeline) OnStreamTerminated(ctx context.Context, err error) error
  • func (p *Pipeline) ProcessBlock(block *pbbstream.Block, obj interface{}) (err error)
  • func (p *Pipeline) ProcessFromExecOutput(ctx context.Context, clock *pbsubstreams.Clock, cursor *bstream.Cursor) (err error)
• type Stores
  • func NewStores(ctx context.Context, storeConfigs store.ConfigMap, ...) *Stores
  • func (s *Stores) SetStoreMap(storeMap store.Map)
• type UndoHandler

Variables

var ErrDone = errors.New("done")

Functions

func BlockToClock

func BlockToClock(block *pbbstream.Block) *pbsubstreams.Clock

func BuildRequestDetails

func BuildRequestDetails(
	ctx context.Context,
	request *pbsubstreamsrpc.Request,
	getRecentFinalBlock getBlockFunc,
	resolveCursor CursorResolver,
	getHeadBlock getBlockFunc,
	segmentSize uint64) (req *reqctx.RequestDetails, undoSignal *pbsubstreamsrpc.BlockUndoSignal, err error)

func BuildRequestDetailsFromSubrequest

func BuildRequestDetailsFromSubrequest(request *pbssinternal.ProcessRangeRequest) (req *reqctx.RequestDetails)

func NewStoreBoundary

func NewStoreBoundary(
	interval uint64,
	requestStartBlockNum uint64,
	requestStopBlock uint64,
) *storeBoundary

Types

type CursorResolver

type CursorResolver func(context.Context, *bstream.Cursor) (reorgJunctionBlock, currentHead bstream.BlockRef, err error)

func NewCursorResolver

func NewCursorResolver(hub *hub.ForkableHub, mergedBlocksStore, forkedBlocksStore dstore.Store) CursorResolver

type ForkHandler

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

TODO(abourget): The scope of this object and the Engine

are not pretty similar, to keep track of certain pieces
of info that are reversible, and handle the back and forth
between undos and redos.
Perhaps what we could have here, is have those undo handlers
live on the Pipeline (where it makes sense)
and have some nested structs handle

func NewForkHandler

func NewForkHandler() *ForkHandler

type Option

type Option func(p *Pipeline)

func WithFinalBlocksOnly

func WithFinalBlocksOnly() Option

func WithHighestStage

func WithHighestStage(stage uint32) Option

func WithPendingUndoMessage

func WithPendingUndoMessage(msg *pbsubstreamsrpc.Response) Option

WithPendingUndoMessage allows sending a message right before we send the first 'BlockScopedData'

func WithPostBlockHook

func WithPostBlockHook(f substreams.BlockHook) Option

func WithPostJobHook

func WithPostJobHook(f substreams.PostJobHook) Option

func WithPreBlockHook

func WithPreBlockHook(f substreams.BlockHook) Option

type Pipeline

type Pipeline struct {
	ModuleExecutors [][]exec.ModuleExecutor // Staged module executors
	// contains filtered or unexported fields
}

func New

func New(
	ctx context.Context,
	execGraph *exec.Graph,
	stores *Stores,
	indices map[string]map[string]*roaring64.Bitmap,
	execoutStorage *execout.Configs,
	wasmRuntime *wasm.Registry,
	execOutputCache *cache.Engine,
	stateBundleSize uint64,
	workerFactory work.WorkerFactory,
	respFunc substreams.ResponseFunc,
	executionTimeout time.Duration,
	opts ...Option,
) *Pipeline

func (*Pipeline) BuildModuleExecutors

func (p *Pipeline) BuildModuleExecutors(ctx context.Context) error

BuildModuleExecutors builds the ModuleExecutors, and the loadedModules.

func (*Pipeline) GetStoreMap

func (p *Pipeline) GetStoreMap() store.Map

func (*Pipeline) Init

func (p *Pipeline) Init(ctx context.Context) (err error)

func (*Pipeline) InitTier1StoresAndBackprocess

func (p *Pipeline) InitTier1StoresAndBackprocess(ctx context.Context, reqPlan *plan.RequestPlan) (err error)

func (*Pipeline) InitTier2Stores

func (p *Pipeline) InitTier2Stores(ctx context.Context) (err error)

func (*Pipeline) OnStreamTerminated

func (p *Pipeline) OnStreamTerminated(ctx context.Context, err error) error

OnStreamTerminated performs flush of store and setting trailers when the stream terminated gracefully from our point of view. If the stream terminated gracefully, we return `nil` otherwise, the original is returned.

func (*Pipeline) ProcessBlock

func (p *Pipeline) ProcessBlock(block *pbbstream.Block, obj interface{}) (err error)

func (*Pipeline) ProcessFromExecOutput

func (p *Pipeline) ProcessFromExecOutput(
	ctx context.Context,
	clock *pbsubstreams.Clock,
	cursor *bstream.Cursor,
) (err error)

type Stores

type Stores struct {
	StoreMap store.Map
	// contains filtered or unexported fields
}

func NewStores

func NewStores(ctx context.Context, storeConfigs store.ConfigMap, storeSnapshotSaveInterval, requestStartBlockNum, stopBlockNum uint64, isTier2Request bool, storesToWrite map[string]struct{}) *Stores

func (*Stores) SetStoreMap

func (s *Stores) SetStoreMap(storeMap store.Map)

type UndoHandler

type UndoHandler func(clock *pbsubstreams.Clock, moduleOutputs []*pbssinternal.ModuleOutput)