Surface index store errors in block_indexer

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.

transform/block_indexer.go

@@ -79,31 +79,36 @@ 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)
 	for _, size := range possibleIndexSizes {
 		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
 }

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

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