From 41a225210085f904a18235e688a5f200fc624b6d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 16 Aug 2026 19:08:16 -0400 Subject: [PATCH 1/4] Fail fast on a cursor no source can resolve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cursor block inside the live buffer's range that the buffer does not know made the hub decline the source, and the file source it fell back to waited for the merged file holding that block number — twenty minutes on Ethereum — only to fail there too. --- CHANGELOG.md | 3 + filesource.go | 33 +++++++++++ interfaces.go | 16 ++++++ joiningsource.go | 60 ++++++++++++++++++++ joiningsource_test.go | 125 ++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 237 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a5e3b27..f1ba3f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,12 +12,15 @@ 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). +- `CheckCursorResolvable`: reports whether a cursor names a block anything can still produce, against a `LiveBlockKnower` (the hub) and an optional `ForkedBlockKnower` (a `FileSourceFactory`). Both are new optional interfaces, implemented by `hub.ForkableHub` and `FileSourceFactory` respectively, so callers resolving cursors outside `JoiningSource` can make the same call. +- `FileSourceFactory.HasForkedBlock`: says whether the forked-blocks store holds a one-block file whose ID ends with a given suffix, over a block range. - `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 - Hub subscriptions with `with_partials=false` no longer stall on flash/partial-block chains. Previously every block with `PartialIndex != 0` (including the closing `LastPartial`) was dropped, so a no-partial subscriber only advanced on separate `PartialIndex==0` full blocks, which can lag the sealed head by tens of seconds. The subscription now drops only intermediate partials and delivers each `LastPartial` as a full block (partial markers cleared on a thin copy that shares the payload; the shared original block is never mutated). - `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). +- `JoiningSource` no longer leaves a stream silent when its cursor names a block nothing can produce. A cursor block inside the live buffer's range whose ID the buffer does not know (a corrupted or forged cursor, or one from a chain the process never saw) made the hub decline the source, and the file source it fell back to then waited for the merged-blocks file holding that block number — a whole bundle, some twenty minutes on Ethereum — before failing anyway. Such a cursor now fails immediately with `ErrResolveCursor`, which the `stream` package already surfaces as an invalid argument. Cursors on a fork the live buffer no longer holds are unaffected: the forked-blocks store is consulted before giving up, and a cursor below the live range still goes to the file source. ### Changed diff --git a/filesource.go b/filesource.go index 2e279c8..6274406 100644 --- a/filesource.go +++ b/filesource.go @@ -16,9 +16,11 @@ package bstream import ( "context" + "errors" "fmt" "io" "sort" + "strings" "sync/atomic" "time" @@ -200,6 +202,37 @@ func (g *FileSourceFactory) SourceThroughCursor(start uint64, cursor *Cursor, h ) } +// HasForkedBlock says whether the forked-blocks store holds a block whose ID ends with +// idSuffix, looking only at the files that could carry it. It is what tells a cursor +// sitting on a fork the live source no longer holds from one naming a block that never +// existed, without waiting for the merged files to catch up to it. +func (g *FileSourceFactory) HasForkedBlock(idSuffix string, fromBlockNum, toBlockNum uint64) (bool, error) { + if g.forkedBlocksStore == nil { + return false, nil + } + + found := false + err := g.forkedBlocksStore.WalkFrom(context.Background(), "", fmt.Sprintf("%010d", fromBlockNum), func(filename string) error { + oneBlockFile, err := NewOneBlockFile(filename) + if err != nil { + return nil + } + if oneBlockFile.Num > toBlockNum { + return dstore.StopIteration + } + if strings.HasSuffix(oneBlockFile.ID, idSuffix) { + found = true + return dstore.StopIteration + } + return nil + }) + if err != nil && !errors.Is(err, dstore.StopIteration) { + return false, err + } + + return found, nil +} + func NewFileSourceFromCursor( mergedBlocksStore dstore.Store, forkedBlocksStore dstore.Store, diff --git a/interfaces.go b/interfaces.go index c4a9c2a..6b36f28 100644 --- a/interfaces.go +++ b/interfaces.go @@ -80,6 +80,22 @@ type LowSourceLimitGetter interface { LowestBlockNum() uint64 } +// LiveBlockKnower is implemented by a live source factory that can say which block range +// it holds and whether a block ID is one of the blocks in it. Over that range the live +// source is authoritative: a block ID it does not know is on no chain it ever saw. +type LiveBlockKnower interface { + LowestBlockNum() uint64 + HeadNum() uint64 + GetBlockByHash(id string) *pbbstream.Block +} + +// ForkedBlockKnower is implemented by a file source factory that can say whether the +// forked-blocks store holds a block, which is the other place a cursor sitting on a fork +// can be resolved from once the live source no longer has it. +type ForkedBlockKnower interface { + HasForkedBlock(idSuffix string, fromBlockNum, toBlockNum uint64) (bool, error) +} + type SourceFactory func(h Handler) Source type SourceFromRefFactory func(startBlockRef BlockRef, h Handler) Source type SourceFromNumFactory func(startBlockNum uint64, h Handler) Source diff --git a/joiningsource.go b/joiningsource.go index 1f87ea5..b4e2f94 100644 --- a/joiningsource.go +++ b/joiningsource.go @@ -121,6 +121,10 @@ func (s *JoiningSource) run() error { s.lowestLiveBlockNum = lowestBlockGetter.LowestBlockNum() } + if err := s.checkCursorResolvable(); err != nil { + return err + } + fileSrc := s.tryGetSource(HandlerFunc(s.fileSourceHandler), s.fileSourceFactory) if fileSrc == nil { @@ -142,6 +146,62 @@ func (s *JoiningSource) run() error { } +func (s *JoiningSource) checkCursorResolvable() error { + live, ok := s.liveSourceFactory.(LiveBlockKnower) + if !ok { + return nil + } + forked, _ := s.fileSourceFactory.(ForkedBlockKnower) + + return CheckCursorResolvable(s.cursor, live, forked, s.logger) +} + +// CheckCursorResolvable says whether a cursor names a block that anything can still +// produce, and returns an ErrResolveCursor error when nothing can. +// +// The live source is authoritative over the range it holds: a cursor block inside that +// range whose ID it does not know is on no chain it ever saw. The one other place such a +// block can come from is the forked-blocks store — a live source restarted after the fork +// happened no longer holds it, while the store still does — so that one is asked before +// giving up. +// +// Both coming back empty is what makes a cursor unresolvable, and saying so here is what +// keeps the caller off the file source, which would otherwise wait for merged files that +// cannot contain that block: a whole bundle on a slow chain — 100 blocks, some twenty +// minutes on Ethereum — and then the same failure anyway. +func CheckCursorResolvable(cursor *Cursor, live LiveBlockKnower, forked ForkedBlockKnower, logger *zap.Logger) error { + if cursor.IsEmpty() || live == nil { + return nil + } + + lowest, head := live.LowestBlockNum(), live.HeadNum() + cursorBlockNum := cursor.Block.Num() + if lowest == 0 || head == 0 || cursorBlockNum < lowest || cursorBlockNum > head { + return nil + } + + if live.GetBlockByHash(cursor.Block.ID()) != nil { + return nil + } + + if forked != nil { + hasForkedBlock, err := forked.HasForkedBlock(TruncateBlockID(cursor.Block.ID()), cursor.LIB.Num(), cursorBlockNum) + if err != nil { + if logger != nil { + logger.Warn("cannot look up the cursor block in the forked blocks store, leaving the cursor to the file source", + zap.Stringer("cursor_block", cursor.Block), zap.Error(err)) + } + return nil + } + if hasForkedBlock { + return nil + } + } + + return fmt.Errorf("%w: block %s sits inside the live range [%d, %d], where neither the live buffer nor the forked blocks hold it", + ErrResolveCursor, cursor.Block, lowest, head) +} + func (s *JoiningSource) tryGetSource(handler Handler, factory ForkableSourceFactory) Source { if s.cursor != nil { if s.cursorIsTarget { diff --git a/joiningsource_test.go b/joiningsource_test.go index ef28fd6..290d69f 100644 --- a/joiningsource_test.go +++ b/joiningsource_test.go @@ -286,3 +286,128 @@ func TestJoiningSource_lowerLimitBackoff(t *testing.T) { assert.Equal(t, 3, liveSourceFactoryCalls) } + +// testLiveKnower is a live source factory that also answers what its buffer holds, the +// way the forkable hub does. +type testLiveKnower struct { + *TestSourceFactory + lowest uint64 + head uint64 + blocks map[string]*pbbstream.Block +} + +func (t *testLiveKnower) LowestBlockNum() uint64 { return t.lowest } +func (t *testLiveKnower) HeadNum() uint64 { return t.head } +func (t *testLiveKnower) GetBlockByHash(id string) *pbbstream.Block { + return t.blocks[id] +} + +// testForkedKnower is a file source factory that also answers what the forked-blocks +// store holds. +type testForkedKnower struct { + *TestSourceFactory + forkedIDSuffixes map[string]bool + err error +} + +func (t *testForkedKnower) HasForkedBlock(idSuffix string, from, to uint64) (bool, error) { + if t.err != nil { + return false, t.err + } + return t.forkedIDSuffixes[idSuffix], nil +} + +func TestJoiningSourceCheckCursorResolvable(t *testing.T) { + knownID := "00000000000000000000000000000000000000000000000000000000000000aa" + unknownID := "00000000000000000000000000000000000000000000000000000000000000bb" + + cursorAt := func(id string, num uint64) *Cursor { + return &Cursor{ + Step: StepNew, + Block: NewBlockRef(id, num), + HeadBlock: NewBlockRef(id, num), + LIB: NewBlockRef(knownID, 100), + } + } + + tests := []struct { + name string + cursor *Cursor + lowest uint64 + head uint64 + forked map[string]bool + forkedErr error + expectErrror bool + }{ + { + name: "no cursor", + cursor: nil, + lowest: 100, head: 200, + }, + { + name: "live buffer holds the cursor block", + cursor: cursorAt(knownID, 150), + lowest: 100, head: 200, + }, + { + name: "cursor block below the live buffer, left to the file source", + cursor: cursorAt(unknownID, 50), + lowest: 100, head: 200, + }, + { + name: "cursor block above the live head, left to the file source", + cursor: cursorAt(unknownID, 250), + lowest: 100, head: 200, + }, + { + name: "live buffer not ready, left to the file source", + cursor: cursorAt(unknownID, 150), + lowest: 0, head: 0, + }, + { + name: "unknown inside the live buffer, forked blocks hold it", + cursor: cursorAt(unknownID, 150), + lowest: 100, head: 200, + forked: map[string]bool{TruncateBlockID(unknownID): true}, + }, + { + name: "unknown inside the live buffer, forked blocks store fails", + cursor: cursorAt(unknownID, 150), + lowest: 100, + head: 200, + forkedErr: errTestMock, + }, + { + name: "unknown inside the live buffer and nowhere else", + cursor: cursorAt(unknownID, 150), + lowest: 100, head: 200, + expectErrror: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + live := &testLiveKnower{ + TestSourceFactory: NewTestSourceFactory(), + lowest: test.lowest, + head: test.head, + blocks: map[string]*pbbstream.Block{knownID: {Id: knownID, Number: 150}}, + } + file := &testForkedKnower{ + TestSourceFactory: NewTestSourceFactory(), + forkedIDSuffixes: test.forked, + err: test.forkedErr, + } + + s := NewJoiningSource(file, live, nil, 100, test.cursor, false, zlog) + + err := s.checkCursorResolvable() + if test.expectErrror { + require.Error(t, err) + assert.ErrorIs(t, err, ErrResolveCursor) + return + } + assert.NoError(t, err) + }) + } +} From 36ebf040b2a00572815be349b6c7f903524327d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 16 Aug 2026 19:57:28 -0400 Subject: [PATCH 2/4] Look the forked block up at its own height A one-block file is named after its block number, so scanning from the cursor's LIB up to it could match that ID suffix on a different block and call an unresolvable cursor resolvable. --- CHANGELOG.md | 2 +- filesource.go | 21 +++++++++++---------- filesource_test.go | 34 ++++++++++++++++++++++++++++++++++ interfaces.go | 2 +- joiningsource.go | 2 +- joiningsource_test.go | 20 ++++++++++++++------ 6 files changed, 62 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f1ba3f3..654d56a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `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). - `CheckCursorResolvable`: reports whether a cursor names a block anything can still produce, against a `LiveBlockKnower` (the hub) and an optional `ForkedBlockKnower` (a `FileSourceFactory`). Both are new optional interfaces, implemented by `hub.ForkableHub` and `FileSourceFactory` respectively, so callers resolving cursors outside `JoiningSource` can make the same call. -- `FileSourceFactory.HasForkedBlock`: says whether the forked-blocks store holds a one-block file whose ID ends with a given suffix, over a block range. +- `FileSourceFactory.HasForkedBlock`: says whether the forked-blocks store holds the block at a given number whose ID ends with a given suffix. - `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 diff --git a/filesource.go b/filesource.go index 6274406..be4f113 100644 --- a/filesource.go +++ b/filesource.go @@ -202,25 +202,26 @@ func (g *FileSourceFactory) SourceThroughCursor(start uint64, cursor *Cursor, h ) } -// HasForkedBlock says whether the forked-blocks store holds a block whose ID ends with -// idSuffix, looking only at the files that could carry it. It is what tells a cursor -// sitting on a fork the live source no longer holds from one naming a block that never -// existed, without waiting for the merged files to catch up to it. -func (g *FileSourceFactory) HasForkedBlock(idSuffix string, fromBlockNum, toBlockNum uint64) (bool, error) { +// HasForkedBlock says whether the forked-blocks store holds the block at blockNum whose +// ID ends with idSuffix. It is what tells a cursor sitting on a fork the live source no +// longer holds from one naming a block that never existed, without waiting for the merged +// files to catch up to it. +// +// One-block files are named after their block number, so the search is that one height: +// the same ID at another height is another block, and a suffix that matches there says +// nothing about the one being looked for. +func (g *FileSourceFactory) HasForkedBlock(idSuffix string, blockNum uint64) (bool, error) { if g.forkedBlocksStore == nil { return false, nil } found := false - err := g.forkedBlocksStore.WalkFrom(context.Background(), "", fmt.Sprintf("%010d", fromBlockNum), func(filename string) error { + err := g.forkedBlocksStore.Walk(context.Background(), fmt.Sprintf("%010d", blockNum), func(filename string) error { oneBlockFile, err := NewOneBlockFile(filename) if err != nil { return nil } - if oneBlockFile.Num > toBlockNum { - return dstore.StopIteration - } - if strings.HasSuffix(oneBlockFile.ID, idSuffix) { + if oneBlockFile.Num == blockNum && strings.HasSuffix(oneBlockFile.ID, idSuffix) { found = true return dstore.StopIteration } diff --git a/filesource_test.go b/filesource_test.go index c8cbda7..c39ee38 100644 --- a/filesource_test.go +++ b/filesource_test.go @@ -455,3 +455,37 @@ func TestFileSource_lookupBlockIndex_LiveFloor(t *testing.T) { assert.True(t, noMoreIndex) assert.Equal(t, uint64(400), baseBlock, "floor 0 disables the early stop") } + +// TestFileSourceFactory_HasForkedBlock covers the one lookup a cursor on a fork needs: +// the block at that exact height. One-block files are named after their block number, so +// the same ID suffix at another height is another block and must not answer for it. +func TestFileSourceFactory_HasForkedBlock(t *testing.T) { + forkedID := "00000000000000000000000000000000000000000000000000000000000000bb" + previousID := "00000000000000000000000000000000000000000000000000000000000000aa" + oneBlockFile := func(num uint64, id string) string { + return fmt.Sprintf("%010d-%s-%s-100-suffix", num, TruncateBlockID(id), TruncateBlockID(previousID)) + } + + forkedBlocksStore := dstore.NewMockStore(nil) + forkedBlocksStore.SetFile(oneBlockFile(149, forkedID), nil) + forkedBlocksStore.SetFile(oneBlockFile(150, forkedID), nil) + forkedBlocksStore.SetFile(oneBlockFile(151, previousID), nil) + + factory := NewFileSourceFactory(dstore.NewMockStore(nil), forkedBlocksStore, zlog) + + found, err := factory.HasForkedBlock(TruncateBlockID(forkedID), 150) + require.NoError(t, err) + assert.True(t, found, "the forked block at its own height") + + found, err = factory.HasForkedBlock(TruncateBlockID(forkedID), 151) + require.NoError(t, err) + assert.False(t, found, "another block holds that height") + + found, err = factory.HasForkedBlock(TruncateBlockID(forkedID), 152) + require.NoError(t, err) + assert.False(t, found, "no file at that height") + + found, err = NewFileSourceFactory(dstore.NewMockStore(nil), nil, zlog).HasForkedBlock(TruncateBlockID(forkedID), 150) + require.NoError(t, err) + assert.False(t, found, "no forked blocks store configured") +} diff --git a/interfaces.go b/interfaces.go index 6b36f28..ea591e6 100644 --- a/interfaces.go +++ b/interfaces.go @@ -93,7 +93,7 @@ type LiveBlockKnower interface { // forked-blocks store holds a block, which is the other place a cursor sitting on a fork // can be resolved from once the live source no longer has it. type ForkedBlockKnower interface { - HasForkedBlock(idSuffix string, fromBlockNum, toBlockNum uint64) (bool, error) + HasForkedBlock(idSuffix string, blockNum uint64) (bool, error) } type SourceFactory func(h Handler) Source diff --git a/joiningsource.go b/joiningsource.go index b4e2f94..68228cd 100644 --- a/joiningsource.go +++ b/joiningsource.go @@ -185,7 +185,7 @@ func CheckCursorResolvable(cursor *Cursor, live LiveBlockKnower, forked ForkedBl } if forked != nil { - hasForkedBlock, err := forked.HasForkedBlock(TruncateBlockID(cursor.Block.ID()), cursor.LIB.Num(), cursorBlockNum) + hasForkedBlock, err := forked.HasForkedBlock(TruncateBlockID(cursor.Block.ID()), cursorBlockNum) if err != nil { if logger != nil { logger.Warn("cannot look up the cursor block in the forked blocks store, leaving the cursor to the file source", diff --git a/joiningsource_test.go b/joiningsource_test.go index 290d69f..6ebc60e 100644 --- a/joiningsource_test.go +++ b/joiningsource_test.go @@ -16,6 +16,7 @@ package bstream import ( "errors" + "fmt" "testing" "time" @@ -306,15 +307,15 @@ func (t *testLiveKnower) GetBlockByHash(id string) *pbbstream.Block { // store holds. type testForkedKnower struct { *TestSourceFactory - forkedIDSuffixes map[string]bool - err error + forkedBlocks map[string]bool // "-" + err error } -func (t *testForkedKnower) HasForkedBlock(idSuffix string, from, to uint64) (bool, error) { +func (t *testForkedKnower) HasForkedBlock(idSuffix string, blockNum uint64) (bool, error) { if t.err != nil { return false, t.err } - return t.forkedIDSuffixes[idSuffix], nil + return t.forkedBlocks[fmt.Sprintf("%d-%s", blockNum, idSuffix)], nil } func TestJoiningSourceCheckCursorResolvable(t *testing.T) { @@ -368,7 +369,14 @@ func TestJoiningSourceCheckCursorResolvable(t *testing.T) { name: "unknown inside the live buffer, forked blocks hold it", cursor: cursorAt(unknownID, 150), lowest: 100, head: 200, - forked: map[string]bool{TruncateBlockID(unknownID): true}, + forked: map[string]bool{"150-" + TruncateBlockID(unknownID): true}, + }, + { + name: "forked blocks hold that ID at another height, which is another block", + cursor: cursorAt(unknownID, 150), + lowest: 100, head: 200, + forked: map[string]bool{"149-" + TruncateBlockID(unknownID): true}, + expectErrror: true, }, { name: "unknown inside the live buffer, forked blocks store fails", @@ -395,7 +403,7 @@ func TestJoiningSourceCheckCursorResolvable(t *testing.T) { } file := &testForkedKnower{ TestSourceFactory: NewTestSourceFactory(), - forkedIDSuffixes: test.forked, + forkedBlocks: test.forked, err: test.forkedErr, } From c931e9cefd45cd8fd649ef04d745920612274405 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 16 Aug 2026 20:17:52 -0400 Subject: [PATCH 3/4] Wait out a cursor the live source has not reached MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cursor above head names a block that exists, on an instance that has not got there yet — a client reconnecting to a fleet member a few blocks behind. Give it five seconds, then report a retryable failure rather than a cursor to discard. --- CHANGELOG.md | 3 ++ joiningsource.go | 73 +++++++++++++++++++++++++++++++++++++++++-- joiningsource_test.go | 64 +++++++++++++++++++++++++++++++------ stream/errors.go | 16 ++++++++++ stream/stream.go | 4 +++ 5 files changed, 147 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 654d56a..71f88d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ 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). +- `stream.ErrUnavailable` / `stream.NewErrUnavailable`: error type for a request this process could not serve yet, for servers to map to their transport's retryable status (`codes.Unavailable`). +- `ErrCursorAboveHead`, `CursorHeadWaitTimeout`: a cursor block above the live source's head is waited for (default 5s) before being reported. - `CheckCursorResolvable`: reports whether a cursor names a block anything can still produce, against a `LiveBlockKnower` (the hub) and an optional `ForkedBlockKnower` (a `FileSourceFactory`). Both are new optional interfaces, implemented by `hub.ForkableHub` and `FileSourceFactory` respectively, so callers resolving cursors outside `JoiningSource` can make the same call. - `FileSourceFactory.HasForkedBlock`: says whether the forked-blocks store holds the block at a given number whose ID ends with a given suffix. - `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`. @@ -24,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- A cursor above head is now retryable, in case we are lagging behind and another instance is already serving that block. - `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. diff --git a/joiningsource.go b/joiningsource.go index 68228cd..70c33b7 100644 --- a/joiningsource.go +++ b/joiningsource.go @@ -15,9 +15,11 @@ package bstream import ( + "context" "errors" "fmt" "sync" + "time" pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" @@ -27,6 +29,21 @@ import ( var stopSourceOnJoin = errors.New("stopping source on join") +// ErrCursorAboveHead is returned for a cursor naming a block above the live source's head +// that did not arrive within CursorHeadWaitTimeout. Nothing about it says the block does +// not exist — only that this process has not reached it — so it is meant to reach the +// client as a retryable failure, never as a bad cursor. +var ErrCursorAboveHead = errors.New("cursor block is above the live source's head") + +// CursorHeadWaitTimeout bounds how long a cursor block above the live source's head is +// waited for. It covers the lag between two instances of a fleet, which is seconds at +// most; a cursor still unreachable after it is reported as ErrCursorAboveHead. +var CursorHeadWaitTimeout = 5 * time.Second + +// cursorHeadWaitInterval is how often the live source's head is polled while waiting. The +// hub advances it on its own goroutine, so polling is what a caller outside it has. +var cursorHeadWaitInterval = 100 * time.Millisecond + // JoiningSource joins an irreversible-only source (file) to a fork-aware source close to HEAD (live) // 1) it tries to get the source from LiveSourceFactory (using startblock or cursor) // 2) if it can't, it will ask the FileSourceFactory for a source of those blocks. @@ -153,7 +170,7 @@ func (s *JoiningSource) checkCursorResolvable() error { } forked, _ := s.fileSourceFactory.(ForkedBlockKnower) - return CheckCursorResolvable(s.cursor, live, forked, s.logger) + return CheckCursorResolvable(context.Background(), s.cursor, live, forked, s.logger) } // CheckCursorResolvable says whether a cursor names a block that anything can still @@ -169,17 +186,30 @@ func (s *JoiningSource) checkCursorResolvable() error { // keeps the caller off the file source, which would otherwise wait for merged files that // cannot contain that block: a whole bundle on a slow chain — 100 blocks, some twenty // minutes on Ethereum — and then the same failure anyway. -func CheckCursorResolvable(cursor *Cursor, live LiveBlockKnower, forked ForkedBlockKnower, logger *zap.Logger) error { +// +// A cursor block above the live source's head is a different thing: nothing says the block +// does not exist, only that this process has not reached it, which is what a client +// reconnecting to an instance a few blocks behind its last one looks like. That one is +// given CursorHeadWaitTimeout to arrive, and reported as ErrCursorAboveHead — meant to +// reach the client as a retryable failure — rather than as a cursor no source can resolve. +func CheckCursorResolvable(ctx context.Context, cursor *Cursor, live LiveBlockKnower, forked ForkedBlockKnower, logger *zap.Logger) error { if cursor.IsEmpty() || live == nil { return nil } lowest, head := live.LowestBlockNum(), live.HeadNum() cursorBlockNum := cursor.Block.Num() - if lowest == 0 || head == 0 || cursorBlockNum < lowest || cursorBlockNum > head { + if lowest == 0 || head == 0 || cursorBlockNum < lowest { return nil } + if cursorBlockNum > head { + if err := waitForLiveHead(ctx, live, cursorBlockNum, logger); err != nil { + return err + } + head = live.HeadNum() + } + if live.GetBlockByHash(cursor.Block.ID()) != nil { return nil } @@ -202,6 +232,43 @@ func CheckCursorResolvable(cursor *Cursor, live LiveBlockKnower, forked ForkedBl ErrResolveCursor, cursor.Block, lowest, head) } +// waitForLiveHead gives the live source CursorHeadWaitTimeout to reach blockNum. +// +// A cursor above head is the normal shape of a client reconnecting to an instance that +// runs a little behind the one that served it — a fleet is rarely in lockstep — and the +// blocks it names do arrive, in the seconds it takes this process to catch up. Waiting +// them out is what keeps that from being reported as a bad cursor, which no client can +// act on: it would have to drop a cursor that was never wrong. +// +// What is left after the wait cannot be told apart from a cursor invented far above head, +// so it is reported as ErrCursorAboveHead for the caller to turn into a retryable failure. +func waitForLiveHead(ctx context.Context, live LiveBlockKnower, blockNum uint64, logger *zap.Logger) error { + deadline := time.After(CursorHeadWaitTimeout) + ticker := time.NewTicker(cursorHeadWaitInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-deadline: + head := live.HeadNum() + if head >= blockNum { + return nil + } + if logger != nil { + logger.Info("cursor block is above the live source's head, which did not reach it in time", + zap.Uint64("cursor_block_num", blockNum), zap.Uint64("live_head_num", head), zap.Duration("waited", CursorHeadWaitTimeout)) + } + return fmt.Errorf("%w: block %d is above the live source's head at %d", ErrCursorAboveHead, blockNum, head) + case <-ticker.C: + if live.HeadNum() >= blockNum { + return nil + } + } + } +} + func (s *JoiningSource) tryGetSource(handler Handler, factory ForkableSourceFactory) Source { if s.cursor != nil { if s.cursorIsTarget { diff --git a/joiningsource_test.go b/joiningsource_test.go index 6ebc60e..1ea86b2 100644 --- a/joiningsource_test.go +++ b/joiningsource_test.go @@ -295,10 +295,21 @@ type testLiveKnower struct { lowest uint64 head uint64 blocks map[string]*pbbstream.Block + + headAfterCatchUp uint64 + catchUpAt time.Time } func (t *testLiveKnower) LowestBlockNum() uint64 { return t.lowest } -func (t *testLiveKnower) HeadNum() uint64 { return t.head } + +// HeadNum reports headAfterCatchUp once catchUpAt has passed, standing in for a live +// source that is behind and catching up while the check waits on it. +func (t *testLiveKnower) HeadNum() uint64 { + if t.headAfterCatchUp != 0 && time.Now().After(t.catchUpAt) { + return t.headAfterCatchUp + } + return t.head +} func (t *testLiveKnower) GetBlockByHash(id string) *pbbstream.Block { return t.blocks[id] } @@ -319,6 +330,11 @@ func (t *testForkedKnower) HasForkedBlock(idSuffix string, blockNum uint64) (boo } func TestJoiningSourceCheckCursorResolvable(t *testing.T) { + // the wait on a lagging live source is bounded by this; keep the test quick + previousTimeout := CursorHeadWaitTimeout + CursorHeadWaitTimeout = 300 * time.Millisecond + defer func() { CursorHeadWaitTimeout = previousTimeout }() + knownID := "00000000000000000000000000000000000000000000000000000000000000aa" unknownID := "00000000000000000000000000000000000000000000000000000000000000bb" @@ -332,13 +348,16 @@ func TestJoiningSourceCheckCursorResolvable(t *testing.T) { } tests := []struct { - name string - cursor *Cursor - lowest uint64 - head uint64 - forked map[string]bool - forkedErr error - expectErrror bool + name string + cursor *Cursor + lowest uint64 + head uint64 + headAfterCatchUp uint64 + catchUpAfter time.Duration + forked map[string]bool + forkedErr error + expectErrror bool + expectAboveHead bool }{ { name: "no cursor", @@ -356,9 +375,25 @@ func TestJoiningSourceCheckCursorResolvable(t *testing.T) { lowest: 100, head: 200, }, { - name: "cursor block above the live head, left to the file source", + name: "cursor block above the live head, reached while waiting", + cursor: cursorAt(knownID, 250), + lowest: 100, head: 200, + headAfterCatchUp: 260, + catchUpAfter: 50 * time.Millisecond, + }, + { + name: "cursor block above the live head, never reached", cursor: cursorAt(unknownID, 250), lowest: 100, head: 200, + expectAboveHead: true, + }, + { + name: "cursor block reached while waiting, and unknown there", + cursor: cursorAt(unknownID, 250), + lowest: 100, head: 200, + headAfterCatchUp: 260, + catchUpAfter: 50 * time.Millisecond, + expectErrror: true, }, { name: "live buffer not ready, left to the file source", @@ -399,7 +434,11 @@ func TestJoiningSourceCheckCursorResolvable(t *testing.T) { TestSourceFactory: NewTestSourceFactory(), lowest: test.lowest, head: test.head, - blocks: map[string]*pbbstream.Block{knownID: {Id: knownID, Number: 150}}, + blocks: map[string]*pbbstream.Block{ + knownID: {Id: knownID, Number: 150}, + }, + headAfterCatchUp: test.headAfterCatchUp, + catchUpAt: time.Now().Add(test.catchUpAfter), } file := &testForkedKnower{ TestSourceFactory: NewTestSourceFactory(), @@ -410,6 +449,11 @@ func TestJoiningSourceCheckCursorResolvable(t *testing.T) { s := NewJoiningSource(file, live, nil, 100, test.cursor, false, zlog) err := s.checkCursorResolvable() + if test.expectAboveHead { + require.Error(t, err) + assert.ErrorIs(t, err, ErrCursorAboveHead) + return + } if test.expectErrror { require.Error(t, err) assert.ErrorIs(t, err, ErrResolveCursor) diff --git a/stream/errors.go b/stream/errors.go index 4465f87..2950a6d 100644 --- a/stream/errors.go +++ b/stream/errors.go @@ -19,4 +19,20 @@ func (e *ErrInvalidArg) Error() string { return e.message } +// ErrUnavailable is a request this process could not serve yet, for servers to map to +// their transport's retryable status (gRPC codes.Unavailable). +type ErrUnavailable struct { + message string +} + +func NewErrUnavailable(m string, args ...any) *ErrUnavailable { + return &ErrUnavailable{ + message: fmt.Sprintf(m, args...), + } +} + +func (e *ErrUnavailable) Error() string { + return e.message +} + var ErrStopBlockReached = errors.New("stop block reached") diff --git a/stream/stream.go b/stream/stream.go index 5a6bc5f..a9f8f07 100644 --- a/stream/stream.go +++ b/stream/stream.go @@ -109,6 +109,10 @@ func (s *Stream) Run(ctx context.Context) error { source.Run() if err := source.Err(); err != nil { s.logger.Debug("source shutting down", zap.Error(err)) + if errors.Is(err, bstream.ErrCursorAboveHead) { + // retryable in case our head is lagging + return NewErrUnavailable("%s", err.Error()) + } if errors.Is(err, bstream.ErrResolveCursor) { return &ErrInvalidArg{message: err.Error()} } From 11b15285a176f23f92de56f5f07a9aa410bb654b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Mon, 17 Aug 2026 14:20:12 -0400 Subject: [PATCH 4/4] add ctx to HasForkedBlock to prevent leaking blocked filewalker --- filesource.go | 4 ++-- filesource_test.go | 10 ++++++---- interfaces.go | 8 ++++++-- joiningsource.go | 2 +- joiningsource_test.go | 3 ++- 5 files changed, 17 insertions(+), 10 deletions(-) diff --git a/filesource.go b/filesource.go index be4f113..4ae979e 100644 --- a/filesource.go +++ b/filesource.go @@ -210,13 +210,13 @@ func (g *FileSourceFactory) SourceThroughCursor(start uint64, cursor *Cursor, h // One-block files are named after their block number, so the search is that one height: // the same ID at another height is another block, and a suffix that matches there says // nothing about the one being looked for. -func (g *FileSourceFactory) HasForkedBlock(idSuffix string, blockNum uint64) (bool, error) { +func (g *FileSourceFactory) HasForkedBlock(ctx context.Context, idSuffix string, blockNum uint64) (bool, error) { if g.forkedBlocksStore == nil { return false, nil } found := false - err := g.forkedBlocksStore.Walk(context.Background(), fmt.Sprintf("%010d", blockNum), func(filename string) error { + err := g.forkedBlocksStore.Walk(ctx, fmt.Sprintf("%010d", blockNum), func(filename string) error { oneBlockFile, err := NewOneBlockFile(filename) if err != nil { return nil diff --git a/filesource_test.go b/filesource_test.go index c39ee38..9348b13 100644 --- a/filesource_test.go +++ b/filesource_test.go @@ -16,6 +16,7 @@ package bstream import ( "bytes" + "context" "fmt" "testing" "time" @@ -473,19 +474,20 @@ func TestFileSourceFactory_HasForkedBlock(t *testing.T) { factory := NewFileSourceFactory(dstore.NewMockStore(nil), forkedBlocksStore, zlog) - found, err := factory.HasForkedBlock(TruncateBlockID(forkedID), 150) + ctx := context.Background() + found, err := factory.HasForkedBlock(ctx, TruncateBlockID(forkedID), 150) require.NoError(t, err) assert.True(t, found, "the forked block at its own height") - found, err = factory.HasForkedBlock(TruncateBlockID(forkedID), 151) + found, err = factory.HasForkedBlock(ctx, TruncateBlockID(forkedID), 151) require.NoError(t, err) assert.False(t, found, "another block holds that height") - found, err = factory.HasForkedBlock(TruncateBlockID(forkedID), 152) + found, err = factory.HasForkedBlock(ctx, TruncateBlockID(forkedID), 152) require.NoError(t, err) assert.False(t, found, "no file at that height") - found, err = NewFileSourceFactory(dstore.NewMockStore(nil), nil, zlog).HasForkedBlock(TruncateBlockID(forkedID), 150) + found, err = NewFileSourceFactory(dstore.NewMockStore(nil), nil, zlog).HasForkedBlock(ctx, TruncateBlockID(forkedID), 150) require.NoError(t, err) assert.False(t, found, "no forked blocks store configured") } diff --git a/interfaces.go b/interfaces.go index ea591e6..0f5b182 100644 --- a/interfaces.go +++ b/interfaces.go @@ -14,7 +14,11 @@ package bstream -import pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" +import ( + "context" + + pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" +) type Shutterer interface { Shutdown(error) @@ -93,7 +97,7 @@ type LiveBlockKnower interface { // forked-blocks store holds a block, which is the other place a cursor sitting on a fork // can be resolved from once the live source no longer has it. type ForkedBlockKnower interface { - HasForkedBlock(idSuffix string, blockNum uint64) (bool, error) + HasForkedBlock(ctx context.Context, idSuffix string, blockNum uint64) (bool, error) } type SourceFactory func(h Handler) Source diff --git a/joiningsource.go b/joiningsource.go index 70c33b7..588d9ce 100644 --- a/joiningsource.go +++ b/joiningsource.go @@ -215,7 +215,7 @@ func CheckCursorResolvable(ctx context.Context, cursor *Cursor, live LiveBlockKn } if forked != nil { - hasForkedBlock, err := forked.HasForkedBlock(TruncateBlockID(cursor.Block.ID()), cursorBlockNum) + hasForkedBlock, err := forked.HasForkedBlock(ctx, TruncateBlockID(cursor.Block.ID()), cursorBlockNum) if err != nil { if logger != nil { logger.Warn("cannot look up the cursor block in the forked blocks store, leaving the cursor to the file source", diff --git a/joiningsource_test.go b/joiningsource_test.go index 1ea86b2..8a1d94c 100644 --- a/joiningsource_test.go +++ b/joiningsource_test.go @@ -15,6 +15,7 @@ package bstream import ( + "context" "errors" "fmt" "testing" @@ -322,7 +323,7 @@ type testForkedKnower struct { err error } -func (t *testForkedKnower) HasForkedBlock(idSuffix string, blockNum uint64) (bool, error) { +func (t *testForkedKnower) HasForkedBlock(ctx context.Context, idSuffix string, blockNum uint64) (bool, error) { if t.err != nil { return false, t.err }