diff --git a/CHANGELOG.md b/CHANGELOG.md index 172165d..fd7c245 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/filesource.go b/filesource.go index 33b4517..2e279c8 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(), @@ -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 } diff --git a/filesource_test.go b/filesource_test.go index edcb63e..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( @@ -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 diff --git a/hub/hub.go b/hub/hub.go index 7b223ca..9f97fe0 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,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 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..a39477d 100644 --- a/registry.go +++ b/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 } 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 }