pipeline

package
v1.25.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 8, 2026 License: Apache-2.0 Imports: 64 Imported by: 2

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrDone = errors.New("done")
View Source 
var ErrShuttingDown = errors.New("endpoint is shutting down, please reconnect")
View Source 
var MaxLinearHandoffLagSegments = parseMaxLinearHandoffLagSegments(os.Getenv("SUBSTREAMS_MAX_LINEAR_HANDOFF_LAG_SEGMENTS"))

MaxLinearHandoffLagSegments is how many segments a production request may trail the last final block (rounded down to a segment), either when backprocessing completes or while streaming final blocks. Past that, the client is disconnected so that its reconnection back-processes the gap in parallel.

View Source 
var PrintStack = os.Getenv("SUBSTREAMS_PRINT_STACK") == "true" || os.Getenv("SUBSTREAMS_PRINT_STACK") == "1"

Functions

func BlockToClock added in v1.8.1 

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

func BuildRequestDetails added in v0.1.0 

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 added in v1.0.2 

func BuildRequestDetailsFromSubrequest(ctx context.Context, request *pbssinternal.ProcessRangeRequest) (req *reqctx.RequestDetails)

func LinearHandoffLagTarget added in v1.23.0 

func LinearHandoffLagTarget(linearHandoff, stopBlock, lastFinalBlock, segmentSize, maxLagSegments uint64) (target uint64, tooFarBehind bool)

LinearHandoffLagTarget returns the block the linear handoff would be set to if the request was planned now, and whether the given handoff trails it by more than maxLagSegments.

func NewStoreBoundary added in v0.0.21 

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

Types

type CursorResolver added in v1.0.2 

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

func NewCursorResolver added in v1.0.2 

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

type ForkHandler added in v0.0.20 

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 added in v0.0.21 

func NewForkHandler() *ForkHandler

type Option 

type Option func(p *Pipeline)

func WithExecOutPrefetch added in v1.23.0 

func WithExecOutPrefetch(cfg execout.PrefetchConfig) Option

WithExecOutPrefetch bounds how far ahead tier1 downloads cached execution output files while streaming them to the client.

func WithFinalBlockLagCheck added in v1.23.0 

func WithFinalBlockLagCheck(getRecentFinalBlock func() (uint64, error), linearHandoff uint64) Option

WithFinalBlockLagCheck disconnects the request with ErrShuttingDown when, at a segment boundary, its final block trails the last final block of the chain by more than MaxLinearHandoffLagSegments.

func WithFinalBlocksOnly added in v1.1.1 

func WithFinalBlocksOnly() Option

func WithHeadBlockGetter added in v1.14.3 

func WithHeadBlockGetter(getter func() (uint64, error)) Option

func WithHighestStage added in v1.1.9 

func WithHighestStage(stage uint32) Option

func WithPendingUndoMessage added in v1.1.1 

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 {

	// StagedModuleExecutors represents all the modules within a stage that should be executed. The
	// first level of the 2D list represents layer within a stage to execute sequentially.
	// The second level contains modules to execute within a layer, those can be executed concurrently.
	StagedModuleExecutors [][]exec.ModuleExecutor
	ModuleBlockIndexes    map[string]*index.BlockIndex
	// contains filtered or unexported fields
}

func New 

func New(
	ctx context.Context,
	isTier1 bool,
	execGraph *exec.Graph,
	blockType string,
	stores *Stores,
	indices map[string]map[string]*roaring64.Bitmap,
	execoutStorage *execout.Configs,
	wasmRuntime *wasm.Registry,
	execOutputCache *cache.Engine,
	stateBundleSize uint64,
	workerPoolFactory work.WorkerPoolFactory,
	respFunc substreams.ResponseFunc,
	executionTimeout time.Duration,
	checkPendingShutdown func() bool,
	foundationalEndpoints map[string]string,
	outputBufferSize int,
	supportBuffering bool,
	opts ...Option,
) *Pipeline

func (*Pipeline) BuildModuleExecutors added in v1.6.0 

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

BuildModuleExecutors builds the ModuleExecutors, and the loadedModules.

func (*Pipeline) Drain added in v1.23.0 

func (p *Pipeline) Drain(reason string) error

Drain waits for the block being processed (if any) to finish, quick-saves the stores and makes every later ProcessBlock call return ErrShuttingDown. It lets the caller end the stream on a timer without waiting for the next block, which can take a long time on slow chains. The caller is expected to cancel the stream right after.

func (*Pipeline) GetStoreMap added in v0.1.0 

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

func (*Pipeline) Init added in v0.0.14 

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

func (*Pipeline) InitTier1StoresAndBackprocess added in v1.1.9 

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

func (*Pipeline) InitTier2Stores added in v1.1.9 

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

func (*Pipeline) LastCursor added in v1.16.5 

func (p *Pipeline) LastCursor() *bstream.Cursor

func (*Pipeline) OnStreamTerminated added in v0.1.0 

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 added in v0.0.14 

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

func (*Pipeline) ProcessFromExecOutput added in v1.4.0 

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

func (*Pipeline) SendProgressSnapshot added in v1.22.0 

func (p *Pipeline) SendProgressSnapshot() error

SendProgressSnapshot emits an internal progress update from outside the block loop.

returnInternalModuleProgressOutputs only runs once a block finished processing, so a module stuck inside a single block — typically blocked on an external call that is retrying against an unreachable endpoint — reports nothing at all for as long as it is stuck, and tier1 sees an idle job with no external call metrics. The stats already account for calls still in flight; they just need to be sent. Safe to call concurrently with the block loop: the tier2 response function serializes sends.

type Stores added in v0.1.0 

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

func NewStores added in v0.1.0 

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

func (*Stores) Close added in v1.20.0 

func (s *Stores) Close()

func (*Stores) SetStoreMap added in v0.1.0 

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

type UndoHandler added in v0.0.21 

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

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.