Add Rust RUDP reconnect loop

docs/cultnet-transport-parity.md

@@ -247,6 +247,10 @@ Current progress:
   callers: the caller-owned receive loop reports closure, the shared controller
   owns retry delay/exhaustion state, and the caller-provided factory opens the
   next `CultNetRudpSocketTransportConnection`.
+- Rust now has a poll-driven `CultNetRudpReconnectLoop` for substrate callers:
+  service loops report closure, ask `reconnect_if_due(now_ms)` inside their own
+  reactor, and the shared controller owns retry delay/exhaustion state while the
+  caller-owned factory opens the next socket transport.
 - Kotlin now has the same RUDP reconnect-loop shape for JVM/Android callers:
   the app-owned receive loop reports closure, `CultNetRudpReconnectLoop`
   schedules retry through `CultNetReconnectController`, and the caller-provided

docs/runtime-parity-scope.md

@@ -33,7 +33,7 @@ CultMesh.
 | C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path, shared transport profiles, shared RUDP socket transport, RUDP reconnect policy/controller, LiteNetLib client reconnect-controller adoption, CultMesh node/session/catalog/stream/simulation surfaces | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption |
 | Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, schema catalogs, shard catalogs/logs, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe |
 | TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, transport profiles, TCP-framed transport port, shared RUDP socket transport, RUDP reconnect policy/controller, RUDP reconnect-loop helper, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack |
-| Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, RUDP reconnect policy/controller, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for peer cards, authority leases, RUDP endpoint parsing, and peer-card RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server |
+| Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for peer cards, authority leases, RUDP endpoint parsing, and peer-card RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server |
 | Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs, authority leases, stream catalog, branded `CultMesh` facade, RUDP client/server sugar, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout |
 
 ## Evidence Gates
@@ -63,7 +63,7 @@ CultMesh.
 - Extend RUDP interop evidence through daemon/socket reconnect loops and the
   wider schema-message matrix. RUDP profiles now advertise the portable
   reconnect policy, runtime-local tests prove the shared controller across the
-  targeted runtimes, and the C# LiteNetLib client plus TypeScript/Python/Kotlin
+  targeted runtimes, and the C# LiteNetLib client plus TypeScript/Rust/Python/Kotlin
   RUDP reconnect loops use that controller; broader service adoption remains a
   separate claim.
 - Keep Kotlin ergonomic parity moving with Android/JVM-shaped APIs instead of

packages/cultnet-rs/README.md

@@ -112,6 +112,31 @@ let mut client = CultMesh::create_rudp_client_for_peer(
 client.connect(b"join".to_vec())?;
 ```
 
+For caller-owned reconnect loops, `CultNetRudpReconnectLoop` keeps transport
+construction outside the library while sharing the portable retry controller:
+
+```rust
+use cultnet_rs::{
+    CultNetReconnectPolicyOptions, CultNetRudpReconnectLoop, create_reconnect_policy,
+};
+
+let policy = create_reconnect_policy(CultNetReconnectPolicyOptions::default());
+let mut reconnect = CultNetRudpReconnectLoop::new(policy, b"join".to_vec(), || {
+    CultMesh::create_rudp_client_for_peer(
+        "rust-client",
+        0x1020_3040,
+        &peer,
+        CultMeshRudpSocketOptions::default(),
+    )
+});
+
+reconnect.start()?;
+let decision = reconnect.handle_closed(now_ms, 0);
+if decision.is_some() {
+    reconnect.reconnect_if_due(now_ms)?;
+}
+```
+
 `CultMesh::parse_rudp_endpoint(...)` handles advertised `rudp://host:port`
 contact hints, and `CultMeshAuthorityLeaseCatalog` keeps trust separate from
 discovery. A peer card can say where to dial; a lease decides whether that peer

packages/cultnet-rs/src/rudp.rs

@@ -9,6 +9,8 @@ use std::net::UdpSocket;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
 
+use crate::CultNetReconnectController;
+use crate::CultNetReconnectDecision;
 use crate::CultNetReconnectPolicy;
 use crate::CultNetTransportChannel;
 use crate::CultNetTransportDelivery;
@@ -996,6 +998,98 @@ impl CultNetRudpSocketTransportConnection {
     }
 }
 
