feat: implement subnet topology fork transition with centralized routing

anchor/client/src/lib.rs

@@ -43,7 +43,7 @@ use sensitive_url::SensitiveUrl;
 use signature_collector::SignatureCollectorManager;
 use slashing_protection::SlashingDatabase;
 use slot_clock::{SlotClock, SystemTimeSlotClock};
-use subnet_service::{SUBNET_COUNT, SubnetId, start_subnet_service};
+use subnet_service::{ForkSchedule, SUBNET_COUNT_NZ, SubnetId, SubnetRouter, start_subnet_service};
 use task_executor::TaskExecutor;
 use tokio::{
     net::TcpListener,
@@ -412,13 +412,26 @@ impl Client {
         ));
         duties_tracker.clone().start(executor.clone());
 
-        let message_validator = Validator::new(
+        // Create ForkSchedule from config
+        let fork_schedule = match config.global_config.ssv_network.subnet_topology_fork_epoch {
+            Some(fork_epoch) => ForkSchedule::new(fork_epoch),
+            None => ForkSchedule::pre_fork(),
+        };
+
+        // Create centralized SubnetRouter for fork-aware subnet calculations
+        let router = Arc::new(SubnetRouter::new(
             database.watch(),
+            fork_schedule.clone(),
+            slot_clock.clone(),
             E::slots_per_epoch(),
+            SUBNET_COUNT_NZ,
+        ));
+
+        let message_validator = Validator::new(
             spec.epochs_per_sync_committee_period.as_u64(),
             E::sync_committee_size(),
             duties_tracker.clone(),
-            slot_clock.clone(),
+            router.clone(),
             &executor,
         );
 
@@ -447,12 +460,15 @@ impl Client {
                     private_key: key.clone(),
                     operator_id: operator_id.clone(),
                     validator: Some(message_validator.clone()),
-                    subnet_count: SUBNET_COUNT,
                     is_synced: is_synced.clone(),
+                    router: router.clone(),
                 },
             )?)
         } else {
-            Arc::new(ImpostorMessageSender::new(network_tx.clone(), SUBNET_COUNT))
+            Arc::new(ImpostorMessageSender::new(
+                network_tx.clone(),
+                SUBNET_COUNT_NZ,
+            ))
         };
 
         // Create the signature collector
@@ -478,9 +494,10 @@ impl Client {
         // Start the subnet service now that we have slot_clock
         let subnet_service = start_subnet_service::<E>(
             database.watch(),
-            SUBNET_COUNT,
+            SUBNET_COUNT_NZ,
             config.network.subscribe_all_subnets,
             config.network.disable_gossipsub_topic_scoring,
+            fork_schedule.clone(),
             &executor,
             slot_clock.clone(),
             spec.clone(),
@@ -497,6 +514,7 @@ impl Client {
             outcome_tx,
             message_validator,
             doppelganger_service.clone(),
+            router,
         );
 
         // Start the p2p network

anchor/common/ssv_network_config/Cargo.toml

@@ -10,3 +10,4 @@ enr = { workspace = true }
 eth2_network_config = { workspace = true }
 serde_yaml = { workspace = true }
 ssv_types = { workspace = true }
+types = { workspace = true }

anchor/common/ssv_network_config/src/lib.rs

@@ -8,6 +8,7 @@ use alloy::primitives::Address;
 use enr::{CombinedKey, Enr};
 use eth2_network_config::Eth2NetworkConfig;
 use ssv_types::domain_type::DomainType;
+use types::Epoch;
 
 macro_rules! include_str_for_net {
     ($network:ident, $file:literal) => {
@@ -38,6 +39,9 @@ pub struct SsvNetworkConfig {
     pub ssv_contract: Address,
     pub ssv_contract_block: u64,
     pub ssv_domain_type: DomainType,
+    /// The epoch at which the subnet topology fork activates.
+    /// `None` means pre-fork operation (no fork configured).
+    pub subnet_topology_fork_epoch: Option<Epoch>,
 }
 
 impl SsvNetworkConfig {
@@ -65,6 +69,7 @@ impl SsvNetworkConfig {
             ssv_domain_type: domain_type
                 .parse()
                 .map_err(|e| format!("Unable to parse built-in domain type: {e}"))?,
+            subnet_topology_fork_epoch: None,
         }))
     }
 
@@ -82,12 +87,19 @@ impl SsvNetworkConfig {
             })
             .transpose()?;
 
+        let subnet_topology_fork_epoch_path = base_dir.join("ssv_subnet_topology_fork_epoch.txt");
+        let subnet_topology_fork_epoch = subnet_topology_fork_epoch_path
+            .exists()
+            .then(|| read(&subnet_topology_fork_epoch_path))
+            .transpose()?;
+
         Ok(Self {
             ssv_boot_nodes,
             ssv_contract: read(&base_dir.join("ssv_contract_address.txt"))?,
             ssv_contract_block: read(&base_dir.join("ssv_contract_block.txt"))?,
             ssv_domain_type: read(&base_dir.join("ssv_domain_type.txt"))?,
             eth2_network: Self::load_eth2_network_config(base_dir)?,
+            subnet_topology_fork_epoch,
         })
     }
 

