Add --semi-supernode support (#8254)

Squashed commit of the following:

commit 45a4315
Author: Jimmy Chen <[email protected]>
Date:   Tue Oct 21 23:15:58 2025 +1100

    Add tests for flags

commit 15569bc
Author: Jimmy Chen <[email protected]>
Date:   Tue Oct 21 22:40:49 2025 +1100

    Revert some changes to preserve existing behaviour. Full node with no validator should have `validator_custody_at_head` set to 0.

commit 7767a2e
Author: Jimmy Chen <[email protected]>
Date:   Tue Oct 21 16:47:40 2025 +1100

    More test fixes and update help text.

commit 0932f36
Author: Jimmy Chen <[email protected]>
Date:   Tue Oct 21 16:14:46 2025 +1100

    Add tests for restoring custody context from persisted.

commit 5cc3186
Author: Jimmy Chen <[email protected]>
Date:   Tue Oct 21 16:07:47 2025 +1100

    Implement semi-supernode.

    Co-authored-by: pawanjay176 <[email protected]>

Author: jimmygchen

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -21,6 +21,7 @@ use crate::block_verification_types::{
 };
 pub use crate::canonical_head::CanonicalHead;
 use crate::chain_config::ChainConfig;
+use crate::custody_context::CustodyContextSsz;
 use crate::data_availability_checker::{
     Availability, AvailabilityCheckError, AvailableBlock, AvailableBlockData,
     DataAvailabilityChecker, DataColumnReconstructionResult,
@@ -64,7 +65,6 @@ use crate::shuffling_cache::{BlockShufflingIds, ShufflingCache};
 use crate::sync_committee_verification::{
     Error as SyncCommitteeError, VerifiedSyncCommitteeMessage, VerifiedSyncContribution,
 };
-use crate::validator_custody::CustodyContextSsz;
 use crate::validator_monitor::{
     HISTORIC_EPOCHS as VALIDATOR_MONITOR_HISTORIC_EPOCHS, ValidatorMonitor, get_slot_delay_ms,
     timestamp_now,

beacon_node/beacon_chain/src/builder.rs

@@ -4,6 +4,7 @@ use crate::beacon_chain::{
     BEACON_CHAIN_DB_KEY, CanonicalHead, LightClientProducerEvent, OP_POOL_DB_KEY,
 };
 use crate::beacon_proposer_cache::BeaconProposerCache;
+use crate::custody_context::NodeCustodyType;
 use crate::data_availability_checker::DataAvailabilityChecker;
 use crate::fork_choice_signal::ForkChoiceSignalTx;
 use crate::fork_revert::{reset_fork_choice_to_finalization, revert_to_fork_boundary};
@@ -100,7 +101,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     kzg: Arc<Kzg>,
     task_executor: Option<TaskExecutor>,
     validator_monitor_config: Option<ValidatorMonitorConfig>,
-    import_all_data_columns: bool,
+    node_custody_type: NodeCustodyType,
     rng: Option<Box<dyn RngCore + Send>>,
 }
 
@@ -139,7 +140,7 @@ where
             kzg,
             task_executor: None,
             validator_monitor_config: None,
-            import_all_data_columns: false,
+            node_custody_type: NodeCustodyType::Fullnode,
             rng: None,
         }
     }
@@ -640,9 +641,9 @@ where
         self
     }
 
-    /// Sets whether to require and import all data columns when importing block.
-    pub fn import_all_data_columns(mut self, import_all_data_columns: bool) -> Self {
-        self.import_all_data_columns = import_all_data_columns;
+    /// Sets the node custody type for data column import.
+    pub fn node_custody_type(mut self, node_custody_type: NodeCustodyType) -> Self {
+        self.node_custody_type = node_custody_type;
         self
     }
 
@@ -935,10 +936,11 @@ where
         {
             Arc::new(CustodyContext::new_from_persisted_custody_context(
                 custody,
-                self.import_all_data_columns,
+                self.node_custody_type,
+                &self.spec,
             ))
         } else {
-            Arc::new(CustodyContext::new(self.import_all_data_columns))
+            Arc::new(CustodyContext::new(self.node_custody_type, &self.spec))
         };
         debug!(?custody_context, "Loading persisted custody context");
 

beacon_node/beacon_chain/src/chain_config.rs

@@ -1,3 +1,4 @@
+use crate::custody_context::NodeCustodyType;
 pub use proto_array::{DisallowedReOrgOffsets, ReOrgThreshold};
 use serde::{Deserialize, Serialize};
 use std::str::FromStr;
@@ -118,6 +119,8 @@ pub struct ChainConfig {
     pub invalid_block_roots: HashSet<Hash256>,
     /// Disable the getBlobs optimisation to fetch blobs from the EL mempool.
     pub disable_get_blobs: bool,
+    /// The node's custody type, determining how many data columns to custody and sample.
+    pub node_custody_type: NodeCustodyType,
 }
 
 impl Default for ChainConfig {
@@ -158,6 +161,7 @@ impl Default for ChainConfig {
             data_column_publishing_delay: None,
             invalid_block_roots: HashSet::new(),
             disable_get_blobs: false,
+            node_custody_type: NodeCustodyType::Fullnode,
         }
     }
 }