Skip to content

Commit 1790aa6

Browse files
sudeepdino008sudeepdino008
authored
enforce block-snapshots cap inside aggregator collation (#20852)
- https://github.com/erigontech/erigon/pull/20701/changes was added, wherein the cap was passed to BuildFiles* api - but BuildFilesInBackground can ignore the passed txnum value (considering `lastTxNumInDb` etc.) - this PR moves the state collation cap logic inside the aggregator; the aggregator then needs a `SetFrozenBlocksProvider` which is called to find block snapshots progress, and use it to cap the state collation.
1 parent ab501fc commit 1790aa6

7 files changed

Lines changed: 215 additions & 24 deletions

File tree

‎cmd/integration/commands/stages.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -980,6 +980,7 @@ func allSnapshots(ctx context.Context, db kv.RoDB, logger log.Logger) (*freezebl
980980
_aggSingleton = dbstate.New(dirs).Logger(logger).WithErigonDBSettings(erigonDBSettings).MustOpen(ctx, db)
981981

982982
_aggSingleton.SetProduceMod(snapCfg.ProduceE3)
983+
_aggSingleton.SetFrozenBlocksProvider(blockReader)
983984

984985
g := &errgroup.Group{}
985986
g.Go(func() error {

‎cmd/utils/app/snapshots_cmd.go‎

Lines changed: 4 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,6 @@ import (
6767
"github.com/erigontech/erigon/db/rawdb/blockio"
6868
"github.com/erigontech/erigon/db/recsplit"
6969
"github.com/erigontech/erigon/db/seg"
70-
"github.com/erigontech/erigon/db/services"
7170
"github.com/erigontech/erigon/db/snapshotsync"
7271
"github.com/erigontech/erigon/db/snapshotsync/freezeblocks"
7372
"github.com/erigontech/erigon/db/snaptype"
@@ -2907,6 +2906,9 @@ func doRetireCommand(cliCtx *cli.Context, dirs datadir.Dirs) error {
29072906
blockSnapBuildSema := semaphore.NewWeighted(int64(runtime.NumCPU()))
29082907
agg.SetSnapshotBuildSema(blockSnapBuildSema)
29092908

2909+
blockReader, _ := br.IO()
2910+
agg.SetFrozenBlocksProvider(blockReader)
2911+
29102912
agg.PresetOfflineMerge()
29112913
agg.PeriodicalyPrintProcessSet(ctx)
29122914

@@ -2927,8 +2929,6 @@ func doRetireCommand(cliCtx *cli.Context, dirs datadir.Dirs) error {
29272929
return err
29282930
}
29292931

2930-
blockReader, _ := br.IO()
2931-
29322932
blocksInSnapshots := blockReader.FrozenBlocks()
29332933
if chainConfig.Bor != nil {
29342934
blocksInSnapshots = min(blocksInSnapshots, blockReader.FrozenBorBlocks(false))
@@ -2977,15 +2977,7 @@ func doRetireCommand(cliCtx *cli.Context, dirs datadir.Dirs) error {
29772977
if err := db.Update(ctx, func(tx kv.RwTx) error {
29782978
execProgress, _ := stages.GetStageProgress(tx, stages.Execution)
29792979
lastTxNum, err = txNumsReader.Max(ctx, tx, execProgress)
2980-
if err != nil {
2981-
return err
2982-
}
2983-
maxCollatable, err := services.MaxCollatableTxNum(ctx, tx, blockReader)
2984-
if err != nil {
2985-
return err
2986-
}
2987-
lastTxNum = min(lastTxNum, maxCollatable)
2988-
return nil
2980+
return err
29892981
}); err != nil {
29902982
return err
29912983
}

‎db/services/snapshot_progress.go‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,8 @@ import (
2222
"github.com/erigontech/erigon/db/kv"
2323
)
2424

25-
// MaxCollatableTxNum returns the upper bound txNum that state collation may
26-
// target without running ahead of block snapshot files. Callers of
27-
// Aggregator.BuildFiles / BuildFilesInBackground must cap their target txNum
28-
// by this value — otherwise state files may advance past block files, an
29-
// unrecoverable state that requires manual `erigon seg rm-state --latest` to
30-
// release.
25+
// MaxCollatableTxNum: upper bound txNum that state collation may target.
26+
// Aggregator enforces this internally via SetFrozenBlocksProvider.
3127
func MaxCollatableTxNum(ctx context.Context, tx kv.Tx, blockReader FullBlockReader) (uint64, error) {
3228
return blockReader.TxnumReader().Max(ctx, tx, blockReader.FrozenBlocks())
3329
}

‎db/state/aggregator.go‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,13 @@ type Aggregator struct {
105105
configured bool
106106
savedSalt *uint32
107107
disableFsync bool
108+
109+
// nil = no cap. See #20701.
110+
frozenBlocks FrozenBlocksProvider
111+
}
112+
113+
type FrozenBlocksProvider interface {
114+
FrozenBlocks() uint64
108115
}
109116

110117
func newAggregator(ctx context.Context, dirs datadir.Dirs, reorgBlockDepth uint64, db kv.RoDB, logger log.Logger) (*Aggregator, error) {
@@ -935,6 +942,24 @@ func (a *Aggregator) readyForCollation(ctx context.Context, step kv.Step) (lastB
935942
if a.reorgBlockDepth == 0 {
936943
return 0, 0, 0, true, nil
937944
}
945+
946+
if a.frozenBlocks != nil {
947+
var capTxNum uint64
948+
if err = a.db.View(ctx, func(tx kv.Tx) error {
949+
var err error
950+
capTxNum, err = rawdbv3.TxNums.Max(ctx, tx, a.frozenBlocks.FrozenBlocks())
951+
return err
952+
}); err != nil {
953+
return 0, 0, 0, false, fmt.Errorf("read max collatable txNum: %w", err)
954+
}
955+
if uint64(step+1)*a.StepSize() > capTxNum {
956+
a.logger.Info("[snapshots] holding state collation at block snapshot boundary",
957+
"step", step,
958+
"stepEndTxNum", fmt.Sprintf("%d (step %d)", uint64(step+1)*a.StepSize(), uint64(step+1)),
959+
"blockSnapshotsTxNum", fmt.Sprintf("%d (step %d)", capTxNum, capTxNum/a.StepSize()))
960+
return 0, 0, 0, false, nil
961+
}
962+
}
938963
err = a.db.View(ctx, func(tx kv.Tx) error {
939964
lastBlockInStep, ok, err = rawdbv3.TxNums.FindBlockNum(ctx, tx, lastTxNumOfStep(step, a.stepSize))
940965
if err != nil {
@@ -1805,6 +1830,13 @@ func (a *Aggregator) SetSnapshotBuildSema(semaphore *semaphore.Weighted) {
18051830
func (a *Aggregator) SetProduceMod(produce bool) {
18061831
a.produce = produce
18071832
}
1833+
1834+
// SetFrozenBlocksProvider caps state collation at the block-snapshots boundary.
1835+
// Without it, state files may advance past block files — recovery requires
1836+
// `erigon seg rm-state --latest`. See #20701.
1837+
func (a *Aggregator) SetFrozenBlocksProvider(p FrozenBlocksProvider) {
1838+
a.frozenBlocks = p
1839+
}
18081840
func (a *Aggregator) BuildFilesInBackground(txNum uint64) chan struct{} {
18091841
return a.buildFilesInBackground(txNum, true)
18101842
}

‎db/test/aggregator_ext_test.go‎

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -769,6 +769,180 @@ func TestAggregatorV3_BuildFiles_WithReorgDepth(t *testing.T) {
769769
require.Equal(t, kv.Step(6), kv.Step(agg.EndTxNumMinimax()/agg.StepSize()))
770770
}
771771

772+
// With txnsPerBlock=1, fakeFrozenBlocks(N) yields capTxNum=N.
773+
type fakeFrozenBlocks uint64
774+
775+
func (f fakeFrozenBlocks) FrozenBlocks() uint64 { return uint64(f) }
776+
777+
// Regression for #20701: cap must clamp the loop, not just early-return.
778+
func TestAggregatorV3_BuildFiles_RespectsMaxCollatableTxNumCap(t *testing.T) {
779+
if testing.Short() {
780+
t.Skip("slow test")
781+
}
782+
ctx := t.Context()
783+
logger := log.New()
784+
dirs := datadir.New(t.TempDir())
785+
db := mdbx.New(dbcfg.ChainDB, logger).InMem(t, dirs.Chaindata).GrowthStep(32 * datasize.MB).MapSize(2 * datasize.GB).MustOpen()
786+
t.Cleanup(db.Close)
787+
agg := state.NewTest(dirs).ReorgBlockDepth(1).StepSize(2).Logger(logger).MustOpen(ctx, db)
788+
t.Cleanup(agg.Close)
789+
require.NoError(t, agg.OpenFolder())
790+
791+
// Cap at 4 steps; write 9 steps' worth into the DB.
792+
const capTxNum = uint64(8)
793+
agg.SetFrozenBlocksProvider(fakeFrozenBlocks(capTxNum))
794+
795+
tdb, err := temporal.New(db, agg)
796+
require.NoError(t, err)
797+
t.Cleanup(tdb.Close)
798+
tx, err := tdb.BeginTemporalRw(ctx)
799+
require.NoError(t, err)
800+
t.Cleanup(tx.Rollback)
801+
doms, err := execctx.NewSharedDomains(context.Background(), tx, logger)
802+
require.NoError(t, err)
803+
t.Cleanup(doms.Close)
804+
805+
const txnNums = uint64(18)
806+
const txnsPerBlock = uint64(1)
807+
for i := uint64(0); i < txnNums/txnsPerBlock; i++ {
808+
require.NoError(t, rawdbv3.TxNums.Append(tx, i+1, (i+1)*txnsPerBlock))
809+
}
810+
generateSharedDomainsUpdates(t, doms, tx, txnNums, newRnd(0), length.Addr, 10, txnsPerBlock)
811+
require.NoError(t, doms.Flush(ctx, tx))
812+
require.NoError(t, tx.Commit())
813+
814+
require.NoError(t, agg.BuildFiles(txnNums))
815+
816+
require.Equal(t, capTxNum, agg.EndTxNumMinimax(),
817+
"EndTxNumMinimax must be clamped to cap (%d), not driven by lastInDB", capTxNum)
818+
require.Equal(t, kv.Step(capTxNum/agg.StepSize()), kv.Step(agg.EndTxNumMinimax()/agg.StepSize()))
819+
}
820+
821+
// nil provider = no cap; behavior unchanged.
822+
func TestAggregatorV3_BuildFiles_NoCapHookBuildsAll(t *testing.T) {
823+
if testing.Short() {
824+
t.Skip("slow test")
825+
}
826+
ctx := t.Context()
827+
logger := log.New()
828+
dirs := datadir.New(t.TempDir())
829+
db := mdbx.New(dbcfg.ChainDB, logger).InMem(t, dirs.Chaindata).GrowthStep(32 * datasize.MB).MapSize(2 * datasize.GB).MustOpen()
830+
t.Cleanup(db.Close)
831+
agg := state.NewTest(dirs).ReorgBlockDepth(1).StepSize(2).Logger(logger).MustOpen(ctx, db)
832+
t.Cleanup(agg.Close)
833+
require.NoError(t, agg.OpenFolder())
834+
835+
tdb, err := temporal.New(db, agg)
836+
require.NoError(t, err)
837+
t.Cleanup(tdb.Close)
838+
tx, err := tdb.BeginTemporalRw(ctx)
839+
require.NoError(t, err)
840+
t.Cleanup(tx.Rollback)
841+
doms, err := execctx.NewSharedDomains(context.Background(), tx, logger)
842+
require.NoError(t, err)
843+
t.Cleanup(doms.Close)
844+
845+
const txnNums = uint64(18)
846+
const txnsPerBlock = uint64(1)
847+
for i := uint64(0); i < txnNums/txnsPerBlock; i++ {
848+
require.NoError(t, rawdbv3.TxNums.Append(tx, i+1, (i+1)*txnsPerBlock))
849+
}
850+
generateSharedDomainsUpdates(t, doms, tx, txnNums, newRnd(0), length.Addr, 10, txnsPerBlock)
851+
require.NoError(t, doms.Flush(ctx, tx))
852+
require.NoError(t, tx.Commit())
853+
854+
require.NoError(t, agg.BuildFiles(txnNums))
855+
856+
// reorg-depth=1 holds the last block back; cap unset.
857+
require.Equal(t, uint64(16), agg.EndTxNumMinimax())
858+
}
859+
860+
// BuildFiles2 path (stage_custom_trace, squeeze) must honor the cap too.
861+
func TestAggregatorV3_BuildFiles2_RespectsMaxCollatableTxNumCap(t *testing.T) {
862+
if testing.Short() {
863+
t.Skip("slow test")
864+
}
865+
ctx := t.Context()
866+
logger := log.New()
867+
dirs := datadir.New(t.TempDir())
868+
db := mdbx.New(dbcfg.ChainDB, logger).InMem(t, dirs.Chaindata).GrowthStep(32 * datasize.MB).MapSize(2 * datasize.GB).MustOpen()
869+
t.Cleanup(db.Close)
870+
agg := state.NewTest(dirs).ReorgBlockDepth(1).StepSize(2).Logger(logger).MustOpen(ctx, db)
871+
t.Cleanup(agg.Close)
872+
require.NoError(t, agg.OpenFolder())
873+
874+
const capTxNum = uint64(8)
875+
agg.SetFrozenBlocksProvider(fakeFrozenBlocks(capTxNum))
876+
877+
tdb, err := temporal.New(db, agg)
878+
require.NoError(t, err)
879+
t.Cleanup(tdb.Close)
880+
tx, err := tdb.BeginTemporalRw(ctx)
881+
require.NoError(t, err)
882+
t.Cleanup(tx.Rollback)
883+
doms, err := execctx.NewSharedDomains(context.Background(), tx, logger)
884+
require.NoError(t, err)
885+
t.Cleanup(doms.Close)
886+
887+
const txnNums = uint64(18)
888+
for i := uint64(0); i < txnNums; i++ {
889+
require.NoError(t, rawdbv3.TxNums.Append(tx, i+1, i+1))
890+
}
891+
generateSharedDomainsUpdates(t, doms, tx, txnNums, newRnd(0), length.Addr, 10, 1)
892+
require.NoError(t, doms.Flush(ctx, tx))
893+
require.NoError(t, tx.Commit())
894+
895+
require.NoError(t, agg.BuildFiles2(ctx, 0, kv.Step(txnNums/agg.StepSize()), false))
896+
agg.WaitForFiles()
897+
898+
require.Equal(t, capTxNum, agg.EndTxNumMinimax(),
899+
"BuildFiles2 must clamp to cap (%d), not toStep", capTxNum)
900+
}
901+
902+
// Cap below visibleFilesMinimaxTxNum: defensive only — no rollback.
903+
func TestAggregatorV3_BuildFiles_CapBelowVisibleIsNoop(t *testing.T) {
904+
if testing.Short() {
905+
t.Skip("slow test")
906+
}
907+
ctx := t.Context()
908+
logger := log.New()
909+
dirs := datadir.New(t.TempDir())
910+
db := mdbx.New(dbcfg.ChainDB, logger).InMem(t, dirs.Chaindata).GrowthStep(32 * datasize.MB).MapSize(2 * datasize.GB).MustOpen()
911+
t.Cleanup(db.Close)
912+
agg := state.NewTest(dirs).ReorgBlockDepth(1).StepSize(2).Logger(logger).MustOpen(ctx, db)
913+
t.Cleanup(agg.Close)
914+
require.NoError(t, agg.OpenFolder())
915+
916+
tdb, err := temporal.New(db, agg)
917+
require.NoError(t, err)
918+
t.Cleanup(tdb.Close)
919+
tx, err := tdb.BeginTemporalRw(ctx)
920+
require.NoError(t, err)
921+
t.Cleanup(tx.Rollback)
922+
doms, err := execctx.NewSharedDomains(context.Background(), tx, logger)
923+
require.NoError(t, err)
924+
t.Cleanup(doms.Close)
925+
926+
const txnNums = uint64(18)
927+
const txnsPerBlock = uint64(1)
928+
for i := uint64(0); i < txnNums/txnsPerBlock; i++ {
929+
require.NoError(t, rawdbv3.TxNums.Append(tx, i+1, (i+1)*txnsPerBlock))
930+
}
931+
generateSharedDomainsUpdates(t, doms, tx, txnNums, newRnd(0), length.Addr, 10, txnsPerBlock)
932+
require.NoError(t, doms.Flush(ctx, tx))
933+
require.NoError(t, tx.Commit())
934+
935+
// build state up to txNum 8.
936+
agg.SetFrozenBlocksProvider(fakeFrozenBlocks(8))
937+
require.NoError(t, agg.BuildFiles(txnNums))
938+
require.Equal(t, uint64(8), agg.EndTxNumMinimax())
939+
940+
// lower cap below current visible — must be a no-op, no rollback.
941+
agg.SetFrozenBlocksProvider(fakeFrozenBlocks(4))
942+
require.NoError(t, agg.BuildFiles(txnNums))
943+
require.Equal(t, uint64(8), agg.EndTxNumMinimax())
944+
}
945+
772946
func compareMapsBytes(t *testing.T, m1, m2 map[string][]byte) {
773947
t.Helper()
774948
for k, v := range m1 {

‎execution/stagedsync/exec3.go‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,6 @@ import (
3838
"github.com/erigontech/erigon/db/rawdb"
3939
"github.com/erigontech/erigon/db/rawdb/rawdbhelpers"
4040
"github.com/erigontech/erigon/db/rawdb/rawtemporaldb"
41-
"github.com/erigontech/erigon/db/services"
4241
dbstate "github.com/erigontech/erigon/db/state"
4342
"github.com/erigontech/erigon/db/state/execctx"
4443
"github.com/erigontech/erigon/execution/commitment"
@@ -142,11 +141,7 @@ func ExecV3(ctx context.Context,
142141
}
143142

144143
if execStage.SyncMode() == stages.ModeApplyingBlocks {
145-
maxCollatable, err := services.MaxCollatableTxNum(ctx, applyTx, cfg.blockReader)
146-
if err != nil {
147-
return err
148-
}
149-
agg.BuildFilesInBackground(min(initialTxNum, maxCollatable))
144+
agg.BuildFilesInBackground(initialTxNum)
150145
}
151146

152147
var (

‎node/eth/backend.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1433,6 +1433,7 @@ func SetUpBlockReader(ctx context.Context, db kv.RwDB, dirs datadir.Dirs, snConf
14331433
}
14341434
agg.SetSnapshotBuildSema(blockSnapBuildSema)
14351435
agg.SetProduceMod(snConfig.Snapshot.ProduceE3)
1436+
agg.SetFrozenBlocksProvider(blockReader)
14361437

14371438
allSegmentsDownloadComplete, err := rawdb.AllSegmentsDownloadCompleteFromDB(db)
14381439
if err != nil {

0 commit comments

Comments
 (0)