feat(infra): add P2P handlers, WASM host functions, and distributed storage for decentralized challenges

bins/validator-node/Cargo.toml

@@ -45,4 +45,5 @@ parking_lot = { workspace = true }
 sp-core = { workspace = true }
 uuid = { workspace = true }
 chrono = { workspace = true }
-bincode = { workspace = true }
+bincode = { workspace = true }
+sha2 = { workspace = true }

bins/validator-node/src/challenge_storage.rs

@@ -0,0 +1,55 @@
+use platform_distributed_storage::{
+    DistributedStore, GetOptions as DGetOptions, LocalStorage, PutOptions as DPutOptions,
+    StorageKey as DStorageKey,
+};
+use sha2::{Digest, Sha256};
+use std::sync::Arc;
+use wasm_runtime_interface::{StorageBackend, StorageHostError};
+
+pub struct ChallengeStorageBackend {
+    storage: Arc<LocalStorage>,
+}
+
+impl ChallengeStorageBackend {
+    pub fn new(storage: Arc<LocalStorage>) -> Self {
+        Self { storage }
+    }
+}
+
+impl StorageBackend for ChallengeStorageBackend {
+    fn get(&self, challenge_id: &str, key: &[u8]) -> Result<Option<Vec<u8>>, StorageHostError> {
+        let storage_key = DStorageKey::new(challenge_id, hex::encode(key));
+        let result = tokio::runtime::Handle::current()
+            .block_on(self.storage.get(&storage_key, DGetOptions::default()))
+            .map_err(|e| StorageHostError::StorageError(e.to_string()))?;
+        Ok(result.map(|v| v.data))
+    }
+
+    fn propose_write(
+        &self,
+        challenge_id: &str,
+        key: &[u8],
+        value: &[u8],
+    ) -> Result<[u8; 32], StorageHostError> {
+        let storage_key = DStorageKey::new(challenge_id, hex::encode(key));
+        tokio::runtime::Handle::current()
+            .block_on(
+                self.storage
+                    .put(storage_key, value.to_vec(), DPutOptions::default()),
+            )
+            .map_err(|e| StorageHostError::StorageError(e.to_string()))?;
+
+        let mut hasher = Sha256::new();
+        hasher.update(challenge_id.as_bytes());
+        hasher.update(key);
+        hasher.update(value);
+        Ok(hasher.finalize().into())
+    }
+
+    fn delete(&self, challenge_id: &str, key: &[u8]) -> Result<bool, StorageHostError> {
+        let storage_key = DStorageKey::new(challenge_id, hex::encode(key));
+        tokio::runtime::Handle::current()
+            .block_on(self.storage.delete(&storage_key))
+            .map_err(|e| StorageHostError::StorageError(e.to_string()))
+    }
+}

bins/validator-node/src/main.rs

@@ -4,6 +4,7 @@
 //! Uses libp2p for gossipsub consensus and Kademlia DHT for storage.
 //! Submits weights to Bittensor at epoch boundaries.
 
+mod challenge_storage;
 mod wasm_executor;
 
 use anyhow::Result;
@@ -25,8 +26,9 @@ use platform_distributed_storage::{
     DistributedStoreExt, LocalStorage, LocalStorageBuilder, StorageKey,
 };
 use platform_p2p_consensus::{
-    ChainState, ConsensusEngine, EvaluationMessage, EvaluationMetrics, EvaluationRecord,
-    NetworkEvent, P2PConfig, P2PMessage, P2PNetwork, StateManager, ValidatorRecord, ValidatorSet,
+    ChainState, ConsensusEngine, EvaluationMessage, EvaluationMetrics, EvaluationRecord, JobRecord,
+    JobStatus, NetworkEvent, P2PConfig, P2PMessage, P2PNetwork, StateManager, TaskProgressRecord,
+    ValidatorRecord, ValidatorSet,
 };
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
@@ -419,6 +421,7 @@ async fn main() -> Result<()> {
     let mut state_persist_interval = tokio::time::interval(Duration::from_secs(60));
     let mut checkpoint_interval = tokio::time::interval(Duration::from_secs(300)); // 5 minutes
     let mut wasm_eval_interval = tokio::time::interval(Duration::from_secs(5));
+    let mut stale_job_interval = tokio::time::interval(Duration::from_secs(120));
 
     let (eval_broadcast_tx, mut eval_broadcast_rx) = tokio::sync::mpsc::channel::<P2PMessage>(256);
 
@@ -518,6 +521,15 @@ async fn main() -> Result<()> {
                 }
             }
 
+            // Stale job cleanup
+            _ = stale_job_interval.tick() => {
+                let now = chrono::Utc::now().timestamp_millis();
+                let stale = state_manager.apply(|state| state.cleanup_stale_jobs(now));
+                if !stale.is_empty() {
+                    info!(count = stale.len(), "Cleaned up stale jobs");
+                }
+            }
+
             // Periodic checkpoint
             _ = checkpoint_interval.tick() => {
                 if let Some(handler) = shutdown_handler.as_mut() {
@@ -784,8 +796,160 @@ async fn handle_network_event(
                     }
                 });
             }
