Move updateCustodyInfoInDB from the blockchain service to the sync service and use the justified checkpoint (+ 1).

beacon-chain/blockchain/BUILD.bazel

@@ -74,7 +74,6 @@ go_library(
         "//beacon-chain/state:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",
         "//beacon-chain/verification:go_default_library",
-        "//cmd/beacon-chain/flags:go_default_library",
         "//config/features:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
@@ -174,7 +173,6 @@ go_test(
         "//beacon-chain/state/state-native:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",
         "//beacon-chain/verification:go_default_library",
-        "//cmd/beacon-chain/flags:go_default_library",
         "//config/features:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",

beacon-chain/blockchain/service.go

@@ -14,7 +14,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/cache"
 	statefeed "github.com/OffchainLabs/prysm/v7/beacon-chain/core/feed/state"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/helpers"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
 	coreTime "github.com/OffchainLabs/prysm/v7/beacon-chain/core/time"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/transition"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db"
@@ -30,7 +29,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/startup"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state/stategen"
-	"github.com/OffchainLabs/prysm/v7/cmd/beacon-chain/flags"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/interfaces"
@@ -291,19 +289,6 @@ func (s *Service) StartFromSavedState(saved state.BeaconState) error {
 		return errors.Wrap(err, "failed to initialize blockchain service")
 	}
 
-	if !params.FuluEnabled() {
-		return nil
-	}
-
-	earliestAvailableSlot, custodySubnetCount, err := s.updateCustodyInfoInDB(saved.Slot())
-	if err != nil {
-		return errors.Wrap(err, "could not get and save custody group count")
-	}
-
-	if _, _, err := s.cfg.P2P.UpdateCustodyInfo(earliestAvailableSlot, custodySubnetCount); err != nil {
-		return errors.Wrap(err, "update custody info")
-	}
-
 	return nil
 }
 
@@ -468,73 +453,6 @@ func (s *Service) removeStartupState() {
 	s.cfg.FinalizedStateAtStartUp = nil
 }
 
-// UpdateCustodyInfoInDB updates the custody information in the database.
-// It returns the (potentially updated) custody group count and the earliest available slot.
-func (s *Service) updateCustodyInfoInDB(slot primitives.Slot) (primitives.Slot, uint64, error) {
-	isSupernode := flags.Get().Supernode
-	isSemiSupernode := flags.Get().SemiSupernode
-
-	cfg := params.BeaconConfig()
-	custodyRequirement := cfg.CustodyRequirement
-
-	// Check if the node was previously subscribed to all data subnets, and if so,
-	// store the new status accordingly.
-	wasSupernode, err := s.cfg.BeaconDB.UpdateSubscribedToAllDataSubnets(s.ctx, isSupernode)
-	if err != nil {
-		return 0, 0, errors.Wrap(err, "update subscribed to all data subnets")
-	}
-
-	// Compute the target custody group count based on current flag configuration.
-	targetCustodyGroupCount := custodyRequirement
-
-	// Supernode: custody all groups (either currently set or previously enabled)
-	if isSupernode {
-		targetCustodyGroupCount = cfg.NumberOfCustodyGroups
-	}
-
-	// Semi-supernode: custody minimum needed for reconstruction, or custody requirement if higher
-	if isSemiSupernode {
-		semiSupernodeCustody, err := peerdas.MinimumCustodyGroupCountToReconstruct()
-		if err != nil {
-			return 0, 0, errors.Wrap(err, "minimum custody group count")
-		}
-
-		targetCustodyGroupCount = max(custodyRequirement, semiSupernodeCustody)
-	}
-
-	// Safely compute the fulu fork slot.
-	fuluForkSlot, err := fuluForkSlot()
-	if err != nil {
-		return 0, 0, errors.Wrap(err, "fulu fork slot")
-	}
-
-	// If slot is before the fulu fork slot, then use the earliest stored slot as the reference slot.
-	if slot < fuluForkSlot {
-		slot, err = s.cfg.BeaconDB.EarliestSlot(s.ctx)
-		if err != nil {
-			return 0, 0, errors.Wrap(err, "earliest slot")
-		}
-	}
-
-	earliestAvailableSlot, actualCustodyGroupCount, err := s.cfg.BeaconDB.UpdateCustodyInfo(s.ctx, slot, targetCustodyGroupCount)
-	if err != nil {
-		return 0, 0, errors.Wrap(err, "update custody info")
-	}
-
-	if isSupernode {
-		log.WithFields(logrus.Fields{
-			"current": actualCustodyGroupCount,
-			"target":  cfg.NumberOfCustodyGroups,
-		}).Info("Supernode mode enabled. Will custody all data columns going forward.")
-	}
-
-	if wasSupernode && !isSupernode {
-		log.Warningf("Because the `--%s` flag was previously used, the node will continue to act as a super node.", flags.Supernode.Name)
-	}
-
-	return earliestAvailableSlot, actualCustodyGroupCount, nil
-}
-
 func spawnCountdownIfPreGenesis(ctx context.Context, genesisTime time.Time, db db.HeadAccessDatabase) {
 	currentTime := prysmTime.Now()
 	if currentTime.After(genesisTime) {
@@ -551,19 +469,3 @@ func spawnCountdownIfPreGenesis(ctx context.Context, genesisTime time.Time, db d
 	}
 	go slots.CountdownToGenesis(ctx, genesisTime, uint64(gState.NumValidators()), gRoot)
 }
-
-func fuluForkSlot() (primitives.Slot, error) {
-	cfg := params.BeaconConfig()
-
-	fuluForkEpoch := cfg.FuluForkEpoch
-	if fuluForkEpoch == cfg.FarFutureEpoch {
-		return cfg.FarFutureSlot, nil
-	}
-
-	forkFuluSlot, err := slots.EpochStart(fuluForkEpoch)
-	if err != nil {
-		return 0, errors.Wrap(err, "epoch start")
-	}
-
-	return forkFuluSlot, nil
-}

beacon-chain/blockchain/service_test.go

@@ -23,11 +23,9 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/startup"
 	state_native "github.com/OffchainLabs/prysm/v7/beacon-chain/state/state-native"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state/stategen"
-	"github.com/OffchainLabs/prysm/v7/cmd/beacon-chain/flags"
 	"github.com/OffchainLabs/prysm/v7/config/features"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
-	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	consensusblocks "github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/interfaces"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
@@ -596,218 +594,3 @@ func TestNotifyIndex(t *testing.T) {
 		t.Errorf("Notifier channel did not receive the index")
 	}
 }
-
-func TestUpdateCustodyInfoInDB(t *testing.T) {
-	const (
-		fuluForkEpoch         = 10
-		custodyRequirement    = uint64(4)
-		earliestStoredSlot    = primitives.Slot(12)
-		numberOfCustodyGroups = uint64(64)
-	)
-
-	params.SetupTestConfigCleanup(t)
-	cfg := params.BeaconConfig()
-	cfg.FuluForkEpoch = fuluForkEpoch
-	cfg.CustodyRequirement = custodyRequirement
-	cfg.NumberOfCustodyGroups = numberOfCustodyGroups
-	params.OverrideBeaconConfig(cfg)
-
-	ctx := t.Context()
-	pbBlock := util.NewBeaconBlock()
-	pbBlock.Block.Slot = 12
-	signedBeaconBlock, err := blocks.NewSignedBeaconBlock(pbBlock)
-	require.NoError(t, err)
-
-	roBlock, err := blocks.NewROBlock(signedBeaconBlock)
-	require.NoError(t, err)
-
-	t.Run("CGC increases before fulu", func(t *testing.T) {
-		service, requirements := minimalTestService(t)
-		err = requirements.db.SaveBlock(ctx, roBlock)
-		require.NoError(t, err)
-
-		// Before Fulu
-		// -----------
-		actualEas, actualCgc, err := service.updateCustodyInfoInDB(15)
-		require.NoError(t, err)
-		require.Equal(t, earliestStoredSlot, actualEas)
-		require.Equal(t, custodyRequirement, actualCgc)
-
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(17)
-		require.NoError(t, err)
-		require.Equal(t, earliestStoredSlot, actualEas)
-		require.Equal(t, custodyRequirement, actualCgc)
-
-		resetFlags := flags.Get()
-		gFlags := new(flags.GlobalFlags)
-		gFlags.Supernode = true
-		flags.Init(gFlags)
-		defer flags.Init(resetFlags)
-
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(19)
-		require.NoError(t, err)
-		require.Equal(t, earliestStoredSlot, actualEas)
-		require.Equal(t, numberOfCustodyGroups, actualCgc)
-
-		// After Fulu
-		// ----------
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1)
-		require.NoError(t, err)
-		require.Equal(t, earliestStoredSlot, actualEas)
-		require.Equal(t, numberOfCustodyGroups, actualCgc)
-	})
-
-	t.Run("CGC increases after fulu", func(t *testing.T) {
-		service, requirements := minimalTestService(t)
-		err = requirements.db.SaveBlock(ctx, roBlock)
-		require.NoError(t, err)
-
-		// Before Fulu
-		// -----------
-		actualEas, actualCgc, err := service.updateCustodyInfoInDB(15)
-		require.NoError(t, err)
-		require.Equal(t, earliestStoredSlot, actualEas)
-		require.Equal(t, custodyRequirement, actualCgc)
-
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(17)
-		require.NoError(t, err)
-		require.Equal(t, earliestStoredSlot, actualEas)
-		require.Equal(t, custodyRequirement, actualCgc)
-
-		// After Fulu
-		// ----------
-		resetFlags := flags.Get()
-		gFlags := new(flags.GlobalFlags)
-		gFlags.Supernode = true
-		flags.Init(gFlags)
-		defer flags.Init(resetFlags)
-
-		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(slot)
-		require.NoError(t, err)
-		require.Equal(t, slot, actualEas)
-		require.Equal(t, numberOfCustodyGroups, actualCgc)
-
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(slot + 2)
-		require.NoError(t, err)
-		require.Equal(t, slot, actualEas)
-		require.Equal(t, numberOfCustodyGroups, actualCgc)
-	})
-
-	t.Run("Supernode downgrade prevented", func(t *testing.T) {
-		service, requirements := minimalTestService(t)
-		err = requirements.db.SaveBlock(ctx, roBlock)
-		require.NoError(t, err)
-
-		// Enable supernode
-		resetFlags := flags.Get()
-		gFlags := new(flags.GlobalFlags)
-		gFlags.Supernode = true
-		flags.Init(gFlags)
-
-		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
-		actualEas, actualCgc, err := service.updateCustodyInfoInDB(slot)
-		require.NoError(t, err)
-		require.Equal(t, slot, actualEas)
-		require.Equal(t, numberOfCustodyGroups, actualCgc)
-
-		// Try to downgrade by removing flag
-		gFlags.Supernode = false
-		flags.Init(gFlags)
-		defer flags.Init(resetFlags)
-
-		// Should still be supernode
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(slot + 2)
-		require.NoError(t, err)
-		require.Equal(t, slot, actualEas)
-		require.Equal(t, numberOfCustodyGroups, actualCgc) // Still 64, not downgraded
-	})
-
-	t.Run("Semi-supernode downgrade prevented", func(t *testing.T) {
-		service, requirements := minimalTestService(t)
-		err = requirements.db.SaveBlock(ctx, roBlock)
-		require.NoError(t, err)
-
-		// Enable semi-supernode
-		resetFlags := flags.Get()
-		gFlags := new(flags.GlobalFlags)
-		gFlags.SemiSupernode = true
-		flags.Init(gFlags)
-
-		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
-		actualEas, actualCgc, err := service.updateCustodyInfoInDB(slot)
-		require.NoError(t, err)
-		require.Equal(t, slot, actualEas)
-		semiSupernodeCustody := numberOfCustodyGroups / 2 // 64
-		require.Equal(t, semiSupernodeCustody, actualCgc) // Semi-supernode custodies 64 groups
-
-		// Try to downgrade by removing flag
-		gFlags.SemiSupernode = false
-		flags.Init(gFlags)
-		defer flags.Init(resetFlags)
-
-		// UpdateCustodyInfo should prevent downgrade - custody count should remain at 64
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(slot + 2)
-		require.NoError(t, err)
-		require.Equal(t, slot, actualEas)
-		require.Equal(t, semiSupernodeCustody, actualCgc) // Still 64 due to downgrade prevention by UpdateCustodyInfo
-	})
-
-	t.Run("Semi-supernode to supernode upgrade allowed", func(t *testing.T) {
-		service, requirements := minimalTestService(t)
-		err = requirements.db.SaveBlock(ctx, roBlock)
-		require.NoError(t, err)
-
-		// Start with semi-supernode
-		resetFlags := flags.Get()
-		gFlags := new(flags.GlobalFlags)
-		gFlags.SemiSupernode = true
-		flags.Init(gFlags)
-
-		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
-		actualEas, actualCgc, err := service.updateCustodyInfoInDB(slot)
-		require.NoError(t, err)
-		require.Equal(t, slot, actualEas)
-		semiSupernodeCustody := numberOfCustodyGroups / 2 // 64
-		require.Equal(t, semiSupernodeCustody, actualCgc) // Semi-supernode custodies 64 groups
-
-		// Upgrade to full supernode
-		gFlags.SemiSupernode = false
-		gFlags.Supernode = true
-		flags.Init(gFlags)
-		defer flags.Init(resetFlags)
-
-		// Should upgrade to full supernode
-		upgradeSlot := slot + 2
-		actualEas, actualCgc, err = service.updateCustodyInfoInDB(upgradeSlot)
-		require.NoError(t, err)
-		require.Equal(t, upgradeSlot, actualEas)           // Earliest slot updates when upgrading
-		require.Equal(t, numberOfCustodyGroups, actualCgc) // Upgraded to 128
-	})
-
-	t.Run("Semi-supernode with high validator requirements uses higher custody", func(t *testing.T) {
-		service, requirements := minimalTestService(t)
-		err = requirements.db.SaveBlock(ctx, roBlock)
-		require.NoError(t, err)
-
-		// Enable semi-supernode
-		resetFlags := flags.Get()
-		gFlags := new(flags.GlobalFlags)
-		gFlags.SemiSupernode = true
-		flags.Init(gFlags)
-		defer flags.Init(resetFlags)
-
-		// Mock a high custody requirement (simulating many validators)
-		// We need to override the custody requirement calculation
-		// For this test, we'll verify the logic by checking if custodyRequirement > 64
-		// Since custodyRequirement in minimalTestService is 4, we can't test the high case here
-		// This would require a different test setup with actual validators
-		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
-		actualEas, actualCgc, err := service.updateCustodyInfoInDB(slot)
-		require.NoError(t, err)
-		require.Equal(t, slot, actualEas)
-		semiSupernodeCustody := numberOfCustodyGroups / 2 // 64
-		// With low validator requirements (4), should use semi-supernode minimum (64)
-		require.Equal(t, semiSupernodeCustody, actualCgc)
-	})
-}

