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)
• type Option
  • func WithMaxConcurrentRequests(max uint64) Option
  • func WithMaxWasmFuelPerBlockModule(maxFuel uint64) Option
  • func WithModuleExecutionTracing() Option
  • func WithReadinessFunc(f func(bool)) Option
  • func WithWASMExtensioner(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, 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(runtimeConfig config.RuntimeConfig, streamFactoryFunc StreamFactoryFunc) *Tier2Service
  • 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

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)

Types

type Option

type Option func(anyTierService)

func WithMaxConcurrentRequests

func WithMaxConcurrentRequests(max uint64) Option

func WithMaxWasmFuelPerBlockModule

func WithMaxWasmFuelPerBlockModule(maxFuel uint64) Option

func WithModuleExecutionTracing

func WithModuleExecutionTracing() Option

func WithReadinessFunc

func WithReadinessFunc(f func(bool)) Option

func WithWASMExtensioner

func WithWASMExtensioner(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,

	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(runtimeConfig config.RuntimeConfig, streamFactoryFunc StreamFactoryFunc) *Tier2Service

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) error