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 TestTraceIDParam() store.TraceIDParam
• type Option
  • func WithMaxWasmFuelPerBlockModule(maxFuel uint64) Option
  • func WithModuleExecutionTracing() Option
  • func WithPipelineOptions(f pipeline.PipelineOptioner) Option
  • func WithWASMExtension(ext wasm.WASMExtensioner) Option
• 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
  • func TestNewService(runtimeConfig config.RuntimeConfig, linearHandoffBlockNum uint64, ...) *Tier1Service
  • func (s *Tier1Service) BlockType() string
  • 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, mergedBlocksStore dstore.Store, stateStore dstore.Store, ...) (s *Tier2Service)
  • func TestNewServiceTier2(runtimeConfig config.RuntimeConfig, streamFactoryFunc StreamFactoryFunc) *Tier2Service
  • func (s *Tier2Service) BlockType() string
  • func (s *Tier2Service) ProcessRange(request *pbssinternal.ProcessRangeRequest, ...) (grpcError 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 TestTraceID = "00000000000000000000000000000000"

TestTraceID must be used everywhere a TraceID is required. It must be the same between tier1 and tier2, otherwise tier1 will not find the file produced by tier2 correctly.

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 TestTraceIDParam

func TestTraceIDParam() store.TraceIDParam

Types

type Option

type Option func(anyTierService)

func WithMaxWasmFuelPerBlockModule

func WithMaxWasmFuelPerBlockModule(maxFuel uint64) Option

func WithModuleExecutionTracing

func WithModuleExecutionTracing() Option

func WithPipelineOptions

func WithPipelineOptions(f pipeline.PipelineOptioner) Option

WithPipelineOptions is used to configure pipeline options for consumer outside of the substreams library itself, for example in chain specific Firehose implementations.

func WithWASMExtension

func WithWASMExtension(ext wasm.WASMExtensioner) Option

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,

	blockType string,

	parallelSubRequests uint64,
	stateBundleSize uint64,

	substreamsClientConfig *client.SubstreamsClientConfig,
	opts ...Option,
) *Tier1Service

func TestNewService

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

func (*Tier1Service) BlockType

func (s *Tier1Service) BlockType() string

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,
	mergedBlocksStore dstore.Store,

	stateStore dstore.Store,
	defaultCacheTag string,
	stateBundleSize uint64,

	blockType string,
	opts ...Option,

) (s *Tier2Service)

func TestNewServiceTier2

func TestNewServiceTier2(runtimeConfig config.RuntimeConfig, streamFactoryFunc StreamFactoryFunc) *Tier2Service

func (*Tier2Service) BlockType

func (s *Tier2Service) BlockType() string

func (*Tier2Service) ProcessRange

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

func (*Tier2Service) TestProcessRange

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