refactor(network): generalize paired streams into service sessions

crates/zakura-network/src/zakura/block_sync/service.rs

@@ -1,9 +1,9 @@
 use super::{config::*, events::*, peer_registry::SessionAdmission, wire::*, *};
 use crate::zakura::{
-    handle_pipe_exit, spawn_supervised_pipe, FramedRecv, FramedSend, OrderedSendError,
-    OrderedSessionDemand, OrderedStreamOpening, OrderedStreamPolicy, Peer, PeerStreamSession,
-    Service, ServicePeerSnapshot, SinkReject, Stream, StreamMode, ZakuraBlockSyncCandidateState,
-    ZakuraConnId, ZakuraPeerId, FRAME_HEADER_BYTES,
+    handle_pipe_exit, spawn_supervised_pipe, FramedRecv, FramedSend, OrderedSendError, Peer,
+    PeerStreamSession, Service, ServicePeerSnapshot, SessionDemand, SessionOpening, SessionPolicy,
+    SinkReject, Stream, StreamMode, ZakuraBlockSyncCandidateState, ZakuraConnId, ZakuraPeerId,
+    FRAME_HEADER_BYTES,
 };
 use std::{
     sync::atomic::{AtomicU64, Ordering},
@@ -26,7 +26,7 @@ const BLOCK_SYNC_SERVICE_STREAMS: [Stream; 1] = [Stream {
     version: ZAKURA_BLOCK_SYNC_STREAM_VERSION,
     frame_cap: MAX_BS_FRAME_BYTES,
     capability: ZAKURA_CAP_BLOCK_SYNC,
-    mode: StreamMode::Ordered,
+    mode: StreamMode::Persistent,
 }];
 
 /// Service-declared streams for native block sync.
@@ -451,28 +451,28 @@ impl Service for BlockSyncService {
         block_sync_streams()
     }
 
-    fn ordered_stream_policy(&self, _kind: u16) -> OrderedStreamPolicy {
-        OrderedStreamPolicy {
-            opening: OrderedStreamOpening::EitherSide,
+    fn session_policy(&self) -> SessionPolicy {
+        SessionPolicy {
+            opening: SessionOpening::EitherSide,
             reopen: true,
         }
     }
 
-    fn ordered_session_demand(
+    fn session_demand(
         &self,
         conn_id: ZakuraConnId,
         peer: &ZakuraPeerId,
         _negotiated: u64,
         direction: ServicePeerDirection,
-    ) -> OrderedSessionDemand {
+    ) -> SessionDemand {
         if let Some(deadline) = self.peer_park_deadline(peer) {
-            return OrderedSessionDemand::RetryAt(deadline);
+            return SessionDemand::RetryAt(deadline);
         }
 
         let mut peer_snapshot = self.inner.peer_snapshot.clone();
         peer_snapshot.borrow_and_update();
         if !self.peer_slots_free(direction) {
-            return OrderedSessionDemand::WaitForChange(Box::pin(async move {
+            return SessionDemand::WaitForChange(Box::pin(async move {
                 if peer_snapshot.changed().await.is_err() {
                     std::future::pending::<()>().await;
                 }
@@ -495,7 +495,7 @@ impl Service for BlockSyncService {
                 .is_empty()
             {
                 let mut service_demand = self.service_demand.clone();
-                return OrderedSessionDemand::WaitForChange(Box::pin(async move {
+                return SessionDemand::WaitForChange(Box::pin(async move {
                     if let Some(demand) = service_demand.as_mut() {
                         tokio::select! {
                             changed = candidates.changed() => {
@@ -516,7 +516,7 @@ impl Service for BlockSyncService {
             }
         }
 
-        OrderedSessionDemand::OpenNow
+        SessionDemand::OpenNow
     }
 
     fn wants_peer(
@@ -760,8 +760,8 @@ impl Service for BlockSyncService {
             return false;
         };
         matches!(
-            self.ordered_session_demand(conn_id, peer, ZAKURA_CAP_BLOCK_SYNC, direction),
-            OrderedSessionDemand::OpenNow
+            self.session_demand(conn_id, peer, ZAKURA_CAP_BLOCK_SYNC, direction),
+            SessionDemand::OpenNow
         )
     }
 

crates/zakura-network/src/zakura/block_sync/tests.rs

@@ -32,8 +32,8 @@ use crate::zakura::{
     framed_channel,
     testkit::{await_until, TraceCapture, TraceValue},
     trace::BlockBodySource,
-    FramedRecv, FramedSend, OrderedSessionDemand, Peer, Service, ServicePeerSnapshot,
-    ServiceRegistry, StreamMode, ZakuraBlockSyncCandidateState,
+    FramedRecv, FramedSend, Peer, Service, ServicePeerSnapshot, ServiceRegistry, SessionDemand,
+    StreamMode, ZakuraBlockSyncCandidateState,
 };
 use zakura_chain::{
     fmt::HexDebug,
@@ -6232,7 +6232,7 @@ fn block_sync_stream_declares_kind_capability_version_and_frame_cap() {
     assert_eq!(stream.kind, ZAKURA_STREAM_BLOCK_SYNC);
     assert_eq!(stream.version, ZAKURA_BLOCK_SYNC_STREAM_VERSION);
     assert_eq!(stream.capability, ZAKURA_CAP_BLOCK_SYNC);
-    assert_eq!(stream.mode, StreamMode::Ordered);
+    assert_eq!(stream.mode, StreamMode::Persistent);
     assert_eq!(stream.frame_cap, MAX_BS_FRAME_BYTES);
 }
 
@@ -6255,14 +6255,14 @@ async fn service_registry_routes_block_sync_by_exact_capability_and_version() {
         .is_none());
     assert_eq!(
         registry
-            .ordered_streams_for_negotiated(ZAKURA_CAP_BLOCK_SYNC)
+            .persistent_streams_for_negotiated(ZAKURA_CAP_BLOCK_SYNC)
             .iter()
             .map(|stream| stream.kind)
             .collect::<Vec<_>>(),
         vec![ZAKURA_STREAM_BLOCK_SYNC]
     );
-    assert!(registry.ordered_streams_for_negotiated(0).is_empty());
-    assert!(registry.wants_ordered_stream(
+    assert!(registry.persistent_streams_for_negotiated(0).is_empty());
+    assert!(registry.wants_session(
         ZAKURA_STREAM_BLOCK_SYNC,
         ZAKURA_CAP_BLOCK_SYNC,
         &peer,
@@ -15311,24 +15311,24 @@ async fn parked_connection_cleanup_allows_a_fresh_connection_after_cooldown() {
     handle.park_session_for_test(&peer, old_conn_id, Duration::ZERO);
 
     assert!(matches!(
-        service.ordered_session_demand(
+        service.session_demand(
             old_conn_id,
             &peer,
             ZAKURA_CAP_BLOCK_SYNC,
             ServicePeerDirection::Outbound,
         ),
-        OrderedSessionDemand::WaitForChange(_),
+        SessionDemand::WaitForChange(_),
     ));
 
     service.remove_peer(&peer, old_conn_id);
     assert!(matches!(
-        service.ordered_session_demand(
+        service.session_demand(
             new_conn_id,
             &peer,
             ZAKURA_CAP_BLOCK_SYNC,
             ServicePeerDirection::Outbound,
         ),
-        OrderedSessionDemand::OpenNow
+        SessionDemand::OpenNow
     ));
     reactor_task.abort();
 }
@@ -15353,13 +15353,13 @@ async fn same_connection_block_sync_session_waits_at_tip_then_reopens_for_new_wo
     let conn_id = 17;
     handle.park_session_for_test(&peer, conn_id, Duration::ZERO);
 
-    let demand = service.ordered_session_demand(
+    let demand = service.session_demand(
         conn_id,
         &peer,
         ZAKURA_CAP_BLOCK_SYNC,
         ServicePeerDirection::Outbound,
     );
-    let OrderedSessionDemand::WaitForChange(changed) = demand else {
+    let SessionDemand::WaitForChange(changed) = demand else {
         panic!("a locally parked session must stay absent while block sync is at tip");
     };
 
@@ -15375,13 +15375,13 @@ async fn same_connection_block_sync_session_waits_at_tip_then_reopens_for_new_wo
         .expect("new block work wakes the parked session demand");
 
     assert!(matches!(
-        service.ordered_session_demand(
+        service.session_demand(
             conn_id,
             &peer,
             ZAKURA_CAP_BLOCK_SYNC,
             ServicePeerDirection::Outbound,
         ),
-        OrderedSessionDemand::OpenNow,
+        SessionDemand::OpenNow,
     ));
     reactor_task.abort();
 }
@@ -15420,7 +15420,7 @@ async fn serving_only_coordinator_demand_keeps_block_session_available_during_fa
     let conn_id = 18;
     handle.park_session_for_test(&peer, conn_id, Duration::ZERO);
 
-    let OrderedSessionDemand::WaitForChange(changed) = service.ordered_session_demand(
+    let SessionDemand::WaitForChange(changed) = service.session_demand(
         conn_id,
         &peer,
         ZAKURA_CAP_BLOCK_SYNC,
@@ -15440,13 +15440,13 @@ async fn serving_only_coordinator_demand_keeps_block_session_available_during_fa
         .await
         .expect("fallback service demand wakes the parked ordered session");
     assert!(matches!(
-        service.ordered_session_demand(
+        service.session_demand(
             conn_id,
             &peer,
             ZAKURA_CAP_BLOCK_SYNC,
             ServicePeerDirection::Outbound,
         ),
-        OrderedSessionDemand::OpenNow,
+        SessionDemand::OpenNow,
     ));
     reactor_task.abort();
 }

crates/zakura-network/src/zakura/discovery/service.rs

@@ -24,9 +24,9 @@ use tokio_util::sync::CancellationToken;
 use crate::zakura::{
     handle_pipe_exit, spawn_supervised_peer_task, spawn_supervised_pipe, BlockSyncHandle,
     CloseCause, Event, Flow, Frame, FramedRecv, FramedSend, HeaderSyncHandle, OrderedSendError,
-    OrderedSessionDemand, OrderedStreamOpening, OrderedStreamPolicy, Peer, PeerStreamSession, Pipe,
-    Service, ServiceAdmissionDecision, ServicePeerDirection, SinkReject, Stream, StreamMode,
-    ZakuraConnId, ZakuraPeerId, LOCAL_MAX_CONTROL_FRAME_BYTES, ZAKURA_CAP_DISCOVERY,
+    Peer, PeerStreamSession, Pipe, Service, ServiceAdmissionDecision, ServicePeerDirection,
+    SessionDemand, SessionOpening, SessionPolicy, SinkReject, Stream, StreamMode, ZakuraConnId,
+    ZakuraPeerId, LOCAL_MAX_CONTROL_FRAME_BYTES, ZAKURA_CAP_DISCOVERY,
 };
 
 #[cfg(test)]
@@ -47,7 +47,7 @@ const DISCOVERY_SERVICE_STREAMS: [Stream; 1] = [Stream {
     version: ZAKURA_DISCOVERY_STREAM_VERSION,
     frame_cap: LOCAL_MAX_CONTROL_FRAME_BYTES,
     capability: ZAKURA_CAP_DISCOVERY,
-    mode: StreamMode::Ordered,
+    mode: StreamMode::Persistent,
 }];
 
 /// Service-declared streams for native discovery.
@@ -305,28 +305,28 @@ impl Service for DiscoveryService {
         discovery_streams()
     }
 
-    fn ordered_stream_policy(&self, _kind: u16) -> OrderedStreamPolicy {
-        OrderedStreamPolicy {
-            opening: OrderedStreamOpening::InitiatorOnly,
+    fn session_policy(&self) -> SessionPolicy {
+        SessionPolicy {
+            opening: SessionOpening::InitiatorOnly,
             reopen: true,
         }
     }
 
-    fn ordered_session_demand(
+    fn session_demand(
         &self,
         conn_id: ZakuraConnId,
         peer: &ZakuraPeerId,
         _negotiated: u64,
         direction: ServicePeerDirection,
-    ) -> OrderedSessionDemand {
+    ) -> SessionDemand {
         if self
             .session_states
             .lock()
             .expect("discovery session-state mutex is never poisoned")
             .get(&(peer.clone(), conn_id))
             .is_some_and(|record| record.state == DiscoverySessionState::Retired)
         {
-            return OrderedSessionDemand::Retire;
+            return SessionDemand::Retire;
         }
 
         let mut peers = self.handle.subscribe_peer_snapshot();
@@ -336,14 +336,14 @@ impl Service for DiscoveryService {
             ServicePeerDirection::Outbound => snapshot.outbound_slots_free,
         };
         if slots_free == 0 {
-            return OrderedSessionDemand::WaitForChange(Box::pin(async move {
+            return SessionDemand::WaitForChange(Box::pin(async move {
                 if peers.changed().await.is_err() {
                     std::future::pending::<()>().await;
                 }
             }));
         }
 
-        OrderedSessionDemand::OpenNow
+        SessionDemand::OpenNow
     }
 
     fn wants_peer(
@@ -1970,13 +1970,13 @@ mod tests {
         );
         assert_eq!(record.state, DiscoverySessionState::Retired);
         assert!(matches!(
-            service.ordered_session_demand(
+            service.session_demand(
                 0,
                 &peer_id,
                 ZAKURA_CAP_DISCOVERY,
                 ServicePeerDirection::Inbound,
             ),
-            OrderedSessionDemand::Retire,
+            SessionDemand::Retire,
         ));
         Ok(())
     }
@@ -2313,13 +2313,13 @@ mod tests {
         // A finished discovery-only exchange retires the session so the handler
         // cannot reopen the stream while the connection tears down.
         assert!(matches!(
-            service.ordered_session_demand(
+            service.session_demand(
                 0,
                 &peer_id,
                 ZAKURA_CAP_DISCOVERY | ZAKURA_CAP_BLOCK_SYNC,
                 ServicePeerDirection::Inbound,
             ),
-            OrderedSessionDemand::Retire,
+            SessionDemand::Retire,
         ));
         wait_for_discovery_inbound_peers(&handle, 0).await;
 
@@ -2557,13 +2557,13 @@ mod tests {
         // The refresh loop still owns this session.
         // The handler must keep the stream eligible.
         assert!(!matches!(
-            service.ordered_session_demand(
+            service.session_demand(
                 0,
                 &peer_id,
                 ZAKURA_CAP_DISCOVERY | ZAKURA_CAP_HEADER_SYNC,
                 ServicePeerDirection::Inbound,
             ),
-            OrderedSessionDemand::Retire,
+            SessionDemand::Retire,
         ));
 
         connection_cancel.cancel();