fix: confirm durable writes before advancing indexer state

indexer/beacon/bidcache.go

@@ -148,32 +148,40 @@ func (cache *blockBidCache) checkAndFlush() error {
 	// Calculate the cutoff slot - we'll flush bids older than this
 	cutoffSlot := cache.maxSlot - bidCacheRetainSlots
 
-	// Collect bids to flush (from minSlot to cutoffSlot)
+	// Collect bids to flush (from minSlot to cutoffSlot) without removing them yet
+	flushKeys := make([]bidCacheKey, 0)
 	bidsToFlush := make([]*dbtypes.BlockBid, 0)
 	for key, bid := range cache.bids {
 		if phase0.Slot(bid.Slot) < cutoffSlot {
+			flushKeys = append(flushKeys, key)
 			bidsToFlush = append(bidsToFlush, bid)
-			delete(cache.bids, key)
 		}
 	}
 
-	// Update minSlot
-	cache.minSlot = cutoffSlot
-
 	cache.cacheMutex.Unlock()
 
-	// Write to DB outside of lock
-	if len(bidsToFlush) > 0 {
-		err := db.RunDBTransaction(func(tx *sqlx.Tx) error {
-			return db.InsertBids(bidsToFlush, tx)
-		})
-		if err != nil {
-			cache.indexer.logger.Errorf("error flushing bids to db: %v", err)
-			return err
-		}
-		cache.indexer.logger.Debugf("flushed %d bids to DB (slots < %d)", len(bidsToFlush), cutoffSlot)
+	if len(bidsToFlush) == 0 {
+		return nil
+	}
+
+	// Write to DB before dropping the cached copies, so a failed write keeps the
+	// bids for the next flush instead of losing them.
+	err := db.RunDBTransaction(func(tx *sqlx.Tx) error {
+		return db.InsertBids(bidsToFlush, tx)
+	})
+	if err != nil {
+		cache.indexer.logger.Errorf("error flushing bids to db: %v", err)
+		return err
+	}
+
+	cache.cacheMutex.Lock()
+	for _, key := range flushKeys {
+		delete(cache.bids, key)
 	}
+	cache.minSlot = cutoffSlot
+	cache.cacheMutex.Unlock()
 
+	cache.indexer.logger.Debugf("flushed %d bids to DB (slots < %d)", len(bidsToFlush), cutoffSlot)
 	return nil
 }
 
@@ -192,14 +200,9 @@ func (cache *blockBidCache) flushAll() error {
 		bidsToFlush = append(bidsToFlush, bid)
 	}
 
-	// Clear the cache
-	cache.bids = make(map[bidCacheKey]*dbtypes.BlockBid, 64)
-	cache.minSlot = 0
-	cache.maxSlot = 0
-
 	cache.cacheMutex.Unlock()
 
-	// Write to DB outside of lock
+	// Write to DB before clearing the cache so a failed write does not lose the bids.
 	err := db.RunDBTransaction(func(tx *sqlx.Tx) error {
 		return db.InsertBids(bidsToFlush, tx)
 	})
@@ -208,6 +211,12 @@ func (cache *blockBidCache) flushAll() error {
 		return err
 	}
 
+	cache.cacheMutex.Lock()
+	cache.bids = make(map[bidCacheKey]*dbtypes.BlockBid, 64)
+	cache.minSlot = 0
+	cache.maxSlot = 0
+	cache.cacheMutex.Unlock()
+
 	cache.indexer.logger.Infof("flushed %d bids to DB on shutdown", len(bidsToFlush))
 	return nil
 }

indexer/beacon/bidcache_test.go

