diff --git a/CHANGELOG.md b/CHANGELOG.md index f783232..d2b648c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ 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`. ### Fixed @@ -20,9 +24,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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 +- 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/filesource.go b/filesource.go index 2e279c8..4ae979e 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,38 @@ func (g *FileSourceFactory) SourceThroughCursor(start uint64, cursor *Cursor, h ) } +// 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(ctx context.Context, idSuffix string, blockNum uint64) (bool, error) { + if g.forkedBlocksStore == nil { + return false, nil + } + + found := false + err := g.forkedBlocksStore.Walk(ctx, fmt.Sprintf("%010d", blockNum), func(filename string) error { + oneBlockFile, err := NewOneBlockFile(filename) + if err != nil { + return nil + } + if oneBlockFile.Num == blockNum && 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/filesource_test.go b/filesource_test.go index c8cbda7..9348b13 100644 --- a/filesource_test.go +++ b/filesource_test.go @@ -16,6 +16,7 @@ package bstream import ( "bytes" + "context" "fmt" "testing" "time" @@ -455,3 +456,38 @@ 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) + + 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(ctx, TruncateBlockID(forkedID), 151) + require.NoError(t, err) + assert.False(t, found, "another block holds that height") + + 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(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 c4a9c2a..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) @@ -80,6 +84,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(ctx context.Context, idSuffix string, blockNum 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..588d9ce 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. @@ -121,6 +138,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 +163,112 @@ 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(context.Background(), 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. +// +// 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 { + 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 + } + + if forked != nil { + 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", + 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) +} + +// 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 ef28fd6..8a1d94c 100644 --- a/joiningsource_test.go +++ b/joiningsource_test.go @@ -15,7 +15,9 @@ package bstream import ( + "context" "errors" + "fmt" "testing" "time" @@ -286,3 +288,179 @@ 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 + + headAfterCatchUp uint64 + catchUpAt time.Time +} + +func (t *testLiveKnower) LowestBlockNum() uint64 { return t.lowest } + +// 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] +} + +// testForkedKnower is a file source factory that also answers what the forked-blocks +// store holds. +type testForkedKnower struct { + *TestSourceFactory + forkedBlocks map[string]bool // "-" + err error +} + +func (t *testForkedKnower) HasForkedBlock(ctx context.Context, idSuffix string, blockNum uint64) (bool, error) { + if t.err != nil { + return false, t.err + } + return t.forkedBlocks[fmt.Sprintf("%d-%s", blockNum, idSuffix)], nil +} + +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" + + 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 + headAfterCatchUp uint64 + catchUpAfter time.Duration + forked map[string]bool + forkedErr error + expectErrror bool + expectAboveHead 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, 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", + 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{"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", + 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}, + }, + headAfterCatchUp: test.headAfterCatchUp, + catchUpAt: time.Now().Add(test.catchUpAfter), + } + file := &testForkedKnower{ + TestSourceFactory: NewTestSourceFactory(), + forkedBlocks: test.forked, + err: test.forkedErr, + } + + 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) + return + } + assert.NoError(t, err) + }) + } +} 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()} }