fix: wire P2P backpressure to resolve 'no available capacity' errors

echobt · echobt · commit caf596a2223e · 2026-04-02T08:39:13.000Z
- Add process_retries() periodic task (500ms interval)
- Wire ChannelMetrics with RealP2PSender::with_metrics()
- Replace all 15 try_send() calls with send_timeout() by priority:
  - HIGH (200ms): StateMutationProposal/Vote (2 calls)
  - MEDIUM (100ms): Heartbeat (1 call)
  - LOW (10ms): Storage/Challenge broadcasts (12 calls)
- Fix bug: increment dropped_count when max retries exceeded
- Add retry queue integration tests

Fixes P2P broadcast channel overflow that was dropping messages
without backpressure, causing 'no available capacity' errors.
diff --git a/bins/validator-node/src/main.rs b/bins/validator-node/src/main.rs
@@ -34,7 +34,7 @@ use platform_distributed_storage::{
 
 use platform_p2p_consensus::{
     ChainState, ConsensusEngine, EvaluationMessage, EvaluationMetrics, EvaluationRecord,
-    HeartbeatMessage, JobRecord, JobStatus, NetworkEvent, P2PConfig, P2PMessage, P2PNetwork,
+    HeartbeatMessage, JobRecord, JobStatus, MessagePriority, NetworkEvent, P2PConfig, P2PMessage, P2PNetwork,
     StateManager, StorageProposal, StorageVoteMessage, TaskProgressRecord, ValidatorRecord,
     ValidatorSet,
 };
@@ -1512,7 +1512,7 @@ async fn main() -> Result<()> {
                         };
 
                         let msg = P2PMessage::ChallengeUpdate(update_msg);
-                        if let Err(e) = p2p_broadcast_tx.try_send(platform_p2p_consensus::P2PCommand::Broadcast(msg)) {
+                        if let Err(e) = p2p_broadcast_tx.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(msg), MessagePriority::Low.timeout()).await {
                             error!("Failed to broadcast sudo action: {}", e);
                         } else {
                             info!("Sudo action broadcast to P2P network");
@@ -1543,7 +1543,7 @@ async fn main() -> Result<()> {
                         };
 
                         let msg = P2PMessage::ChallengeUpdate(update_msg);
-                        if let Err(e) = p2p_broadcast_tx.try_send(platform_p2p_consensus::P2PCommand::Broadcast(msg)) {
+                        if let Err(e) = p2p_broadcast_tx.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(msg), MessagePriority::Low.timeout()).await {
                             error!("Failed to broadcast ChallengeUpdate: {}", e);
                         } else {
                             info!(
@@ -1582,7 +1582,7 @@ async fn main() -> Result<()> {
                                 );
 
                                 if let Err(e) = p2p_broadcast_tx
-                                    .try_send(platform_p2p_consensus::P2PCommand::Broadcast(mutation_msg))
+                                    .send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(mutation_msg), MessagePriority::High.timeout()).await
                                 {
                                     warn!(error = %e, "Failed to broadcast StateMutationProposal");
                                 }
@@ -1836,7 +1836,7 @@ async fn main() -> Result<()> {
                         },
                     });
 
-                    if let Err(e) = p2p_broadcast_tx.try_send(platform_p2p_consensus::P2PCommand::Broadcast(heartbeat)) {
+                    if let Err(e) = p2p_broadcast_tx.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(heartbeat), MessagePriority::Medium.timeout()).await {
                         warn!("Failed to broadcast heartbeat: {}", e);
                     }
 
@@ -2171,7 +2171,7 @@ async fn main() -> Result<()> {
                                             }
                                         );
 
-                                        if let Err(e) = p2p_broadcast_tx.try_send(platform_p2p_consensus::P2PCommand::Broadcast(proposal_msg)) {
+                                        if let Err(e) = p2p_broadcast_tx.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(proposal_msg), MessagePriority::Low.timeout()).await {
                                             warn!(error = %e, "Failed to broadcast sync proposal");
                                         }
                                     }
@@ -2676,7 +2676,7 @@ async fn handle_network_event(
                         );
 
                         if let Err(e) =
-                            p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(req))
+                            p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(req), MessagePriority::Low.timeout()).await
                         {
                             warn!(error = %e, "Failed to send core state request");
                         }
@@ -2910,7 +2910,7 @@ async fn handle_network_event(
                             );
 
                             if let Err(e) = p2p_sender
