[performance] cherry-pick: db/snapshotsync: resume caplin state dump per-type to avoid overlapping files (#21901) (#22124)

Cherry-pick of #21901 to `performance`.

Author: sudeepdino008

cl/antiquary/state_antiquary.go

@@ -594,7 +594,6 @@ func (s *Antiquary) IncrementBeaconState(ctx context.Context, to uint64) error {
 		}
 		if err := s.stateSn.DumpCaplinState(
 			ctx,
-			s.stateSn.BlocksAvailable()+1,
 			to,
 			blocksPerStatefulFile,
 			s.sn.Salt,
@@ -605,16 +604,18 @@ func (s *Antiquary) IncrementBeaconState(ctx context.Context, to uint64) error {
 		); err != nil {
 			return err
 		}
-		paths := s.stateSn.SegFileNames(from, to)
+		// Open the new files before collecting paths so the seeder sees them;
+		// seed from 0 since a per-type resume can dump a new type from genesis.
+		if err := s.stateSn.OpenFolder(); err != nil {
+			return err
+		}
 		if s.downloader != nil {
+			paths := s.stateSn.SegFileNames(0, to)
 			// Notify bittorent to seed the new snapshots
 			if err := s.downloader.Seed(s.ctx, paths); err != nil {
 				s.logger.Warn("[Antiquary] Failed to add items to bittorent", "err", err)
 			}
 		}
-		if err := s.stateSn.OpenFolder(); err != nil {
-			return err
-		}
 	}
 
 	return nil

cmd/capcli/cli.go

@@ -1390,7 +1390,7 @@ func (c *DumpStateSnapshots) Run(ctx *Context) error {
 	r, _ := stateSn.Get(kv.BlockRoot, 999424)
 	fmt.Printf("%x\n", r)
 
-	if err := stateSn.DumpCaplinState(ctx, stateSn.BlocksAvailable(), to, c.StepSize, salt, dirs, runtime.NumCPU(), log.LvlInfo, log.Root()); err != nil {
+	if err := stateSn.DumpCaplinState(ctx, to, c.StepSize, salt, dirs, runtime.NumCPU(), log.LvlInfo, log.Root()); err != nil {
 		return err
 	}
 	if err := stateSn.OpenFolder(); err != nil {

db/snapshotsync/caplin_state_dump_test.go

@@ -0,0 +1,127 @@
+// Copyright 2024 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package snapshotsync
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func firstJobFrom(jobs []caplinStateDumpJob, name string) (uint64, bool) {
+	for _, j := range jobs {
+		if j.name == name {
+			return j.from, true
+		}
+	}
+	return 0, false
+}
+
+func countJobs(jobs []caplinStateDumpJob, name string) uint64 {
+	var n uint64
+	for _, j := range jobs {
+		if j.name == name {
+			n++
+		}
+	}
+	return n
+}
+
+// overlapsCoverage reports whether any job lands inside an existing covered range.
+func overlapsCoverage(jobs []caplinStateDumpJob, name string, covered []Range) bool {
+	for _, j := range jobs {
+		if j.name != name {
+			continue
+		}
+		for _, r := range covered {
+			if j.from < r.to && r.from < j.to {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// A new state-snapshot type added to an already-populated datadir must not drag
+// the dump back to genesis for types that are already caught up: only missing
+// ranges are scheduled, so a full base file is never re-sliced into overlaps.
+func TestPlanStateDumpResumesPerType(t *testing.T) {
+	const blocksPerFile = 50_000
+	coverage := map[string][]Range{
+		"Covered": {{from: 0, to: 10_400_000}}, // already at head
+		"Lagging": {{from: 0, to: 2_700_000}},  // newly added type, behind
+		"Empty":   nil,                         // brand-new type, nothing on disk
+	}
+
+	jobs := planStateDump(coverage, 10_400_000, blocksPerFile)
+
+	for _, j := range jobs {
+		require.Equal(t, blocksPerFile, int(j.to-j.from), "every job must be exactly one full file")
+	}
+	for name, cov := range coverage {
+		require.False(t, overlapsCoverage(jobs, name, cov), "type %s scheduled a job overlapping existing coverage", name)
+	}
+
+	require.Zero(t, countJobs(jobs, "Covered"), "caught-up type must not be re-dumped")
+
+	laggingFrom, ok := firstJobFrom(jobs, "Lagging")
+	require.True(t, ok, "lagging type must be dumped")
+	require.Equal(t, uint64(2_700_000), laggingFrom, "lagging type must resume at its coverage end")
+	require.Equal(t, uint64((10_400_000-2_700_000)/blocksPerFile), countJobs(jobs, "Lagging"))
+
+	emptyFrom, ok := firstJobFrom(jobs, "Empty")
+	require.True(t, ok, "empty type must be dumped from genesis")
+	require.Equal(t, uint64(0), emptyFrom)
+	require.Equal(t, uint64(10_400_000/blocksPerFile), countJobs(jobs, "Empty"))
+}
+
+// A gap between existing segments must be filled, without re-dumping (and thus
+// overlapping) a larger already-present segment that follows the gap.
+func TestPlanStateDumpFillsGapWithoutOverlap(t *testing.T) {
+	const blocksPerFile = 50_000
+	covered := []Range{
+		{from: 0, to: 7_150_000},
+		{from: 7_200_000, to: 10_400_000}, // 7.15M..7.2M missing; tail is one large segment
+	}
+	jobs := planStateDump(map[string][]Range{"X": covered}, 10_400_000, blocksPerFile)
+
+	require.Equal(t, uint64(1), countJobs(jobs, "X"), "only the single missing file should be scheduled")
+	from, _ := firstJobFrom(jobs, "X")
+	require.Equal(t, uint64(7_150_000), from)
+	require.False(t, overlapsCoverage(jobs, "X", covered))
+}
+
+// An unaligned coverage tail must resume exactly at its end, never flooring to a
+// file boundary below it (which would overlap the existing tail).
+func TestPlanStateDumpUnalignedResumeHasNoOverlap(t *testing.T) {
+	const blocksPerFile = 50_000
+	covered := []Range{{from: 0, to: 2_725_000}}
+	jobs := planStateDump(map[string][]Range{"X": covered}, 2_825_000, blocksPerFile)
+
+	from, ok := firstJobFrom(jobs, "X")
+	require.True(t, ok)
+	require.Equal(t, uint64(2_725_000), from, "must resume at coverage end, not floor below it")
+	require.False(t, overlapsCoverage(jobs, "X", covered))
+}
+
+func TestMissingRanges(t *testing.T) {
+	require.Equal(t, []Range(nil), missingRanges([]Range{{from: 0, to: 100}}, 100), "fully covered → nothing missing")
+	require.Equal(t, []Range{{from: 0, to: 100}}, missingRanges(nil, 100), "empty → all missing")
+	require.Equal(t, []Range{{from: 70, to: 100}}, missingRanges([]Range{{from: 0, to: 70}}, 100), "trailing tail")
+	require.Equal(t, []Range{{from: 50, to: 60}}, missingRanges([]Range{{from: 0, to: 50}, {from: 60, to: 100}}, 100), "interior gap only")
+	require.Equal(t, []Range(nil), missingRanges([]Range{{from: 0, to: 200}}, 100), "coverage beyond toSlot → nothing missing")
+}

db/snapshotsync/caplin_state_snapshots.go

@@ -25,6 +25,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime/debug"
+	"sort"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -264,6 +265,25 @@ func (s *CaplinStateSnapshots) BlocksAvailable() uint64 {
 	return min(s.segmentsMax.Load(), s.idxMax.Load())
 }
 
+func (s *CaplinStateSnapshots) coveredRangesForType(name string) []Range {
+	s.visibleLock.RLock()
+	defer s.visibleLock.RUnlock()
+
+	v, ok := s.visible.Load(name)
+	if !ok {
+		return nil
+	}
+	segs, ok := v.([]*VisibleSegment)
+	if !ok {
+		return nil
+	}
+	ranges := make([]Range, 0, len(segs))
+	for _, seg := range segs {
+		ranges = append(ranges, seg.Range)
+	}
+	return ranges
+}
+
 func (s *CaplinStateSnapshots) Close() {
 	if s == nil {
 		return
@@ -690,22 +710,67 @@ func simpleIdx(ctx context.Context, sn snaptype.FileInfo, salt uint32, tmpDir st
 	return nil
 }
 
-func (s *CaplinStateSnapshots) DumpCaplinState(ctx context.Context, fromSlot, toSlot, blocksPerFile uint64, salt uint32, dirs datadir.Dirs, workers int, lvl log.Lvl, logger log.Logger) error {
-	fromSlot = (fromSlot / blocksPerFile) * blocksPerFile
+type caplinStateDumpJob struct {
+	name     string
+	from, to uint64
+}
+
+// missingRanges returns the sub-ranges of [0, toSlot) not covered by `covered`
+// (the type's existing segment ranges, sorted by `from`).
+func missingRanges(covered []Range, toSlot uint64) []Range {
+	var missing []Range
+	var cur uint64
+	for _, r := range covered {
+		if r.from > cur {
+			gapEnd := min(r.from, toSlot)
+			missing = append(missing, Range{from: cur, to: gapEnd})
+		}
+		cur = max(cur, r.to)
+		if cur >= toSlot {
+			return missing
+		}
+	}
+	if cur < toSlot {
+		missing = append(missing, Range{from: cur, to: toSlot})
+	}
+	return missing
+}
+
+// planStateDump schedules only the ranges each type is missing within
+// [0, toSlot), starting every full file at a gap boundary so it fills holes and
+// the trailing tail without overlapping an existing segment.
+func planStateDump(coverage map[string][]Range, toSlot, blocksPerFile uint64) []caplinStateDumpJob {
 	toSlot = (toSlot / blocksPerFile) * blocksPerFile
-	for snapName, kvGetter := range s.snapshotTypes.KeyValueGetters {
-		for i := fromSlot; i < toSlot; i += blocksPerFile {
-			if toSlot-i < blocksPerFile {
-				break
-			}
-			// keep beaconblocks here but whatever....
-			to := i + blocksPerFile
-			logger.Log(lvl, "Dumping "+snapName, "from", i, "to", to)
-			if err := dumpCaplinState(ctx, snapName, kvGetter, i, to, blocksPerFile, salt, dirs, workers, lvl, logger, s.snapshotTypes.Compression[snapName]); err != nil {
-				return err
+
+	names := make([]string, 0, len(coverage))
+	for name := range coverage {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	jobs := make([]caplinStateDumpJob, 0)
+	for _, name := range names {
+		for _, gap := range missingRanges(coverage[name], toSlot) {
+			for i := gap.from; i+blocksPerFile <= gap.to; i += blocksPerFile {
+				jobs = append(jobs, caplinStateDumpJob{name: name, from: i, to: i + blocksPerFile})
 			}
 		}
 	}
+	return jobs
+}
+
+func (s *CaplinStateSnapshots) DumpCaplinState(ctx context.Context, toSlot, blocksPerFile uint64, salt uint32, dirs datadir.Dirs, workers int, lvl log.Lvl, logger log.Logger) error {
+	coverage := make(map[string][]Range, len(s.snapshotTypes.KeyValueGetters))
+	for name := range s.snapshotTypes.KeyValueGetters {
+		coverage[name] = s.coveredRangesForType(name)
+	}
+
+	for _, job := range planStateDump(coverage, toSlot, blocksPerFile) {
+		logger.Log(lvl, "Dumping "+job.name, "from", job.from, "to", job.to)
+		if err := dumpCaplinState(ctx, job.name, s.snapshotTypes.KeyValueGetters[job.name], job.from, job.to, blocksPerFile, salt, dirs, workers, lvl, logger, s.snapshotTypes.Compression[job.name]); err != nil {
+			return err
+		}
+	}
 	return nil
 }