fix(storage): make pruning crash-consistent

sei-db/db_engine/pebbledb/batch.go

@@ -29,6 +29,10 @@ func (pb *pebbleBatch) Delete(key []byte) error {
 	return pb.b.Delete(key, nil)
 }
 
+func (pb *pebbleBatch) DeleteRange(start, end []byte) error {
+	return pb.b.DeleteRange(start, end, nil)
+}
+
 func (pb *pebbleBatch) Commit(opts types.WriteOptions) error {
 	writeCount := int64(pb.b.Count())
 	err := pb.b.Commit(toPebbleWriteOpts(opts))

sei-db/ledger_db/block/littblock/litt_block_db.go

@@ -52,11 +52,12 @@ type blockDB struct {
 	mu              sync.Mutex
 	hasBlocks       bool
 	lastBlockNumber types.GlobalBlockNumber
-	hasQC           bool
-	lastQCNext      types.GlobalBlockNumber
-
-	// latestQCStartBlock is the most recently written QC's starting block number.
-	latestQCStartBlock types.GlobalBlockNumber
+	// The newest block covered by a successful Flush (or recovered on open). GC measures retention
+	// from this cursor so it never prunes durable history against a volatile write suffix.
+	hasDurableBlocks       bool
+	lastDurableBlockNumber types.GlobalBlockNumber
+	hasQC                  bool
+	lastQCNext             types.GlobalBlockNumber
 
 	// firstBlockNumber is the lowest block number this handle has seen. Iterator clamps its
 	// start up to it so a scan always opens on a block that exists: the first block may be
@@ -196,11 +197,15 @@ func (s *blockDB) recoverCursors() error {
 				if err != nil {
 					return fmt.Errorf("failed to unmarshal newest qc: %w", err)
 				}
-				s.latestQCStartBlock, s.lastQCNext = coveredRange(qc)
+				_, s.lastQCNext = coveredRange(qc)
 				s.hasQC = true
 			}
 		}
 	}
+	if s.hasBlocks {
+		s.hasDurableBlocks = true
+		s.lastDurableBlockNumber = s.lastBlockNumber
+	}
 	return nil
 }
 
@@ -329,7 +334,6 @@ func (s *blockDB) WriteQC(qc *types.FullCommitQC) error {
 		// discovering it by scanning; a reopen re-derives the same value.
 		s.oldestQCStart = first
 	}
-	s.latestQCStartBlock = first
 	s.lastQCNext = next
 	s.hasQC = true
 	return nil
@@ -345,8 +349,14 @@ func (s *blockDB) PruneBefore(blockHeight types.GlobalBlockNumber) error {
 		return nil
 	}
 