+pub struct CultNetRudpReconnectLoop<F>
+where
+    F: FnMut() -> Result<CultNetRudpSocketTransportConnection>,
+{
+    pub reconnect_controller: CultNetReconnectController,
+    create_transport: F,
+    connect_payload: Vec<u8>,
+    transport: Option<CultNetRudpSocketTransportConnection>,
+    stopped: bool,
+}
+
+impl<F> CultNetRudpReconnectLoop<F>
+where
+    F: FnMut() -> Result<CultNetRudpSocketTransportConnection>,
+{
+    pub fn new(
+        reconnect_policy: CultNetReconnectPolicy,
+        connect_payload: Vec<u8>,
+        create_transport: F,
+    ) -> Self {
+        Self {
+            reconnect_controller: CultNetReconnectController::new(reconnect_policy),
+            create_transport,
+            connect_payload,
+            transport: None,
+            stopped: true,
+        }
+    }
+
+    pub fn with_default_policy(connect_payload: Vec<u8>, create_transport: F) -> Self {
+        Self::new(
+            create_reconnect_policy(Default::default()),
+            connect_payload,
+            create_transport,
+        )
+    }
+
+    pub fn transport(&self) -> Option<&CultNetRudpSocketTransportConnection> {
+        self.transport.as_ref()
+    }
+
+    pub fn transport_mut(&mut self) -> Option<&mut CultNetRudpSocketTransportConnection> {
+        self.transport.as_mut()
+    }
+
+    pub fn start(&mut self) -> Result<&mut CultNetRudpSocketTransportConnection> {
+        self.stopped = false;
+        self.reconnect_controller.reset();
+        self.open_transport()
+    }
+
+    pub fn stop(&mut self) {
+        self.stopped = true;
+        self.transport = None;
+        self.reconnect_controller.reset();
+    }
+
+    pub fn mark_connected(&mut self) {
+        self.reconnect_controller.reset();
+    }
+
+    pub fn handle_closed(
+        &mut self,
+        now_ms: u64,
+        jitter_ms: u64,
+    ) -> Option<CultNetReconnectDecision> {
+        self.transport = None;
+        if self.stopped {
+            return None;
+        }
+        Some(self.reconnect_controller.record_failure(now_ms, jitter_ms))
+    }
+
+    pub fn reconnect_if_due(&mut self, now_ms: u64) -> Result<bool> {
+        if self.stopped || !self.reconnect_controller.can_attempt(now_ms) {
+            return Ok(false);
+        }
+        self.open_transport()?;
+        Ok(true)
+    }
+
+    fn open_transport(&mut self) -> Result<&mut CultNetRudpSocketTransportConnection> {
+        let mut transport = (self.create_transport)()?;
+        transport.connect(self.connect_payload.clone())?;
+        self.transport = Some(transport);
+        Ok(self
+            .transport
+            .as_mut()
+            .expect("RUDP reconnect loop opened a transport"))
+    }
+}
+
 #[derive(Clone, Debug, Default)]
 pub struct RudpTransportProfileOptions {
     pub transport_id: Option<String>,

packages/cultnet-rs/tests/cultnet.rs

@@ -20,6 +20,7 @@ use cultnet_rs::CultNetReconnectController;
 use cultnet_rs::CultNetReconnectPolicyOptions;
 use cultnet_rs::CultNetRudpPacket;
 use cultnet_rs::CultNetRudpPacketType;
+use cultnet_rs::CultNetRudpReconnectLoop;
 use cultnet_rs::CultNetRudpSendOptions;
 use cultnet_rs::CultNetRudpSession;
 use cultnet_rs::CultNetRudpSessionOptions;
@@ -53,7 +54,9 @@ use cultnet_rs::encode_cultnet_message_to_vec;
 use cultnet_rs::encode_frame;
 use cultnet_rs::encode_rudp_packet;
 use pretty_assertions::assert_eq;
+use std::cell::RefCell;
 use std::net::UdpSocket;
+use std::rc::Rc;
 use std::thread;
 use std::time::Duration as StdDuration;
 
@@ -395,6 +398,63 @@ fn reconnect_controller_schedules_attempts_and_reset() {
     assert!(controller.can_attempt(99_000));
 }
 
+#[test]
+fn rudp_reconnect_loop_consumes_shared_controller() -> Result<()> {
+    let server_socket = bind_udp_socket()?;
+    let remote_addr = server_socket.local_addr()?;
+    let opened_local_addrs = Rc::new(RefCell::new(Vec::new()));
+    let opened_for_factory = Rc::clone(&opened_local_addrs);
+    let connection_id = 0x2233_4455;
+    let policy = create_reconnect_policy(CultNetReconnectPolicyOptions {
+        max_attempts: Some(2),
+        ..CultNetReconnectPolicyOptions::default()
+    });
+
+    let mut loop_ = CultNetRudpReconnectLoop::new(policy, b"join".to_vec(), move || {
+        let socket = bind_udp_socket()?;
+        opened_for_factory.borrow_mut().push(socket.local_addr()?);
+        CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions::client(
+            "rust-rudp-reconnect",
+            socket,
+            remote_addr,
+            connection_id,
+        ))
+    });
+
+    {
+        let first = loop_.start()?;
+        assert!(first.stats().bytes_sent > 0);
+    }
+    assert!(loop_.transport().is_some());
+    assert_eq!(opened_local_addrs.borrow().len(), 1);
+
+    let decision = loop_.handle_closed(10_000, 17).expect("retry decision");
+    assert_eq!(decision.attempt, 1);
+    assert!(decision.should_retry);
+    assert_eq!(decision.delay_ms, 1_017);
+    assert_eq!(decision.next_attempt_at_ms, Some(11_017));
+    assert_eq!(loop_.reconnect_controller.attempt(), 1);
+    assert_eq!(
+        loop_.reconnect_controller.next_attempt_at_ms(),
+        Some(11_017)
+    );
+
+    assert!(!loop_.reconnect_if_due(11_016)?);
+    assert_eq!(opened_local_addrs.borrow().len(), 1);
+    assert!(loop_.reconnect_if_due(11_017)?);
+    assert_eq!(opened_local_addrs.borrow().len(), 2);
+    assert!(loop_.transport().is_some());
+
+    loop_.mark_connected();
+    assert_eq!(loop_.reconnect_controller.attempt(), 0);
+
+    loop_.stop();
+    assert!(loop_.transport().is_none());
+    assert_eq!(loop_.reconnect_controller.attempt(), 0);
+
+    Ok(())
+}
+
 #[test]
 fn rudp_session_handshake_acks_reliable_connect_and_accept_packets() -> Result<()> {
     let mut client = CultNetRudpSession::new(CultNetRudpSessionOptions {