Merge branch 'develop' into mardizzone/upstream-v1.16.4

Author: marcello33

core/rawdb/accessors_chain.go

@@ -1123,3 +1123,21 @@ func WriteBlockPruneCursor(db ethdb.KeyValueWriter, cursor uint64) {
 		log.Crit("Failed to store block cursor", "err", err)
 	}
 }
+
+func ReadBlockPruneHead(db ethdb.KeyValueReader) *uint64 {
+	log.Debug("ReadBlockHead")
+	data, err := db.Get(blockPruneHeadKey())
+	if err != nil || len(data) == 0 {
+		return nil
+	}
+
+	number := binary.BigEndian.Uint64(data)
+	return &number
+}
+
+func WriteBlockPruneHead(db ethdb.KeyValueWriter, head uint64) {
+	log.Debug("WriteBlockPruneHead", "head", head)
+	if err := db.Put(blockPruneHeadKey(), encodeBlockNumber(head)); err != nil {
+		log.Crit("Failed to store block head", "err", err)
+	}
+}

core/rawdb/accessors_chain_test.go

@@ -1038,3 +1038,93 @@ func TestHeadersRLPStorage(t *testing.T) {
 	checkSequence(1, 1)    // Only block 1
 	checkSequence(1, 2)    // Genesis + block 1
 }
+
+// mockErrorDB is a mock database that always returns errors on Get operations
+type mockErrorDB struct {
+	getError error
+}
+
+func (m *mockErrorDB) Has(key []byte) (bool, error) {
+	return false, m.getError
+}
+
+func (m *mockErrorDB) Get(key []byte) ([]byte, error) {
+	return nil, m.getError
+}
+
+func TestReadBlockPruneHead(t *testing.T) {
+	db := NewMemoryDatabase()
+
+	// Test reading non-existent entry (should return nil)
+	if entry := ReadBlockPruneHead(db); entry != nil {
+		t.Fatalf("Non-existent block prune head returned: %v", entry)
+	}
+
+	// Test reading when Get returns an error
+	errorDB := &mockErrorDB{getError: fmt.Errorf("database error")}
+	if entry := ReadBlockPruneHead(errorDB); entry != nil {
+		t.Fatalf("Expected nil when Get returns error, got: %v", entry)
+	}
+
+	// Write a valid value and verify read
+	testHead := uint64(12345)
+	WriteBlockPruneHead(db, testHead)
+
+	if entry := ReadBlockPruneHead(db); entry == nil {
+		t.Fatalf("Stored block prune head not found")
+	} else if *entry != testHead {
+		t.Fatalf("Retrieved block prune head mismatch: have %v, want %v", *entry, testHead)
+	}
+
+	// Test reading after writing zero value
+	WriteBlockPruneHead(db, 0)
+	if entry := ReadBlockPruneHead(db); entry == nil {
+		t.Fatalf("Stored block prune head (zero) not found")
+	} else if *entry != 0 {
+		t.Fatalf("Retrieved block prune head mismatch: have %v, want 0", *entry)
+	}
+
+	// Test reading after writing max uint64 value
+	maxHead := uint64(18446744073709551615)
+	WriteBlockPruneHead(db, maxHead)
+	if entry := ReadBlockPruneHead(db); entry == nil {
+		t.Fatalf("Stored block prune head (max uint64) not found")
+	} else if *entry != maxHead {
+		t.Fatalf("Retrieved block prune head mismatch: have %v, want %v", *entry, maxHead)
+	}
+}
+
+func TestWriteBlockPruneHead(t *testing.T) {
+	db := NewMemoryDatabase()
+
+	// Test writing and reading back a value
+	testHead := uint64(54321)
+	WriteBlockPruneHead(db, testHead)
+
+	if entry := ReadBlockPruneHead(db); entry == nil {
+		t.Fatalf("Stored block prune head not found")
+	} else if *entry != testHead {
+		t.Fatalf("Retrieved block prune head mismatch: have %v, want %v", *entry, testHead)
+	}
+
+	// Test overwriting with a different value
+	newHead := uint64(99999)
+	WriteBlockPruneHead(db, newHead)
+
+	if entry := ReadBlockPruneHead(db); entry == nil {
+		t.Fatalf("Updated block prune head not found")
+	} else if *entry != newHead {
+		t.Fatalf("Updated block prune head mismatch: have %v, want %v", *entry, newHead)
+	}
+
+	// Test writing multiple different values in sequence
+	testValues := []uint64{0, 1, 100, 1000, 10000, 18446744073709551615}
+	for _, val := range testValues {
+		WriteBlockPruneHead(db, val)
+		if entry := ReadBlockPruneHead(db); entry == nil {
+			t.Fatalf("Block prune head not found for value %d", val)
+		} else if *entry != val {
+			t.Fatalf("Block prune head mismatch for value %d: have %v, want %v", val, *entry, val)
+		}
+	}
+}

core/rawdb/accessors_state.go

@@ -429,3 +429,21 @@ func WriteTrienodeHistory(db ethdb.AncientWriter, id uint64, header []byte, keyS
 	})
 	return err
 }