anchor/eth/src/event_processor.rs

@@ -190,9 +190,10 @@ impl EventProcessor {
 
         let max_seen = self.db.state().get_max_operator_id_seen();
 
-        // Only check for missing operators if we have a previous max (not a migrated database)
+        // Only check for missing operators if we have a previous max (not a migrated database).
+        // Use checked_sub to avoid underflow if operatorId is 0.
         if let Some(max_seen) = max_seen
-            && max_seen != operatorId - 1
+            && operatorId.checked_sub(1) != Some(max_seen)
         {
             return Err(ExecutionError::InvalidEvent(format!(
                 "Missing OperatorAdded events: database has only seen up to id {max_seen}, \

anchor/message_receiver/Cargo.toml

@@ -16,6 +16,7 @@ qbft_manager = { workspace = true }
 signature_collector = { workspace = true }
 slot_clock = { workspace = true }
 ssv_types = { workspace = true }
+subnet_service = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

anchor/message_receiver/src/manager.rs

@@ -11,6 +11,7 @@ use qbft_manager::QbftManager;
 use signature_collector::SignatureCollectorManager;
 use slot_clock::SlotClock;
 use ssv_types::msgid::DutyExecutor;
+use subnet_service::SubnetRouter;
 use tokio::sync::{mpsc, mpsc::error::TrySendError, watch};
 use tracing::{debug, debug_span, error, trace};
 
@@ -34,6 +35,8 @@ pub struct NetworkMessageReceiver<S: SlotClock, D: DutiesProvider> {
     outcome_tx: mpsc::Sender<Outcome>,
     validator: Arc<Validator<S, D>>,
     doppelganger_service: Option<Arc<OperatorDoppelgangerService>>,
+    /// Centralized router for fork-aware subnet calculations
+    router: Arc<SubnetRouter<S>>,
 }
 
 impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageReceiver<S, D> {
@@ -47,6 +50,7 @@ impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageReceiver<S, D> {
         outcome_tx: mpsc::Sender<Outcome>,
         validator: Arc<Validator<S, D>>,
         doppelganger_service: Option<Arc<OperatorDoppelgangerService>>,
+        router: Arc<SubnetRouter<S>>,
     ) -> Arc<Self> {
         Arc::new(Self {
             processor,
@@ -57,6 +61,7 @@ impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageReceiver<S, D> {
             outcome_tx,
             validator,
             doppelganger_service,
+            router,
         })
     }
 }
@@ -76,7 +81,7 @@ impl<S: SlotClock + 'static, D: DutiesProvider> MessageReceiver
                 let span = debug_span!("message_receiver", msg=%message_id);
                 let _enter = span.enter();
 
-                let result = receiver.validator.validate(&message.data);
+                let result = receiver.validator.validate(&message.data, &message.topic);
 
                 let mut action = MessageAcceptance::from(&result);
 
@@ -104,6 +109,7 @@ impl<S: SlotClock + 'static, D: DutiesProvider> MessageReceiver
                 let ValidatedMessage {
                     signed_ssv_message,
                     ssv_message,
+                    matched_schema,
                 } = match result {
                     ValidationResult::Success(message) => message,
                     ValidationResult::PreDecodeFailure(failure) => {
@@ -120,6 +126,18 @@ impl<S: SlotClock + 'static, D: DutiesProvider> MessageReceiver
                     }
                 };
 
+                // Check if the matched schema should be processed at the current epoch.
+                // During pre-subscribe window (F-2, F-1), PostFork messages are accepted for
+                // gossipsub propagation but not processed for committee work.
+                if !receiver.router.should_process(matched_schema) {
+                    trace!(
+                        gossipsub_message_id = ?message_id,
+                        ?matched_schema,
+                        "Message accepted for propagation but not processed (pre-subscribe window)"
+                    );
+                    return;
+                }
+
                 let msg_id = signed_ssv_message.ssv_message().msg_id().clone();
 
                 match msg_id.duty_executor() {

anchor/message_sender/Cargo.toml

@@ -10,6 +10,7 @@ testing = []
 [dependencies]
 database = { workspace = true }
 ethereum_ssz = { workspace = true }
+gossipsub = { workspace = true }
 message_validator = { workspace = true }
 openssl = { workspace = true }
 processor = { workspace = true }

anchor/message_sender/src/impostor.rs

@@ -1,15 +1,17 @@
+use std::num::NonZeroU64;
+
 use ssv_types::{CommitteeId, consensus::UnsignedSSVMessage, message::SignedSSVMessage};
 use subnet_service::SubnetId;
 use tokio::sync::mpsc;
-use tracing::debug;
+use tracing::{debug, warn};
 
 use crate::{Error, MessageCallback, MessageSender};
 
 #[derive(Clone)]
 pub struct ImpostorMessageSender {
     // we only hold this so network does not get sad over the closed channel lol
     _network_tx: mpsc::Sender<(SubnetId, Vec<u8>)>,
-    subnet_count: usize,
+    subnet_count: NonZeroU64,
 }
 
 impl MessageSender for ImpostorMessageSender {
@@ -19,20 +21,32 @@ impl MessageSender for ImpostorMessageSender {
         committee_id: CommitteeId,
         _additional_message_callback: Option<Box<MessageCallback>>,
     ) -> Result<(), Error> {
-        let subnet = SubnetId::from_committee_alan(committee_id, self.subnet_count);
-        debug!(?msg, ?subnet, "Would send message");
+        match SubnetId::from_committee_alan(committee_id, self.subnet_count) {
+            Ok(subnet) => debug!(?msg, ?subnet, "Would send message"),
+            Err(e) => warn!(
+                ?committee_id,
+                ?e,
+                "Failed to calculate subnet for impostor message"
+            ),
+        }
         Ok(())
     }
 
     fn send(&self, msg: SignedSSVMessage, committee_id: CommitteeId) -> Result<(), Error> {
-        let subnet = SubnetId::from_committee_alan(committee_id, self.subnet_count);
-        debug!(?msg, ?subnet, "Would send message");
+        match SubnetId::from_committee_alan(committee_id, self.subnet_count) {
+            Ok(subnet) => debug!(?msg, ?subnet, "Would send message"),
+            Err(e) => warn!(
+                ?committee_id,
+                ?e,
+                "Failed to calculate subnet for impostor message"
+            ),
+        }
         Ok(())
     }
 }
 
 impl ImpostorMessageSender {
-    pub fn new(network_tx: mpsc::Sender<(SubnetId, Vec<u8>)>, subnet_count: usize) -> Self {
+    pub fn new(network_tx: mpsc::Sender<(SubnetId, Vec<u8>)>, subnet_count: NonZeroU64) -> Self {
         Self {
             _network_tx: network_tx,
             subnet_count,

anchor/message_sender/src/network.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use database::OwnOperatorId;
+use gossipsub::IdentTopic;
 use message_validator::{DutiesProvider, MessageAcceptance, Validator};
 use openssl::{
     hash::MessageDigest,
@@ -13,7 +14,7 @@ use ssv_types::{
     CommitteeId, RSA_SIGNATURE_SIZE, consensus::UnsignedSSVMessage, message::SignedSSVMessage,
 };
 use ssz::Encode;
-use subnet_service::SubnetId;
+use subnet_service::{SubnetId, SubnetRouter};
 use tokio::sync::{mpsc, mpsc::error::TrySendError, watch};
 use tracing::{debug, error, trace, warn};
 
@@ -29,8 +30,9 @@ pub struct NetworkMessageSenderConfig<S: SlotClock, D: DutiesProvider> {
     pub private_key: Rsa<Private>,
     pub operator_id: OwnOperatorId,
     pub validator: Option<Arc<Validator<S, D>>>,
-    pub subnet_count: usize,
     pub is_synced: watch::Receiver<bool>,
+    /// Centralized router for fork-aware subnet calculations
+    pub router: Arc<SubnetRouter<S>>,
 }
 
 pub struct NetworkMessageSender<S: SlotClock, D: DutiesProvider> {
@@ -39,8 +41,9 @@ pub struct NetworkMessageSender<S: SlotClock, D: DutiesProvider> {
     private_key: PKey<Private>,
     operator_id: OwnOperatorId,
     validator: Option<Arc<Validator<S, D>>>,
-    subnet_count: usize,
     is_synced: watch::Receiver<bool>,
+    /// Centralized router for fork-aware subnet calculations
+    router: Arc<SubnetRouter<S>>,
 }
 
 impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMessageSender<S, D>> {
@@ -125,16 +128,32 @@ impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageSender<S, D> {
             private_key,
             operator_id: config.operator_id,
             validator: config.validator,
-            subnet_count: config.subnet_count,
             is_synced: config.is_synced,
+            router: config.router,
         }))
     }
 
     fn do_send(&self, message: SignedSSVMessage, committee_id: CommitteeId) {
         let message_bytes = message.as_ssz_bytes();
 
+        // Calculate subnet using centralized router
+        let subnet = match self.router.publish_subnet(committee_id) {
+            Ok(subnet) => subnet,
+            Err(e) => {
+                error!(
+                    ?committee_id,
+                    ?e,
+                    "Failed to calculate subnet for publishing"
+                );
+                return;
+            }
+        };
+
+        // Construct the topic for validation (format: "ssv.v2.<subnet_id>")
+        let topic = IdentTopic::new(format!("ssv.v2.{}", *subnet)).hash();
+
         if let Some(validator) = self.validator.as_ref()
-            && let Err(err) = validator.validate(&message_bytes).as_result()
+            && let Err(err) = validator.validate(&message_bytes, &topic).as_result()
         {
             // `Reject` is more severe and can be punished by other peers. We should not have
             // created this message ever, while `Ignore` can be triggered simply because the message
@@ -148,7 +167,6 @@ impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageSender<S, D> {
             return;
         }
 
-        let subnet = SubnetId::from_committee_alan(committee_id, self.subnet_count);
         match self.network_tx.try_send((subnet, message_bytes)) {
             Ok(_) => trace!(?subnet, "Successfully sent message to network"),
             Err(TrySendError::Closed(_)) => warn!("Network queue closed (shutting down?)"),