From 775b59feeda3cef049b958b9e7fb83ed6562ca79 Mon Sep 17 00:00:00 2001 From: Matthieu Vachon Date: Fri, 9 Oct 2026 15:22:06 -0400 Subject: [PATCH] Surface index store errors in block_indexer FindNextUnindexed and BlockIndexer.Add used to swallow dstore errors (FileExists, WalkFrom, WriteObject), which made index-builder restart indexing from the start block on transient S3 errors instead of surfacing the failure. Both now return an error; Add leaves currentIndex unchanged on a write failure so a retry stays consistent. writeIndex also backs off between retry attempts and no longer leaks a context per attempt. --- CHANGELOG.md | 2 + transform/block_indexer.go | 74 +++++++++++++++++++++------------ transform/block_indexer_test.go | 50 +++++++++++++++++++++- transform/testing.go | 2 +- 4 files changed, 98 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ce6176..4b3a801 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,9 +30,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - The hub no longer reads the `SOURCE_CHAN_SIZE` environment variable; set `hub.SubscriptionMaxBufferedBlocks` instead. The default goes from `100` to `10000` blocks: on fast chains, a burst of live blocks after a short pause from the live source filled 100 slots before the consumer could send the first block, closing healthy subscriptions. A consumer that cannot keep up is now caught by `SubscriptionCatchUpTimeout` instead. - `OneBlockFile.CanonicalName` no longer includes the LIB number (`num-id-prevID` instead of `num-id-prevID-lib`). Readers whose view of the LIB differs, as on Polygon PoS / Amoy, write the same block with different LIB numbers, and these copies used to be seen as different blocks. - `DBinBlockReader.Read` no longer copies the block payload out of the message it decodes: `Payload.Value` (and the legacy `PayloadBuffer`) now point into the buffer the message was read into. Reading merged blocks is 30-47% faster and allocates half as much. +- BREAKING: `transform.FindNextUnindexed` and `(*transform.BlockIndexer).Add` now return an `error`. Both used to silently swallow index store errors (e.g. a transient S3 `FileExists`/`WalkFrom`/write failure), which made `index-builder` restart indexing from `--index-builder-start-block` instead of resuming. Callers must now handle the returned error. ### Fixed +- `(*transform.BlockIndexer).Add` no longer swallows a failed index write with a warning and moves on to the next range, which left permanent holes in the index; the error now propagates to the caller and `currentIndex` is left unchanged so a retry does not lose data. `writeIndex` also waits with a backoff between retry attempts instead of retrying back-to-back, and no longer leaks a context per attempt until the whole call returns. - A live stream starting at a block number with no block (a missed or skipped slot) now starts at the next block instead of waiting for the merged bundle containing it. - `hub.ForkableHub` no longer breaks when its one-block bootstrap races with the merger deleting the files of a bundle it just merged. Concretely: bootstrap retries a few times when the listing comes back with a gap or a listed file is gone by the time it is opened (both expected during that race), a failed bootstrap no longer leaves the forkable holding a bogus LIB it then can't reconcile with live blocks (the hub now only swaps in the forkable it built once bootstrap actually succeeds, falling back to a fresh one otherwise), and `cannot link block after reconnection` is only fatal once the hub has actually been ready — before that, an unlinkable block is handled like any other, already subject to `WithMaxConsecutiveUnlinkableBlocks`. Previously a consumer starting or restarting during that race could see the hub die or never become ready. - The hub no longer walks the one-block store for a block it already has below LIB. With redundant live sources, a source re-sending a final block that the hub still keeps below LIB (the lowest one, whose parent was already pruned) triggered a store lookup and a `block not linkable after one-block lookup` warning, even though the forkable drops blocks below LIB anyway. The hub now compares against LIB instead of the lowest kept block. `Forkable.LIBNum` is added for this. diff --git a/transform/block_indexer.go b/transform/block_indexer.go index 4be221c..9f001d7 100644 --- a/transform/block_indexer.go +++ b/transform/block_indexer.go @@ -79,22 +79,28 @@ func WithDefinedStartBlock(startBlock uint64) Option { } } -func FindNextUnindexed(ctx context.Context, startBlockNum uint64, possibleIndexSizes []uint64, shortName string, store dstore.Store) (next uint64) { +func FindNextUnindexed(ctx context.Context, startBlockNum uint64, possibleIndexSizes []uint64, shortName string, store dstore.Store) (uint64, error) { if startBlockNum < bstream.GetProtocolFirstStreamableBlock { startBlockNum = bstream.GetProtocolFirstStreamableBlock } + var next uint64 var found bool for _, size := range possibleIndexSizes { base := lowBoundary(startBlockNum, size) // we want to start - if exists, _ := store.FileExists(ctx, toIndexFilename(size, base, shortName)); exists { + filename := toIndexFilename(size, base, shortName) + exists, err := store.FileExists(ctx, filename) + if err != nil { + return 0, fmt.Errorf("checking if index file %q exists: %w", filename, err) + } + if exists { next = base + size found = true break } } if !found { - return startBlockNum + return startBlockNum, nil } sizes := make(map[uint64]bool) @@ -102,8 +108,7 @@ func FindNextUnindexed(ctx context.Context, startBlockNum uint64, possibleIndexS sizes[size] = true } startFrom := fmt.Sprintf("%010d", next) - var skippedCount int - store.WalkFrom(ctx, "", startFrom, func(filename string) (err error) { + if err := store.WalkFrom(ctx, "", startFrom, func(filename string) (err error) { size, blockNum, short, err := parseIndexFilename(filename) if err != nil { zlog.Warn("parsing index files", zap.Error(err), zap.String("filename", filename)) @@ -119,12 +124,13 @@ func FindNextUnindexed(ctx context.Context, startBlockNum uint64, possibleIndexS if blockNum <= next && end > next { next = end zlog.Debug("skipping to next range...", zap.Uint64("next", next), zap.Uint64("index_size", size), zap.String("index_shortname", shortName)) - skippedCount++ } return nil - }) + }); err != nil { + return 0, fmt.Errorf("walking index files from %q: %w", startFrom, err) + } - return + return next, nil } // String returns a summary of the current BlockIndexer @@ -137,7 +143,7 @@ func (i *BlockIndexer) String() string { // Add will populate the BlockIndexer's currentIndex // by adding the specified BlockNum to the bitmaps identified with the provided keys -func (i *BlockIndexer) Add(keys []string, blockNum uint64) { +func (i *BlockIndexer) Add(keys []string, blockNum uint64) error { // init lower bound if i.currentIndex == nil { switch { @@ -156,14 +162,14 @@ func (i *BlockIndexer) Add(keys []string, blockNum uint64) { default: zlog.Warn("couldn't determine boundary for block", zap.Uint64("blk_num", blockNum)) - return + return nil } } // upper bound reached if blockNum >= i.currentIndex.lowBlockNum+i.indexSize { if err := i.writeIndex(); err != nil { - zlog.Warn("couldn't write index", zap.Error(err)) + return fmt.Errorf("writing index: %w", err) } lb := lowBoundary(blockNum, i.indexSize) i.currentIndex = NewBlockIndex(lb, i.indexSize) @@ -172,6 +178,7 @@ func (i *BlockIndexer) Add(keys []string, blockNum uint64) { for _, key := range keys { i.currentIndex.add(key, blockNum) } + return nil } // writeIndex writes the BlockIndexer's currentIndex to a file in the active dstore.Store @@ -190,28 +197,41 @@ func (i *BlockIndexer) writeIndex() error { attempt := 0 for { + attempt++ + ctx, cancel := context.WithTimeout(context.Background(), i.indexOpsTimeout) - defer cancel() - - if err = i.store.WriteObject(ctx, filename, bytes.NewReader(data)); err != nil { - attempt++ - if i.maxAttempts > 0 && attempt >= i.maxAttempts { - return fmt.Errorf("cannot write file to store after %d attempts: %w", attempt, err) - } - zlog.Warn("cannot write index file to store, retrying", - zap.String("filename", filename), - zap.Int("attempt", attempt), - zap.Int("max_attempts", i.maxAttempts), - zap.Error(err), - ) - } else { + err = i.store.WriteObject(ctx, filename, bytes.NewReader(data)) + cancel() + + if err == nil { zlog.Info("wrote file to store", zap.String("filename", filename), zap.Uint64("low_block_num", i.currentIndex.lowBlockNum), ) - break + return nil } + + if i.maxAttempts > 0 && attempt >= i.maxAttempts { + return fmt.Errorf("cannot write file to store after %d attempts: %w", attempt, err) + } + + zlog.Warn("cannot write index file to store, retrying", + zap.String("filename", filename), + zap.Int("attempt", attempt), + zap.Int("max_attempts", i.maxAttempts), + zap.Error(err), + ) + + time.Sleep(writeIndexRetryBackoff(attempt)) } +} - return nil +// writeIndexRetryBackoff returns the delay to wait before the next write attempt, +// growing linearly with the attempt number and capped at 10s. +func writeIndexRetryBackoff(attempt int) time.Duration { + backoff := time.Duration(attempt) * 500 * time.Millisecond + if backoff > 10*time.Second { + backoff = 10 * time.Second + } + return backoff } diff --git a/transform/block_indexer_test.go b/transform/block_indexer_test.go index 298d5c0..005a269 100644 --- a/transform/block_indexer_test.go +++ b/transform/block_indexer_test.go @@ -2,6 +2,7 @@ package transform import ( "context" + "errors" "fmt" "io" "testing" @@ -69,12 +70,38 @@ func Test_FindNextUnindexed(t *testing.T) { indexStore.SetFile(name, nil) } ctx := context.Background() - next := FindNextUnindexed(ctx, test.startBlock, test.indexSizes, "test", indexStore) + next, err := FindNextUnindexed(ctx, test.startBlock, test.indexSizes, "test", indexStore) + require.NoError(t, err) assert.EqualValues(t, int(test.expectNext), int(next)) }) } } +func Test_FindNextUnindexed_FileExistsError(t *testing.T) { + indexStore := dstore.NewMockStore(nil) + boom := errors.New("boom") + indexStore.FileExistsFunc = func(ctx context.Context, base string) (bool, error) { + return false, boom + } + + _, err := FindNextUnindexed(context.Background(), 0, []uint64{10000}, "test", indexStore) + require.Error(t, err) + assert.ErrorIs(t, err, boom) +} + +func Test_FindNextUnindexed_WalkFromError(t *testing.T) { + indexStore := dstore.NewMockStore(nil) + indexStore.SetFile("0000020000.10000.test.idx", nil) + boom := errors.New("boom") + indexStore.WalkFromFunc = func(ctx context.Context, prefix, startingPoint string, f func(filename string) error) error { + return boom + } + + _, err := FindNextUnindexed(context.Background(), 22222, []uint64{10000}, "test", indexStore) + require.Error(t, err) + assert.ErrorIs(t, err, boom) +} + func TestBlockIndexer_String(t *testing.T) { indexStore := dstore.NewMockStore(func(base string, f io.Reader) error { return nil @@ -149,7 +176,7 @@ func TestBlockIndexer_writeIndex(t *testing.T) { // spawn an indexer and feed it indexer := NewBlockIndexer(indexStore, test.indexSize, "test") for blockNum, keys := range test.kv { - indexer.Add(keys, blockNum) + require.NoError(t, indexer.Add(keys, blockNum)) } // write the index to dstore @@ -167,3 +194,22 @@ func TestBlockIndexer_writeIndex(t *testing.T) { }) } } + +func TestBlockIndexer_Add_WriteError(t *testing.T) { + boom := errors.New("boom") + indexStore := dstore.NewMockStore(func(base string, f io.Reader) error { + return boom + }) + + indexer := NewBlockIndexer(indexStore, 10, "test", WithMaxAttempts(1)) + + require.NoError(t, indexer.Add([]string{"aaaa"}, 10)) + + err := indexer.Add([]string{"bbbb"}, 20) + require.Error(t, err) + assert.ErrorIs(t, err, boom) + + // currentIndex must not have advanced to the next range on write failure + require.NotNil(t, indexer.currentIndex) + assert.EqualValues(t, 10, indexer.currentIndex.lowBlockNum) +} diff --git a/transform/testing.go b/transform/testing.go index 4dda8d3..66b0417 100644 --- a/transform/testing.go +++ b/transform/testing.go @@ -32,7 +32,7 @@ func testMockstoreWithFiles(t *testing.T, blocks []map[uint64][]string, indexSiz for _, blk := range blocks { // feed the indexer for k, v := range blk { - indexer.Add(v, k) + require.NoError(t, indexer.Add(v, k)) } }