feat: add challenge sync mechanism

- Add sync() method to Challenge trait for periodic state sync
- Add WasmSyncResult type for consensus
- Add ChallengeSyncProposal/ChallengeSyncVote P2P messages
- Add execute_sync() in wasm_executor
- Add sync ticker every 60 blocks in validator main loop
- Validators sync all active challenges and broadcast proposals

Author: echobt

bins/validator-node/src/main.rs

@@ -912,6 +912,9 @@ async fn main() -> Result<()> {
     let mut stale_job_interval = tokio::time::interval(Duration::from_secs(120));
     let mut weight_check_interval = tokio::time::interval(Duration::from_secs(30));
     let mut last_weight_submission_epoch: u64 = 0; // Local tracking of weight submissions
+    let mut challenge_sync_interval = tokio::time::interval(Duration::from_secs(60)); // Check every minute
+    let mut last_sync_block: u64 = 0; // Last block where sync was triggered
+    let sync_block_interval: u64 = 60; // Sync every 60 blocks
 
     // Clone p2p_cmd_tx for use in the loop
     let p2p_broadcast_tx = p2p_cmd_tx.clone();
@@ -1497,6 +1500,78 @@ async fn main() -> Result<()> {
                 }
             }
 
+            // Challenge sync ticker - every 60 blocks, sync all challenges
+            _ = challenge_sync_interval.tick() => {
+                if !is_bootnode {
+                    let current_block = state_manager.apply(|state| state.bittensor_block);
+
+                    // Check if we should sync (every 60 blocks)
+                    if current_block / sync_block_interval > last_sync_block / sync_block_interval {
+                        last_sync_block = current_block;
+
+                        // Get all active challenges
+                        let challenges: Vec<_> = {
+                            let cs = chain_state.read();
+                            cs.wasm_challenge_configs
+                                .iter()
+                                .filter(|(_, cfg)| cfg.is_active)
+                                .map(|(id, _)| *id)
+                                .collect()
+                        };
+
+                        for challenge_id in challenges {
+                            let challenge_id_str = challenge_id.to_string();
+                            let module_path = format!("{}.wasm", challenge_id_str);
+
+                            if let Some(ref executor) = wasm_executor {
+                                match executor.execute_sync(&module_path) {
+                                    Ok(sync_result) => {
+                                        info!(
+                                            challenge_id = %challenge_id,
+                                            block = current_block,
+                                            total_users = sync_result.total_users,
+                                            "Challenge sync completed, broadcasting proposal"
+                                        );
+
+                                        // Broadcast sync proposal
+                                        let timestamp = chrono::Utc::now().timestamp_millis();
+                                        let proposal_data = bincode::serialize(&(
+                                            &challenge_id,
+                                            &sync_result.leaderboard_hash,
+                                            current_block,
+                                            timestamp
+                                        )).unwrap_or_default();
+                                        let signature = keypair.sign_bytes(&proposal_data).unwrap_or_default();
+
+                                        let proposal_msg = P2PMessage::ChallengeSyncProposal(
+                                            platform_p2p_consensus::ChallengeSyncProposalMessage {
+                                                challenge_id,
+                                                sync_result_hash: sync_result.leaderboard_hash,
+                                                proposer: keypair.hotkey(),
+                                                block_number: current_block,
+                                                timestamp,
+                                                signature,
+                                            }
+                                        );
+
+                                        if let Err(e) = p2p_broadcast_tx.send(platform_p2p_consensus::P2PCommand::Broadcast(proposal_msg)).await {
+                                            warn!(error = %e, "Failed to broadcast sync proposal");
+                                        }
+                                    }
+                                    Err(e) => {
+                                        debug!(
+                                            challenge_id = %challenge_id,
+                                            error = %e,
+                                            "Failed to execute sync (WASM may not support sync)"
+                                        );
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+
             // Ctrl+C
             _ = tokio::signal::ctrl_c() => {
                 info!("Received shutdown signal, persisting state...");
@@ -3002,6 +3077,21 @@ async fn handle_network_event(
                     }
                 }
             }
+            P2PMessage::ChallengeSyncProposal(proposal) => {
+                info!(
+                    challenge_id = %proposal.challenge_id,
+                    block = proposal.block_number,
+                    proposer = %proposal.proposer.to_ss58(),
+                    "Received challenge sync proposal (consensus voting not yet implemented)"
+                );
+            }
+            P2PMessage::ChallengeSyncVote(vote) => {
+                debug!(
+                    challenge_id = %vote.challenge_id,
+                    voter = %vote.voter.to_ss58(),
+                    "Received challenge sync vote (consensus voting not yet implemented)"
+                );
+            }
         },
         NetworkEvent::PeerConnected(peer_id) => {
             info!("Peer connected: {}", peer_id);

bins/validator-node/src/wasm_executor.rs

@@ -985,6 +985,72 @@ impl WasmChallengeExecutor {
         Ok(weights)
     }
 
+    /// Execute sync on a WASM challenge module.
+    /// Returns WasmSyncResult with leaderboard hash and stats for consensus.
+    pub fn execute_sync(
+        &self,
+        module_path: &str,
+    ) -> Result<platform_challenge_sdk_wasm::WasmSyncResult> {
+        let start = Instant::now();
+
+        let module = self
+            .load_module(module_path)
+            .context("Failed to load WASM module")?;
+
+        let instance_config = InstanceConfig {
+            challenge_id: module_path.to_string(),
+            validator_id: "validator".to_string(),
+            storage_host_config: StorageHostConfig {
+                allow_direct_writes: true,
+                require_consensus: false,
+                ..self.config.storage_host_config.clone()
+            },
+            storage_backend: Arc::clone(&self.config.storage_backend),
+            consensus_policy: ConsensusPolicy::read_only(),
+            ..Default::default()
+        };
+
+        let mut instance = self
+            .runtime
+            .instantiate(&module, instance_config, None)
+            .map_err(|e| anyhow::anyhow!("WASM instantiation failed: {}", e))?;
+
+        let result = instance
+            .call_return_i64("sync")
+            .map_err(|e| anyhow::anyhow!("WASM sync call failed: {}", e))?;
+
+        let out_len = (result >> 32) as i32;
+        let out_ptr = (result & 0xFFFF_FFFF) as i32;
+
+        if out_ptr <= 0 || out_len <= 0 {
+            return Ok(platform_challenge_sdk_wasm::WasmSyncResult {
+                leaderboard_hash: [0u8; 32],
+                total_users: 0,
+                total_valid_issues: 0,
+                total_invalid_issues: 0,
+                total_pending_issues: 0,
+                sync_timestamp: 0,
+            });
+        }
+
+        let result_data = instance
+            .read_memory(out_ptr as usize, out_len as usize)
+            .map_err(|e| anyhow::anyhow!("failed to read WASM memory for sync output: {}", e))?;
+
+        let sync_result: platform_challenge_sdk_wasm::WasmSyncResult =
+            bincode::deserialize(&result_data)
+                .context("Failed to deserialize sync output as WasmSyncResult")?;
+
+        info!(
+            module = module_path,
+            total_users = sync_result.total_users,
+            execution_time_ms = start.elapsed().as_millis() as u64,
+            "WASM sync completed"
+        );
+
+        Ok(sync_result)
+    }
+
     #[allow(dead_code)]
     pub fn execute_validate_storage_write(
         &self,

crates/challenge-sdk-wasm/src/lib.rs

@@ -13,7 +13,9 @@ pub use types::{
 };
 pub use types::{ContainerRunRequest, ContainerRunResponse};
 pub use types::{EvaluationInput, EvaluationOutput};
-pub use types::{WasmRouteDefinition, WasmRouteRequest, WasmRouteResponse, WeightEntry};
+pub use types::{
+    WasmRouteDefinition, WasmRouteRequest, WasmRouteResponse, WasmSyncResult, WeightEntry,
+};
 
 pub trait Challenge {
     fn name(&self) -> &'static str;
@@ -62,6 +64,13 @@ pub trait Challenge {
     fn validate_storage_write(&self, _key: &[u8], _value: &[u8]) -> bool {
         true
     }
+
+    /// Perform a periodic sync operation and return serialized sync result.
+    /// Called by validators at regular intervals for consensus-based state sync.
+    /// The default implementation returns an empty vector.
+    fn sync(&self) -> alloc::vec::Vec<u8> {
+        alloc::vec::Vec::new()
+    }
 }
 
 /// Pack a pointer and length into a single i64 value.
@@ -311,5 +320,21 @@ macro_rules! register_challenge {
                 0
             }
         }
+
+        #[no_mangle]
+        pub extern "C" fn sync() -> i64 {
+            let output = <$ty as $crate::Challenge>::sync(&_CHALLENGE);
+            if output.is_empty() {
+                return $crate::pack_ptr_len(0, 0);
+            }
+            let ptr = $crate::alloc_impl::sdk_alloc(output.len());
+            if ptr.is_null() {
+                return $crate::pack_ptr_len(0, 0);
+            }
+            unsafe {
+                core::ptr::copy_nonoverlapping(output.as_ptr(), ptr, output.len());
+            }
+            $crate::pack_ptr_len(ptr as i32, output.len() as i32)
+        }
     };
 }

crates/challenge-sdk-wasm/src/types.rs

@@ -199,3 +199,16 @@ pub struct WeightEntry {
     pub uid: u16,
     pub weight: u16,
 }
+
+/// Result of a sync operation for consensus.
+///
+/// Returned by [`Challenge::sync`] for validators to reach consensus on state.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct WasmSyncResult {
+    pub leaderboard_hash: [u8; 32],
+    pub total_users: u32,
+    pub total_valid_issues: u32,
+    pub total_invalid_issues: u32,
+    pub total_pending_issues: u32,
+    pub sync_timestamp: i64,
+}

crates/p2p-consensus/src/lib.rs

@@ -45,15 +45,16 @@ pub mod validator;
 pub use config::{P2PConfig, DEFAULT_BOOTSTRAP_NODES, DEFAULT_P2P_PORT};
 pub use consensus::{ConsensusDecision, ConsensusEngine, ConsensusError, ConsensusPhase};
 pub use messages::{
-    ChallengeUpdateMessage, CommitMessage, ConsensusProposal, CoreStateRequestMessage,
-    CoreStateResponseMessage, DataRequestMessage, DataResponseMessage, EvaluationMessage,
-    EvaluationMetrics, HeartbeatMessage, JobAssignmentMessage, JobClaimMessage,
-    LeaderboardRequestMessage, LeaderboardResponseMessage, MerkleNode, MerkleProof, NewViewMessage,
-    P2PMessage, PeerAnnounceMessage, PrePrepare, PrepareMessage, PreparedProof, ProposalContent,
-    RoundId, SequenceNumber, SignedP2PMessage, StateChangeType, StateMutationProposalMessage,
-    StateMutationType, StateMutationVoteMessage, StateRequest, StateResponse,
-    StorageProposalMessage, StorageRootSyncMessage, StorageVoteMessage, SubmissionMessage,
-    TaskProgressMessage, TaskResultMessage, ViewChangeMessage, ViewNumber, WeightVoteMessage,
+    ChallengeSyncProposalMessage, ChallengeSyncVoteMessage, ChallengeUpdateMessage, CommitMessage,
+    ConsensusProposal, CoreStateRequestMessage, CoreStateResponseMessage, DataRequestMessage,
+    DataResponseMessage, EvaluationMessage, EvaluationMetrics, HeartbeatMessage,
+    JobAssignmentMessage, JobClaimMessage, LeaderboardRequestMessage, LeaderboardResponseMessage,
+    MerkleNode, MerkleProof, NewViewMessage, P2PMessage, PeerAnnounceMessage, PrePrepare,
+    PrepareMessage, PreparedProof, ProposalContent, RoundId, SequenceNumber, SignedP2PMessage,
+    StateChangeType, StateMutationProposalMessage, StateMutationType, StateMutationVoteMessage,
+    StateRequest, StateResponse, StorageProposalMessage, StorageRootSyncMessage,
+    StorageVoteMessage, SubmissionMessage, TaskProgressMessage, TaskResultMessage,
+    ViewChangeMessage, ViewNumber, WeightVoteMessage,
 };
 pub use network::{
     CombinedBehaviour, CombinedEvent, NetworkBehaviour, NetworkError, NetworkEvent, NetworkRunner,

crates/p2p-consensus/src/messages.rs

@@ -74,6 +74,10 @@ pub enum P2PMessage {
     CoreStateRequest(CoreStateRequestMessage),
     /// Core state sync response
     CoreStateResponse(CoreStateResponseMessage),
+    /// Challenge sync proposal for periodic consensus
+    ChallengeSyncProposal(ChallengeSyncProposalMessage),
+    /// Vote on a challenge sync proposal
+    ChallengeSyncVote(ChallengeSyncVoteMessage),
 }
 
 impl P2PMessage {
@@ -134,6 +138,8 @@ impl P2PMessage {
             P2PMessage::StateMutationVote(_) => "StateMutationVote",
             P2PMessage::CoreStateRequest(_) => "CoreStateRequest",
             P2PMessage::CoreStateResponse(_) => "CoreStateResponse",
+            P2PMessage::ChallengeSyncProposal(_) => "ChallengeSyncProposal",
+            P2PMessage::ChallengeSyncVote(_) => "ChallengeSyncVote",
         }
     }
 }
@@ -983,3 +989,39 @@ pub struct StorageRootSyncMessage {
     /// Signature
     pub signature: Vec<u8>,
 }
+
+/// Challenge sync proposal for periodic consensus-based state sync
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ChallengeSyncProposalMessage {
+    /// Challenge being synced
+    pub challenge_id: ChallengeId,
+    /// Hash of the sync result for consensus comparison
+    pub sync_result_hash: [u8; 32],
+    /// Proposer validator
+    pub proposer: Hotkey,
+    /// Block number at which sync was triggered
+    pub block_number: u64,
+    /// Timestamp
+    pub timestamp: i64,
+    /// Signature
+    pub signature: Vec<u8>,
+}
+
+/// Vote on a challenge sync proposal
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ChallengeSyncVoteMessage {
+    /// Challenge being synced
+    pub challenge_id: ChallengeId,
+    /// Hash being voted on
+    pub sync_result_hash: [u8; 32],
+    /// Block number
+    pub block_number: u64,
+    /// Voter validator
+    pub voter: Hotkey,
+    /// Whether the voter agrees with the hash
+    pub approve: bool,
+    /// Timestamp
+    pub timestamp: i64,
+    /// Signature
+    pub signature: Vec<u8>,
+}

crates/p2p-consensus/src/network.rs

@@ -1158,6 +1158,8 @@ fn expected_signer(message: &P2PMessage) -> Option<&Hotkey> {
         P2PMessage::StateMutationVote(msg) => Some(&msg.voter),
         P2PMessage::CoreStateRequest(msg) => Some(&msg.requester),
         P2PMessage::CoreStateResponse(msg) => Some(&msg.responder),
+        P2PMessage::ChallengeSyncProposal(msg) => Some(&msg.proposer),
+        P2PMessage::ChallengeSyncVote(msg) => Some(&msg.voter),
     }
 }