+
+func ReadWitnessPruneHead(db ethdb.KeyValueReader) *uint64 {
+	log.Debug("ReadWitnessHead")
+	data, err := db.Get(witnessPruneHeadKey())
+	if err != nil || len(data) == 0 {
+		return nil
+	}
+
+	number := binary.BigEndian.Uint64(data)
+	return &number
+}
+
+func WriteWitnessPruneHead(db ethdb.KeyValueWriter, head uint64) {
+	log.Debug("WriteWitnessPruneHead", "head", head)
+	if err := db.Put(witnessPruneHeadKey(), encodeBlockNumber(head)); err != nil {
+		log.Crit("Failed to store witness Head", "err", err)
+	}
+}

core/rawdb/pruner.go

@@ -8,6 +8,8 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
+const MaxDeleteRangeSize = uint64(50_000)
+
 type Strategy interface {
 	Name() string
 	RetentionBlocks() uint64
@@ -17,6 +19,10 @@ type Strategy interface {
 	ReadCursor(db ethdb.KeyValueReader) *uint64
 	WriteCursor(db ethdb.KeyValueWriter, cur uint64)
 
+	// Head Persistency
+	ReadPrunerHead(db ethdb.KeyValueReader) *uint64
+	WritePrunerHead(db ethdb.KeyValueWriter, head uint64)
+
 	// Find earliest height <= cutoff that has data for this domain.
 	FindEarliest(db ethdb.Database, cutoff uint64) (earliest uint64, ok bool)
 
@@ -37,10 +43,13 @@ type pruner struct {
 	strategy      Strategy
 }
 
-// This pruner implements an online solution of pruning data. It demands the data to be pruned
-// to be on the KV database. That's why, for example, if you prune blocks you need to disable ancient db
 func NewPruner(db ethdb.Database, s Strategy) *pruner {
-	return &pruner{db: db, strategy: s, quit: make(chan struct{}), stopped: make(chan struct{})}
+	return &pruner{
+		db:       db,
+		strategy: s,
+		quit:     make(chan struct{}),
+		stopped:  make(chan struct{}),
+	}
 }
 
 func (p *pruner) Start() {
@@ -60,6 +69,7 @@ func (p *pruner) Start() {
 	}()
 	log.Info(p.strategy.Name()+": started", "retentionBlocks", p.strategy.RetentionBlocks(), "interval", p.strategy.Interval().String())
 }
+
 func (p *pruner) Close() error {
 	close(p.quit)
 	<-p.stopped
@@ -74,12 +84,35 @@ func (p *pruner) prune() {
 	}
 	latest := head.Number.Uint64()
 
+	prevHeadPtr := p.strategy.ReadPrunerHead(p.db)
+	prevHead := latest
+	if prevHeadPtr != nil {
+		prevHead = *prevHeadPtr
+	}
+
+	if latest < prevHead {
+		// Reorg between prevHead and latest (could have happened while offline).
+		if err := p.handleReorg(latest, prevHead); err != nil {
+			log.Error(p.strategy.Name()+": reorg cleanup failed", "newHead", latest, "oldHead", prevHead, "err", err)
+			// Do not update stored head; we want to try again next run.
+			return
+		}
+	}
+
 	var cutoff uint64
 	if rb := p.strategy.RetentionBlocks(); latest > rb {
 		cutoff = latest - rb
 	}
 
 	cur := p.strategy.ReadCursor(p.db)
+
+	if cur != nil && *cur > latest {
+		log.Warn(p.strategy.Name()+": cursor beyond head; clamping", "cursor", *cur, "head", latest)
+		p.strategy.WriteCursor(p.db, latest)
+		tmp := latest
+		cur = &tmp
+	}
+
 	if cur == nil {
 		if e, ok := p.strategy.FindEarliest(p.db, cutoff); ok {
 			log.Info(p.strategy.Name()+": no cursor stored", "earliestFound", e)
@@ -90,45 +123,108 @@ func (p *pruner) prune() {
 			log.Info(p.strategy.Name()+": no data ≤ cutoff; starting at cutoff", "cutoff", cutoff)
 		}
 	}
+
 	if *cur >= cutoff {
-		batch := p.db.NewBatch()
-		p.strategy.WriteCursor(batch, *cur)
-		_ = batch.Write()
-		log.Info(p.strategy.Name()+": no data to prune", "cursor", *cur, "cutoff", cutoff)
+		p.strategy.WritePrunerHead(p.db, latest)
 		return
 	}
 
-	// Chunk to keep batches reasonable.
-	const step = uint64(50_000)
+	// Normal pruning: delete [cur .. cutoff-1] inclusive, then move cursor to cutoff.
 	from := *cur
-	for from < cutoff {
-		to := from + step
-		if to > cutoff {
-			to = cutoff
+	to := cutoff - 1
+
+	if err := p.deleteRange(from, to); err != nil {
+		log.Error(p.strategy.Name()+": batch write error during prune", "from", from, "to", to, "err", err)
+		return
+	}
+
+	p.strategy.WriteCursor(p.db, cutoff)
+	log.Info(p.strategy.Name()+": successfully pruned", "count", cutoff-from, "from", from, "to", to)
+	p.strategy.WritePrunerHead(p.db, latest)
+}
+
+func (p *pruner) handleReorg(newHead, oldHead uint64) error {
+	log.Warn(p.strategy.Name()+": reorg detected", "newHead", newHead, "oldHead", oldHead)
+
+	cur := p.strategy.ReadCursor(p.db)
+	if cur == nil {
+		cur = new(uint64)
+		*cur = 0 // meaning no pruned data yet
+	}
+
+	// Range of non-canonical heights
+	deleteFrom := max(*cur, newHead+1)
+	deleteTo := oldHead
+
+	if deleteFrom <= deleteTo {
+		log.Warn(p.strategy.Name()+": deleting non-canonical data",
+			"from", deleteFrom, "to", deleteTo)
+
+		if err := p.deleteRange(deleteFrom, deleteTo); err != nil {
+			log.Error(p.strategy.Name()+": reorg cleanup failed; keeping cursor unchanged",
+				"from", deleteFrom, "to", deleteTo, "err", err)
+			return err
 		}
-		nhs := p.strategy.ReadNumberHashes(p.db, from, to-1)
+	}
+
+	if *cur > newHead {
+		p.strategy.WriteCursor(p.db, newHead)
+		log.Warn(p.strategy.Name()+": cursor rolled back", "oldCursor", *cur, "newCursor", newHead)
+	}
+
+	return nil
+}
+
+func max(a, b uint64) uint64 {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+func (p *pruner) deleteRange(from, to uint64) error {
+	if from > to {
+		return nil
+	}
+
+	const step = MaxDeleteRangeSize
+	start := from
+
+	for start <= to {
+		end := start + step - 1
+		if end > to {
+			end = to
+		}
+
+		nhs := p.strategy.ReadNumberHashes(p.db, start, end)
 
 		batch := p.db.NewBatch()
-		var lastNum = ^uint64(0) // sentinel
+		var lastNum = ^uint64(0)
+
 		for _, nh := range nhs {
 			p.strategy.DeletePerHash(batch, nh.Number, nh.Hash)
+
 			if nh.Number != lastNum {
-				if lastNum != ^uint64(0) { // avoid multiple deletes per same height in case of multiple hashs in same height
+				if lastNum != ^uint64(0) {
 					p.strategy.DeletePerHeight(batch, lastNum)
 				}
 				lastNum = nh.Number
 			}
 		}
-		// flush last height if any rows existed
+
 		if lastNum != ^uint64(0) {
 			p.strategy.DeletePerHeight(batch, lastNum)
 		}
-		p.strategy.WriteCursor(batch, to)
+
 		if err := batch.Write(); err != nil {
-			log.Error(p.strategy.Name()+": batch write error", "from", from, "to", to-1, "err", err)
-			return
+			log.Error(p.strategy.Name()+": batch write error during reorg cleanup",
+				"from", start, "to", end, "err", err)
+			return err
 		}
-		log.Info(p.strategy.Name()+": successfully pruned", "count", (to-1)-from, "from", from, "to", to-1)
-		from = to
+
+		log.Info(p.strategy.Name()+": removed reverted data", "from", start, "to", end)
+
+		start = end + 1
 	}
+	return nil
 }

core/rawdb/pruner_strategy_block.go

@@ -33,6 +33,13 @@ func (b *BlockStrategy) WriteCursor(db ethdb.KeyValueWriter, cur uint64) {
 	WriteBlockPruneCursor(db, cur)
 }
 
+func (b *BlockStrategy) ReadPrunerHead(db ethdb.KeyValueReader) *uint64 {
+	return ReadBlockPruneHead(db)
+}
+func (b *BlockStrategy) WritePrunerHead(db ethdb.KeyValueWriter, cur uint64) {
+	WriteBlockPruneHead(db, cur)
+}
+
 func (b *BlockStrategy) FindEarliest(db ethdb.Database, cutoff uint64) (uint64, bool) {
 	// earliest canonical header with data
 	return findEarliestBlockWithData(db, cutoff)

core/rawdb/pruner_strategy_wit.go

@@ -35,6 +35,13 @@ func (w *WitnessStrategy) WriteCursor(db ethdb.KeyValueWriter, cur uint64) {
 	WriteWitnessPruneCursor(db, cur)
 }
 
+func (w *WitnessStrategy) ReadPrunerHead(db ethdb.KeyValueReader) *uint64 {
+	return ReadWitnessPruneHead(db)
+}
+func (w *WitnessStrategy) WritePrunerHead(db ethdb.KeyValueWriter, cur uint64) {
+	WriteWitnessPruneHead(db, cur)
+}
+
 func (w *WitnessStrategy) FindEarliest(db ethdb.Database, cutoff uint64) (uint64, bool) {
 	// same as your findEarliestWitness
 	return findEarliestWitness(db, cutoff)