service

Documentation

Index

• Variables
• func GetCommonServerOptions(listenAddr string, logger *zap.Logger, healthcheck dgrpcserver.HealthCheck) []dgrpcserver.Option
• func ListenTier1(addr string, svc *Tier1Service, auth dauth.Authenticator, logger *zap.Logger, ...) error
• func ListenTier2(addr string, serviceDiscoveryURL *url.URL, svc *Tier2Service, ...) (err error)
• func RequestBackProcessing(ctx context.Context, logger *zap.Logger, blockRange *block.Range, ...)
• func ValidateTier1Request(request *pbsubstreamsrpc.Request, blockType string) error
• func ValidateTier2Request(request *pbssinternal.ProcessRangeRequest) error
• type BlockFilter
• type ExecutionPlan
  • func GetExecutionPlan(ctx context.Context, logger *zap.Logger, execGraph *exec.Graph, stage uint32, ...) (*ExecutionPlan, error)
• type LiveBackFiller
  • func NewLiveBackFiller(nextHandler bstream.Handler, logger *zap.Logger, stageToProcess int, ...) *LiveBackFiller
  • func (l *LiveBackFiller) ProcessBlock(blk *pbbstream.Block, obj interface{}) (err error)
  • func (l *LiveBackFiller) Start(ctx context.Context)
• type ModuleExecutionConfig
• type NoopHandler
  • func NewNoopHandler(respFunc substreams.ResponseFunc) *NoopHandler
  • func (n NoopHandler) ProcessBlock(blk *pbbstream.Block, obj interface{}) (err error)
• type Option
  • func WithMaxConcurrentRequests(max uint64) Option
  • func WithModuleExecutionTracing() Option
  • func WithReadinessFunc(f func(bool)) Option
  • func WithWASMExtensioner(ext wasm.WASMExtensioner) Option
• type RequestBackProcessingFunc
• type StreamFactory
  • func (s *StreamFactory) GetHeadBlock() (uint64, error)
  • func (s *StreamFactory) GetRecentFinalBlock() (uint64, error)
  • func (sf *StreamFactory) New(ctx context.Context, h bstream.Handler, startBlockNum int64, ...) (Streamable, error)
• type StreamFactoryFunc
• type Streamable
• type Tier1Service
  • func NewTier1(logger *zap.Logger, mergedBlocksStore dstore.Store, ...) (*Tier1Service, error)
  • func TestNewService(runtimeConfig config.RuntimeConfig, linearHandoffBlockNum uint64, ...) *Tier1Service
  • func (s *Tier1Service) Blocks(ctx context.Context, req *connect.Request[pbsubstreamsrpc.Request], ...) error
  • func (s *Tier1Service) TestBlocks(ctx context.Context, isSubRequest bool, request *pbsubstreamsrpc.Request, ...) error
• type Tier2Service
  • func NewTier2(logger *zap.Logger, opts ...Option) (*Tier2Service, error)
  • func TestNewServiceTier2(moduleExecutionTracing bool, streamFactoryFunc StreamFactoryFunc) *Tier2Service
  • func (s *Tier2Service) ProcessRange(request *pbssinternal.ProcessRangeRequest, ...) error
  • func (s *Tier2Service) TestProcessRange(ctx context.Context, request *pbssinternal.ProcessRangeRequest, ...) error

Variables

var FailureBlacklistDuration = time.Minute * 10

var FailureBlacklistMinimalCount = 0

fail fast when the exact same request has already failed twice, preventing waste of tier2 resources

var FailureForcedBackoffIncrement = time.Millisecond * 500

hold the incoming request for this duration before answering with an error if the client keeps retrying

var FailureForcedBackoffLimit = time.Second * 30

