Caplin: Automatic retirement of state tables to their own snapshot files

cl/antiquary/antiquary.go

@@ -50,6 +50,7 @@ type Antiquary struct {
 	downloader                     proto_downloader.DownloaderClient
 	logger                         log.Logger
 	sn                             *freezeblocks.CaplinSnapshots
+	stateSn                        *freezeblocks.CaplinStateSnapshots
 	snReader                       freezeblocks.BeaconSnapshotReader
 	snBuildSema                    *semaphore.Weighted // semaphore for building only one type (blocks, caplin, v3) at a time
 	ctx                            context.Context
@@ -65,7 +66,7 @@ type Antiquary struct {
 	balances32   []byte
 }
 
-func NewAntiquary(ctx context.Context, blobStorage blob_storage.BlobStorage, genesisState *state.CachingBeaconState, validatorsTable *state_accessors.StaticValidatorTable, cfg *clparams.BeaconChainConfig, dirs datadir.Dirs, downloader proto_downloader.DownloaderClient, mainDB kv.RwDB, sn *freezeblocks.CaplinSnapshots, reader freezeblocks.BeaconSnapshotReader, logger log.Logger, states, blocks, blobs, snapgen bool, snBuildSema *semaphore.Weighted) *Antiquary {
+func NewAntiquary(ctx context.Context, blobStorage blob_storage.BlobStorage, genesisState *state.CachingBeaconState, validatorsTable *state_accessors.StaticValidatorTable, cfg *clparams.BeaconChainConfig, dirs datadir.Dirs, downloader proto_downloader.DownloaderClient, mainDB kv.RwDB, stateSn *freezeblocks.CaplinStateSnapshots, sn *freezeblocks.CaplinSnapshots, reader freezeblocks.BeaconSnapshotReader, logger log.Logger, states, blocks, blobs, snapgen bool, snBuildSema *semaphore.Weighted) *Antiquary {
 	backfilled := &atomic.Bool{}
 	blobBackfilled := &atomic.Bool{}
 	backfilled.Store(false)
@@ -89,6 +90,7 @@ func NewAntiquary(ctx context.Context, blobStorage blob_storage.BlobStorage, gen
 		blocks:          blocks,
 		blobs:           blobs,
 		snapgen:         snapgen,
+		stateSn:         stateSn,
 	}
 }
 

cl/antiquary/state_antiquary.go

@@ -20,13 +20,15 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"runtime"
 	"sync"
 	"time"
 
 	"github.com/klauspost/compress/zstd"
 
 	"github.com/erigontech/erigon-lib/common"
 	libcommon "github.com/erigontech/erigon-lib/common"
+	"github.com/erigontech/erigon-lib/downloader/snaptype"
 	"github.com/erigontech/erigon-lib/etl"
 	"github.com/erigontech/erigon-lib/kv"
 	"github.com/erigontech/erigon-lib/log/v3"
@@ -111,6 +113,9 @@ func (s *Antiquary) readHistoricalProcessingProgress(ctx context.Context) (progr
 	if err != nil {
 		return
 	}
+	if s.stateSn != nil {
+		progress = max(progress, s.stateSn.BlocksAvailable())
+	}
 
 	finalized, err = beacon_indicies.ReadHighestFinalized(tx)
 	if err != nil {
@@ -119,17 +124,84 @@ func (s *Antiquary) readHistoricalProcessingProgress(ctx context.Context) (progr
 	return
 }
 
+func (s *Antiquary) fillStaticValidatorsTable(ctx context.Context) error {
+	if s.stateSn == nil || s.stateSn.BlocksAvailable() <= s.validatorsTable.Slot() {
+		return nil
+	}
+	if err := s.stateSn.OpenFolder(); err != nil {
+		return err
+	}
+	blocksAvaiable := s.stateSn.BlocksAvailable()
+	stateSnRoTx := s.stateSn.View()
+	defer stateSnRoTx.Close()
+	if s.genesisState == nil {
+		return fmt.Errorf("genesis state is nil")
+	}
+	startSlot := s.validatorsTable.Slot() + 1
+	if startSlot == 1 {
+		startSlot = 0
+	}
+
+	start := time.Now()
+	for slot := s.validatorsTable.Slot() + 1; slot <= blocksAvaiable; slot++ {
+		seg, ok := stateSnRoTx.VisibleSegment(slot, kv.StateEvents)
+		if !ok {
+			return fmt.Errorf("segment not found for slot %d", slot)
+		}
+		buf, err := seg.Get(slot)
+		if err != nil {
+			return err
+		}
+		if len(buf) == 0 {
+			continue
+		}
+		event := state_accessors.NewStateEventsFromBytes(buf)
+		state_accessors.ReplayEvents(
+			func(validatorIndex uint64, validator solid.Validator) error {
+				return s.validatorsTable.AddValidator(validator, validatorIndex, slot)
+			},
+			func(validatorIndex uint64, exitEpoch uint64) error {
+				return s.validatorsTable.AddExitEpoch(validatorIndex, slot, exitEpoch)
+			},
+			func(validatorIndex uint64, withdrawableEpoch uint64) error {
+				return s.validatorsTable.AddWithdrawableEpoch(validatorIndex, slot, withdrawableEpoch)
+			},
+			func(validatorIndex uint64, withdrawalCredentials libcommon.Hash) error {
+				return s.validatorsTable.AddWithdrawalCredentials(validatorIndex, slot, withdrawalCredentials)
+			},
+			func(validatorIndex uint64, activationEpoch uint64) error {
+				return s.validatorsTable.AddActivationEpoch(validatorIndex, slot, activationEpoch)
+			},
+			func(validatorIndex uint64, activationEligibilityEpoch uint64) error {
+				return s.validatorsTable.AddActivationEligibility(validatorIndex, slot, activationEligibilityEpoch)
+			},
+			func(validatorIndex uint64, slashed bool) error {
+				return s.validatorsTable.AddSlashed(validatorIndex, slot, slashed)
+			},
+			event,
+		)
+		s.validatorsTable.SetSlot(slot)
+	}
+	s.logger.Info("[Antiquary] Filled static validators table", "slots", blocksAvaiable, "elapsed", time.Since(start))
+	return nil
+}
+
 func (s *Antiquary) IncrementBeaconState(ctx context.Context, to uint64) error {
 	var tx kv.Tx
 
+	// Check if you need to fill the static validators table
+	if err := s.fillStaticValidatorsTable(ctx); err != nil {
+		return err
+	}
+
 	tx, err := s.mainDB.BeginRo(ctx)
 	if err != nil {
 		return err
 	}
 	defer tx.Rollback()
 
 	// maps which validators changes
-	changedValidators := make(map[uint64]struct{})
+	var changedValidators sync.Map
 
 	stateAntiquaryCollector := newBeaconStatesCollector(s.cfg, s.dirs.Tmp, s.logger)
 	defer stateAntiquaryCollector.close()
@@ -144,7 +216,7 @@ func (s *Antiquary) IncrementBeaconState(ctx context.Context, to uint64) error {
 		}
 		// Mark all validators as touched because we just initizialized the whole state.
 		s.currentState.ForEachValidator(func(v solid.Validator, index, total int) bool {
-			changedValidators[uint64(index)] = struct{}{}
+			changedValidators.Store(uint64(index), struct{}{})
 			if err = s.validatorsTable.AddValidator(v, uint64(index), 0); err != nil {
 				return false
 			}
@@ -175,37 +247,37 @@ func (s *Antiquary) IncrementBeaconState(ctx context.Context, to uint64) error {
 			return stateAntiquaryCollector.collectIntraEpochRandaoMix(slot, mix)
 		},
 		OnNewValidator: func(index int, v solid.Validator, balance uint64) error {
-			changedValidators[uint64(index)] = struct{}{}
+			changedValidators.Store(uint64(index), struct{}{})
 			events.AddValidator(uint64(index), v)
 			return s.validatorsTable.AddValidator(v, uint64(index), slot)
 		},
 		OnNewValidatorActivationEpoch: func(index int, epoch uint64) error {
-			changedValidators[uint64(index)] = struct{}{}
+			changedValidators.Store(uint64(index), struct{}{})
 			events.ChangeActivationEpoch(uint64(index), epoch)
 			return s.validatorsTable.AddActivationEpoch(uint64(index), slot, epoch)
 		},
 		OnNewValidatorExitEpoch: func(index int, epoch uint64) error {
-			changedValidators[uint64(index)] = struct{}{}
+			changedValidators.Store(uint64(index), struct{}{})
 			events.ChangeExitEpoch(uint64(index), epoch)
 			return s.validatorsTable.AddExitEpoch(uint64(index), slot, epoch)
 		},
 		OnNewValidatorWithdrawableEpoch: func(index int, epoch uint64) error {
-			changedValidators[uint64(index)] = struct{}{}
+			changedValidators.Store(uint64(index), struct{}{})
 			events.ChangeWithdrawableEpoch(uint64(index), epoch)
 			return s.validatorsTable.AddWithdrawableEpoch(uint64(index), slot, epoch)
 		},
 		OnNewValidatorSlashed: func(index int, newSlashed bool) error {
-			changedValidators[uint64(index)] = struct{}{}
+			changedValidators.Store(uint64(index), struct{}{})
 			events.ChangeSlashed(uint64(index), newSlashed)
 			return s.validatorsTable.AddSlashed(uint64(index), slot, newSlashed)
 		},
 		OnNewValidatorActivationEligibilityEpoch: func(index int, epoch uint64) error {
-			changedValidators[uint64(index)] = struct{}{}
+			changedValidators.Store(uint64(index), struct{}{})
 			events.ChangeActivationEligibilityEpoch(uint64(index), epoch)
 			return s.validatorsTable.AddActivationEligibility(uint64(index), slot, epoch)
 		},
 		OnNewValidatorWithdrawalCredentials: func(index int, wc []byte) error {
-			changedValidators[uint64(index)] = struct{}{}
+			changedValidators.Store(uint64(index), struct{}{})
 			events.ChangeWithdrawalCredentials(uint64(index), libcommon.BytesToHash(wc))
 			return s.validatorsTable.AddWithdrawalCredentials(uint64(index), slot, libcommon.BytesToHash(wc))
 		},
@@ -389,7 +461,7 @@ func (s *Antiquary) IncrementBeaconState(ctx context.Context, to uint64) error {
 
 	buf := &bytes.Buffer{}
 	s.validatorsTable.ForEach(func(validatorIndex uint64, validator *state_accessors.StaticValidator) bool {
-		if _, ok := changedValidators[validatorIndex]; !ok {
+		if _, ok := changedValidators.Load(validatorIndex); !ok {
 			return true
 		}
 		buf.Reset()
@@ -413,6 +485,41 @@ func (s *Antiquary) IncrementBeaconState(ctx context.Context, to uint64) error {
 		return err
 	}
 	log.Info("Historical states antiquated", "slot", s.currentState.Slot(), "root", libcommon.Hash(stateRoot), "latency", endTime)
+	if s.snapgen {
+		if err := s.stateSn.OpenFolder(); err != nil {
+			return err
+		}
+		blocksPerStatefulFile := uint64(snaptype.CaplinMergeLimit * 5)
+		from := s.stateSn.BlocksAvailable() + 1
+		if from+blocksPerStatefulFile+safetyMargin > s.currentState.Slot() {
+			return nil
+		}
+		to := s.currentState.Slot()
+		if to < (safetyMargin + blocksPerStatefulFile) {
+			return nil
+		}
+		to = to - (safetyMargin + blocksPerStatefulFile)
+		if from >= to {
+			return nil
+		}
+		if err := s.stateSn.DumpCaplinState(
+			ctx,
+			s.stateSn.BlocksAvailable()+1,
+			to,
+			blocksPerStatefulFile,
+			s.sn.Salt,
+			s.dirs,
+			runtime.NumCPU(),
+			log.LvlInfo,
+			s.logger,
+		); err != nil {
+			return err
+		}
+		if err := s.stateSn.OpenFolder(); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -439,12 +546,15 @@ func (s *Antiquary) initializeStateAntiquaryIfNeeded(ctx context.Context, tx kv.
 	if err != nil {
 		return err
 	}
+	if s.stateSn != nil {
+		targetSlot = max(targetSlot, s.stateSn.BlocksAvailable())
+	}
 	// We want to backoff by some slots until we get a correct state from DB.
 	// we start from 10 * clparams.SlotsPerDump.
 	backoffStrides := uint64(10)
 	backoffStep := backoffStrides
 
-	historicalReader := historical_states_reader.NewHistoricalStatesReader(s.cfg, s.snReader, s.validatorsTable, s.genesisState)
+	historicalReader := historical_states_reader.NewHistoricalStatesReader(s.cfg, s.snReader, s.validatorsTable, s.genesisState, s.stateSn)
 
 	for {
 		attempt, err := computeSlotToBeRequested(tx, s.cfg, s.genesisState.Slot(), targetSlot, backoffStep)
@@ -465,6 +575,7 @@ func (s *Antiquary) initializeStateAntiquaryIfNeeded(ctx context.Context, tx kv.
 		if err != nil {
 			return fmt.Errorf("failed to read historical state at slot %d: %w", attempt, err)
 		}
+		fmt.Println("currentState", s.currentState, attempt)
 		if s.currentState == nil {
 			log.Warn("historical state not found, backoff more and try again", "slot", attempt)
 			backoffStep += backoffStrides

cl/beacon/handler/attestation_rewards.go

@@ -178,26 +178,30 @@ func (a *ApiHandler) PostEthV1BeaconRewardsAttestations(w http.ResponseWriter, r
 	if lastSlot > stateProgress {
 		return nil, beaconhttp.NewEndpointError(http.StatusNotFound, errors.New("requested range is not yet processed or the node is not archivial"))
 	}
+	snRoTx := a.caplinStateSnapshots.View()
+	defer snRoTx.Close()
 
-	epochData, err := state_accessors.ReadEpochData(tx, a.beaconChainCfg.RoundSlotToEpoch(lastSlot))
+	stateGetter := state_accessors.GetValFnTxAndSnapshot(tx, snRoTx)
+
+	epochData, err := state_accessors.ReadEpochData(stateGetter, a.beaconChainCfg.RoundSlotToEpoch(lastSlot))
 	if err != nil {
 		return nil, err
 	}
 
-	validatorSet, err := a.stateReader.ReadValidatorsForHistoricalState(tx, lastSlot)
+	validatorSet, err := a.stateReader.ReadValidatorsForHistoricalState(tx, stateGetter, lastSlot)
 	if err != nil {
 		return nil, err
 	}
 	if validatorSet == nil {
 		return nil, beaconhttp.NewEndpointError(http.StatusNotFound, errors.New("no validator set found for this epoch"))
 	}
 
-	_, previousIdx, err := a.stateReader.ReadParticipations(tx, lastSlot)
+	_, previousIdx, err := a.stateReader.ReadParticipations(tx, stateGetter, lastSlot)
 	if err != nil {
 		return nil, err
 	}
 
-	_, _, finalizedCheckpoint, ok, err := state_accessors.ReadCheckpoints(tx, epoch*a.beaconChainCfg.SlotsPerEpoch)
+	_, _, finalizedCheckpoint, ok, err := state_accessors.ReadCheckpoints(stateGetter, epoch*a.beaconChainCfg.SlotsPerEpoch)
 	if err != nil {
 		return nil, err
 	}
@@ -212,7 +216,7 @@ func (a *ApiHandler) PostEthV1BeaconRewardsAttestations(w http.ResponseWriter, r
 		return resp.WithFinalized(true).WithOptimistic(a.forkchoiceStore.IsRootOptimistic(root)), nil
 	}
 	inactivityScores := solid.NewUint64ListSSZ(int(a.beaconChainCfg.ValidatorRegistryLimit))
-	if err := a.stateReader.ReconstructUint64ListDump(tx, lastSlot, kv.InactivityScores, validatorSet.Length(), inactivityScores); err != nil {
+	if err := a.stateReader.ReconstructUint64ListDump(stateGetter, lastSlot, kv.InactivityScores, validatorSet.Length(), inactivityScores); err != nil {
 		return nil, err
 	}
 	resp, err := a.computeAttestationsRewardsForAltair(

cl/beacon/handler/committees.go

@@ -119,8 +119,13 @@ func (a *ApiHandler) getCommittees(w http.ResponseWriter, r *http.Request) (*bea
 		}
 		return newBeaconResponse(resp).WithFinalized(isFinalized).WithOptimistic(isOptimistic), nil
 	}
+	snRoTx := a.caplinStateSnapshots.View()
+	defer snRoTx.Close()
+	stateGetter := state_accessors.GetValFnTxAndSnapshot(tx, snRoTx)
 	// finality case
-	activeIdxs, err := state_accessors.ReadActiveIndicies(tx, epoch*a.beaconChainCfg.SlotsPerEpoch)
+	activeIdxs, err := state_accessors.ReadActiveIndicies(
+		stateGetter,
+		epoch*a.beaconChainCfg.SlotsPerEpoch)
 	if err != nil {
 		return nil, err
 	}
@@ -134,7 +139,7 @@ func (a *ApiHandler) getCommittees(w http.ResponseWriter, r *http.Request) (*bea
 	}
 
 	mixPosition := (epoch + a.beaconChainCfg.EpochsPerHistoricalVector - a.beaconChainCfg.MinSeedLookahead - 1) % a.beaconChainCfg.EpochsPerHistoricalVector
-	mix, err := a.stateReader.ReadRandaoMixBySlotAndIndex(tx, epoch*a.beaconChainCfg.SlotsPerEpoch, mixPosition)
+	mix, err := a.stateReader.ReadRandaoMixBySlotAndIndex(tx, stateGetter, epoch*a.beaconChainCfg.SlotsPerEpoch, mixPosition)
 	if err != nil {
 		return nil, beaconhttp.NewEndpointError(http.StatusNotFound, fmt.Errorf("could not read randao mix: %v", err))
 	}

cl/beacon/handler/duties_attester.go

@@ -149,8 +149,15 @@ func (a *ApiHandler) getAttesterDuties(w http.ResponseWriter, r *http.Request) (
 	if (epoch)*a.beaconChainCfg.SlotsPerEpoch >= stageStateProgress {
 		return nil, beaconhttp.NewEndpointError(http.StatusBadRequest, fmt.Errorf("epoch %d is too far in the future", epoch))
 	}
+
+	snRoTx := a.caplinStateSnapshots.View()
+	defer snRoTx.Close()
+
+	stateGetter := state_accessors.GetValFnTxAndSnapshot(tx, snRoTx)
 	// finality case
-	activeIdxs, err := state_accessors.ReadActiveIndicies(tx, epoch*a.beaconChainCfg.SlotsPerEpoch)
+	activeIdxs, err := state_accessors.ReadActiveIndicies(
+		stateGetter,
+		epoch*a.beaconChainCfg.SlotsPerEpoch)
 	if err != nil {
 		return nil, err
 	}
@@ -164,7 +171,7 @@ func (a *ApiHandler) getAttesterDuties(w http.ResponseWriter, r *http.Request) (
 	}
 
 	mixPosition := (epoch + a.beaconChainCfg.EpochsPerHistoricalVector - a.beaconChainCfg.MinSeedLookahead - 1) % a.beaconChainCfg.EpochsPerHistoricalVector
-	mix, err := a.stateReader.ReadRandaoMixBySlotAndIndex(tx, epoch*a.beaconChainCfg.SlotsPerEpoch, mixPosition)
+	mix, err := a.stateReader.ReadRandaoMixBySlotAndIndex(tx, stateGetter, epoch*a.beaconChainCfg.SlotsPerEpoch, mixPosition)
 	if err != nil {
 		return nil, beaconhttp.NewEndpointError(http.StatusNotFound, fmt.Errorf("could not read randao mix: %v", err))
 	}

cl/beacon/handler/duties_sync.go

@@ -81,9 +81,13 @@ func (a *ApiHandler) getSyncDuties(w http.ResponseWriter, r *http.Request) (*bea
 	if !ok {
 		_, syncCommittee, ok = a.forkchoiceStore.GetSyncCommittees(period - 1)
 	}
+	snRoTx := a.caplinStateSnapshots.View()
+	defer snRoTx.Close()
 	// Read them from the archive node if we do not have them in the fast-access storage
 	if !ok {
-		syncCommittee, err = state_accessors.ReadCurrentSyncCommittee(tx, a.beaconChainCfg.RoundSlotToSyncCommitteePeriod(startSlotAtEpoch))
+		syncCommittee, err = state_accessors.ReadCurrentSyncCommittee(
+			state_accessors.GetValFnTxAndSnapshot(tx, snRoTx),
+			a.beaconChainCfg.RoundSlotToSyncCommitteePeriod(startSlotAtEpoch))
 		if syncCommittee == nil {
 			log.Warn("could not find sync committee for epoch", "epoch", epoch, "period", period)
 			return nil, beaconhttp.NewEndpointError(http.StatusNotFound, fmt.Errorf("could not find sync committee for epoch %d", epoch))