sink

package

Go to main page

Versions in this module

v1

v1.18.2

Feb 26, 2026

v1.18.1

Feb 20, 2026

v1.18.0

Feb 19, 2026

v1.17.11

Jan 21, 2026

v1.17.10

Jan 15, 2026

Changes in this version

type SinkerConfig

+ func (config *SinkerConfig) CloneWithNewModule(expectedOutputModuleType string, manifestPath string, outputModuleName string) (*SinkerConfig, error)

v1.17.9

Jan 9, 2026

Changes in this version

+ const FlagPartialBlocks

type SinkerConfig

+ PartialBlocks bool

v1.17.8

Dec 12, 2025

Changes in this version

+ const FlagIncludePartialBlocks

+ const FlagPartialBlocksOnly

type SinkerConfig

+ IncludePartialBlocks bool

+ PartialBlocksOnly bool

type SinkerHandler

+ func NewSinkerFullHandlersWithPartial(...) SinkerHandler

+ type SinkerPartialBlockHandler interface

+ HandlePartialBlockData func(ctx context.Context, data *pbsubstreamsrpc.PartialBlockData) error

v1.17.8-partialblocks

Dec 10, 2025

v1.17.8-joinedpartials

Jan 8, 2026

v1.17.5

Nov 23, 2025

v1.17.4

Nov 21, 2025

Changes in this version

+ const FlagSkipCheckModuleBinariesExist

v1.17.3

Nov 21, 2025

v1.17.2

Nov 12, 2025

v1.17.1

Oct 23, 2025

v1.17.0

Oct 21, 2025

Changes in this version

+ const FlagForceProtocolVersion

+ func RegisterMetrics()

+ type FlagInclusionExclusion interface

+ IsExplicitlyExcluded func(flag string) bool

+ IsExplicitlyIncluded func(flag string) bool

+ func FlagExcludeDefault(in ...string) FlagInclusionExclusion

+ func FlagIncludeOptional(in ...string) FlagInclusionExclusion

+ type Option func(s *Sinker)

+ func WithAgent(agent string) Option

+ func WithBlockDataBuffer(bufferSize int) Option

+ func WithBlockRange(blockRange *bstream.Range) Option

+ func WithExtraHeaders(headers []string) Option

+ func WithFinalBlocksOnly() Option

+ func WithInfiniteRetry() Option

+ func WithLivenessChecker(livenessChecker LivenessChecker) Option

+ func WithRetryBackOff(backOff backoff.BackOff) Option

type Sinker

+ func NewFromConfig(config *SinkerConfig) (*Sinker, error)

+ func (s *Sinker) BlockRange() *bstream.Range

type SinkerConfig

+ func (c *SinkerConfig) BlockRange() *bstream.Range

type Stats

+ func (s *Stats) SetNoop()

v1.16.6

Sep 24, 2025

Changes in this version

type SinkerConfig

+ SupportIndexOutputProductionMode bool

type Stats

+ func (s *Stats) SetIndexOutputProductionMode()

v1.16.5

Sep 11, 2025

Changes in this version

+ const FlagMaxRetries

type SinkerConfig

+ MaxRetries int

v1.16.4

Aug 25, 2025

Changes in this version

+ type CursorBasedLivenessChecker struct

+ func NewCursorBasedLivenessChecker() *CursorBasedLivenessChecker

+ func (t *CursorBasedLivenessChecker) CheckCursor(cursor string)

+ func (t *CursorBasedLivenessChecker) IsLive(clock *pbsubstreams.Clock) bool

v1.16.3

Aug 14, 2025

v1.16.2

Aug 1, 2025

Changes in this version

+ const FlagAPIKeyEnvvar

+ const FlagAPITokenEnvvar

+ const FlagCursor

+ const FlagDevelopmentMode

+ const FlagEndpoint

+ const FlagExtraHeaders

+ const FlagFinalBlocksOnly

+ const FlagInfiniteRetry

+ const FlagInsecure

+ const FlagLiveBlockTimeDelta

+ const FlagNetwork

+ const FlagNoopMode

+ const FlagParams

+ const FlagPlaintext

+ const FlagPrometheusAddr

+ const FlagProtoDescriptorSet

+ const FlagProtoPath

+ const FlagSkipPackageValidation

+ const FlagStartBlock

+ const FlagStopBlock

+ const FlagUndoBufferSize

+ const IgnoreOutputModuleType

+ const InferOutputModuleFromPackage

+ const ShortFlagCursor

+ const ShortFlagEndoint