@@ -0,0 +1,80 @@
+package beacon
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/sirupsen/logrus"
+
+	"github.com/ethpandaops/dora/db"
+	"github.com/ethpandaops/dora/dbtypes"
+	"github.com/ethpandaops/dora/types"
+)
+
+func newBidTestDB(t *testing.T) {
+	t.Helper()
+	cfg := &types.DatabaseConfig{
+		Engine: "sqlite",
+		Sqlite: &types.SqliteDatabaseConfig{File: filepath.Join(t.TempDir(), "test.sqlite")},
+	}
+	db.MustInitDB(cfg)
+	if err := db.ApplyEmbeddedDbSchema(-2); err != nil {
+		t.Fatalf("apply schema: %v", err)
+	}
+	t.Cleanup(db.MustCloseDB)
+}
+
+// makeBidCache builds a cache whose slot span (0..20) exceeds the flush threshold,
+// so checkAndFlush flushes every bid below the cutoff (maxSlot - retain = 10).
+func makeBidCache() *blockBidCache {
+	cache := &blockBidCache{
+		indexer: &Indexer{logger: logrus.New()},
+		bids:    make(map[bidCacheKey]*dbtypes.BlockBid, 64),
+	}
+	for slot := 0; slot <= 20; slot++ {
+		bid := &dbtypes.BlockBid{
+			ParentRoot:   []byte{1},
+			ParentHash:   []byte{2},
+			BlockHash:    []byte{byte(slot)}, // unique per slot -> unique primary key
+			FeeRecipient: []byte{3},
+			Slot:         uint64(slot),
+		}
+		cache.bids[bidCacheKey{BlockHash: [32]byte{byte(slot)}}] = bid
+	}
+	cache.minSlot = 0
+	cache.maxSlot = 20
+	return cache
+}
+
+// TestCheckAndFlush verifies that bids are dropped from the cache only after a
+// successful db write, and kept for retry when the write fails.
+func TestCheckAndFlush(t *testing.T) {
+	newBidTestDB(t)
+
+	// Success: slots 0..9 are below the cutoff and flushed; slots 10..20 stay.
+	ok := makeBidCache()
+	if err := ok.checkAndFlush(); err != nil {
+		t.Fatalf("checkAndFlush: %v", err)
+	}
+	if len(ok.bids) != 11 {
+		t.Errorf("cached bids after flush = %d, want 11 (slots >= cutoff retained)", len(ok.bids))
+	}
+
+	// Failure: drop the table so the write fails; the bids must be kept.
+	if err := db.RunDBTransaction(func(tx *sqlx.Tx) error {
+		_, err := tx.Exec("DROP TABLE block_bids")
+		return err
+	}); err != nil {
+		t.Fatalf("drop table: %v", err)
+	}
+
+	fail := makeBidCache()
+	before := len(fail.bids)
+	if err := fail.checkAndFlush(); err == nil {
+		t.Fatal("expected error from checkAndFlush when the db write fails")
+	}
+	if len(fail.bids) != before {
+		t.Errorf("cached bids after failed flush = %d, want %d (retained)", len(fail.bids), before)
+	}
+}