beacon-chain/sync/custody.go

@@ -10,20 +10,72 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p"
 	"github.com/OffchainLabs/prysm/v7/cmd/beacon-chain/flags"
 	"github.com/OffchainLabs/prysm/v7/config/params"
+	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
 var nilFinalizedStateError = errors.New("finalized state is nil")
 
-func (s *Service) maintainCustodyInfo() {
+func (s *Service) maintainCustodyInfo() error {
+	// Rationale of slot choice:
+	// - If syncing with an empty DB from genesis, then justifiedSlot = finalizedSlot = 0,
+	//   and the node starts to sync from slot 0 ==> Using justifiedSlot is correct.
+	// - If syncing with an empty DB from a checkpoint, then justifiedSlot = finalizedSlot = checkpointSlot,
+	//   and the node starts to sync from checkpointSlot ==> Using justifiedSlot is correct.
+	// - If syncing with a non-empty DB, then justifiedSlot > finalizedSlot,
+	//   and the node starts to sync from justifiedSlot + 1 ==> Using justifiedSlot + 1 is correct.
 	const interval = 1 * time.Minute
 
+	finalizedCheckpoint, err := s.cfg.beaconDB.FinalizedCheckpoint(s.ctx)
+	if err != nil {
+		return errors.Wrap(err, "finalized checkpoint")
+	}
+
+	if finalizedCheckpoint == nil {
+		return errors.New("finalized checkpoint is nil")
+	}
+
+	finalizedSlot, err := slots.EpochStart(finalizedCheckpoint.Epoch)
+	if err != nil {
+		return errors.Wrap(err, "epoch start for finalized slot")
+	}
+
+	justifiedCheckpoint, err := s.cfg.beaconDB.JustifiedCheckpoint(s.ctx)
+	if err != nil {
+		return errors.Wrap(err, "justified checkpoint")
+	}
+
+	if justifiedCheckpoint == nil {
+		return errors.New("justified checkpoint is nil")
+	}
+
+	justifiedSlot, err := slots.EpochStart(justifiedCheckpoint.Epoch)
+	if err != nil {
+		return errors.Wrap(err, "epoch start for justified slot")
+	}
+
+	slot := justifiedSlot
+	if justifiedSlot > finalizedSlot {
+		slot++
+	}
+
+	earliestAvailableSlot, custodySubnetCount, err := s.updateCustodyInfoInDB(slot)
+	if err != nil {
+		return errors.Wrap(err, "could not get and save custody group count")
+	}
+
+	if _, _, err := s.cfg.p2p.UpdateCustodyInfo(earliestAvailableSlot, custodySubnetCount); err != nil {
+		return errors.Wrap(err, "update custody info")
+	}
+
 	async.RunEvery(s.ctx, interval, func() {
 		if err := s.updateCustodyInfoIfNeeded(); err != nil {
 			log.WithError(err).Error("Failed to update custody info")
 		}
 	})
+
+	return nil
 }
 
 func (s *Service) updateCustodyInfoIfNeeded() error {

beacon-chain/sync/rpc_status_test.go

@@ -12,6 +12,7 @@ import (
 	mock "github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/testing"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/transition"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db/kv"
+	dbTest "github.com/OffchainLabs/prysm/v7/beacon-chain/db/testing"
 	testingDB "github.com/OffchainLabs/prysm/v7/beacon-chain/db/testing"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/peers"
@@ -934,6 +935,7 @@ func TestStatusRPCRequest_BadPeerHandshake(t *testing.T) {
 	r := &Service{
 		cfg: &config{
 			p2p:           p1,
+			beaconDB:      dbTest.SetupDB(t),
 			chain:         chain,
 			stateNotifier: chain.StateNotifier(),
 			initialSync:   &mockSync.Sync{IsSyncing: false},

beacon-chain/sync/service.go

@@ -17,6 +17,7 @@ import (
 	blockfeed "github.com/OffchainLabs/prysm/v7/beacon-chain/core/feed/block"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/feed/operation"
 	statefeed "github.com/OffchainLabs/prysm/v7/beacon-chain/core/feed/state"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db/filesystem"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/execution"
@@ -33,9 +34,11 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/sync/backfill/coverage"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/verification"
 	lruwrpr "github.com/OffchainLabs/prysm/v7/cache/lru"
+	"github.com/OffchainLabs/prysm/v7/cmd/beacon-chain/flags"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/interfaces"
+	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	leakybucket "github.com/OffchainLabs/prysm/v7/container/leaky-bucket"
 	"github.com/OffchainLabs/prysm/v7/crypto/rand"
 	"github.com/OffchainLabs/prysm/v7/runtime"
@@ -275,11 +278,6 @@ func (s *Service) Start() {
 
 	s.processPendingBlocksQueue()
 	s.maintainPeerStatuses()
-
-	if params.FuluEnabled() {
-		s.maintainCustodyInfo()
-	}
-
 	s.resyncIfBehind()
 
 	// Update sync metrics.
@@ -287,6 +285,15 @@ func (s *Service) Start() {
 
 	// Prune data column cache periodically on finalization.
 	async.RunEvery(s.ctx, 30*time.Second, s.pruneDataColumnCache)
+
+	if !params.FuluEnabled() {
+		return
+	}
+
+	if err := s.maintainCustodyInfo(); err != nil {
+		log.WithError(err).Error("Failed to maintain custody info")
+	}
+
 }
 
 // Stop the regular sync service.
@@ -452,6 +459,89 @@ func (s *Service) waitForInitialSync(ctx context.Context) error {
 	}
 }
 
+// UpdateCustodyInfoInDB updates the custody information in the database.
+// It returns the (potentially updated) custody group count and the earliest available slot.
+func (s *Service) updateCustodyInfoInDB(slot primitives.Slot) (primitives.Slot, uint64, error) {
+	isSupernode := flags.Get().Supernode
+	isSemiSupernode := flags.Get().SemiSupernode
+
+	cfg := params.BeaconConfig()
+	custodyRequirement := cfg.CustodyRequirement
+
+	// Check if the node was previously subscribed to all data subnets, and if so,
+	// store the new status accordingly.
+	wasSupernode, err := s.cfg.beaconDB.UpdateSubscribedToAllDataSubnets(s.ctx, isSupernode)
+	if err != nil {
+		return 0, 0, errors.Wrap(err, "update subscribed to all data subnets")
+	}
+
+	// Compute the target custody group count based on current flag configuration.
+	targetCustodyGroupCount := custodyRequirement
+
+	// Supernode: custody all groups (either currently set or previously enabled)
+	if isSupernode {
+		targetCustodyGroupCount = cfg.NumberOfCustodyGroups
+	}
+
+	// Semi-supernode: custody minimum needed for reconstruction, or custody requirement if higher
+	if isSemiSupernode {
+		semiSupernodeCustody, err := peerdas.MinimumCustodyGroupCountToReconstruct()
+		if err != nil {
+			return 0, 0, errors.Wrap(err, "minimum custody group count")
+		}
+
+		targetCustodyGroupCount = max(custodyRequirement, semiSupernodeCustody)
+	}
+
+	// Safely compute the fulu fork slot.
+	fuluForkSlot, err := fuluForkSlot()
+	if err != nil {
+		return 0, 0, errors.Wrap(err, "fulu fork slot")
+	}
+
+	// If slot is before the fulu fork slot, then use the earliest stored slot as the reference slot.
+	if slot < fuluForkSlot {
+		slot, err = s.cfg.beaconDB.EarliestSlot(s.ctx)
+		if err != nil {
+			return 0, 0, errors.Wrap(err, "earliest slot")
+		}
+	}
+
+	earliestAvailableSlot, actualCustodyGroupCount, err := s.cfg.beaconDB.UpdateCustodyInfo(s.ctx, slot, targetCustodyGroupCount)
+	if err != nil {
+		return 0, 0, errors.Wrap(err, "update custody info")
+	}
+
+	if isSupernode {
+		log.WithFields(logrus.Fields{
+			"current": actualCustodyGroupCount,
+			"target":  cfg.NumberOfCustodyGroups,
+		}).Info("Supernode mode enabled. Will custody all data columns going forward.")
+	}
+
+	if wasSupernode && !isSupernode {
+		log.Warningf("Because the `--%s` flag was previously used, the node will continue to act as a super node.", flags.Supernode.Name)
+	}
+
+	return earliestAvailableSlot, actualCustodyGroupCount, nil
+}
+
+func fuluForkSlot() (primitives.Slot, error) {
+	cfg := params.BeaconConfig()
+
+	fuluForkEpoch := cfg.FuluForkEpoch
+	if fuluForkEpoch == cfg.FarFutureEpoch {
+		return cfg.FarFutureSlot, nil
+	}
+
+	forkFuluSlot, err := slots.EpochStart(fuluForkEpoch)
+	if err != nil {
+		return 0, errors.Wrap(err, "epoch start")
+	}
+
+	return forkFuluSlot, nil
+}
+
 // Checker defines a struct which can verify whether a node is currently
 // synchronizing a chain with the rest of peers in the network.
 type Checker interface {

beacon-chain/sync/service_test.go

@@ -16,6 +16,9 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/startup"
 	state_native "github.com/OffchainLabs/prysm/v7/beacon-chain/state/state-native"
 	mockSync "github.com/OffchainLabs/prysm/v7/beacon-chain/sync/initial-sync/testing"
+	"github.com/OffchainLabs/prysm/v7/cmd/beacon-chain/flags"
+	"github.com/OffchainLabs/prysm/v7/config/params"
+	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	leakybucket "github.com/OffchainLabs/prysm/v7/container/leaky-bucket"
 	"github.com/OffchainLabs/prysm/v7/crypto/bls"
@@ -440,3 +443,224 @@ func TestService_Stop_ConcurrentGoodbyeMessages(t *testing.T) {
 
 	require.Equal(t, false, util.WaitTimeout(&wg, 2*time.Second))
 }
+
+func TestUpdateCustodyInfoInDB(t *testing.T) {
+	const (
+		fuluForkEpoch         = 10
+		custodyRequirement    = uint64(4)
+		earliestStoredSlot    = primitives.Slot(12)
+		numberOfCustodyGroups = uint64(64)
+	)
+
+	params.SetupTestConfigCleanup(t)
+	cfg := params.BeaconConfig()
+	cfg.FuluForkEpoch = fuluForkEpoch
+	cfg.CustodyRequirement = custodyRequirement
+	cfg.NumberOfCustodyGroups = numberOfCustodyGroups
+	params.OverrideBeaconConfig(cfg)
+
+	ctx := t.Context()
+	pbBlock := util.NewBeaconBlock()
+	pbBlock.Block.Slot = 12
+	signedBeaconBlock, err := blocks.NewSignedBeaconBlock(pbBlock)
+	require.NoError(t, err)
+
+	roBlock, err := blocks.NewROBlock(signedBeaconBlock)
+	require.NoError(t, err)
+
+	t.Run("CGC increases before fulu", func(t *testing.T) {
+		beaconDB := dbTest.SetupDB(t)
+		service := Service{cfg: &config{beaconDB: beaconDB}}
+		err = beaconDB.SaveBlock(ctx, roBlock)
+		require.NoError(t, err)
+
+		// Before Fulu
+		// -----------
+		actualEas, actualCgc, err := service.updateCustodyInfoInDB(15)
+		require.NoError(t, err)
+		require.Equal(t, earliestStoredSlot, actualEas)
+		require.Equal(t, custodyRequirement, actualCgc)
+
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(17)
+		require.NoError(t, err)
+		require.Equal(t, earliestStoredSlot, actualEas)
+		require.Equal(t, custodyRequirement, actualCgc)
+
+		resetFlags := flags.Get()
+		gFlags := new(flags.GlobalFlags)
+		gFlags.Supernode = true
+		flags.Init(gFlags)
+		defer flags.Init(resetFlags)
+
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(19)
+		require.NoError(t, err)
+		require.Equal(t, earliestStoredSlot, actualEas)
+		require.Equal(t, numberOfCustodyGroups, actualCgc)
+
+		// After Fulu
+		// ----------
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1)
+		require.NoError(t, err)
+		require.Equal(t, earliestStoredSlot, actualEas)
+		require.Equal(t, numberOfCustodyGroups, actualCgc)
+	})
+
+	t.Run("CGC increases after fulu", func(t *testing.T) {
+		beaconDB := dbTest.SetupDB(t)
+		service := Service{cfg: &config{beaconDB: beaconDB}}
+		err = beaconDB.SaveBlock(ctx, roBlock)
+		require.NoError(t, err)
+
+		// Before Fulu
+		// -----------
+		actualEas, actualCgc, err := service.updateCustodyInfoInDB(15)
+		require.NoError(t, err)
+		require.Equal(t, earliestStoredSlot, actualEas)
+		require.Equal(t, custodyRequirement, actualCgc)
+
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(17)
+		require.NoError(t, err)
+		require.Equal(t, earliestStoredSlot, actualEas)
+		require.Equal(t, custodyRequirement, actualCgc)
+
+		// After Fulu
+		// ----------
+		resetFlags := flags.Get()
+		gFlags := new(flags.GlobalFlags)
+		gFlags.Supernode = true
+		flags.Init(gFlags)
+		defer flags.Init(resetFlags)
+
+		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(slot)
+		require.NoError(t, err)
+		require.Equal(t, slot, actualEas)
+		require.Equal(t, numberOfCustodyGroups, actualCgc)
+
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(slot + 2)
+		require.NoError(t, err)
+		require.Equal(t, slot, actualEas)
+		require.Equal(t, numberOfCustodyGroups, actualCgc)
+	})
+
+	t.Run("Supernode downgrade prevented", func(t *testing.T) {
+		beaconDB := dbTest.SetupDB(t)
+		service := Service{cfg: &config{beaconDB: beaconDB}}
+		err = beaconDB.SaveBlock(ctx, roBlock)
+		require.NoError(t, err)
+
+		// Enable supernode
+		resetFlags := flags.Get()
+		gFlags := new(flags.GlobalFlags)
+		gFlags.Supernode = true
+		flags.Init(gFlags)
+
+		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
+		actualEas, actualCgc, err := service.updateCustodyInfoInDB(slot)
+		require.NoError(t, err)
+		require.Equal(t, slot, actualEas)
+		require.Equal(t, numberOfCustodyGroups, actualCgc)
+
+		// Try to downgrade by removing flag
+		gFlags.Supernode = false
+		flags.Init(gFlags)
+		defer flags.Init(resetFlags)
+
+		// Should still be supernode
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(slot + 2)
+		require.NoError(t, err)
+		require.Equal(t, slot, actualEas)
+		require.Equal(t, numberOfCustodyGroups, actualCgc) // Still 64, not downgraded
+	})
+
+	t.Run("Semi-supernode downgrade prevented", func(t *testing.T) {
+		beaconDB := dbTest.SetupDB(t)
+		service := Service{cfg: &config{beaconDB: beaconDB}}
+		err = beaconDB.SaveBlock(ctx, roBlock)
+		require.NoError(t, err)
+
+		// Enable semi-supernode
+		resetFlags := flags.Get()
+		gFlags := new(flags.GlobalFlags)
+		gFlags.SemiSupernode = true
+		flags.Init(gFlags)
+
+		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
+		actualEas, actualCgc, err := service.updateCustodyInfoInDB(slot)
+		require.NoError(t, err)
+		require.Equal(t, slot, actualEas)
+		semiSupernodeCustody := numberOfCustodyGroups / 2 // 64
+		require.Equal(t, semiSupernodeCustody, actualCgc) // Semi-supernode custodies 64 groups
+
+		// Try to downgrade by removing flag
+		gFlags.SemiSupernode = false
+		flags.Init(gFlags)
+		defer flags.Init(resetFlags)
+
+		// UpdateCustodyInfo should prevent downgrade - custody count should remain at 64
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(slot + 2)
+		require.NoError(t, err)
+		require.Equal(t, slot, actualEas)
+		require.Equal(t, semiSupernodeCustody, actualCgc) // Still 64 due to downgrade prevention by UpdateCustodyInfo
+	})
+
+	t.Run("Semi-supernode to supernode upgrade allowed", func(t *testing.T) {
+		beaconDB := dbTest.SetupDB(t)
+		service := Service{cfg: &config{beaconDB: beaconDB}}
+		err = beaconDB.SaveBlock(ctx, roBlock)
+		require.NoError(t, err)
+
+		// Start with semi-supernode
+		resetFlags := flags.Get()
+		gFlags := new(flags.GlobalFlags)
+		gFlags.SemiSupernode = true
+		flags.Init(gFlags)
+
+		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
+		actualEas, actualCgc, err := service.updateCustodyInfoInDB(slot)
+		require.NoError(t, err)
+		require.Equal(t, slot, actualEas)
+		semiSupernodeCustody := numberOfCustodyGroups / 2 // 64
+		require.Equal(t, semiSupernodeCustody, actualCgc) // Semi-supernode custodies 64 groups
+
+		// Upgrade to full supernode
+		gFlags.SemiSupernode = false
+		gFlags.Supernode = true
+		flags.Init(gFlags)
+		defer flags.Init(resetFlags)
+
+		// Should upgrade to full supernode
+		upgradeSlot := slot + 2
+		actualEas, actualCgc, err = service.updateCustodyInfoInDB(upgradeSlot)
+		require.NoError(t, err)
+		require.Equal(t, upgradeSlot, actualEas)           // Earliest slot updates when upgrading
+		require.Equal(t, numberOfCustodyGroups, actualCgc) // Upgraded to 128
+	})
+
+	t.Run("Semi-supernode with high validator requirements uses higher custody", func(t *testing.T) {
+		beaconDB := dbTest.SetupDB(t)
+		service := Service{cfg: &config{beaconDB: beaconDB}}
+		err = beaconDB.SaveBlock(ctx, roBlock)
+		require.NoError(t, err)
+
+		// Enable semi-supernode
+		resetFlags := flags.Get()
+		gFlags := new(flags.GlobalFlags)
+		gFlags.SemiSupernode = true
+		flags.Init(gFlags)
+		defer flags.Init(resetFlags)
+
+		// Mock a high custody requirement (simulating many validators)
+		// We need to override the custody requirement calculation
+		// For this test, we'll verify the logic by checking if custodyRequirement > 64
+		// Since custodyRequirement in minimalTestService is 4, we can't test the high case here
+		// This would require a different test setup with actual validators
+		slot := fuluForkEpoch*primitives.Slot(cfg.SlotsPerEpoch) + 1
+		actualEas, actualCgc, err := service.updateCustodyInfoInDB(slot)
+		require.NoError(t, err)
+		require.Equal(t, slot, actualEas)
+		semiSupernodeCustody := numberOfCustodyGroups / 2 // 64
+		// With low validator requirements (4), should use semi-supernode minimum (64)
+		require.Equal(t, semiSupernodeCustody, actualCgc)
+	})
+}

changelog/manu-eas.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- When adding the `--[semi-]supernode` flag, update the ealiest available slot accordingly.