Update Earliest available slot when pruning (#15694)

* Update Earliest available slot when pruning

* bazel run //:gazelle -- fix

* custodyUpdater interface to avoid import cycle

* bazel run //:gazelle -- fix

* simplify test

* separation of concerns

* debug log for updating eas

* UpdateEarliestAvailableSlot function in CustodyManager

* fix test

* UpdateEarliestAvailableSlot function for FakeP2P

* lint

* UpdateEarliestAvailableSlot instead of UpdateCustodyInfo + check for Fulu

* fix test and lint

* bugfix: enforce minimum retention period in pruner

* remove MinEpochsForBlockRequests function and use from config

* remove modifying earliest_available_slot after data column pruning

* correct earliestAvailableSlot validation: allow backfill decrease but prevent increase within MIN_EPOCHS_FOR_BLOCK_REQUESTS

* lint

* bazel run //:gazelle -- fix

* lint and remove unwanted debug logs

* Return a wrapped error, and let the caller decide what to do

* fix tests because updateEarliestSlot returns error now

* avoid re-doing computation in the test function

* lint and correct changelog

* custody updater should be a mandatory part of the pruner service

* ensure never increase eas if we are in the block requests window

* slot level granularity edge case

* update the value stored in the DB

* log tidy up

* use errNoCustodyInfo

* allow earliestAvailableSlot edit when custodyGroupCount doesnt change

* undo the minimal config change

* add context to CustodyGroupCount after merging from develop

* cosmetic change

* shift responsibility from caller to callee, protection for updateEarliestSlot. UpdateEarliestAvailableSlot returns cgc

* allow increase in earliestAvailableSlot only when custodyGroupCount also increases

* remove CustodyGroupCount as it is no longer needed as UpdateEarliestAvailableSlot returns cgc now

* proper place for log and name refactor

* test for Nil custody info

* allow decreasing earliest slot in DB (just like in memory)

* invert if statement to make more readable

* UpdateEarliestAvailableSlot for DB (equivalent of p2p's UpdateEarliestAvailableSlot) & undo changes made to UpdateCustodyInfo

* in UpdateEarliestAvailableSlot, no need to return unused values

* no need to log stored group count

* log.WithField instead of log.WithFields

Author: satushh

beacon-chain/blockchain/setup_test.go

@@ -130,6 +130,14 @@ func (dch *mockCustodyManager) UpdateCustodyInfo(earliestAvailableSlot primitive
 	return earliestAvailableSlot, custodyGroupCount, nil
 }
 
+func (dch *mockCustodyManager) UpdateEarliestAvailableSlot(earliestAvailableSlot primitives.Slot) error {
+	dch.mut.Lock()
+	defer dch.mut.Unlock()
+
+	dch.earliestAvailableSlot = earliestAvailableSlot
+	return nil
+}
+
 func (dch *mockCustodyManager) CustodyGroupCountFromPeer(peer.ID) uint64 {
 	return 0
 }

beacon-chain/core/helpers/weak_subjectivity.go

@@ -201,14 +201,3 @@ func ParseWeakSubjectivityInputString(wsCheckpointString string) (*v1alpha1.Chec
 		Root:  bRoot,
 	}, nil
 }
-
-// MinEpochsForBlockRequests computes the number of epochs of block history that we need to maintain,
-// relative to the current epoch, per the p2p specs. This is used to compute the slot where backfill is complete.
-// value defined:
-// https://github.com/ethereum/consensus-specs/blob/master/specs/phase0/p2p-interface.md#configuration
-// MIN_VALIDATOR_WITHDRAWABILITY_DELAY + CHURN_LIMIT_QUOTIENT // 2 (= 33024, ~5 months)
-// detailed rationale: https://github.com/ethereum/consensus-specs/blob/master/specs/phase0/p2p-interface.md#why-are-blocksbyrange-requests-only-required-to-be-served-for-the-latest-min_epochs_for_block_requests-epochs
-func MinEpochsForBlockRequests() primitives.Epoch {
-	return params.BeaconConfig().MinValidatorWithdrawabilityDelay +
-		primitives.Epoch(params.BeaconConfig().ChurnLimitQuotient/2)
-}

beacon-chain/core/helpers/weak_subjectivity_test.go

@@ -286,20 +286,3 @@ func genState(t *testing.T, valCount, avgBalance uint64) state.BeaconState {
 	return beaconState
 }
 
-func TestMinEpochsForBlockRequests(t *testing.T) {
-	helpers.ClearCache()
-
-	params.SetActiveTestCleanup(t, params.MainnetConfig())
-	var expected primitives.Epoch = 33024
-	// expected value of 33024 via spec commentary:
-	// https://github.com/ethereum/consensus-specs/blob/master/specs/phase0/p2p-interface.md#why-are-blocksbyrange-requests-only-required-to-be-served-for-the-latest-min_epochs_for_block_requests-epochs
-	//	MIN_EPOCHS_FOR_BLOCK_REQUESTS is calculated using the arithmetic from compute_weak_subjectivity_period found in the weak subjectivity guide. Specifically to find this max epoch range, we use the worst case event of a very large validator size (>= MIN_PER_EPOCH_CHURN_LIMIT * CHURN_LIMIT_QUOTIENT).
-	//
-	//	MIN_EPOCHS_FOR_BLOCK_REQUESTS = (
-	//	    MIN_VALIDATOR_WITHDRAWABILITY_DELAY
-	//	    + MAX_SAFETY_DECAY * CHURN_LIMIT_QUOTIENT // (2 * 100)
-	//	)
-	//
-	//	Where MAX_SAFETY_DECAY = 100 and thus MIN_EPOCHS_FOR_BLOCK_REQUESTS = 33024 (~5 months).
-	require.Equal(t, expected, helpers.MinEpochsForBlockRequests())
-}

beacon-chain/db/iface/interface.go

@@ -129,6 +129,7 @@ type NoHeadAccessDatabase interface {
 	// Custody operations.
 	UpdateSubscribedToAllDataSubnets(ctx context.Context, subscribed bool) (bool, error)
 	UpdateCustodyInfo(ctx context.Context, earliestAvailableSlot primitives.Slot, custodyGroupCount uint64) (primitives.Slot, uint64, error)
+	UpdateEarliestAvailableSlot(ctx context.Context, earliestAvailableSlot primitives.Slot) error
 
 	// P2P Metadata operations.
 	SaveMetadataSeqNum(ctx context.Context, seqNum uint64) error

beacon-chain/db/kv/custody.go

@@ -2,16 +2,19 @@ package kv
 
 import (
 	"context"
+	"time"
 
+	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v6/encoding/bytesutil"
 	"github.com/OffchainLabs/prysm/v6/monitoring/tracing/trace"
+	"github.com/OffchainLabs/prysm/v6/time/slots"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	bolt "go.etcd.io/bbolt"
 )
 
-// UpdateCustodyInfo atomically updates the custody group count only it is greater than the stored one.
+// UpdateCustodyInfo atomically updates the custody group count only if it is greater than the stored one.
 // In this case, it also updates the earliest available slot with the provided value.
 // It returns the (potentially updated) custody group count and earliest available slot.
 func (s *Store) UpdateCustodyInfo(ctx context.Context, earliestAvailableSlot primitives.Slot, custodyGroupCount uint64) (primitives.Slot, uint64, error) {
@@ -70,6 +73,79 @@ func (s *Store) UpdateCustodyInfo(ctx context.Context, earliestAvailableSlot pri
 	return storedEarliestAvailableSlot, storedGroupCount, nil
 }
 
+// UpdateEarliestAvailableSlot updates the earliest available slot.
+func (s *Store) UpdateEarliestAvailableSlot(ctx context.Context, earliestAvailableSlot primitives.Slot) error {
+	_, span := trace.StartSpan(ctx, "BeaconDB.UpdateEarliestAvailableSlot")
+	defer span.End()
+
+	storedEarliestAvailableSlot := primitives.Slot(0)
+	if err := s.db.Update(func(tx *bolt.Tx) error {
+		// Retrieve the custody bucket.
+		bucket, err := tx.CreateBucketIfNotExists(custodyBucket)
+		if err != nil {
+			return errors.Wrap(err, "create custody bucket")
+		}
+
+		// Retrieve the stored earliest available slot.
+		storedEarliestAvailableSlotBytes := bucket.Get(earliestAvailableSlotKey)
+		if len(storedEarliestAvailableSlotBytes) != 0 {
+			storedEarliestAvailableSlot = primitives.Slot(bytesutil.BytesToUint64BigEndian(storedEarliestAvailableSlotBytes))
+		}
+
+		// Allow decrease (for backfill scenarios)
+		if earliestAvailableSlot <= storedEarliestAvailableSlot {
+			storedEarliestAvailableSlot = earliestAvailableSlot
+			bytes := bytesutil.Uint64ToBytesBigEndian(uint64(earliestAvailableSlot))
+			if err := bucket.Put(earliestAvailableSlotKey, bytes); err != nil {
+				return errors.Wrap(err, "put earliest available slot")
+			}
+			return nil
+		}
+
+		// Prevent increase within the MIN_EPOCHS_FOR_BLOCK_REQUESTS period
+		// This ensures we don't voluntarily refuse to serve mandatory block data
+		genesisTime := time.Unix(int64(params.BeaconConfig().MinGenesisTime+params.BeaconConfig().GenesisDelay), 0)
+		currentSlot := slots.CurrentSlot(genesisTime)
+		currentEpoch := slots.ToEpoch(currentSlot)
+		minEpochsForBlocks := primitives.Epoch(params.BeaconConfig().MinEpochsForBlockRequests)
+
+		// Calculate the minimum required epoch (or 0 if we're early in the chain)
+		minRequiredEpoch := primitives.Epoch(0)
+		if currentEpoch > minEpochsForBlocks {
+			minRequiredEpoch = currentEpoch - minEpochsForBlocks
+		}
+
+		// Convert to slot to ensure we compare at slot-level granularity
+		minRequiredSlot, err := slots.EpochStart(minRequiredEpoch)
+		if err != nil {
+			return errors.Wrap(err, "calculate minimum required slot")
+		}
+
+		// Prevent any increase that would put earliest available slot beyond the minimum required slot
+		if earliestAvailableSlot > minRequiredSlot {
+			return errors.Errorf(
+				"cannot increase earliest available slot to %d (epoch %d) as it exceeds minimum required slot %d (epoch %d)",
+				earliestAvailableSlot, slots.ToEpoch(earliestAvailableSlot),
+				minRequiredSlot, minRequiredEpoch,
+			)
+		}
+
+		storedEarliestAvailableSlot = earliestAvailableSlot
+		bytes := bytesutil.Uint64ToBytesBigEndian(uint64(earliestAvailableSlot))
+		if err := bucket.Put(earliestAvailableSlotKey, bytes); err != nil {
+			return errors.Wrap(err, "put earliest available slot")
+		}
+
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	log.WithField("earliestAvailableSlot", storedEarliestAvailableSlot).Debug("Updated earliest available slot")
+
+	return nil
+}
+
 // UpdateSubscribedToAllDataSubnets updates the "subscribed to all data subnets" status in the database
 // only if `subscribed` is `true`.
 // It returns the previous subscription status.

beacon-chain/db/kv/custody_test.go

@@ -3,10 +3,13 @@ package kv
 import (
 	"context"
 	"testing"
+	"time"
 
+	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v6/encoding/bytesutil"
 	"github.com/OffchainLabs/prysm/v6/testing/require"
+	"github.com/OffchainLabs/prysm/v6/time/slots"
 	bolt "go.etcd.io/bbolt"
 )
 
@@ -132,6 +135,131 @@ func TestUpdateCustodyInfo(t *testing.T) {
 	})
 }
 
+func TestUpdateEarliestAvailableSlot(t *testing.T) {
+	ctx := t.Context()
+
+	t.Run("allow decreasing earliest slot (backfill scenario)", func(t *testing.T) {
+		const (
+			initialSlot  = primitives.Slot(300)
+			initialCount = uint64(10)
+			earliestSlot = primitives.Slot(200) // Lower than initial (backfill discovered earlier blocks)
+		)
+
+		db := setupDB(t)
+
+		// Initialize custody info
+		_, _, err := db.UpdateCustodyInfo(ctx, initialSlot, initialCount)
+		require.NoError(t, err)
+
+		// Update with a lower slot (should update for backfill)
+		err = db.UpdateEarliestAvailableSlot(ctx, earliestSlot)
+		require.NoError(t, err)
+
+		storedSlot, storedCount := getCustodyInfoFromDB(t, db)
+		require.Equal(t, earliestSlot, storedSlot)
+		require.Equal(t, initialCount, storedCount)
+	})
+
+	t.Run("allow increasing slot within MIN_EPOCHS_FOR_BLOCK_REQUESTS (pruning scenario)", func(t *testing.T) {
+		db := setupDB(t)
+
+		// Calculate the current slot and minimum required slot based on actual current time
+		genesisTime := time.Unix(int64(params.BeaconConfig().MinGenesisTime+params.BeaconConfig().GenesisDelay), 0)
+		currentSlot := slots.CurrentSlot(genesisTime)
+		currentEpoch := slots.ToEpoch(currentSlot)
+		minEpochsForBlocks := primitives.Epoch(params.BeaconConfig().MinEpochsForBlockRequests)
+
+		var minRequiredEpoch primitives.Epoch
+		if currentEpoch > minEpochsForBlocks {
+			minRequiredEpoch = currentEpoch - minEpochsForBlocks
+		} else {
+			minRequiredEpoch = 0
+		}
+
+		minRequiredSlot, err := slots.EpochStart(minRequiredEpoch)
+		require.NoError(t, err)
+
+		// Initial setup: set earliest slot well before minRequiredSlot
+		const groupCount = uint64(5)
+		initialSlot := primitives.Slot(1000)
+
+		_, _, err = db.UpdateCustodyInfo(ctx, initialSlot, groupCount)
+		require.NoError(t, err)
+
+		// Try to increase to a slot that's still BEFORE minRequiredSlot (should succeed)
+		validSlot := minRequiredSlot - 100
+
+		err = db.UpdateEarliestAvailableSlot(ctx, validSlot)
+		require.NoError(t, err)
+
+		// Verify the database was updated
+		storedSlot, storedCount := getCustodyInfoFromDB(t, db)
+		require.Equal(t, validSlot, storedSlot)
+		require.Equal(t, groupCount, storedCount)
+	})
+
+	t.Run("prevent increasing slot beyond MIN_EPOCHS_FOR_BLOCK_REQUESTS", func(t *testing.T) {
+		db := setupDB(t)
+
+		// Calculate the current slot and minimum required slot based on actual current time
+		genesisTime := time.Unix(int64(params.BeaconConfig().MinGenesisTime+params.BeaconConfig().GenesisDelay), 0)
+		currentSlot := slots.CurrentSlot(genesisTime)
+		currentEpoch := slots.ToEpoch(currentSlot)
+		minEpochsForBlocks := primitives.Epoch(params.BeaconConfig().MinEpochsForBlockRequests)
+
+		var minRequiredEpoch primitives.Epoch
+		if currentEpoch > minEpochsForBlocks {
+			minRequiredEpoch = currentEpoch - minEpochsForBlocks
+		} else {
+			minRequiredEpoch = 0
+		}
+
+		minRequiredSlot, err := slots.EpochStart(minRequiredEpoch)
+		require.NoError(t, err)
+
+		// Initial setup: set a valid earliest slot (well before minRequiredSlot)
+		const initialCount = uint64(5)
+		initialSlot := primitives.Slot(1000)
+
+		_, _, err = db.UpdateCustodyInfo(ctx, initialSlot, initialCount)
+		require.NoError(t, err)
+
+		// Try to set earliest slot beyond the minimum required slot
+		invalidSlot := minRequiredSlot + 100
+
+		// This should fail
+		err = db.UpdateEarliestAvailableSlot(ctx, invalidSlot)
+		require.ErrorContains(t, "cannot increase earliest available slot", err)
+		require.ErrorContains(t, "exceeds minimum required slot", err)
+
+		// Verify the database wasn't updated
+		storedSlot, storedCount := getCustodyInfoFromDB(t, db)
+		require.Equal(t, initialSlot, storedSlot)
+		require.Equal(t, initialCount, storedCount)
+	})
+
+	t.Run("no change when slot equals current slot", func(t *testing.T) {
+		const (
+			initialSlot  = primitives.Slot(100)
+			initialCount = uint64(5)
+		)
+
+		db := setupDB(t)
+
+		// Initialize custody info
+		_, _, err := db.UpdateCustodyInfo(ctx, initialSlot, initialCount)
+		require.NoError(t, err)
+
+		// Update with the same slot
+		err = db.UpdateEarliestAvailableSlot(ctx, initialSlot)
+		require.NoError(t, err)
+
+		storedSlot, storedCount := getCustodyInfoFromDB(t, db)
+		require.Equal(t, initialSlot, storedSlot)
+		require.Equal(t, initialCount, storedCount)
+	})
+}
+
 func TestUpdateSubscribedToAllDataSubnets(t *testing.T) {
 	ctx := context.Background()
 

beacon-chain/db/pruner/BUILD.bazel

@@ -8,7 +8,6 @@ go_library(
         "//beacon-chain:__subpackages__",
     ],
     deps = [
-        "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/iface:go_default_library",
         "//config/params:go_default_library",
@@ -29,6 +28,7 @@ go_test(
         "//consensus-types/blocks:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
+        "//testing/assert:go_default_library",
         "//testing/require:go_default_library",
         "//testing/util:go_default_library",
         "//time/slots/testing:go_default_library",