indexer/beacon/finalization.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sort"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/ethpandaops/dora/blockdb"
@@ -575,6 +576,7 @@ func (indexer *Indexer) finalizeEpoch(epoch phase0.Epoch, justifiedRoot phase0.R
 	t1 = time.Now()
 
 	// save block bodies to blockdb
+	var blockDbWriteFailed atomic.Bool
 	if blockdb.GlobalBlockDb != nil && !indexer.disableBlockDbWrite {
 		var wg sync.WaitGroup
 		for _, block := range canonicalBlocks {
@@ -583,6 +585,7 @@ func (indexer *Indexer) finalizeEpoch(epoch phase0.Epoch, justifiedRoot phase0.R
 				defer wg.Done()
 				if err := b.writeToBlockDb(indexer.ctx); err != nil {
 					indexer.logger.Errorf("error writing block %v to blockdb: %v", b.Root.String(), err)
+					blockDbWriteFailed.Store(true)
 				}
 			}(block)
 		}
@@ -612,6 +615,14 @@ func (indexer *Indexer) finalizeEpoch(epoch phase0.Epoch, justifiedRoot phase0.R
 	}
 	t3dur := time.Since(t1)
 
+	// The unfinalized_blocks rows were deleted in the committed transaction above and
+	// the block bodies are about to be freed from the cache, so a failed blockdb write
+	// would leave those bodies unrecoverable. Trigger a resync of the epoch instead of
+	// advancing; the cached bodies stay in place for it to rewrite.
+	if blockDbWriteFailed.Load() {
+		return false, fmt.Errorf("failed writing epoch %v blocks to blockdb", epoch)
+	}
+
 	indexer.lastFinalizedEpoch = epoch + 1
 
 	// sleep 500 ms to give running UI threads time to fetch data from cache

indexer/execution/system_contracts/transaction_matcher.go

@@ -121,20 +121,15 @@ func (ds *transactionMatcher[MatchType]) runTransactionMatcher(indexerBlock uint
 			return err
 		}
 
-		ds.state.MatchHeight = matchTarget
-
 		if len(matches) > 0 {
-			// persist matches to the database
-			err := db.RunDBTransaction(func(tx *sqlx.Tx) error {
-				ds.options.persistMatches(tx, matches)
-
-				persistedHeight = matchTarget
-				return ds.persistState(tx)
-			})
-
-			if err != nil {
+			// Advance only if the matches and the new height persist together; a failed
+			// range must be retried, not skipped with its rows left unmatched.
+			if err := ds.persistMatchRange(matches, matchTarget); err != nil {
 				return err
 			}
+			persistedHeight = matchTarget
+		} else {
+			ds.state.MatchHeight = matchTarget
 		}
 	}
 
@@ -172,6 +167,25 @@ func (ds *transactionMatcher[_]) persistState(tx *sqlx.Tx) error {
 	return nil
 }
 
+// persistMatchRange persists the matches for a range together with the advanced
+// match height in a single transaction. The height is rolled back if either the
+// matches or the state fail to persist, so the range is retried on the next run.
+func (ds *transactionMatcher[MatchType]) persistMatchRange(matches []*MatchType, matchTarget uint64) error {
+	prevHeight := ds.state.MatchHeight
+	ds.state.MatchHeight = matchTarget
+
+	err := db.RunDBTransaction(func(tx *sqlx.Tx) error {
+		if err := ds.options.persistMatches(tx, matches); err != nil {
+			return err
+		}
+		return ds.persistState(tx)
+	})
+	if err != nil {
+		ds.state.MatchHeight = prevHeight
+	}
+	return err
+}
+
 // getFinalizedBlockNumber gets highest available finalized el block number
 // available means that the block is processed by the finalization routine, so all request operations are available in the db
 func (ds *transactionMatcher[_]) getFinalizedBlockNumber() uint64 {

indexer/execution/system_contracts/transaction_matcher_test.go

@@ -0,0 +1,85 @@
+package system_contracts
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"testing"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/sirupsen/logrus"
+
+	"github.com/ethpandaops/dora/db"
+	"github.com/ethpandaops/dora/indexer/execution"
+	"github.com/ethpandaops/dora/types"
+)
+
+func newTestDB(t *testing.T) {
+	t.Helper()
+	cfg := &types.DatabaseConfig{
+		Engine: "sqlite",
+		Sqlite: &types.SqliteDatabaseConfig{File: filepath.Join(t.TempDir(), "test.sqlite")},
+	}
+	db.MustInitDB(cfg)
+	if err := db.ApplyEmbeddedDbSchema(-2); err != nil {
+		t.Fatalf("apply schema: %v", err)
+	}
+	t.Cleanup(db.MustCloseDB)
+}
+
+type stubMatch struct{}
+
+func newTestMatcher(persist func(tx *sqlx.Tx, matches []*stubMatch) error, startHeight uint64) *transactionMatcher[stubMatch] {
+	return &transactionMatcher[stubMatch]{
+		indexer: &execution.IndexerCtx{Ctx: context.Background()},
+		logger:  logrus.New(),
+		options: &transactionMatcherOptions[stubMatch]{
+			stateKey:       "test.matcher",
+			persistMatches: persist,
+		},
+		state: &transactionMatcherState{MatchHeight: startHeight},
+	}
+}
+
+func storedHeight(t *testing.T) uint64 {
+	t.Helper()
+	state := transactionMatcherState{}
+	if _, err := db.GetExplorerState(context.Background(), "test.matcher", &state); err != nil {
+		return 0 // no state persisted yet
+	}
+	return state.MatchHeight
+}
+
+// TestPersistMatchRange checks that a failed match persist rolls the height back
+// so the range is retried, and that a successful persist advances and stores it.
+func TestPersistMatchRange(t *testing.T) {
+	newTestDB(t)
+
+	// Failure: persistMatches errors, so the height must revert and nothing persist.
+	failing := newTestMatcher(func(tx *sqlx.Tx, matches []*stubMatch) error {
+		return errors.New("transient error")
+	}, 100)
+	if err := failing.persistMatchRange([]*stubMatch{{}}, 200); err == nil {
+		t.Fatal("expected error from persistMatchRange")
+	}
+	if failing.state.MatchHeight != 100 {
+		t.Errorf("match height after failure = %d, want 100 (reverted)", failing.state.MatchHeight)
+	}
+	if h := storedHeight(t); h != 0 {
+		t.Errorf("stored height after failure = %d, want 0 (nothing persisted)", h)
+	}
+
+	// Success: persistMatches ok, so the height advances and is stored.
+	ok := newTestMatcher(func(tx *sqlx.Tx, matches []*stubMatch) error {
+		return nil
+	}, 100)
+	if err := ok.persistMatchRange([]*stubMatch{{}}, 200); err != nil {
+		t.Fatalf("persistMatchRange: %v", err)
+	}
+	if ok.state.MatchHeight != 200 {
+		t.Errorf("match height after success = %d, want 200", ok.state.MatchHeight)
+	}
+	if h := storedHeight(t); h != 200 {
+		t.Errorf("stored height after success = %d, want 200", h)
+	}
+}