From f45ff841df213e02eaf859dddfba35461b72a3eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 14:49:15 -0400 Subject: [PATCH 1/4] Support configurable merged-blocks bundle size Adds DefaultMergedBlocksBundleSize (process-wide default, like GetProtocolFirstStreamableBlock) and stream.WithMergedBlocksBundleSize for per-stream override. FileSource now errors out when a file contains blocks beyond the configured bundle size. Hub bootstrap rounding follows the configured size. Co-Authored-By: Claude Fable 5 --- filesource.go | 6 +++- filesource_test.go | 79 ++++++++++++++++++++++++++++++++++++++++++ hub/hub.go | 9 +++-- hub/rounding_test.go | 38 ++++++++++++++++++++ registry.go | 7 ++++ stream/options.go | 10 ++++++ stream/options_test.go | 13 +++++++ stream/stream.go | 5 +++ 8 files changed, 163 insertions(+), 4 deletions(-) create mode 100644 hub/rounding_test.go create mode 100644 stream/options_test.go diff --git a/filesource.go b/filesource.go index 33b4517..62800a8 100644 --- a/filesource.go +++ b/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(), @@ -514,6 +514,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 } diff --git a/filesource_test.go b/filesource_test.go index edcb63e..eb0d3a9 100644 --- a/filesource_test.go +++ b/filesource_test.go @@ -213,6 +213,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 diff --git a/hub/hub.go b/hub/hub.go index 7b223ca..585c5c1 100644 --- a/hub/hub.go +++ b/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,17 @@ 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 { 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 diff --git a/hub/rounding_test.go b/hub/rounding_test.go new file mode 100644 index 0000000..950e6a8 --- /dev/null +++ b/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") +} diff --git a/registry.go b/registry.go index f5b696f..0a96c3c 100644 --- a/registry.go +++ b/registry.go @@ -7,6 +7,13 @@ 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) var NormalizeBlockID = func(in string) string { // some chains have block IDs that optionally start with 0x or are case insensitive return in } diff --git a/stream/options.go b/stream/options.go index daa8495..f32c2ac 100644 --- a/stream/options.go +++ b/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 diff --git a/stream/options_test.go b/stream/options_test.go new file mode 100644 index 0000000..68f2908 --- /dev/null +++ b/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) +} diff --git a/stream/stream.go b/stream/stream.go index 09e5022..5a6bc5f 100644 --- a/stream/stream.go +++ b/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 } From a7847c4d9394ca514457c0a082440c29f281ae30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 15:25:45 -0400 Subject: [PATCH 2/4] Add variable merged-blocks bundle size to CHANGELOG --- CHANGELOG.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 172165d..b669913 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,14 @@ 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). + +### 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. From 787469bfed7183d6c8422f9ceafb58c9c979d60d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Mon, 6 Jul 2026 18:31:07 -0400 Subject: [PATCH 3/4] Guard merged-blocks bundle size against 0 --- CHANGELOG.md | 1 + filesource.go | 2 ++ hub/hub.go | 2 ++ registry.go | 13 +++++++++++++ 4 files changed, 18 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b669913..6ddff2d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `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`. ### Changed diff --git a/filesource.go b/filesource.go index 62800a8..4950716 100644 --- a/filesource.go +++ b/filesource.go @@ -284,6 +284,8 @@ func NewFileSource( option(s) } + s.bundleSize = SanitizeBundleSize(s.bundleSize) + return s } diff --git a/hub/hub.go b/hub/hub.go index 585c5c1..9f97fe0 100644 --- a/hub/hub.go +++ b/hub/hub.go @@ -551,6 +551,8 @@ func (h *ForkableHub) reconnect(err error) { // 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 diff --git a/registry.go b/registry.go index 0a96c3c..a39477d 100644 --- a/registry.go +++ b/registry.go @@ -14,6 +14,19 @@ var GetMaxNormalLIBDistance = uint64(1000) // 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 } From 109c03dc9bd547dab514857143b07b9f0ed43ec8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Tue, 7 Jul 2026 09:25:24 -0400 Subject: [PATCH 4/4] Drain FileSource blocks before erroring on missing file FileSourceErrorOnMissingMergedBlocksFile called Shutdown() the moment the walker hit a missing file. Shutdown() fires Terminating() right away, so the consuming loop and the per-file reader goroutines abort mid-flight and discard blocks from files that were already read but not yet processed, truncating the output. Push the error through the ordered fileStream like the stop-block path does, so the consumer drains every queued block first and only then surfaces the missing-file error. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 4 ++++ filesource.go | 11 ++++++++++- filesource_test.go | 38 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 52 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ddff2d..fd7c245 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `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`. diff --git a/filesource.go b/filesource.go index 4950716..2e279c8 100644 --- a/filesource.go +++ b/filesource.go @@ -652,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 } diff --git a/filesource_test.go b/filesource_test.go index eb0d3a9..c8cbda7 100644 --- a/filesource_test.go +++ b/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(