-            _ => {
-                debug!("Unhandled P2P message type");
+            P2PMessage::StateRequest(req) => {
+                debug!(
+                    requester = %req.requester.to_hex(),
+                    sequence = req.current_sequence,
+                    "Received state request"
+                );
+            }
+            P2PMessage::StateResponse(resp) => {
+                debug!(
+                    responder = %resp.responder.to_hex(),
+                    sequence = resp.sequence,
+                    "Received state response"
+                );
+            }
+            P2PMessage::WeightVote(wv) => {
+                debug!(
+                    validator = %wv.validator.to_hex(),
+                    epoch = wv.epoch,
+                    "Received weight vote"
+                );
+            }
+            P2PMessage::PeerAnnounce(pa) => {
+                debug!(
+                    validator = %pa.validator.to_hex(),
+                    peer_id = %pa.peer_id,
+                    addresses = pa.addresses.len(),
+                    "Received peer announce"
+                );
+            }
+            P2PMessage::JobClaim(claim) => {
+                info!(
+                    validator = %claim.validator.to_hex(),
+                    challenge_id = %claim.challenge_id,
+                    max_jobs = claim.max_jobs,
+                    "Received job claim"
+                );
+            }
+            P2PMessage::JobAssignment(assignment) => {
+                info!(
+                    submission_id = %assignment.submission_id,
+                    challenge_id = %assignment.challenge_id,
+                    assigned_validator = %assignment.assigned_validator.to_hex(),
+                    assigner = %assignment.assigner.to_hex(),
+                    "Received job assignment"
+                );
+                let job = JobRecord {
+                    submission_id: assignment.submission_id.clone(),
+                    challenge_id: assignment.challenge_id,
+                    assigned_validator: assignment.assigned_validator,
+                    assigned_at: assignment.timestamp,
+                    timeout_at: assignment.timestamp + 300_000,
+                    status: JobStatus::Pending,
+                };
+                state_manager.apply(|state| {
+                    state.assign_job(job);
+                });
+            }
+            P2PMessage::DataRequest(req) => {
+                debug!(
+                    request_id = %req.request_id,
+                    requester = %req.requester.to_hex(),
+                    challenge_id = %req.challenge_id,
+                    data_type = %req.data_type,
+                    "Received data request"
+                );
+            }
+            P2PMessage::DataResponse(resp) => {
+                debug!(
+                    request_id = %resp.request_id,
+                    responder = %resp.responder.to_hex(),
+                    challenge_id = %resp.challenge_id,
+                    data_bytes = resp.data.len(),
+                    "Received data response"
+                );
+            }
+            P2PMessage::TaskProgress(progress) => {
+                debug!(
+                    submission_id = %progress.submission_id,
+                    challenge_id = %progress.challenge_id,
+                    validator = %progress.validator.to_hex(),
+                    task_index = progress.task_index,
+                    total_tasks = progress.total_tasks,
+                    progress_pct = progress.progress_pct,
+                    "Received task progress"
+                );
+                let record = TaskProgressRecord {
+                    submission_id: progress.submission_id.clone(),
+                    challenge_id: progress.challenge_id,
+                    validator: progress.validator,
+                    task_index: progress.task_index,
+                    total_tasks: progress.total_tasks,
+                    status: progress.status,
+                    progress_pct: progress.progress_pct,
+                    updated_at: progress.timestamp,
+                };
+                state_manager.apply(|state| {
+                    state.update_task_progress(record);
+                });
+            }
+            P2PMessage::TaskResult(result) => {
+                info!(
+                    submission_id = %result.submission_id,
+                    challenge_id = %result.challenge_id,
+                    validator = %result.validator.to_hex(),
+                    task_id = %result.task_id,
+                    passed = result.passed,
+                    score = result.score,
+                    execution_time_ms = result.execution_time_ms,
+                    "Received task result"
+                );
+            }
+            P2PMessage::LeaderboardRequest(req) => {
+                debug!(
+                    requester = %req.requester.to_hex(),
+                    challenge_id = %req.challenge_id,
+                    limit = req.limit,
+                    offset = req.offset,
+                    "Received leaderboard request"
+                );
+            }
+            P2PMessage::LeaderboardResponse(resp) => {
+                debug!(
+                    responder = %resp.responder.to_hex(),
+                    challenge_id = %resp.challenge_id,
+                    total_count = resp.total_count,
+                    "Received leaderboard response"
+                );
+            }
+            P2PMessage::ChallengeUpdate(update) => {
+                info!(
+                    challenge_id = %update.challenge_id,
+                    updater = %update.updater.to_hex(),
+                    update_type = %update.update_type,
+                    data_bytes = update.data.len(),
+                    "Received challenge update"
+                );
+            }
+            P2PMessage::StorageProposal(proposal) => {
+                debug!(
+                    proposal_id = %hex::encode(&proposal.proposal_id[..8]),
+                    challenge_id = %proposal.challenge_id,
+                    proposer = %proposal.proposer.to_hex(),
+                    key_len = proposal.key.len(),
+                    value_len = proposal.value.len(),
+                    "Received storage proposal"
+                );
+            }
+            P2PMessage::StorageVote(vote) => {
+                debug!(
+                    proposal_id = %hex::encode(&vote.proposal_id[..8]),
+                    voter = %vote.voter.to_hex(),
+                    approve = vote.approve,
+                    "Received storage vote"
+                );
             }
         },
         NetworkEvent::PeerConnected(peer_id) => {

bins/validator-node/src/wasm_executor.rs

@@ -7,9 +7,10 @@ use std::sync::Arc;
 use std::time::Instant;
 use tracing::{debug, info};
 use wasm_runtime_interface::{
-    ExecPolicy, InMemoryStorageBackend, InstanceConfig, NetworkHostFunctions, NetworkPolicy,
-    NoopStorageBackend, RuntimeConfig, SandboxHostFunctions, SandboxPolicy, StorageHostConfig,
-    StorageHostState, TimePolicy, WasmModule, WasmRuntime, WasmRuntimeError,
+    ConsensusPolicy, ExecPolicy, InMemoryStorageBackend, InstanceConfig, NetworkHostFunctions,
+    NetworkPolicy, NoopStorageBackend, RuntimeConfig, SandboxHostFunctions, SandboxPolicy,
+    StorageHostConfig, StorageHostState, TerminalPolicy, TimePolicy, WasmModule, WasmRuntime,
+    WasmRuntimeError,
 };
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -393,6 +394,8 @@ impl WasmChallengeExecutor {
             storage_host_config: StorageHostConfig::default(),
             storage_backend: Arc::new(InMemoryStorageBackend::new()),
             fixed_timestamp_ms: None,
+            consensus_policy: ConsensusPolicy::default(),
+            terminal_policy: TerminalPolicy::default(),
         };
 
         let mut instance = self
@@ -473,6 +476,8 @@ impl WasmChallengeExecutor {
             storage_host_config: StorageHostConfig::default(),
             storage_backend: Arc::new(InMemoryStorageBackend::new()),
             fixed_timestamp_ms: None,
+            consensus_policy: ConsensusPolicy::default(),
+            terminal_policy: TerminalPolicy::default(),
         };
 
         let mut instance = self

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

@@ -215,3 +215,63 @@ pub fn host_get_timestamp() -> i64 {
 pub fn host_log(level: u8, msg: &str) {
     unsafe { log_message(level as i32, msg.as_ptr() as i32, msg.len() as i32) }
 }
+
+#[link(wasm_import_module = "platform_consensus")]
+extern "C" {
+    fn consensus_get_epoch() -> i64;
+    fn consensus_get_validators(buf_ptr: i32, buf_len: i32) -> i32;
+    fn consensus_propose_weight(uid: i32, weight: i32) -> i32;
+    fn consensus_get_votes(buf_ptr: i32, buf_len: i32) -> i32;
+    fn consensus_get_state_hash(buf_ptr: i32) -> i32;
+    fn consensus_get_submission_count() -> i32;
+    fn consensus_get_block_height() -> i64;
+}
+
+pub fn host_consensus_get_epoch() -> i64 {
+    unsafe { consensus_get_epoch() }
+}
+
+pub fn host_consensus_get_validators() -> Result<Vec<u8>, i32> {
+    let mut buf = vec![0u8; 65536];
+    let status = unsafe { consensus_get_validators(buf.as_mut_ptr() as i32, buf.len() as i32) };
+    if status < 0 {
+        return Err(status);
+    }
+    buf.truncate(status as usize);
+    Ok(buf)
+}
+
+pub fn host_consensus_propose_weight(uid: i32, weight: i32) -> Result<(), i32> {
+    let status = unsafe { consensus_propose_weight(uid, weight) };
+    if status < 0 {
+        return Err(status);
+    }
+    Ok(())
+}
+
+pub fn host_consensus_get_votes() -> Result<Vec<u8>, i32> {
+    let mut buf = vec![0u8; 65536];
+    let status = unsafe { consensus_get_votes(buf.as_mut_ptr() as i32, buf.len() as i32) };
+    if status < 0 {
+        return Err(status);
+    }
+    buf.truncate(status as usize);
+    Ok(buf)
+}
+
+pub fn host_consensus_get_state_hash() -> Result<[u8; 32], i32> {
+    let mut buf = [0u8; 32];
+    let status = unsafe { consensus_get_state_hash(buf.as_mut_ptr() as i32) };
+    if status < 0 {
+        return Err(status);
+    }
+    Ok(buf)
+}
+
+pub fn host_consensus_get_submission_count() -> i32 {
+    unsafe { consensus_get_submission_count() }
+}
+
+pub fn host_consensus_get_block_height() -> i64 {
+    unsafe { consensus_get_block_height() }
+}