-	if ceiling := min(s.latestQCStartBlock, s.lastBlockNumber); blockHeight > ceiling {
-		blockHeight = ceiling
+	// Never make the store depend on an unflushed suffix to remain non-empty. GC and callers may
+	// prune concurrently with the persistence loop; a process crash can lose every write above this
+	// cursor, so the newest durable block's whole QC cohort is the highest safe prune boundary.
+	if !s.hasDurableBlocks {
+		return nil
+	}
+	if blockHeight > s.lastDurableBlockNumber {
+		blockHeight = s.lastDurableBlockNumber
 	}
 
 	// Round the watermark down to the start of a QC's range, to avoid pruning a QC before its blocks.
@@ -405,9 +415,24 @@ func (s *blockDB) gcFilter(key []byte, _ bool) (bool, error) {
 }
 
 func (s *blockDB) Flush() error {
+	// Snapshot the write cursor before flushing. Table.Flush guarantees every Put that completed
+	// before it began, while overlapping writes may or may not be durable; recording this snapshot
+	// is therefore conservative without blocking the persistence loop for the duration of the I/O.
+	s.mu.Lock()
+	hasBlocks := s.hasBlocks
+	lastBlockNumber := s.lastBlockNumber
+	s.mu.Unlock()
+
 	if err := s.table.Flush(); err != nil {
 		return fmt.Errorf("failed to flush ledger table: %w", err)
 	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if hasBlocks && (!s.hasDurableBlocks || lastBlockNumber > s.lastDurableBlockNumber) {
+		s.hasDurableBlocks = true
+		s.lastDurableBlockNumber = lastBlockNumber
+	}
 	return nil
 }
 

sei-db/ledger_db/block/littblock/litt_block_gc.go

@@ -41,16 +41,15 @@ func (s *blockDB) GetRollbackFloor(rollbackWindow uint64) uint64 {
 	return head - rollbackWindow
 }
 
-// GetLatestBlock returns the newest block number written, or 0 when none has been. It reports the
-// written cursor rather than the flushed one, so a block a crash would lose still counts as ingested.
+// GetLatestBlock returns the newest crash-recoverable block number, or 0 when none has been.
 //
 // Global block numbers start at genesis block 0, so a store holding only that block is
 // indistinguishable from an empty one.
 func (s *blockDB) GetLatestBlock() (uint64, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	if !s.hasBlocks {
+	if !s.hasDurableBlocks {
 		return 0, nil
 	}
-	return uint64(s.lastBlockNumber), nil
+	return uint64(s.lastDurableBlockNumber), nil
 }

sei-db/ledger_db/block/littblock/litt_block_gc_test.go

@@ -36,11 +36,10 @@ func openForGC(t *testing.T, dir string) (types.BlockDB, gc.PrunableStore) {
 	return db, store
 }
 
-// The head the collector reads is the newest block written — not the newest QC's coverage, since
-// a QC is written before the blocks it covers — and 0 while nothing has been ingested, so an
-// empty store drops out of the head minimum instead of dragging the lookback floor to 0. The reopen
-// at the end covers recovery: a store that reported 0 after a restart would let the other stores
-// prune past a height this one still holds.
+// The head the collector reads is the newest crash-recoverable block — not the newest written
+// block and not the newest QC's coverage. Counting an unflushed suffix would let the collector
+// prune durable history against records a process crash can lose, leaving less than the configured
+// rollback window. The reopen at the end covers recovery.
 func TestGCLatestBlock(t *testing.T) {
 	dir := t.TempDir()
 	rng := utils.TestRngFromSeed(1)
@@ -53,6 +52,11 @@ func TestGCLatestBlock(t *testing.T) {
 	writeSyntheticBatches(t, db, rng, 4, 5) // blocks 0..19, QCs [0,5)..[15,20)
 	latest, err = store.GetLatestBlock()
 	require.NoError(t, err)
+	require.Equal(t, uint64(0), latest, "written but unflushed blocks must not advance the pruning head")
+
+	require.NoError(t, db.Flush())
+	latest, err = store.GetLatestBlock()
+	require.NoError(t, err)
 	require.Equal(t, uint64(19), latest)
 
 	// A QC covering 20..24 is written but none of its blocks are, so the head must not move.
@@ -62,7 +66,14 @@ func TestGCLatestBlock(t *testing.T) {
 	require.Equal(t, uint64(19), latest, "a QC ahead of its blocks must not advance the head")
 
 	require.NoError(t, db.WriteBlock(20, types.GenBlock(rng)))
+	latest, err = store.GetLatestBlock()
+	require.NoError(t, err)
+	require.Equal(t, uint64(19), latest, "an unflushed block must not advance the pruning head")
+
 	require.NoError(t, db.Flush())
+	latest, err = store.GetLatestBlock()
+	require.NoError(t, err)
+	require.Equal(t, uint64(20), latest)
 	require.NoError(t, db.Close())
 
 	_, reopened := openForGC(t, dir)
@@ -107,6 +118,7 @@ func TestGCRollbackFloorAndPruneHistory(t *testing.T) {
 	require.Equal(t, uint64(0), impl.watermark.Load())
 
 	writeSyntheticBatches(t, db, rng, 4, 5) // blocks 0..19, QCs [0,5),[5,10),[10,15),[15,20)
+	require.NoError(t, db.Flush())
 	require.Equal(t, uint64(19), store.GetRollbackFloor(0), "the whole store is inside a window of 0")
 	require.Equal(t, uint64(7), store.GetRollbackFloor(12))
 	// A window deeper than the store's own head is a rollback promise reaching past genesis, so
@@ -132,6 +144,7 @@ func TestGCPruneHistoryAboveHeadIsCapped(t *testing.T) {
 	rng := utils.TestRngFromSeed(3)
 
 	writeSyntheticBatches(t, db, rng, 4, 5) // blocks 0..19, newest cohort QC[15,20)
+	require.NoError(t, db.Flush())
 	require.NoError(t, store.PruneHistory(1_000))
 	require.Equal(t, uint64(15), db.(*blockDB).watermark.Load(), "a prune past the head is capped to the newest cohort")
 
@@ -142,6 +155,29 @@ func TestGCPruneHistoryAboveHeadIsCapped(t *testing.T) {
 	}
 }
 
+// The never-empty cap is measured from the durable tip, not the newest write. Otherwise a concurrent
+// unflushed suffix could let pruning reclaim every crash-recoverable block; a process crash would
+// then lose the suffix and reopen without the history the cap promised to preserve.
+func TestGCPruneHistoryAboveHeadIsCappedToDurableCohort(t *testing.T) {
+	db, store := openForGC(t, t.TempDir())
+	rng := utils.TestRngFromSeed(5)
+
+	writeSyntheticBatches(t, db, rng, 2, 5) // durable blocks 0..9, newest durable cohort QC[5,10)
+	require.NoError(t, db.Flush())
+	for i := 2; i < 4; i++ {
+		first := types.GlobalBlockNumber(i * 5)
+		next := first + 5
+		require.NoError(t, db.WriteQC(types.GenFullCommitQCRange(rng, first, next)))
+		for n := first; n < next; n++ {
+			require.NoError(t, db.WriteBlock(n, types.GenBlock(rng)))
+		}
+	}
+
+	require.NoError(t, store.PruneHistory(1_000))
+	require.Equal(t, uint64(5), db.(*blockDB).watermark.Load(),
+		"the store must retain the newest durable cohort, not rely on an unflushed suffix")
+}
+
 func TestConfigValidateRetentionTime(t *testing.T) {
 	cfg, err := DefaultConfig(t.TempDir())
 	require.NoError(t, err)

sei-db/ledger_db/receipt/export_test.go

@@ -20,5 +20,14 @@ func GetLogsForTx(receipt *types.Receipt, logStartIndex uint) []*ethtypes.Log {
 // block below cutoff. Test-only hook so prune behavior can be asserted without
 // waiting on the background interval.
 func PruneLittIdx(store ReceiptStore, cutoff uint64) error {
-	return store.(*littReceiptStore).pruneBlocksBelow(cutoff)
+	s := store.(*littReceiptStore)
+	if err := s.flushReceipts(); err != nil {
+		return err
+	}
+	return s.pruneBlocksBelow(cutoff)
+}
+
+// FlushLittIdx makes every receipt body written before the call crash recoverable.
+func FlushLittIdx(store ReceiptStore) error {
+	return store.(*littReceiptStore).flushReceipts()
 }

sei-db/ledger_db/receipt/litt_receipt_gc.go

@@ -43,9 +43,10 @@ func (s *littReceiptStore) GetRollbackFloor(rollbackWindow uint64) uint64 {
 	return head - rollbackWindow
 }
 
-// GetLatestBlock returns the newest block whose receipts have been written, or 0 when none have.
+// GetLatestBlock returns the newest block whose receipt bodies are crash recoverable, or 0 when none
+// have been.
 func (s *littReceiptStore) GetLatestBlock() (uint64, error) {
-	latest := s.latestVersion.Load()
+	latest := s.latestDurableVersion.Load()
 	if latest <= 0 {
 		return 0, nil
 	}

sei-db/ledger_db/receipt/litt_receipt_gc_test.go

@@ -120,6 +120,7 @@ func TestReceiptGCAnswersDoNotDependOnKeepRecent(t *testing.T) {
 		for block := uint64(1); block <= 10; block++ {
 			writeLitBlock(t, store, ctx, block, litReceipt(block, 0, addr, topic))
 		}
+		require.NoError(t, receipt.FlushLittIdx(store))
 		require.Equal(t, uint64(7), prunable.GetRollbackFloor(3),
 			"keepRecent %d must not change the floor (head 10 - window 3)", keepRecent)
 	}
@@ -158,12 +159,42 @@ func TestReceiptGCLatestBlock(t *testing.T) {
 	writeLitBlock(t, store, ctx, 2, litReceipt(2, 0, addr, topic))
 	writeLitBlock(t, store, ctx, 3, litReceipt(3, 0, addr, topic))
 
+	require.NoError(t, receipt.FlushLittIdx(store))
 	latest, err = prunable.GetLatestBlock()
 	require.NoError(t, err)
 	require.Equal(t, uint64(3), latest)
 	require.Equal(t, int64(3), store.LatestVersion(), "the head reported to the collector must agree with the store's own version")
 }
 
+func TestReceiptGCLatestBlockSurvivesReopen(t *testing.T) {
+	dir := t.TempDir()
+	storeKey := storetypes.NewKVStoreKey("evm")
+	tkey := storetypes.NewTransientStoreKey("evm_transient")
+	ctx := testutil.DefaultContext(storeKey, tkey).WithBlockHeight(1)
+	cfg := dbconfig.DefaultReceiptStoreConfig()
+	cfg.Backend = "littidx"
+	cfg.DBDirectory = dir
+	cfg.ExternalPruning = true
+
+	store, err := receipt.NewReceiptStore(cfg, storeKey)
+	require.NoError(t, err)
+	addr := common.HexToAddress("0xabcd")
+	topic := common.HexToHash("0x1111")
+	for block := uint64(1); block <= 3; block++ {
+		writeLitBlock(t, store, ctx, block, litReceipt(block, 0, addr, topic))
+	}
+	require.NoError(t, receipt.FlushLittIdx(store))
+	require.NoError(t, store.Close())
+
+	store, err = receipt.NewReceiptStore(cfg, storeKey)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, store.Close()) })
+	prunable := store.(gc.PrunableStore)
+	latest, err := prunable.GetLatestBlock()
+	require.NoError(t, err)
+	require.Equal(t, uint64(3), latest, "the durable head must be recovered with the receipt bodies")
+}
+
 // A contiguous store resolves the window against its own head, and the prune that follows moves the
 // retention floor to the height it is given — which is what makes the receipts below it stop being
 // served. Reclaiming their bodies lags that, since litt also waits for the TTL, but it can no
@@ -180,6 +211,7 @@ func TestReceiptGCRollbackFloorAndPruneHistory(t *testing.T) {
 	for block := uint64(1); block <= 3; block++ {
 		writeLitBlock(t, store, ctx, block, litReceipt(block, 0, addr, topic))
 	}
+	require.NoError(t, receipt.FlushLittIdx(store))
 	require.Equal(t, uint64(1), prunable.GetRollbackFloor(2))
 	// A window deeper than its own head is a rollback promise reaching past genesis, so nothing here
 	// is eligible for pruning yet.
@@ -219,6 +251,7 @@ func TestReceiptGCPruneHistoryAboveHead(t *testing.T) {
 		for block := uint64(1); block <= 3; block++ {
 			writeLitBlock(t, store, ctx, block, litReceipt(block, 0, addr, topic))
 		}
+		require.NoError(t, receipt.FlushLittIdx(store))
 
 		require.NoError(t, prunable.PruneHistory(1_000))
 		require.Equal(t, int64(3), store.EarliestVersion(), "the floor stops at the head, not the request")
@@ -228,3 +261,22 @@ func TestReceiptGCPruneHistoryAboveHead(t *testing.T) {
 		require.Equal(t, uint64(3), kept.BlockNumber)
 	})
 }
+
+func TestReceiptGCPruneHistoryCapsAtDurableHead(t *testing.T) {
+	store, prunable, ctx := setupLittIdxForGC(t, 0)
+	addr := common.HexToAddress("0xabcd")
+	topic := common.HexToHash("0x1111")
+	for block := uint64(1); block <= 3; block++ {
+		writeLitBlock(t, store, ctx, block, litReceipt(block, 0, addr, topic))
+	}
+	require.NoError(t, receipt.FlushLittIdx(store))
+	writeLitBlock(t, store, ctx, 4, litReceipt(4, 0, addr, topic))
+
+	require.NoError(t, prunable.PruneHistory(1_000))
+	require.Equal(t, int64(3), store.EarliestVersion(),
+		"the floor must not rely on the unflushed block remaining after a crash")
+
+	kept, err := store.GetReceiptFromStore(ctx, litTxHash(3, 0))
+	require.NoError(t, err)
+	require.Equal(t, uint64(3), kept.BlockNumber)
+}