tweaks(yamux): Improvements to reducer

node/src/action_kind.rs

@@ -460,6 +460,10 @@ pub enum ActionKind {
     P2pNetworkSelectTimeout,
     P2pNetworkYamuxIncomingData,
     P2pNetworkYamuxIncomingFrame,
+    P2pNetworkYamuxIncomingFrameData,
+    P2pNetworkYamuxIncomingFrameGoAway,
+    P2pNetworkYamuxIncomingFramePing,
+    P2pNetworkYamuxIncomingFrameWindowUpdate,
     P2pNetworkYamuxOpenStream,
     P2pNetworkYamuxOutgoingData,
     P2pNetworkYamuxOutgoingFrame,
@@ -720,7 +724,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 610;
+    pub const COUNT: u16 = 614;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1948,6 +1952,12 @@ impl ActionKindGet for P2pNetworkYamuxAction {
             Self::IncomingData { .. } => ActionKind::P2pNetworkYamuxIncomingData,
             Self::OutgoingData { .. } => ActionKind::P2pNetworkYamuxOutgoingData,
             Self::IncomingFrame { .. } => ActionKind::P2pNetworkYamuxIncomingFrame,
+            Self::IncomingFrameData { .. } => ActionKind::P2pNetworkYamuxIncomingFrameData,
+            Self::IncomingFramePing { .. } => ActionKind::P2pNetworkYamuxIncomingFramePing,
+            Self::IncomingFrameWindowUpdate { .. } => {
+                ActionKind::P2pNetworkYamuxIncomingFrameWindowUpdate
+            }
+            Self::IncomingFrameGoAway { .. } => ActionKind::P2pNetworkYamuxIncomingFrameGoAway,
             Self::OutgoingFrame { .. } => ActionKind::P2pNetworkYamuxOutgoingFrame,
             Self::PingStream { .. } => ActionKind::P2pNetworkYamuxPingStream,
             Self::OpenStream { .. } => ActionKind::P2pNetworkYamuxOpenStream,

p2p/src/network/scheduler/p2p_network_scheduler_state.rs

@@ -8,7 +8,9 @@ use malloc_size_of_derive::MallocSizeOf;
 use redux::Timestamp;
 use serde::{Deserialize, Serialize};
 
-use crate::{disconnection::P2pDisconnectionReason, identity::PublicKey, PeerId};
+use crate::{
+    disconnection::P2pDisconnectionReason, identity::PublicKey, yamux::YamuxStreamState, PeerId,
+};
 
 use super::super::*;
 
@@ -183,6 +185,17 @@ impl P2pNetworkConnectionState {
             SelectKind::Stream(_, stream_id) => Some(&self.streams.get(stream_id)?.select),
         }
     }
+
+    pub fn get_yamux_stream(&self, stream_id: StreamId) -> Option<&YamuxStreamState> {
+        self.yamux_state()?.streams.get(&stream_id)
+    }
+
+    pub fn incoming_streams_count(&self) -> usize {
+        self.streams
+            .iter()
+            .filter(|(_, state)| state.select.is_incoming())
+            .count()
+    }
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, thiserror::Error)]

p2p/src/network/yamux/mod.rs

@@ -3,8 +3,11 @@ pub use self::p2p_network_yamux_actions::*;
 
 mod p2p_network_yamux_state;
 pub use self::p2p_network_yamux_state::{
-    P2pNetworkYamuxState, StreamId, YamuxFlags, YamuxPing, YamuxStreamKind,
+    P2pNetworkYamuxState, StreamId, YamuxFlags, YamuxPing, YamuxStreamKind, YamuxStreamState,
 };
 
 #[cfg(feature = "p2p-libp2p")]
 mod p2p_network_yamux_reducer;
+
+#[cfg(test)]
+mod tests;

p2p/src/network/yamux/p2p_network_yamux_actions.rs

@@ -22,6 +22,22 @@ pub enum P2pNetworkYamuxAction {
         addr: ConnectionAddr,
     },
     #[action_event(level = trace)]
+    IncomingFrameData {
+        addr: ConnectionAddr,
+    },
+    #[action_event(level = trace)]
+    IncomingFramePing {
+        addr: ConnectionAddr,
+    },
+    #[action_event(level = trace)]
+    IncomingFrameWindowUpdate {
+        addr: ConnectionAddr,
+    },
+    #[action_event(level = trace)]
+    IncomingFrameGoAway {
+        addr: ConnectionAddr,
+    },
+    #[action_event(level = trace)]
     OutgoingFrame {
         addr: ConnectionAddr,
         frame: YamuxFrame,
@@ -43,6 +59,10 @@ impl P2pNetworkYamuxAction {
             Self::IncomingData { addr, .. } => addr,
             Self::OutgoingData { addr, .. } => addr,
             Self::IncomingFrame { addr, .. } => addr,
+            Self::IncomingFrameData { addr, .. } => addr,
+            Self::IncomingFramePing { addr, .. } => addr,
+            Self::IncomingFrameWindowUpdate { addr, .. } => addr,
+            Self::IncomingFrameGoAway { addr, .. } => addr,
             Self::OutgoingFrame { addr, .. } => addr,
             Self::PingStream { addr, .. } => addr,
             Self::OpenStream { addr, .. } => addr,
@@ -72,7 +92,11 @@ impl redux::EnablingCondition<P2pState> for P2pNetworkYamuxAction {
             P2pNetworkYamuxAction::OutgoingData { stream_id, .. } => {
                 yamux_state.streams.contains_key(stream_id)
             }
-            P2pNetworkYamuxAction::IncomingFrame { .. } => true,
+            P2pNetworkYamuxAction::IncomingFrame { .. }
+            | P2pNetworkYamuxAction::IncomingFrameData { .. }
+            | P2pNetworkYamuxAction::IncomingFramePing { .. }
+            | P2pNetworkYamuxAction::IncomingFrameWindowUpdate { .. }
+            | P2pNetworkYamuxAction::IncomingFrameGoAway { .. } => !yamux_state.incoming.is_empty(),
             P2pNetworkYamuxAction::OutgoingFrame { .. } => true,
             P2pNetworkYamuxAction::PingStream { .. } => true,
             P2pNetworkYamuxAction::OpenStream { .. } => true,