-                                .try_send(platform_p2p_consensus::P2PCommand::Broadcast(response))
+                                .send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(response), MessagePriority::Low.timeout()).await
                             {
                                 warn!(error = %e, "Failed to send storage data response");
                             } else {
@@ -3072,7 +3072,7 @@ async fn handle_network_event(
                         },
                     );
                     if let Err(e) =
-                        p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(resp))
+                        p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(resp), MessagePriority::Low.timeout()).await
                     {
                         warn!(error = %e, "Failed to send leaderboard response");
                     }
@@ -3574,7 +3574,7 @@ async fn handle_network_event(
                     }
 
                     if let Err(e) =
-                        p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(vote_msg))
+                        p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(vote_msg), MessagePriority::Low.timeout()).await
                     {
                         warn!(error = %e, "Failed to broadcast storage vote");
                     }
@@ -3776,7 +3776,7 @@ async fn handle_network_event(
                             );
 
                             if let Err(e) = p2p_sender
-                                .try_send(platform_p2p_consensus::P2PCommand::Broadcast(req))
+                                .send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(req), MessagePriority::Low.timeout()).await
                             {
                                 warn!(error = %e, "Failed to send storage sync request");
                             }
@@ -3847,7 +3847,7 @@ async fn handle_network_event(
                     );
 
                     if let Err(e) =
-                        p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(vote_msg))
+                        p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(vote_msg), MessagePriority::High.timeout()).await
                     {
                         warn!(error = %e, "Failed to broadcast state mutation vote");
                     }
@@ -4032,7 +4032,7 @@ async fn handle_network_event(
                     );
 
                     if let Err(e) =
-                        p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(response))
+                        p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(response), MessagePriority::Low.timeout()).await
                     {
                         warn!(error = %e, "Failed to send core state response");
                     }
@@ -4171,8 +4171,8 @@ async fn handle_network_event(
                     );
 
                     if let Err(e) =
-                        p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(req))
-                    {
+                            p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(req), MessagePriority::Low.timeout()).await
+                        {
                         warn!(error = %e, "Failed to send storage sync request");
                     }
                 }
@@ -4283,7 +4283,7 @@ async fn handle_network_event(
                                 );
 
                                 if let Err(e) = p2p_sender
-                                    .try_send(platform_p2p_consensus::P2PCommand::Broadcast(resp))
+                                    .send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(resp), MessagePriority::Low.timeout()).await
                                 {
                                     warn!(error = %e, "Failed to send storage sync response");
                                 } else {
@@ -5812,7 +5812,7 @@ async fn process_wasm_evaluations(
             signature,
             timestamp,
         });
