From 7c6d53ebd2a12a1eb235b3674791fb5afcca5fef Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 2 Jul 2026 21:58:09 +0200 Subject: [PATCH 1/2] db/snapshotsync: share segment open/close helpers with caplin snapshots The two caplin snapshot containers re-implemented RoSnapshots' OpenList machinery: the find-open-segment walk, the optimistic-open error classification (twice inline in freezeblocks' CaplinSnapshots, once per its two segment types), and closeWhatNotInList. The copies had drifted unsafely: neither honored the refcount guard that keeps a segment alive while a View still reads it, and freezeblocks' close path used the embedded Decompressor's Close instead of DirtySegment.close, leaking index handles. Extract FindOpenSegment/ClassifyOpenErr/CloseSegmentsNotInList in snapshotsync, use them from RoSnapshots and both caplin containers, and drop the vestigial 'processed' flag and its stray comment. Caplin close paths now inherit the refcount guard (pinned by TestCloseAndDropNotProtected), full index+seg closing, and removal of stale unopened stubs. --- db/snapshotsync/caplin_state_snapshots.go | 70 ++----- .../freezeblocks/caplin_snapshots.go | 173 ++++-------------- db/snapshotsync/snapshots.go | 115 +++++++----- db/snapshotsync/snapshots_test.go | 29 +++ 4 files changed, 148 insertions(+), 239 deletions(-) diff --git a/db/snapshotsync/caplin_state_snapshots.go b/db/snapshotsync/caplin_state_snapshots.go index adfcd354968..a8291baa130 100644 --- a/db/snapshotsync/caplin_state_snapshots.go +++ b/db/snapshotsync/caplin_state_snapshots.go @@ -316,32 +316,15 @@ func (s *CaplinStateSnapshots) OpenList(fileNames []string, optimistic bool) err s.closeWhatNotInList(fileNames) var segmentsMax uint64 var segmentsMaxSet bool -Loop: for _, fName := range fileNames { f, _, _ := snaptype.ParseFileName(s.dir, fName) - var processed bool = true - var exists bool - var sn *DirtySegment - dirtySegments, ok := s.dirty[f.CaplinTypeString] if !ok { continue } filePath := filepath.Join(s.dir, fName) - dirtySegments.Walk(func(segments []*DirtySegment) bool { - for _, sn2 := range segments { - if sn2.Decompressor == nil { // it's ok if some segment was not able to open - continue - } - if filePath == sn2.filePath { - sn = sn2 - exists = true - break - } - } - return true - }) + sn, exists := findOpenSegment(dirtySegments, func(sn2 *DirtySegment) bool { return sn2.filePath == filePath }) if !exists { sn = &DirtySegment{ // segType: f.Type, Unsupported @@ -352,19 +335,17 @@ Loop: } } if err := s.openSegIfNeed(sn, filePath); err != nil { - if errors.Is(err, os.ErrNotExist) { - if optimistic { - continue Loop - } else { - break Loop - } + _, stop, failErr := ClassifyOpenErr(err, optimistic) + if failErr != nil { + return failErr } - if optimistic { + if stop { + break + } + if !errors.Is(err, os.ErrNotExist) { s.logger.Warn("[snapshots] open segment", "err", err) - continue Loop - } else { - return err } + continue } if !exists { @@ -375,15 +356,12 @@ Loop: if err := openIdxForCaplinStateIfNeeded(sn, filePath, optimistic); err != nil { return err } - // Only bob sidecars count for progression - if processed { - if f.To > 0 { - segmentsMax = f.To - 1 - } else { - segmentsMax = 0 - } - segmentsMaxSet = true + if f.To > 0 { + segmentsMax = f.To - 1 + } else { + segmentsMax = 0 } + segmentsMaxSet = true } if segmentsMaxSet { @@ -545,24 +523,10 @@ func (s *CaplinStateSnapshots) closeWhatNotInList(l []string) { } for _, dirtySegments := range s.dirty { - toClose := make([]*DirtySegment, 0) - dirtySegments.Walk(func(segments []*DirtySegment) bool { - for _, sn := range segments { - if sn.Decompressor == nil { - continue - } - _, name := filepath.Split(sn.FilePath()) - if _, ok := protectFiles[name]; ok { - continue - } - toClose = append(toClose, sn) - } - return true + closeAndDropNotProtected(dirtySegments, protectFiles, func(sn *DirtySegment) string { + _, name := filepath.Split(sn.FilePath()) + return name }) - for _, sn := range toClose { - sn.close() - dirtySegments.Delete(sn) - } } } diff --git a/db/snapshotsync/freezeblocks/caplin_snapshots.go b/db/snapshotsync/freezeblocks/caplin_snapshots.go index cda59ff0df2..66878e9265b 100644 --- a/db/snapshotsync/freezeblocks/caplin_snapshots.go +++ b/db/snapshotsync/freezeblocks/caplin_snapshots.go @@ -180,119 +180,51 @@ func (s *CaplinSnapshots) OpenList(fileNames []string, optimistic bool) error { var segmentsMax uint64 var segmentsMaxSet bool -Loop: for _, fName := range fileNames { f, _, ok := snaptype.ParseFileName(s.dir, fName) if !ok { continue } - var processed bool = true - switch f.Type.Enum() { - case snaptype.CaplinEnums.BeaconBlocks: - var sn *snapshotsync.DirtySegment - var exists bool - s.dirty[snaptype.BeaconBlocks.Enum()].Walk(func(segments []*snapshotsync.DirtySegment) bool { - for _, sn2 := range segments { - if sn2.Decompressor == nil { // it's ok if some segment was not able to open - continue - } - if fName == sn2.FileName() { - sn = sn2 - exists = true - break - } - } - return true - }) - if !exists { - sn = snapshotsync.NewDirtySegment( - snaptype.BeaconBlocks, - f.Version, - f.From, f.To, - true) - } - if err := sn.Open(s.dir); err != nil { - if errors.Is(err, os.ErrNotExist) { - if optimistic { - continue Loop - } else { - break Loop - } - } - if optimistic { - s.logger.Warn("[snapshots] open segment", "err", err) - continue Loop - } else { - return err - } - } + typ := f.Type.Enum() + if typ != snaptype.CaplinEnums.BeaconBlocks && typ != snaptype.CaplinEnums.BlobSidecars { + continue + } + tree := s.dirty[typ] - if !exists { - // it's possible to iterate over .seg file even if you don't have index - // then make segment available even if index open may fail - s.dirty[snaptype.BeaconBlocks.Enum()].Set(sn) - } - if err := sn.OpenIdxIfNeed(s.dir, optimistic, dirEntries); err != nil { - return err - } - // Only bob sidecars count for progression - if processed { - if f.To > 0 { - segmentsMax = f.To - 1 - } else { - segmentsMax = 0 - } - segmentsMaxSet = true + sn, exists := snapshotsync.FindOpenSegment(tree, fName) + if !exists { + sn = snapshotsync.NewDirtySegment(f.Type, f.Version, f.From, f.To, true) + } + if err := sn.Open(s.dir); err != nil { + _, stop, failErr := snapshotsync.ClassifyOpenErr(err, optimistic) + if failErr != nil { + return failErr } - case snaptype.CaplinEnums.BlobSidecars: - var sn *snapshotsync.DirtySegment - var exists bool - s.dirty[snaptype.BlobSidecars.Enum()].Walk(func(segments []*snapshotsync.DirtySegment) bool { - for _, sn2 := range segments { - if sn2.Decompressor == nil { // it's ok if some segment was not able to open - continue - } - if fName == sn2.FileName() { - sn = sn2 - exists = true - break - } - } - return true - }) - if !exists { - sn = snapshotsync.NewDirtySegment( - snaptype.BlobSidecars, - f.Version, - f.From, f.To, - true) + if stop { + break } - if err := sn.Open(s.dir); err != nil { - if errors.Is(err, os.ErrNotExist) { - if optimistic { - continue Loop - } else { - break Loop - } - } - if optimistic { - s.logger.Warn("[snapshots] open segment", "err", err) - continue Loop - } else { - return err - } + if !errors.Is(err, os.ErrNotExist) { + s.logger.Warn("[snapshots] open segment", "err", err) } + continue + } - if !exists { - // it's possible to iterate over .seg file even if you don't have index - // then make segment available even if index open may fail - s.dirty[snaptype.BlobSidecars.Enum()].Set(sn) - } - if err := sn.OpenIdxIfNeed(s.dir, optimistic, dirEntries); err != nil { - return err + if !exists { + // it's possible to iterate over .seg file even if you don't have index + // then make segment available even if index open may fail + tree.Set(sn) + } + if err := sn.OpenIdxIfNeed(s.dir, optimistic, dirEntries); err != nil { + return err + } + if typ == snaptype.CaplinEnums.BeaconBlocks { + if f.To > 0 { + segmentsMax = f.To - 1 + } else { + segmentsMax = 0 } + segmentsMaxSet = true } - } if segmentsMaxSet { s.segmentsMax.Store(segmentsMax) @@ -339,43 +271,8 @@ func (s *CaplinSnapshots) closeWhatNotInList(l []string) { for _, fName := range l { protectFiles[fName] = struct{}{} } - toClose := make([]*snapshotsync.DirtySegment, 0) - s.dirty[snaptype.BeaconBlocks.Enum()].Walk(func(segments []*snapshotsync.DirtySegment) bool { - for _, sn := range segments { - if sn.Decompressor == nil { - continue - } - _, name := filepath.Split(sn.FilePath()) - if _, ok := protectFiles[name]; ok { - continue - } - toClose = append(toClose, sn) - } - return true - }) - for _, sn := range toClose { - sn.Close() - s.dirty[snaptype.BeaconBlocks.Enum()].Delete(sn) - } - - toClose = make([]*snapshotsync.DirtySegment, 0) - s.dirty[snaptype.BlobSidecars.Enum()].Walk(func(segments []*snapshotsync.DirtySegment) bool { - for _, sn := range segments { - if sn.Decompressor == nil { - continue - } - _, name := filepath.Split(sn.FilePath()) - if _, ok := protectFiles[name]; ok { - continue - } - toClose = append(toClose, sn) - } - return true - }) - for _, sn := range toClose { - sn.Close() - s.dirty[snaptype.BlobSidecars.Enum()].Delete(sn) - } + snapshotsync.CloseSegmentsNotInList(s.dirty[snaptype.BeaconBlocks.Enum()], protectFiles) + snapshotsync.CloseSegmentsNotInList(s.dirty[snaptype.BlobSidecars.Enum()], protectFiles) } type CaplinView struct { diff --git a/db/snapshotsync/snapshots.go b/db/snapshotsync/snapshots.go index 213b4d6261f..b215aa679e1 100644 --- a/db/snapshotsync/snapshots.go +++ b/db/snapshotsync/snapshots.go @@ -1174,40 +1174,21 @@ func (s *RoSnapshots) openSegments(fileNames []string, open bool, optimistic boo continue } - var sn *DirtySegment - var exists bool - segtype.Walk(func(segs []*DirtySegment) bool { - for _, sn2 := range segs { - if sn2.Decompressor == nil { // it's ok if some segment was not able to open - continue - } - if fName == sn2.FileName() { - sn = sn2 - exists = true - return false - } - } - return true - }) - + sn, exists := FindOpenSegment(segtype, fName) if !exists { sn = &DirtySegment{segType: f.Type, version: f.Version, Range: Range{f.From, f.To}, frozen: s.snCfg.IsFrozen(f)} } if open { if err := sn.Open(s.dir); err != nil { - if errors.Is(err, os.ErrNotExist) { - if optimistic { - continue - } else { - break - } + _, stop, failErr := ClassifyOpenErr(err, optimistic) + if failErr != nil { + return failErr } - if optimistic { - continue - } else { - return err + if stop { + break } + continue } } @@ -1315,38 +1296,76 @@ func (s *RoSnapshots) closeWhatNotInList(l []string) { for _, f := range l { protectFiles[f] = struct{}{} } - toClose := make(map[snaptype.Enum][]*DirtySegment, 0) for _, t := range s.enums { - s.dirty[t].Walk(func(segs []*DirtySegment) bool { + CloseSegmentsNotInList(s.dirty[t], protectFiles) + } +} - for _, seg := range segs { - if _, ok := protectFiles[seg.FileName()]; ok { - continue - } - if _, ok := toClose[t]; !ok { - toClose[t] = make([]*DirtySegment, 0) - } - toClose[t] = append(toClose[t], seg) +// CloseSegmentsNotInList closes and drops tree segments whose file name is not +// in protectFiles, keeping any that a live reader still references. +func CloseSegmentsNotInList(tree *btree.BTreeG[*DirtySegment], protectFiles map[string]struct{}) { + closeAndDropNotProtected(tree, protectFiles, (*DirtySegment).FileName) +} + +func closeAndDropNotProtected(tree *btree.BTreeG[*DirtySegment], protectFiles map[string]struct{}, nameOf func(*DirtySegment) string) { + var toClose []*DirtySegment + tree.Walk(func(segs []*DirtySegment) bool { + for _, seg := range segs { + if _, ok := protectFiles[nameOf(seg)]; ok { + continue } + toClose = append(toClose, seg) + } + return true + }) - return true - }) + for _, delSeg := range toClose { + if delSeg.refcount.Load() > 0 { + // A live reader (View/RoTx) still holds this segment. Closing it + // now would nil its decompressor out from under that reader and + // turn the reader's later closeAndRemoveFiles into a crash. Leave + // it; it is reaped on a later pass once the reader releases it. + continue + } + delSeg.close() + tree.Delete(delSeg) } +} - for segtype, delSegments := range toClose { - dirtyFiles := s.dirty[segtype] - for _, delSeg := range delSegments { - if delSeg.refcount.Load() > 0 { - // A live reader (View/RoTx) still holds this segment. Closing it - // now would nil its decompressor out from under that reader and - // turn the reader's later closeAndRemoveFiles into a crash. Leave - // it; it is reaped on a later pass once the reader releases it. +// FindOpenSegment returns tree's already-open segment named fName, if any. +func FindOpenSegment(tree *btree.BTreeG[*DirtySegment], fName string) (*DirtySegment, bool) { + return findOpenSegment(tree, func(sn *DirtySegment) bool { return sn.FileName() == fName }) +} + +func findOpenSegment(tree *btree.BTreeG[*DirtySegment], match func(*DirtySegment) bool) (sn *DirtySegment, ok bool) { + tree.Walk(func(segs []*DirtySegment) bool { + for _, sn2 := range segs { + if sn2.Decompressor == nil { // it's ok if some segment was not able to open continue } - delSeg.close() - dirtyFiles.Delete(delSeg) + if match(sn2) { + sn, ok = sn2, true + return false + } + } + return true + }) + return sn, ok +} + +// ClassifyOpenErr says how a snapshot-listing loop proceeds after a segment +// open error: skip just this file, stop the whole listing, or fail hard. +func ClassifyOpenErr(err error, optimistic bool) (skip, stop bool, failErr error) { + if errors.Is(err, os.ErrNotExist) { + if optimistic { + return true, false, nil } + return false, true, nil + } + if optimistic { + return true, false, nil } + return false, false, err } func (s *RoSnapshots) RemoveOverlaps(onDelete func(l []string) error) error { diff --git a/db/snapshotsync/snapshots_test.go b/db/snapshotsync/snapshots_test.go index d6e5fc47df3..0cc073c6e54 100644 --- a/db/snapshotsync/snapshots_test.go +++ b/db/snapshotsync/snapshots_test.go @@ -24,6 +24,7 @@ import ( "testing/fstest" "github.com/stretchr/testify/require" + "github.com/tidwall/btree" dir2 "github.com/erigontech/erigon/common/dir" "github.com/erigontech/erigon/common/log/v3" @@ -1166,3 +1167,31 @@ func TestOverlapNoTruncation(t *testing.T) { require.Equal(uint64(1_500_000), visibleTxn[1].to) require.Equal(uint64(1_500_000-1), s.SegmentsMax()) } + +func TestCloseAndDropNotProtected(t *testing.T) { + tree := btree.NewBTreeGOptions[*DirtySegment](DirtySegmentLess, btree.Options{Degree: 4, NoLocks: false}) + protected := &DirtySegment{Range: Range{0, 1000}} + held := &DirtySegment{Range: Range{1000, 2000}} + held.refcount.Store(1) + stale := &DirtySegment{Range: Range{2000, 3000}} + tree.Set(protected) + tree.Set(held) + tree.Set(stale) + + closeAndDropNotProtected(tree, map[string]struct{}{"keep": {}}, func(sn *DirtySegment) string { + if sn == protected { + return "keep" + } + return "drop" + }) + + require.Equal(t, 2, tree.Len(), "protected and reader-held segments must survive") + var survivors []*DirtySegment + tree.Walk(func(segs []*DirtySegment) bool { + survivors = append(survivors, segs...) + return true + }) + require.Contains(t, survivors, protected) + require.Contains(t, survivors, held, "segment with live readers must not be dropped") + require.NotContains(t, survivors, stale) +} From c6b71a3fdf0851535e0e4f35432c3d44a9f15b4d Mon Sep 17 00:00:00 2001 From: yperbasis Date: Tue, 7 Jul 2026 20:51:09 +0200 Subject: [PATCH 2/2] db/snapshotsync: use caplin-state filePath field in close path; tighten ClassifyOpenErr The closeWhatNotInList nameOf closure called the promoted Decompressor.FilePath(), which nil-panics on an unopened stub; use the filePath field, which is set for every tree member. Pinned by TestCaplinStateCloseWhatNotInListDropsUnopenedStub. Also drop ClassifyOpenErr's unused skip return value. --- db/snapshotsync/caplin_state_snapshots.go | 6 ++- .../caplin_state_snapshots_test.go | 41 +++++++++++++++++++ .../freezeblocks/caplin_snapshots.go | 2 +- db/snapshotsync/snapshots.go | 16 ++++---- 4 files changed, 53 insertions(+), 12 deletions(-) create mode 100644 db/snapshotsync/caplin_state_snapshots_test.go diff --git a/db/snapshotsync/caplin_state_snapshots.go b/db/snapshotsync/caplin_state_snapshots.go index a8291baa130..831873cb2fe 100644 --- a/db/snapshotsync/caplin_state_snapshots.go +++ b/db/snapshotsync/caplin_state_snapshots.go @@ -335,7 +335,7 @@ func (s *CaplinStateSnapshots) OpenList(fileNames []string, optimistic bool) err } } if err := s.openSegIfNeed(sn, filePath); err != nil { - _, stop, failErr := ClassifyOpenErr(err, optimistic) + stop, failErr := ClassifyOpenErr(err, optimistic) if failErr != nil { return failErr } @@ -524,7 +524,9 @@ func (s *CaplinStateSnapshots) closeWhatNotInList(l []string) { for _, dirtySegments := range s.dirty { closeAndDropNotProtected(dirtySegments, protectFiles, func(sn *DirtySegment) string { - _, name := filepath.Split(sn.FilePath()) + // sn.filePath, not the promoted Decompressor.FilePath(): the field is + // set for every tree member, incl. stubs whose Decompressor is nil. + _, name := filepath.Split(sn.filePath) return name }) } diff --git a/db/snapshotsync/caplin_state_snapshots_test.go b/db/snapshotsync/caplin_state_snapshots_test.go new file mode 100644 index 00000000000..b7ed527d8e4 --- /dev/null +++ b/db/snapshotsync/caplin_state_snapshots_test.go @@ -0,0 +1,41 @@ +// Copyright 2026 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 . + +package snapshotsync + +import ( + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + "github.com/tidwall/btree" +) + +func TestCaplinStateCloseWhatNotInListDropsUnopenedStub(t *testing.T) { + tree := btree.NewBTreeGOptions[*DirtySegment](DirtySegmentLess, btree.Options{Degree: 4, NoLocks: false}) + kept := &DirtySegment{Range: Range{0, 1000}, filePath: filepath.Join("snapshots", "keep.seg")} + stale := &DirtySegment{Range: Range{1000, 2000}, filePath: filepath.Join("snapshots", "drop.seg")} + tree.Set(kept) + tree.Set(stale) + s := &CaplinStateSnapshots{dirty: map[string]*btree.BTreeG[*DirtySegment]{"test": tree}} + + s.closeWhatNotInList([]string{"keep.seg"}) + + require.Equal(t, 1, tree.Len(), "unopened stale stub must be dropped") + survivor, ok := tree.Min() + require.True(t, ok) + require.Same(t, kept, survivor) +} diff --git a/db/snapshotsync/freezeblocks/caplin_snapshots.go b/db/snapshotsync/freezeblocks/caplin_snapshots.go index 66878e9265b..a09661bb99f 100644 --- a/db/snapshotsync/freezeblocks/caplin_snapshots.go +++ b/db/snapshotsync/freezeblocks/caplin_snapshots.go @@ -196,7 +196,7 @@ func (s *CaplinSnapshots) OpenList(fileNames []string, optimistic bool) error { sn = snapshotsync.NewDirtySegment(f.Type, f.Version, f.From, f.To, true) } if err := sn.Open(s.dir); err != nil { - _, stop, failErr := snapshotsync.ClassifyOpenErr(err, optimistic) + stop, failErr := snapshotsync.ClassifyOpenErr(err, optimistic) if failErr != nil { return failErr } diff --git a/db/snapshotsync/snapshots.go b/db/snapshotsync/snapshots.go index b215aa679e1..e62c6554690 100644 --- a/db/snapshotsync/snapshots.go +++ b/db/snapshotsync/snapshots.go @@ -1181,7 +1181,7 @@ func (s *RoSnapshots) openSegments(fileNames []string, open bool, optimistic boo if open { if err := sn.Open(s.dir); err != nil { - _, stop, failErr := ClassifyOpenErr(err, optimistic) + stop, failErr := ClassifyOpenErr(err, optimistic) if failErr != nil { return failErr } @@ -1354,18 +1354,16 @@ func findOpenSegment(tree *btree.BTreeG[*DirtySegment], match func(*DirtySegment } // ClassifyOpenErr says how a snapshot-listing loop proceeds after a segment -// open error: skip just this file, stop the whole listing, or fail hard. -func ClassifyOpenErr(err error, optimistic bool) (skip, stop bool, failErr error) { +// open error: stop the whole listing (stop), fail hard (failErr), or, when +// neither, skip just this file. +func ClassifyOpenErr(err error, optimistic bool) (stop bool, failErr error) { if errors.Is(err, os.ErrNotExist) { - if optimistic { - return true, false, nil - } - return false, true, nil + return !optimistic, nil } if optimistic { - return true, false, nil + return false, nil } - return false, false, err + return false, err } func (s *RoSnapshots) RemoveOverlaps(onDelete func(l []string) error) error {