Resolve or refuse a cursor without waiting on merged files

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.

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,

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")
+}

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

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 {