From 3fde29cc0409f9c63cb15d271eded57e4f05379e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 06:43:22 +0100 Subject: [PATCH] Add Rust RUDP reconnect loop --- docs/cultnet-transport-parity.md | 4 ++ docs/runtime-parity-scope.md | 4 +- packages/cultnet-rs/README.md | 25 ++++++++ packages/cultnet-rs/src/rudp.rs | 94 ++++++++++++++++++++++++++++ packages/cultnet-rs/tests/cultnet.rs | 60 ++++++++++++++++++ 5 files changed, 185 insertions(+), 2 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index a52961de..6fba4f35 100644 --- a/docs/cultnet-transport-parity.md +++ b/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 diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index 49df7c81..214bd649 100644 --- a/docs/runtime-parity-scope.md +++ b/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 diff --git a/packages/cultnet-rs/README.md b/packages/cultnet-rs/README.md index da328fce..2c8db475 100644 --- a/packages/cultnet-rs/README.md +++ b/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 diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index cb5fbef3..744858ff 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/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 +where + F: FnMut() -> Result, +{ + pub reconnect_controller: CultNetReconnectController, + create_transport: F, + connect_payload: Vec, + transport: Option, + stopped: bool, +} + +impl CultNetRudpReconnectLoop +where + F: FnMut() -> Result, +{ + pub fn new( + reconnect_policy: CultNetReconnectPolicy, + connect_payload: Vec, + 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, 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 { + 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 { + 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, diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 6a48548d..9098abb2 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/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 {