diff --git a/indexer/beacon/bidcache.go b/indexer/beacon/bidcache.go index d4349e08..b9e82d49 100644 --- a/indexer/beacon/bidcache.go +++ b/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 } diff --git a/indexer/beacon/bidcache_test.go b/indexer/beacon/bidcache_test.go new file mode 100644 index 00000000..16ae74be --- /dev/null +++ b/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) + } +} diff --git a/indexer/beacon/finalization.go b/indexer/beacon/finalization.go index 0353af30..b62a39b3 100644 --- a/indexer/beacon/finalization.go +++ b/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 diff --git a/indexer/execution/system_contracts/transaction_matcher.go b/indexer/execution/system_contracts/transaction_matcher.go index 0ac42ddd..4a57e247 100644 --- a/indexer/execution/system_contracts/transaction_matcher.go +++ b/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 { diff --git a/indexer/execution/system_contracts/transaction_matcher_test.go b/indexer/execution/system_contracts/transaction_matcher_test.go new file mode 100644 index 00000000..77ad7390 --- /dev/null +++ b/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) + } +}