Support configurable merged-blocks bundle size

CHANGELOG.md

@@ -8,6 +8,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `DefaultMergedBlocksBundleSize`: new package variable (default `100`) controlling the number of blocks per merged-blocks file assumed by readers when no explicit size is given. Like `GetProtocolFirstStreamableBlock`, it is meant to be set once at process startup.
+- `stream.WithMergedBlocksBundleSize`: new `stream` option to set the merged-blocks bundle size for a single stream (overrides the process-wide default; used by substreams tier2 which serves multiple chains at once).
+- `FileSource` now fails fast with a clear error when a merged-blocks file contains a block beyond the configured bundle size (store files bigger than the configured size).
+- `SanitizeBundleSize`: guards the merged-blocks math against a bundle size of `0` (misconfigured `DefaultMergedBlocksBundleSize` or `FileSourceWithBundleSize(0)`), which would otherwise divide-by-zero panic in the hub or loop forever in `FileSource`; falls back to `100`.
+
+### Fixed
+
+- `FileSource` with `FileSourceErrorOnMissingMergedBlocksFile` no longer truncates its output: on a missing file it now drains every already-read block through the ordered stream before surfacing the error, instead of calling `Shutdown()` immediately (which aborted in-flight reader goroutines and discarded blocks).
+
+### Changed
+
+- `ForkableHub` bootstrap now rounds its lowest kept block down to the configured merged-blocks bundle size instead of a hardcoded `100`.
+
 - `BlockTimestampGate`: new gate that lets blocks through once a block's timestamp meets or exceeds a given `time.Time`, supporting both inclusive and exclusive gate types.
 - `hub.WithLogger`: new `ForkableHub` option to set the logger used by the hub (and, by default, propagated to its inner forkable). Previously the hub always logged under the package-level `bstream` logger, making lines such as `processing block` indistinguishable across components (relayer, firehose, tier1, ...). Callers should pass their component logger.
 

filesource.go

