Let streams skip partial blocks of a shared hub

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `ForkableHub.WithoutPartials` and `stream.WithoutPartialBlocks`: stream from a hub fed with partial blocks without getting them, the last partial of each block arriving as the complete block, as for a relayer client that does not ask for partials. Lets an app that does not handle partial blocks share its hub with one that does.
 - `hub.SubscriptionCatchUpTimeout` (default `30s`): a hub subscription is now closed with `hub.ErrSubscriptionBehind` when its consumer has not emptied its waiting blocks for that long, which catches a consumer that is stuck or slower than the chain. A consumer draining a burst within the timeout is not affected. `ErrSubscriptionBehind` matches `ErrSubscriptionChannelFull` with `errors.Is`, so callers that reconnect on one reconnect on both.
 - `hub.SubscriptionMaxBufferedBlocks` (default `10000`): how many blocks a hub subscription can have waiting before it is closed with `ErrSubscriptionChannelFull`. Both are meant to be set once at process startup.
 - `MultiplexedSourceWithRetryIntervals`: new option setting, per source, the minimum time between two connection attempts, rounded up to the next 5s increment since sources are checked every 5s. Useful for a rescuer or fallback source that is down most of the time and would otherwise be redialed every 5s.

hub/hub.go

@@ -238,12 +238,16 @@ func (h *ForkableHub) unsubscribe(removeSub *Subscription) {
 }
 
 func (h *ForkableHub) SourceFromBlockNum(num uint64, handler bstream.Handler) (out bstream.Source) {
+	return h.sourceFromBlockNum(num, handler, true)
+}
+
+func (h *ForkableHub) sourceFromBlockNum(num uint64, handler bstream.Handler, withPartials bool) (out bstream.Source) {
 	if h == nil {
 		return nil
 	}
 
 	err := h.forkable.CallWithBlocksFromNum(num, func(blocks []*bstream.PreprocessedBlock) { // Running callback func while forkable is locked
-		out = h.subscribe(handler, blocks, true)
+		out = h.subscribe(handler, blocks, withPartials)
 	}, false)
 	if err != nil {
 		h.logger.Debug("error getting source_from_block_num", zap.Error(err))
@@ -268,12 +272,16 @@ func (h *ForkableHub) SourceFromBlockNumWithForks(num uint64, handler bstream.Ha
 }
 
 func (h *ForkableHub) SourceFromCursor(cursor *bstream.Cursor, handler bstream.Handler) (out bstream.Source) {
+	return h.sourceFromCursor(cursor, handler, true)
+}
+
+func (h *ForkableHub) sourceFromCursor(cursor *bstream.Cursor, handler bstream.Handler, withPartials bool) (out bstream.Source) {
 	if h == nil {
 		return nil
 	}
 
 	err := h.forkable.CallWithBlocksFromCursor(cursor, func(blocks []*bstream.PreprocessedBlock) { // Running callback func while forkable is locked
-		out = h.subscribe(handler, blocks, true)
+		out = h.subscribe(handler, blocks, withPartials)
 	})
 	if err != nil {
 		h.logger.Debug("error getting source_from_cursor", zap.Error(err))
@@ -283,17 +291,21 @@ func (h *ForkableHub) SourceFromCursor(cursor *bstream.Cursor, handler bstream.H
 }
 
 func (h *ForkableHub) SourceThroughCursor(startBlock uint64, cursor *bstream.Cursor, handler bstream.Handler) (out bstream.Source) {
+	return h.sourceThroughCursor(startBlock, cursor, handler, true)
+}
+
+func (h *ForkableHub) sourceThroughCursor(startBlock uint64, cursor *bstream.Cursor, handler bstream.Handler, withPartials bool) (out bstream.Source) {
 	if h == nil {
 		return nil
 	}
 
 	// cursor has already passed, ignoring it
 	if cursor.Block.Num() < startBlock {
-		return h.SourceFromBlockNum(startBlock, handler)
+		return h.sourceFromBlockNum(startBlock, handler, withPartials)
 	}
 
 	err := h.forkable.CallWithBlocksThroughCursor(startBlock, cursor, func(blocks []*bstream.PreprocessedBlock) { // Running callback func while forkable is locked
-		out = h.subscribe(handler, blocks, true)
+		out = h.subscribe(handler, blocks, withPartials)
 	})
 	if err != nil {
 		h.logger.Debug("error getting source_from_cursor", zap.Error(err))
@@ -302,6 +314,30 @@ func (h *ForkableHub) SourceThroughCursor(startBlock uint64, cursor *bstream.Cur
 	return
 }
 
+// WithoutPartials returns the hub as a source factory whose sources leave out
+// partial blocks, delivering the last partial of a block as the complete block,
+// like the relayer does for a client that does not ask for partial blocks. It
+// lets an app that does not handle partial blocks share a hub fed with them.
+func (h *ForkableHub) WithoutPartials() bstream.ForkableSourceFactory {
+	return withoutPartials{hub: h}
+}
+
+type withoutPartials struct {
+	hub *ForkableHub
+}
+
+func (w withoutPartials) SourceFromBlockNum(num uint64, handler bstream.Handler) bstream.Source {
+	return w.hub.sourceFromBlockNum(num, handler, false)
+}
+
+func (w withoutPartials) SourceFromCursor(cursor *bstream.Cursor, handler bstream.Handler) bstream.Source {
+	return w.hub.sourceFromCursor(cursor, handler, false)
+}
+
+func (w withoutPartials) SourceThroughCursor(startBlock uint64, cursor *bstream.Cursor, handler bstream.Handler) bstream.Source {
+	return w.hub.sourceThroughCursor(startBlock, cursor, handler, false)
+}
+
 // bootstrapMaxAttempts bounds the retries bootstrap() performs when the one-block
 // listing looks torn by a concurrent merger deletion (see bootstrapRetryDelay).
 const bootstrapMaxAttempts = 3

hub/hub_test.go

@@ -1112,3 +1112,48 @@ func TestForkableHub_ProcessBlock_UnlinkableCountSkipsNonFinalFlashBlocks(t *tes
 		require.ErrorIs(t, fh.ProcessBlock(unlinkable(24, 0, false), nil), errRestartRequired)
 	})
 }
+
+func TestForkableHub_WithoutPartials(t *testing.T) {
+	fh := &ForkableHub{
+		Shutter: shutter.New(),
+		logger:  zlog,
+	}
+	fh.forkable = forkable.New(bstream.HandlerFunc(fh.broadcastBlock),
+		forkable.HoldBlocksUntilLIB(),
+		forkable.WithKeptFinalBlocks(100),
+	)
+	for _, blk := range []*pbbstream.Block{
+		bstream.TestBlockWithLIBNum("00000003", "00000002", 2),
+		bstream.TestBlockWithLIBNum("00000004", "00000003", 3),
+		bstream.TestBlockWithLIBNum("00000005", "00000004", 3),
+	} {
+		require.NoError(t, fh.forkable.ProcessBlock(blk, nil))
+	}
+
+	handler := bstream.HandlerFunc(func(blk *pbbstream.Block, obj any) error { return nil })
+	cursor := &bstream.Cursor{
+		Step:      bstream.StepNew,
+		Block:     bstream.NewBlockRefFromID("00000004"),
+		HeadBlock: bstream.NewBlockRefFromID("00000004"),
+		LIB:       bstream.NewBlockRefFromID("00000003"),
+	}
+
+	withPartial := func(src bstream.Source) bool {
+		t.Helper()
+		require.NotNil(t, src)
+		return src.(*Subscription).withPartial
+	}
+
+	view := fh.WithoutPartials()
+	assert.False(t, withPartial(view.SourceFromBlockNum(4, handler)))
+	assert.False(t, withPartial(view.SourceFromCursor(cursor, handler)))
+	assert.False(t, withPartial(view.SourceThroughCursor(4, cursor, handler)))
+	assert.False(t, withPartial(view.SourceThroughCursor(5, cursor, handler)), "cursor already passed")
+
+	assert.True(t, withPartial(fh.SourceFromBlockNum(4, handler)))
+	assert.True(t, withPartial(fh.SourceFromCursor(cursor, handler)))
+	assert.True(t, withPartial(fh.SourceThroughCursor(4, cursor, handler)))
+
+	var nilHub *ForkableHub
+	assert.Nil(t, nilHub.WithoutPartials().SourceFromBlockNum(4, handler))
+}

stream/options.go

@@ -32,6 +32,14 @@ func WithFinalBlocksOnly() Option {
 	}
 }
 
+// WithoutPartialBlocks makes the stream leave out the partial blocks of the hub,
+// receiving each block once, complete.
+func WithoutPartialBlocks() Option {
+	return func(s *Stream) {
+		s.withoutPartialBlocks = true
+	}
+}
+
 func WithCustomStepTypeFilter(step bstream.StepType) Option {
 	return func(s *Stream) {
 		s.customStepTypeFilter = &step

stream/stream.go

@@ -36,6 +36,7 @@ type Stream struct {
 	mergedBlocksBundleSize uint64
 
 	finalBlocksOnly      bool
+	withoutPartialBlocks bool
 	customStepTypeFilter *bstream.StepType
 
 	logger *zap.Logger
@@ -61,6 +62,10 @@ func New(
 		option(s)
 	}
 
+	if s.withoutPartialBlocks {
+		s.liveSourceFactory = hub.WithoutPartials()
+	}
+
 	var fileSourceOptions []bstream.FileSourceOption
 	if s.mergedBlocksBundleSize != 0 {
 		fileSourceOptions = append(fileSourceOptions, bstream.FileSourceWithBundleSize(s.mergedBlocksBundleSize))