+ const ShortFlagStartBlock

+ const ShortFlagStopBlock

+ var AvgBlockTimeDelta = dmetrics.NewAvgDurationCounter(30 * time.Second, time.Second, "Average duration for BlockTimeDelta")

+ var AvgBlockWaitTime = dmetrics.NewAvgDurationCounter(30 * time.Second, time.Second, "Average duration for BlockWaitTime")

+ var AvgLocalProcessingTime = dmetrics.NewAvgDurationCounter(30 * time.Second, time.Second, "Average duration for LocalProcessingTime")

+ var BackprocessingCompletion = Metrics.NewGauge("substreams_sink_backprocessing_completion", ...)

+ var BlockTimeDelta = Metrics.NewGauge("substreams_sink_block_time_delta", ...)

+ var BlockWaitTime = Metrics.NewGauge("substreams_sink_block_wait_time", ...)

+ var DataMessageCount = Metrics.NewCounter("substreams_sink_data_message", "The number of data message received")

+ var ErrBackOffExpired = errors.New("unable to complete work within backoff time limit")

+ var ErrHandlerNotImplemented = errors.New("handler not implemented")

+ var HeadBlockNumber = Metrics.NewHeadBlockNumber("substreams_sink")

+ var HeadBlockTimeDrift = Metrics.NewHeadTimeDrift("substreams_sink")

+ var LocalProcessingTime = Metrics.NewGauge("substreams_sink_local_processing_time", ...)

+ var MessageSizeBytes = Metrics.NewCounter("substreams_sink_message_size_bytes", ...)

+ var Metrics = dmetrics.NewSet()

+ var ProcessedBlocks = Metrics.NewGauge("substreams_sink_processed_blocks", ...)

+ var ProcessedBytes = Metrics.NewGauge("substreams_sink_processed_bytes", ...)

+ var ProgressMessageCount = Metrics.NewGauge("substreams_sink_progress_message", "The number of progress message received")

+ var ProgressMessageLastBlock = Metrics.NewGaugeVec("substreams_sink_progress_message_last_block", []string{ ... }, ...)

+ var ProgressMessageLastContiguousBlock = Metrics.NewGaugeVec("substreams_sink_progress_message_last_contiguous_block", []string{ ... }, ...)

+ var ProgressMessageRunningJobs = Metrics.NewGaugeVec("substreams_sink_progress_message_running_jobs", []string{ ... }, ...)

+ var ServerEgressBytes = Metrics.NewCounter("substreams_sink_server_egress_bytes", ...)

+ var SubstreamsErrorCount = Metrics.NewCounter("substreams_sink_error", ...)

+ var UndoMessageCount = Metrics.NewCounter("substreams_sink_undo_message", "The number of block undo message received")

+ var UnknownMessageCount = Metrics.NewCounter("substreams_sink_unknown_message", "The number of unknown message received")

+ func AddFlagsToSet(flags *pflag.FlagSet, ignore ...FlagIgnored)

+ func LoadSubstreamsAuthEnvFile(manifestPath string)

+ func ReadManifestAndModule(manifestPath string, network string, paramsStrings []string, ...) (pkg *pbsubstreams.Package, module *pbsubstreams.Module, ...)

+ func SubstreamsModeNames() []string

+ func WriteCursor(filename string, cursor *Cursor) error

+ type BackOffStringer struct

+ func (s BackOffStringer) String() string

+ type BytesRepresentation int

+ const BytesAsBase58

+ const BytesAsBase64

+ const BytesAsHex

+ const BytesAsString

+ func BytesEncodingToRepresentation(enc string) BytesRepresentation

+ func InferBytesRepresentation(network string, endpoint string) BytesRepresentation

+ type Cursor struct

+ func MustNewCursor(cursor string) *Cursor

+ func NewBlankCursor() *Cursor

+ func NewCursor(cursor string) (*Cursor, error)

+ func ReadCursor(filename string) (*Cursor, error)

+ func (c *Cursor) Block() bstream.BlockRef

+ func (c *Cursor) IsBlank() bool

+ func (c *Cursor) IsEqualTo(other *Cursor) bool

+ func (c *Cursor) MarshalLogObject(encoder zapcore.ObjectEncoder) error

+ func (c *Cursor) String() string

+ type DeltaLivenessChecker struct

+ func NewDeltaLivenessChecker(delta time.Duration) *DeltaLivenessChecker

+ func (t *DeltaLivenessChecker) IsLive(clock *pbsubstreams.Clock) bool