var IsValidCacheTag = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`).MatchString

var ValidateRequest = ValidateTier1Request

Deprecated: use ValidateTier1Request

Functions

func GetCommonServerOptions

func GetCommonServerOptions(listenAddr string, logger *zap.Logger, healthcheck dgrpcserver.HealthCheck) []dgrpcserver.Option

func ListenTier1

func ListenTier1(
	addr string,
	svc *Tier1Service,
	auth dauth.Authenticator,
	logger *zap.Logger,
	healthcheck dgrpcserver.HealthCheck,
) error

func ListenTier2

func ListenTier2(
	addr string,
	serviceDiscoveryURL *url.URL,
	svc *Tier2Service,
	auth dauth.Authenticator,
	logger *zap.Logger,
	healthcheck dgrpcserver.HealthCheck,
) (err error)

func RequestBackProcessing

func RequestBackProcessing(ctx context.Context, logger *zap.Logger, blockRange *block.Range, stageToProcess int, clientFactory client.InternalClientFactory, jobResult chan error)

func ValidateTier1Request

func ValidateTier1Request(request *pbsubstreamsrpc.Request, blockType string) error

ValidateTier1Request is run by the server code.

func ValidateTier2Request

func ValidateTier2Request(request *pbssinternal.ProcessRangeRequest) error

Types

type BlockFilter

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

type ExecutionPlan

type ExecutionPlan struct {
	ExistingExecOuts map[string]*execout.File
	ExecoutWriters   map[string]*execout.Writer
	ExistingIndices  map[string]map[string]*roaring64.Bitmap
	IndexWriters     map[string]*index.Writer
	RequiredModules  map[string]*pbsubstreams.Module
	StoresToWrite    map[string]struct{}
}

func GetExecutionPlan

func GetExecutionPlan(
	ctx context.Context,
	logger *zap.Logger,
	execGraph *exec.Graph,
	stage uint32,
	startBlock uint64,
	stopBlock uint64,
	outputModule string,
	execoutConfigs *execout.Configs,
	indexConfigs *index.Configs,
	storeConfigs store.ConfigMap,
) (*ExecutionPlan, error)

type LiveBackFiller

type LiveBackFiller struct {
	RequestBackProcessing RequestBackProcessingFunc
	NextHandler           bstream.Handler
	// contains filtered or unexported fields
}

func NewLiveBackFiller

func NewLiveBackFiller(nextHandler bstream.Handler, logger *zap.Logger, stageToProcess int, segmentSize uint64, linearHandoff uint64, clientFactory client.InternalClientFactory, requestBackProcessing RequestBackProcessingFunc) *LiveBackFiller

func (*LiveBackFiller) ProcessBlock

func (l *LiveBackFiller) ProcessBlock(blk *pbbstream.Block, obj interface{}) (err error)

func (*LiveBackFiller) Start

func (l *LiveBackFiller) Start(ctx context.Context)

type ModuleExecutionConfig

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

type NoopHandler

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

func NewNoopHandler

func NewNoopHandler(respFunc substreams.ResponseFunc) *NoopHandler

func (NoopHandler) ProcessBlock

func (n NoopHandler) ProcessBlock(blk *pbbstream.Block, obj interface{}) (err error)

type Option

type Option func(anyTierService)

func WithMaxConcurrentRequests

func WithMaxConcurrentRequests(max uint64) Option

func WithModuleExecutionTracing

func WithModuleExecutionTracing() Option

func WithReadinessFunc

func WithReadinessFunc(f func(bool)) Option

func WithWASMExtensioner

func WithWASMExtensioner(ext wasm.WASMExtensioner) Option

type RequestBackProcessingFunc

type RequestBackProcessingFunc = func(ctx context.Context, logger *zap.Logger, blockRange *block.Range, stageToProcess int, clientFactory client.InternalClientFactory, jobCompleted chan error)

type StreamFactory

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

func (*StreamFactory) GetHeadBlock

func (s *StreamFactory) GetHeadBlock() (uint64, error)

func (*StreamFactory) GetRecentFinalBlock

func (s *StreamFactory) GetRecentFinalBlock() (uint64, error)

func (*StreamFactory) New

func (sf *StreamFactory) New(
	ctx context.Context,
	h bstream.Handler,
	startBlockNum int64,
	stopBlockNum uint64,
	cursor string,
	finalBlocksOnly bool,
	cursorIsTarget bool,
	logger *zap.Logger,
) (Streamable, error)

type StreamFactoryFunc

type StreamFactoryFunc func(ctx context.Context,
	h bstream.Handler,
	startBlockNum int64,
	stopBlockNum uint64,
	cursor string,
	finalBlocksOnly bool,
	cursorIsTarget bool,
	logger *zap.Logger) (Streamable, error)

type Streamable

type Streamable interface {
	Run(ctx context.Context) error
}

type Tier1Service

type Tier1Service struct {
	*shutter.Shutter
	ssconnect.UnimplementedStreamHandler
	// contains filtered or unexported fields
}

func NewTier1

func NewTier1(
	logger *zap.Logger,
	mergedBlocksStore dstore.Store,
	forkedBlocksStore dstore.Store,
	hub *hub.ForkableHub,

	stateStore dstore.Store,
	defaultCacheTag string,

	parallelSubRequests uint64,
	stateBundleSize uint64,
	blockType string,

	substreamsClientConfig *client.SubstreamsClientConfig,
	tier2RequestParameters reqctx.Tier2RequestParameters,
	opts ...Option,
) (*Tier1Service, error)

func TestNewService

func TestNewService(runtimeConfig config.RuntimeConfig, linearHandoffBlockNum uint64, streamFactoryFunc StreamFactoryFunc) *Tier1Service

func (*Tier1Service) Blocks

func (s *Tier1Service) Blocks(
	ctx context.Context,
	req *connect.Request[pbsubstreamsrpc.Request],
	stream *connect.ServerStream[pbsubstreamsrpc.Response],
) error

func (*Tier1Service) TestBlocks

func (s *Tier1Service) TestBlocks(ctx context.Context, isSubRequest bool, request *pbsubstreamsrpc.Request, respFunc substreams.ResponseFunc) error

type Tier2Service

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

func NewTier2

func NewTier2(
	logger *zap.Logger,
	opts ...Option,
) (*Tier2Service, error)

func TestNewServiceTier2

func TestNewServiceTier2(moduleExecutionTracing bool, streamFactoryFunc StreamFactoryFunc) *Tier2Service

func (*Tier2Service) ProcessRange

func (s *Tier2Service) ProcessRange(request *pbssinternal.ProcessRangeRequest, streamSrv pbssinternal.Substreams_ProcessRangeServer) error

func (*Tier2Service) TestProcessRange

func (s *Tier2Service) TestProcessRange(ctx context.Context, request *pbssinternal.ProcessRangeRequest, respFunc substreams.ResponseFunc) error