@@ -270,7 +270,7 @@ func NewFileSource(
 ) *FileSource {
 	s := &FileSource{
 		startBlockNum:             startBlockNum,
-		bundleSize:                100,
+		bundleSize:                DefaultMergedBlocksBundleSize,
 		blocksStore:               blocksStore,
 		fileStream:                make(chan *incomingBlocksFile, 1),
 		Shutter:                   shutter.New(),
@@ -284,6 +284,8 @@ func NewFileSource(
 		option(s)
 	}
 
+	s.bundleSize = SanitizeBundleSize(s.bundleSize)
+
 	return s
 }
 
@@ -514,6 +516,10 @@ func (s *FileSource) streamReader(blockReader *DBinBlockReader, prevLastBlockRea
 			continue
 		}
 
+		if blockNum >= incomingBlockFile.baseNum+s.bundleSize {
+			return fmt.Errorf("merged blocks file %q contains block %d, beyond the configured bundle size of %d blocks: the store most likely contains files bigger than the configured bundle size, check your merged-blocks-bundle-size configuration", incomingBlockFile.filename, blockNum, s.bundleSize)
+		}
+
 		if !incomingBlockFile.PassesFilter(blockNum) {
 			continue
 		}
@@ -646,7 +652,16 @@ func (s *FileSource) launchReader() {
 
 		if !exists {
 			if s.errorOutOnMissingFile {
-				s.Shutdown(fmt.Errorf("filesource: missing file %q", s.blocksStore.ObjectPath(baseFilename)))
+				// Send the error in-band instead of calling Shutdown() directly: Shutdown() fires
+				// Terminating() immediately, which makes the consuming loop and the per-file reader
+				// goroutines abort mid-flight, discarding blocks from files that were already read
+				// but not yet processed. Pushing the error through the ordered fileStream (like the
+				// stop-block path does) lets the consumer drain every queued block first and only
+				// then surface the missing-file error.
+				select {
+				case <-s.Terminating():
+				case s.fileStream <- &incomingBlocksFile{err: fmt.Errorf("filesource: missing file %q", s.blocksStore.ObjectPath(baseFilename))}:
+				}
 				return
 			}
 

filesource_test.go

@@ -156,6 +156,44 @@ func TestFileSource_Run(t *testing.T) {
 	fs.Shutdown(nil)
 }
 
+func TestFileSource_ErrorOnMissingFile_DrainsBeforeError(t *testing.T) {
+	bs := dstore.NewMockStore(nil)
+	bs.SetFile(base(0), testBlocks(
+		TestBlockWithNumbers("1a", "00", 1, 0),
+		TestBlockWithNumbers("2a", "1a", 2, 0),
+	))
+	bs.SetFile(base(100), testBlocks(
+		TestBlockWithNumbers("103a", "2a", 103, 0),
+		TestBlockWithNumbers("104a", "103a", 104, 0),
+	))
+	// base(200) is missing on purpose.
+
+	var processed []uint64
+	handler := HandlerFunc(func(blk *pbbstream.Block, obj any) error {
+		processed = append(processed, blk.Number)
+		return nil
+	})
+
+	fs := NewFileSource(bs, 1, handler, zlog, FileSourceErrorOnMissingMergedBlocksFile())
+
+	testDone := make(chan any)
+	go func() {
+		fs.Run()
+		close(testDone)
+	}()
+
+	select {
+	case <-testDone:
+	case <-time.After(time.Second):
+		t.Fatal("Test timeout")
+	}
+
+	// Every block from the available files must be processed before the missing-file error surfaces.
+	assert.Equal(t, []uint64{1, 2, 103, 104}, processed)
+	require.Error(t, fs.Err())
+	assert.Contains(t, fs.Err().Error(), "missing file")
+}
+
 func TestFileSourceFromCursor(t *testing.T) {
 	bs := dstore.NewMockStore(nil)
 	bs.SetFile(base(0), testBlocks(
@@ -213,6 +251,85 @@ func TestFileSourceFromCursor(t *testing.T) {
 	fs.Shutdown(nil)
 }
 
+func TestFileSource_Run_BundleSize1000(t *testing.T) {
+	bs := dstore.NewMockStore(nil)
+	bs.SetFile(base(0), testBlocks(
+		TestBlockWithNumbers("1a", "00", 1, 0),
+		TestBlockWithNumbers("2a", "1a", 2, 0),
+		TestBlockWithNumbers("998a", "2a", 998, 0),
+		TestBlockWithNumbers("999a", "998a", 999, 0),
+	))
+	bs.SetFile(base(1000), testBlocks(
+		TestBlockWithNumbers("1000a", "999a", 1000, 0),
+		TestBlockWithNumbers("1001a", "1000a", 1001, 0),
+	))
+
+	expectedBlocks := []uint64{1, 2, 998, 999, 1000, 1001}
+
+	testDone := make(chan any)
+	handlerCount := 0
+	handler := HandlerFunc(func(blk *pbbstream.Block, obj any) error {
+		require.Equal(t, expectedBlocks[handlerCount], blk.Number)
+		if handlerCount >= len(expectedBlocks)-1 {
+			close(testDone)
+		}
+		handlerCount++
+		return nil
+	})
+
+	fs := NewFileSource(bs, 1, handler, zlog, FileSourceWithBundleSize(1000))
+	go fs.Run()
+
+	select {
+	case <-testDone:
+		require.Equal(t, len(expectedBlocks), handlerCount)
+	case <-time.After(100 * time.Millisecond):
+		t.Error("Test timeout")
+	}
+	fs.Shutdown(nil)
+}
+
+func TestFileSource_DefaultMergedBlocksBundleSize(t *testing.T) {
+	prev := DefaultMergedBlocksBundleSize
+	defer func() { DefaultMergedBlocksBundleSize = prev }()
+
+	DefaultMergedBlocksBundleSize = 1000
+	fs := NewFileSource(dstore.NewMockStore(nil), 0, nil, zlog)
+	assert.Equal(t, uint64(1000), fs.bundleSize)
+
+	fs = NewFileSource(dstore.NewMockStore(nil), 0, nil, zlog, FileSourceWithBundleSize(200))
+	assert.Equal(t, uint64(200), fs.bundleSize, "explicit option wins over default")
+}
+
+// A file containing a block at or beyond baseNum+bundleSize means the store
+// holds bigger files than the configured bundle size: fail loudly instead of
+// streaming out-of-bundle blocks.
+func TestFileSource_Run_BundleSizeSmallerThanFiles(t *testing.T) {
+	bs := dstore.NewMockStore(nil)
+	bs.SetFile(base(0), testBlocks(
+		TestBlockWithNumbers("1a", "00", 1, 0),
+		TestBlockWithNumbers("150a", "1a", 150, 0), // beyond bundle [0,100)
+	))
+
+	handler := HandlerFunc(func(blk *pbbstream.Block, obj any) error { return nil })
+
+	fs := NewFileSource(bs, 1, handler, zlog)
+
+	testDone := make(chan struct{})
+	go func() {
+		fs.Run()
+		close(testDone)
+	}()
+	select {
+	case <-testDone:
+	case <-time.After(100 * time.Millisecond):
+		t.Error("Test timeout")
+	}
+
+	require.Error(t, fs.Err())
+	require.Contains(t, fs.Err().Error(), "beyond the configured bundle size")
+}
+
 func TestFileSource_lookupBlockIndex(t *testing.T) {
 	tests := []struct {
 		name                        string

hub/hub.go

@@ -305,7 +305,7 @@ func (h *ForkableHub) bootstrap() error {
 	if err != nil {
 		return fmt.Errorf("parsing filename: %w", err)
 	}
-	lowestBlockNum := substractAndRoundDownBlocks(refLibNum, uint64(h.keepFinalBlocks))
+	lowestBlockNum := substractAndRoundDownBlocks(refLibNum, uint64(h.keepFinalBlocks), bstream.DefaultMergedBlocksBundleSize)
 
 	oneBlocksAboveLibRef := make([]*pbbstream.Block, 0)
 	for _, filename := range sortedOneBlocksFiles {
@@ -547,14 +547,19 @@ func (h *ForkableHub) reconnect(err error) {
 	go liveSource.Run()
 }
 
-func substractAndRoundDownBlocks(blknum, sub uint64) uint64 {
+// substractAndRoundDownBlocks rounds down to a merged-blocks bundle boundary so
+// the hub's lowest buffered block lines up with a file boundary, letting the
+// joining source hand off from a merged-blocks file.
+func substractAndRoundDownBlocks(blknum, sub, bundleSize uint64) uint64 {
+	bundleSize = bstream.SanitizeBundleSize(bundleSize)
+
 	var out uint64
 	if blknum < sub {
 		out = 0
 	} else {
 		out = blknum - sub
 	}
-	out = out / 100 * 100
+	out = out / bundleSize * bundleSize
 
 	if out < bstream.GetProtocolFirstStreamableBlock {
 		return bstream.GetProtocolFirstStreamableBlock

hub/rounding_test.go

@@ -0,0 +1,38 @@
+package hub
+
+import (
+	"testing"
+
+	"github.com/streamingfast/bstream"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSubstractAndRoundDownBlocks(t *testing.T) {
+	prev := bstream.GetProtocolFirstStreamableBlock
+	defer func() { bstream.GetProtocolFirstStreamableBlock = prev }()
+	bstream.GetProtocolFirstStreamableBlock = 0
+
+	tests := []struct {
+		name       string
+		blknum     uint64
+		sub        uint64
+		bundleSize uint64
+		expect     uint64
+	}{
+		{"round to 100", 12345, 100, 100, 12200},
+		{"exact boundary 100", 12400, 100, 100, 12300},
+		{"underflow clamps to 0", 50, 100, 100, 0},
+		{"round to 1000", 12345, 100, 1000, 12000},
+		{"exact boundary 1000", 13000, 1000, 1000, 12000},
+		{"underflow clamps to 0 at 1000", 500, 1000, 1000, 0},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.expect, substractAndRoundDownBlocks(test.blknum, test.sub, test.bundleSize))
+		})
+	}
+
+	bstream.GetProtocolFirstStreamableBlock = 2
+	assert.Equal(t, uint64(2), substractAndRoundDownBlocks(50, 100, 100), "clamps to first streamable block")
+}

registry.go

@@ -7,6 +7,26 @@ package bstream
 // var GetBlockWriterHeaderLen int
 var GetProtocolFirstStreamableBlock = uint64(0)
 var GetMaxNormalLIBDistance = uint64(1000)
+
+// DefaultMergedBlocksBundleSize is the number of blocks per merged-blocks file
+// assumed by readers when no explicit bundle size is provided (see
+// FileSourceWithBundleSize). Like GetProtocolFirstStreamableBlock, it is meant
+// to be set once at process startup. It must match the size of the files
+// actually present in the merged-blocks store.
+var DefaultMergedBlocksBundleSize = uint64(100)
+
+// SanitizeBundleSize protects the merged-blocks math against a bundle size of 0,
+// which would otherwise cause an integer divide-by-zero panic (see
+// hub.substractAndRoundDownBlocks) or an infinite loop while walking merged
+// files. A 0 typically means DefaultMergedBlocksBundleSize was left unset by a
+// misconfigured process; we fall back to the standard 100-blocks bundle.
+func SanitizeBundleSize(bundleSize uint64) uint64 {
+	if bundleSize == 0 {
+		return 100
+	}
+	return bundleSize
+}
+
 var NormalizeBlockID = func(in string) string { // some chains have block IDs that optionally start with 0x or are case insensitive
 	return in
 }

stream/options.go

@@ -62,6 +62,16 @@ func WithStopBlock(stopBlockNum uint64) Option { //inclusive
 	}
 }
 
+// WithMergedBlocksBundleSize overrides the number of blocks per merged-blocks
+// file for this stream. When not set, bstream.DefaultMergedBlocksBundleSize
+// applies. The value must match the size of the files actually present in the
+// merged-blocks store.
+func WithMergedBlocksBundleSize(bundleSize uint64) Option {
+	return func(s *Stream) {
+		s.mergedBlocksBundleSize = bundleSize
+	}
+}
+
 func WithLiveSourceHandlerMiddleware(mw func(source bstream.Handler) bstream.Handler) Option {
 	return func(s *Stream) {
 		s.liveSourceHandlerMiddleware = mw

stream/options_test.go

@@ -0,0 +1,13 @@
+package stream
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWithMergedBlocksBundleSize(t *testing.T) {
+	s := &Stream{}
+	WithMergedBlocksBundleSize(1000)(s)
+	assert.Equal(t, uint64(1000), s.mergedBlocksBundleSize)
+}

stream/stream.go

@@ -33,6 +33,8 @@ type Stream struct {
 
 	blockIndexProvider bstream.BlockIndexProvider
 
+	mergedBlocksBundleSize uint64
+
 	finalBlocksOnly      bool
 	customStepTypeFilter *bstream.StepType
 
@@ -60,6 +62,9 @@ func New(
 	}
 
 	var fileSourceOptions []bstream.FileSourceOption
+	if s.mergedBlocksBundleSize != 0 {
+		fileSourceOptions = append(fileSourceOptions, bstream.FileSourceWithBundleSize(s.mergedBlocksBundleSize))
+	}
 	if s.stopBlockNum != 0 {
 		fileSourceOptions = append(fileSourceOptions, bstream.FileSourceWithStopBlock(s.stopBlockNum)) // more efficient than using our own
 	}