+ type FlagIgnored interface

+ IsIgnored func(flag string) bool

+ func FlagIgnore(in ...string) FlagIgnored

+ type LivenessChecker interface

+ IsLive func(block *pbsubstreams.Clock) bool

+ type Sinker struct

+ func New(config *SinkerConfig) *Sinker

+ func NewFromViper(cmd *cobra.Command, expectedOutputModuleType string, manifestPath string, ...) (*Sinker, error)

+ func (s *Sinker) ApiToken() string

+ func (s *Sinker) BytesRepresentation() BytesRepresentation

+ func (s *Sinker) ClientConfig() *client.SubstreamsClientConfig

+ func (s *Sinker) EndpointConfig() (endpoint string, plaintext bool, insecure bool)

+ func (s *Sinker) OutputModuleHash() string

+ func (s *Sinker) OutputModuleName() string

+ func (s *Sinker) OutputModuleTypePrefixed() (prefixed string)

+ func (s *Sinker) OutputModuleTypeUnprefixed() (unprefixed string)

+ func (s *Sinker) Package() *pbsubstreams.Package

+ func (s *Sinker) PrintStats()

+ func (s *Sinker) Request() *pbsubstreamsrpc.Request

+ func (s *Sinker) Run(ctx context.Context, cursor *Cursor, handler SinkerHandler)

+ func (s *Sinker) StartBlock() int64

+ func (s *Sinker) StopBlock() uint64

+ type SinkerCompletionHandler interface

+ HandleBlockRangeCompletion func(ctx context.Context, cursor *Cursor) error

+ type SinkerConfig struct

+ BackOff backoff.BackOff

+ ClientConfig *client.SubstreamsClientConfig

+ DevOutputModules []string

+ DevOutputSnapshots []string

+ ExtraHeaders []string

+ FinalBlocksOnly bool

+ InfiniteRetry bool

+ LimitProcessedBlocks uint64

+ LiveBlockTimeDelta time.Duration

+ LivenessChecker LivenessChecker

+ Logger *zap.Logger

+ Mode SubstreamsMode

+ Network string

+ NoopMode bool

+ OutputModule *pbsubstreams.Module

+ OutputModuleHash manifest.ModuleHash

+ Params []string

+ Pkg *pbsubstreams.Package

+ PrometheusAddr string

+ SkipPackageValidation bool

+ StartBlock int64

+ StopBlock uint64

+ Tracer logging.Tracer

+ UndoBufferSize int

+ func ConfigFromViper(cmd *cobra.Command, expectedOutputModuleType string, manifestPath string, ...) (*SinkerConfig, error)

+ func (c *SinkerConfig) ExtractDefaultParams() []string

+ type SinkerErrorHandler interface

+ HandleError func(ctx context.Context, err error)

+ type SinkerHandler interface

+ HandleBlockScopedData func(ctx context.Context, data *pbsubstreamsrpc.BlockScopedData, isLive *bool, ...) error

+ HandleBlockUndoSignal func(ctx context.Context, undoSignal *pbsubstreamsrpc.BlockUndoSignal, ...) error

+ func NewSinkerFullHandlers(...) SinkerHandler

+ func NewSinkerHandlers(...) SinkerHandler

+ type SinkerProgressHandler interface

+ HandleProgress func(ctx context.Context, progress *pbsubstreamsrpc.ModulesProgress)

+ type SinkerSessionInitHandler interface

+ HandleSessionInit func(ctx context.Context, req *pbsubstreamsrpc.Request, ...) error

+ type SinkerSnapshotHandler interface

+ HandleInitialSnapshotComplete func(ctx context.Context, snapshotComplete *pbsubstreamsrpc.InitialSnapshotComplete) error

+ HandleInitialSnapshotData func(ctx context.Context, snapshotData *pbsubstreamsrpc.InitialSnapshotData) error

+ type Stats struct

+ func (s *Stats) Close()

+ func (s *Stats) LogNow()

+ func (s *Stats) RecordBlock(block bstream.BlockRef)

+ func (s *Stats) SetLiveness(isLive *bool)

+ func (s *Stats) Start(each time.Duration)

+ type SubstreamsMode uint

+ const SubstreamsModeDevelopment

+ const SubstreamsModeProduction

+ func ParseSubstreamsMode(name string) (SubstreamsMode, error)

+ func (x *SubstreamsMode) UnmarshalText(text []byte) error

+ func (x SubstreamsMode) MarshalText() ([]byte, error)

+ func (x SubstreamsMode) String() string

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL