Caplin: Added more logs to --caplin.archive (#13074)

Co-authored-by: Kewei <kewei.train@gmail.com>

Author: Giulio2002

cl/antiquary/antiquary.go

@@ -217,7 +217,13 @@ func (a *Antiquary) Loop() error {
 		}
 	}
 
-	a.stateSn.OpenFolder()
+	if a.stateSn != nil {
+		if err := a.stateSn.OpenFolder(); err != nil {
+			return err
+		}
+	}
+	log.Info("[Caplin] Stat", "blocks-static", a.sn.BlocksAvailable(), "states-static", a.stateSn.BlocksAvailable(), "blobs-static", a.sn.FrozenBlobs(),
+		"state-history-enabled", a.states, "block-history-enabled", a.blocks, "blob-history-enabled", a.blobs, "snapgen", a.snapgen)
 
 	frozenSlots := a.sn.BlocksAvailable()
 	if frozenSlots != 0 {

cl/antiquary/state_antiquary.go

@@ -57,11 +57,11 @@ func (s *Antiquary) loopStates(ctx context.Context) {
 	// Execute this each second
 	reqRetryTimer := time.NewTicker(100 * time.Millisecond)
 	defer reqRetryTimer.Stop()
+
 	if !initial_state.IsGenesisStateSupported(clparams.NetworkType(s.cfg.DepositNetworkID)) {
 		s.logger.Warn("Genesis state is not supported for this network, no historical states data will be available")
 		return
 	}
-
 	_, beforeFinalized, err := s.readHistoricalProcessingProgress(ctx)
 	if err != nil {
 		s.logger.Error("Failed to read historical processing progress", "err", err)
@@ -81,14 +81,16 @@ func (s *Antiquary) loopStates(ctx context.Context) {
 				s.logger.Error("Failed to read historical processing progress", "err", err)
 				continue
 			}
+			if s.sn == nil || s.syncedData.Syncing() {
+				continue
+			}
+
 			// We wait for updated finality.
 			if finalized == beforeFinalized {
 				continue
 			}
 			beforeFinalized = finalized
-			if s.sn == nil || s.syncedData.Syncing() {
-				continue
-			}
+
 			if err := s.IncrementBeaconState(ctx, finalized); err != nil {
 				if s.currentState != nil {
 					s.logger.Warn("Could not to increment beacon state, trying again later", "err", err, "slot", s.currentState.Slot())
@@ -140,9 +142,16 @@ func FillStaticValidatorsTableIfNeeded(ctx context.Context, logger log.Logger, s
 	blocksAvaiable := stateSn.BlocksAvailable()
 	stateSnRoTx := stateSn.View()
 	defer stateSnRoTx.Close()
-
+	log.Info("[Caplin-Archive] filling validators table", "from", 0, "to", stateSn.BlocksAvailable())
+	logTicker := time.NewTicker(10 * time.Second)
+	defer logTicker.Stop()
 	start := time.Now()
 	for slot := uint64(0); slot <= stateSn.BlocksAvailable(); slot++ {
+		select {
+		case <-logTicker.C:
+			log.Info("[Caplin-Archive] Filled validators table", "progress", fmt.Sprintf("%d/%d", slot, stateSn.BlocksAvailable()))
+		default:
+		}
 		seg, ok := stateSnRoTx.VisibleSegment(slot, kv.StateEvents)
 		if !ok {
 			return false, fmt.Errorf("segment not found for slot %d", slot)
@@ -437,7 +446,7 @@ func (s *Antiquary) IncrementBeaconState(ctx context.Context, to uint64) error {
 		// We now do some post-processing on the state.
 		select {
 		case <-progressTimer.C:
-			log.Log(logLvl, "State processing progress", "slot", slot, "blk/sec", fmt.Sprintf("%.2f", float64(slot-prevSlot)/60))
+			log.Log(logLvl, "[Caplin-Archive] Historical States reconstruction", "slot", slot, "blk/sec", fmt.Sprintf("%.2f", float64(slot-prevSlot)/60))
 			prevSlot = slot
 		default:
 		}
@@ -491,7 +500,7 @@ 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)
+	log.Info("[Caplin-Archive] Historical states antiquated", "slot", s.currentState.Slot(), "root", libcommon.Hash(stateRoot), "latency", endTime)
 	if s.stateSn != nil {
 		if err := s.stateSn.OpenFolder(); err != nil {
 			return err

cl/phase1/stages/forkchoice.go

@@ -307,6 +307,9 @@ func postForkchoiceOperations(ctx context.Context, tx kv.RwTx, logger log.Logger
 	if _, err = cfg.attestationDataProducer.ProduceAndCacheAttestationData(tx, headState, headRoot, headState.Slot(), 0); err != nil {
 		logger.Warn("failed to produce and cache attestation data", "err", err)
 	}
+	if err := beacon_indicies.WriteHighestFinalized(tx, cfg.forkChoice.FinalizedSlot()); err != nil {
+		return err
+	}
 	start := time.Now()
 	cfg.forkChoice.SetSynced(true) // Now we are synced
 	// Update the head state with the new head state