-        if let Err(e) = p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(eval_msg))
+        if let Err(e) = p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(eval_msg), MessagePriority::Low.timeout()).await
         {
             warn!(
                 submission_id = %submission_id,
diff --git a/crates/p2p-consensus/src/network.rs b/crates/p2p-consensus/src/network.rs
@@ -465,6 +465,9 @@ impl RealP2PSender {
                 Err(e) => {
                     let next_attempt = attempt + 1;
                     if !self.retry_queue.enqueue(cmd, next_attempt) {
+                        if let Some(ref metrics) = self.metrics {
+                            metrics.increment_dropped();
+                        }
                         error!(
                             attempt = next_attempt,
                             error = %e,
diff --git a/tests/Cargo.toml b/tests/Cargo.toml
@@ -53,6 +53,10 @@ required-features = ["wasm-tests"]
 name = "channel_overflow_test"
 path = "channel_overflow_test.rs"
 
+[[test]]
+name = "retry_queue_integration_test"
+path = "retry_queue_integration_test.rs"
+
 [dependencies]
 platform-core = { path = "../crates/core" }
 platform-storage = { path = "../crates/storage" }
diff --git a/tests/retry_queue_integration_test.rs b/tests/retry_queue_integration_test.rs

Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,7 @@ use platform_distributed_storage::{`
`34`	`34`
`35`	`35`	`use platform_p2p_consensus::{`
`36`	`36`	`ChainState, ConsensusEngine, EvaluationMessage, EvaluationMetrics, EvaluationRecord,`
`37`		`- HeartbeatMessage, JobRecord, JobStatus, NetworkEvent, P2PConfig, P2PMessage, P2PNetwork,`
	`37`	`+ HeartbeatMessage, JobRecord, JobStatus, MessagePriority, NetworkEvent, P2PConfig, P2PMessage, P2PNetwork,`
`38`	`38`	`StateManager, StorageProposal, StorageVoteMessage, TaskProgressRecord, ValidatorRecord,`
`39`	`39`	`ValidatorSet,`
`40`	`40`	`};`
`@@ -1512,7 +1512,7 @@ async fn main() -> Result<()> {`
`1512`	`1512`	`};`
`1513`	`1513`
`1514`	`1514`	`let msg = P2PMessage::ChallengeUpdate(update_msg);`
`1515`		`- if let Err(e) = p2p_broadcast_tx.try_send(platform_p2p_consensus::P2PCommand::Broadcast(msg)) {`
	`1515`	`+ if let Err(e) = p2p_broadcast_tx.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(msg), MessagePriority::Low.timeout()).await {`
`1516`	`1516`	`error!("Failed to broadcast sudo action: {}", e);`
`1517`	`1517`	`} else {`
`1518`	`1518`	`info!("Sudo action broadcast to P2P network");`
`@@ -1543,7 +1543,7 @@ async fn main() -> Result<()> {`
`1543`	`1543`	`};`
`1544`	`1544`
`1545`	`1545`	`let msg = P2PMessage::ChallengeUpdate(update_msg);`
`1546`		`- if let Err(e) = p2p_broadcast_tx.try_send(platform_p2p_consensus::P2PCommand::Broadcast(msg)) {`
	`1546`	`+ if let Err(e) = p2p_broadcast_tx.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(msg), MessagePriority::Low.timeout()).await {`
`1547`	`1547`	`error!("Failed to broadcast ChallengeUpdate: {}", e);`
`1548`	`1548`	`} else {`
`1549`	`1549`	`info!(`
`@@ -1582,7 +1582,7 @@ async fn main() -> Result<()> {`
`1582`	`1582`	`);`
`1583`	`1583`
`1584`	`1584`	`if let Err(e) = p2p_broadcast_tx`
`1585`		`- .try_send(platform_p2p_consensus::P2PCommand::Broadcast(mutation_msg))`
	`1585`	`+ .send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(mutation_msg), MessagePriority::High.timeout()).await`
`1586`	`1586`	`{`
`1587`	`1587`	`warn!(error = %e, "Failed to broadcast StateMutationProposal");`
`1588`	`1588`	`}`
`@@ -1836,7 +1836,7 @@ async fn main() -> Result<()> {`
`1836`	`1836`	`},`
`1837`	`1837`	`});`
`1838`	`1838`
`1839`		`- if let Err(e) = p2p_broadcast_tx.try_send(platform_p2p_consensus::P2PCommand::Broadcast(heartbeat)) {`
	`1839`	`+ if let Err(e) = p2p_broadcast_tx.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(heartbeat), MessagePriority::Medium.timeout()).await {`
`1840`	`1840`	`warn!("Failed to broadcast heartbeat: {}", e);`
`1841`	`1841`	`}`
`1842`	`1842`
`@@ -2171,7 +2171,7 @@ async fn main() -> Result<()> {`
`2171`	`2171`	`}`
`2172`	`2172`	`);`
`2173`	`2173`
`2174`		`- if let Err(e) = p2p_broadcast_tx.try_send(platform_p2p_consensus::P2PCommand::Broadcast(proposal_msg)) {`
	`2174`	`+ if let Err(e) = p2p_broadcast_tx.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(proposal_msg), MessagePriority::Low.timeout()).await {`
`2175`	`2175`	`warn!(error = %e, "Failed to broadcast sync proposal");`
`2176`	`2176`	`}`
`2177`	`2177`	`}`
`@@ -2676,7 +2676,7 @@ async fn handle_network_event(`
`2676`	`2676`	`);`
`2677`	`2677`
`2678`	`2678`	`if let Err(e) =`
`2679`		`- p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(req))`
	`2679`	`+ p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(req), MessagePriority::Low.timeout()).await`
`2680`	`2680`	`{`
`2681`	`2681`	`warn!(error = %e, "Failed to send core state request");`
`2682`	`2682`	`}`
`@@ -2910,7 +2910,7 @@ async fn handle_network_event(`
`2910`	`2910`	`);`
`2911`	`2911`
`2912`	`2912`	`if let Err(e) = p2p_sender`
`2913`		`- .try_send(platform_p2p_consensus::P2PCommand::Broadcast(response))`
	`2913`	`+ .send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(response), MessagePriority::Low.timeout()).await`
`2914`	`2914`	`{`
`2915`	`2915`	`warn!(error = %e, "Failed to send storage data response");`
`2916`	`2916`	`} else {`
`@@ -3072,7 +3072,7 @@ async fn handle_network_event(`
`3072`	`3072`	`},`
`3073`	`3073`	`);`
`3074`	`3074`	`if let Err(e) =`
`3075`		`- p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(resp))`
	`3075`	`+ p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(resp), MessagePriority::Low.timeout()).await`
`3076`	`3076`	`{`
`3077`	`3077`	`warn!(error = %e, "Failed to send leaderboard response");`
`3078`	`3078`	`}`
`@@ -3574,7 +3574,7 @@ async fn handle_network_event(`
`3574`	`3574`	`}`
`3575`	`3575`
`3576`	`3576`	`if let Err(e) =`
`3577`		`- p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(vote_msg))`
	`3577`	`+ p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(vote_msg), MessagePriority::Low.timeout()).await`
`3578`	`3578`	`{`
`3579`	`3579`	`warn!(error = %e, "Failed to broadcast storage vote");`
`3580`	`3580`	`}`
`@@ -3776,7 +3776,7 @@ async fn handle_network_event(`
`3776`	`3776`	`);`
`3777`	`3777`
`3778`	`3778`	`if let Err(e) = p2p_sender`
`3779`		`- .try_send(platform_p2p_consensus::P2PCommand::Broadcast(req))`
	`3779`	`+ .send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(req), MessagePriority::Low.timeout()).await`
`3780`	`3780`	`{`
`3781`	`3781`	`warn!(error = %e, "Failed to send storage sync request");`
`3782`	`3782`	`}`
`@@ -3847,7 +3847,7 @@ async fn handle_network_event(`
`3847`	`3847`	`);`
`3848`	`3848`
`3849`	`3849`	`if let Err(e) =`
`3850`		`- p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(vote_msg))`
	`3850`	`+ p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(vote_msg), MessagePriority::High.timeout()).await`
`3851`	`3851`	`{`
`3852`	`3852`	`warn!(error = %e, "Failed to broadcast state mutation vote");`
`3853`	`3853`	`}`
`@@ -4032,7 +4032,7 @@ async fn handle_network_event(`
`4032`	`4032`	`);`
`4033`	`4033`
`4034`	`4034`	`if let Err(e) =`
`4035`		`- p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(response))`
	`4035`	`+ p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(response), MessagePriority::Low.timeout()).await`
`4036`	`4036`	`{`
`4037`	`4037`	`warn!(error = %e, "Failed to send core state response");`
`4038`	`4038`	`}`
`@@ -4171,8 +4171,8 @@ async fn handle_network_event(`
`4171`	`4171`	`);`
`4172`	`4172`
`4173`	`4173`	`if let Err(e) =`
`4174`		`- p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(req))`
`4175`		`- {`
	`4174`	`+ p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(req), MessagePriority::Low.timeout()).await`
	`4175`	`+ {`
`4176`	`4176`	`warn!(error = %e, "Failed to send storage sync request");`
`4177`	`4177`	`}`
`4178`	`4178`	`}`
`@@ -4283,7 +4283,7 @@ async fn handle_network_event(`
`4283`	`4283`	`);`
`4284`	`4284`
`4285`	`4285`	`if let Err(e) = p2p_sender`
`4286`		`- .try_send(platform_p2p_consensus::P2PCommand::Broadcast(resp))`
	`4286`	`+ .send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(resp), MessagePriority::Low.timeout()).await`
`4287`	`4287`	`{`
`4288`	`4288`	`warn!(error = %e, "Failed to send storage sync response");`
`4289`	`4289`	`} else {`
`@@ -5812,7 +5812,7 @@ async fn process_wasm_evaluations(`
`5812`	`5812`	`signature,`
`5813`	`5813`	`timestamp,`
`5814`	`5814`	`});`
`5815`		`- if let Err(e) = p2p_sender.try_send(platform_p2p_consensus::P2PCommand::Broadcast(eval_msg))`
	`5815`	`+ if let Err(e) = p2p_sender.send_timeout(platform_p2p_consensus::P2PCommand::Broadcast(eval_msg), MessagePriority::Low.timeout()).await`
`5816`	`5816`	`{`
`5817`	`5817`	`warn!(`
`5818`	`5818`	`submission_id = %submission_id,`