Bound the RUDP receive path and refuse hostile sequences

Clears the cultnet half of the retained debt. The two unported commits from the
retiring GameCult/cultnet-rs lineage were not portable as patches: this lineage
had reworked the same receive path with sequence domains and ACK horizons, and
applying them raised nine conflicts that would have reverted that work. Most of
what they added already existed here by another route -- received-sequence
history, fragment-set count and reassembly bytes are all bounded already. So
the gaps were taken rather than the patches.

Three were real. The per-channel maps grew without limit, and a peer picks its
own channel ids, so ordered and sequenced channel tracking is now capped at 64
each. The out-of-order hold buffer grew without limit, and a peer picks its own
send order, so it is capped at 1024 frames and 4 MB; sizes are computed from the
buffer rather than tracked in counters, which cannot drift.

The fourth was the sharpest. remember_received ran before the duplicate
early-return, so a single forged packet far past the current highest advanced
the receive state beyond every sequence still in flight and starved the
session. Such a packet is now refused without being remembered.

Reliable sequence exhaustion aborted the process through expect(). The capacity
gate that already guards the send window now also refuses when the sequence
space cannot cover the request, and allocation saturates rather than panicking
so no future path can reintroduce the abort.

Five tests cover the bounds and both refusals. cultnet-rs passes 103,
cultcache-rs 48, cultmesh-rs 7, and the TypeScript-Rust RUDP interop lanes pass
5 of 5 with the hardened peer, so legitimate traffic is unchanged.

The parity ledger now records what this actually leaves owed: these are
refusals, not wire-format changes, so the interop lanes stay green whether or
not the other four runtimes match. The gap will not announce itself.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

docs/runtime-parity-scope.md

@@ -89,14 +89,12 @@ server body, is deliberately not made.
   outside any conformance harness, and its stream, negotiation, and shared-memory frame-ring
   surfaces have never been checked against the browser, Kotlin, and TypeScript CultMesh bodies
   they must interoperate with. Runtime-local tests pass; that is not a parity claim.
-- Finish porting RUDP hardening from the retiring GameCult/cultnet-rs lineage. Its exact
-  acknowledgement receipts are in. Two commits are not: bounded session memory (4a54f46) and
-  hostile sequence rollover rejection (91609ff). They rebuild the receive path -- stale-sequence
-  rejection, per-channel tracking caps, reassembly and ordered-buffer byte accounting -- in the
-  same code this lineage reworked with sequence domains and ACK horizons, and they change when
-  packets are dropped and reassembly refused. That is wire-observable, so the interop lanes are
-  the gate, not a local build. Until they land, this runtime accepts sequence and reassembly
-  input the other lineage rejects. Do not delete that repository before this is resolved.
+- Carry the RUDP receive-path bounds to the other runtimes. Rust now caps tracked ordered
+  and sequenced channels, caps the out-of-order hold buffer by frames and bytes, refuses a
+  packet far ahead of the highest received so a forged sequence cannot starve a session, and
+  reports reliable sequence exhaustion instead of aborting. C#, TypeScript, Python and Kotlin
+  still accept all of it. These are refusals rather than wire-format changes, so the interop
+  lanes stay green either way -- which is exactly why the gap will not announce itself.
 - Keep performance claims evidence-bound: C# owns SoA; Python has benchmark
   comparison gates; other runtimes should advertise measured native hot paths
   only when those paths exist.

packages/cultnet-rs/src/rudp.rs

@@ -33,6 +33,15 @@ const RUDP_FIXED_HEADER_BYTES: usize = 36;
 pub const CULTNET_RUDP_DEFAULT_MAX_PAYLOAD_BYTES: usize = 16 * 1024 * 1024;
 pub const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS: usize = 32;
 const RUDP_RECEIVED_SEQUENCE_WINDOW: usize = 4_096;
+/// A peer chooses its own channel ids and its own send order, so the per-channel
+/// maps and the out-of-order hold buffer are both attacker-sized unless capped.
+const RUDP_MAX_TRACKED_CHANNELS: usize = 64;
+/// A peer cannot be trusted to pick the next sequence honestly. One packet far
+/// ahead of the current highest would otherwise advance the receive state past
+/// every legitimate sequence still in flight and starve the session.
+const RUDP_RECEIVE_AHEAD_WINDOW: u32 = 1_024;
+const RUDP_MAX_ORDERED_BUFFERED_FRAMES: usize = 1_024;
+const RUDP_MAX_ORDERED_BUFFERED_BYTES: usize = 4 * 1024 * 1024;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum CultNetRudpPacketType {
@@ -529,6 +538,25 @@ impl CultNetRudpSession {
             });
         }
 
+        let far_ahead = packet.reliable
+            && self.highest_received_sequence.is_some_and(|highest| {
+                packet.sequence > highest
+                    && packet.sequence - highest > RUDP_RECEIVE_AHEAD_WINDOW
+            });
+        if far_ahead {
+            // Refuse without remembering. Recording it would move the highest
+            // received sequence out of reach of everything still in flight.
+            return Ok(CultNetRudpReceiveResult {
+                delivered: Vec::new(),
+                ready_to_send,
+                reply: None,
+                pong: false,
+                pong_payload: Vec::new(),
+                disconnected: false,
+                disconnect_reason: Vec::new(),
+            });
+        }
+
         let duplicate = packet.reliable
             && (self.received_sequences.contains(&packet.sequence)
                 || self.highest_received_sequence.is_some_and(|highest| {
@@ -562,8 +590,15 @@ impl CultNetRudpSession {
             });
         };
         let delivered = if ordered {
-            self.deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized)
+            self.deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized)?
         } else if packet.sequenced {
+            if !self
+                .latest_sequenced_by_channel
+                .contains_key(&frame.channel_id)
+                && self.latest_sequenced_by_channel.len() >= RUDP_MAX_TRACKED_CHANNELS
+            {
+                return Err(anyhow!("RUDP session has too many sequenced channels"));
+            }
             let latest = self
                 .latest_sequenced_by_channel
                 .get(&frame.channel_id)
@@ -724,8 +759,7 @@ impl CultNetRudpSession {
             let sequence = self.next_sequence;
             self.next_sequence = self
                 .next_sequence
-                .checked_add(1)
-                .expect("reliable sequence overflow");
+                .saturating_add(1);
             sequence
         } else if sequenced {
             let next = self
@@ -733,7 +767,7 @@ impl CultNetRudpSession {
                 .entry(channel_id.to_string())
                 .or_insert(1);
             let sequence = *next;
-            *next = next.checked_add(1).expect("sequenced channel overflow");
+            *next = next.saturating_add(1);
             sequence
         } else {
             0
@@ -808,6 +842,14 @@ impl CultNetRudpSession {
                 return Err(anyhow!("RUDP reliable send queue is full"));
             }
         }
+        // Refuse before allocation rather than at it. A session that runs out of
+        // reliable sequence space must report it, not abort the process.
+        let required = u32::try_from(packet_count)
+            .ok()
+            .filter(|required| self.next_sequence.checked_add(*required).is_some());
+        if required.is_none() {
+            return Err(anyhow!("RUDP reliable sequence space is exhausted"));
+        }
         Ok(())
     }
 
@@ -958,13 +1000,30 @@ impl CultNetRudpSession {
         Ok(())
     }
 
+    fn ordered_buffered_frames(&self) -> usize {
+        self.ordered_buffers.values().map(BTreeMap::len).sum()
+    }
+
+    fn ordered_buffered_bytes(&self) -> usize {
+        self.ordered_buffers
+            .values()
+            .flat_map(BTreeMap::values)
+            .map(|pending| pending.frame.payload.len())
+            .sum()
+    }
+
     fn deliver_ordered(
         &mut self,
         frame: CultNetRudpDeliveredFrame,
         next_sequence_after_frame: u32,
         expected_sequence_if_uninitialized: u32,
-    ) -> Vec<CultNetRudpDeliveredFrame> {
+    ) -> Result<Vec<CultNetRudpDeliveredFrame>> {
         let channel_id = frame.channel_id.clone();
+        if !self.ordered_next_sequence_by_channel.contains_key(&channel_id)
+            && self.ordered_next_sequence_by_channel.len() >= RUDP_MAX_TRACKED_CHANNELS
+        {
+            return Err(anyhow!("RUDP session has too many ordered channels"));
+        }
         let mut next = if let Some(next) = self
             .ordered_next_sequence_by_channel
             .get(&channel_id)
@@ -992,10 +1051,16 @@ impl CultNetRudpSession {
         }
 
         if frame.sequence < next {
-            return Vec::new();
+            return Ok(Vec::new());
         }
 
         if frame.sequence > next {
+            if self.ordered_buffered_frames() >= RUDP_MAX_ORDERED_BUFFERED_FRAMES
+                || self.ordered_buffered_bytes() + frame.payload.len()
+                    > RUDP_MAX_ORDERED_BUFFERED_BYTES
+            {
+                return Err(anyhow!("RUDP ordered hold buffer is full"));
+            }
             self.ordered_buffers.entry(channel_id).or_default().insert(
                 frame.sequence,
                 PendingOrderedFrame {
@@ -1003,14 +1068,14 @@ impl CultNetRudpSession {
                     next_sequence: next_sequence_after_frame,
                 },
             );
-            return Vec::new();
+            return Ok(Vec::new());
         }
 
         self.ordered_next_sequence_by_channel
             .insert(channel_id.clone(), next_sequence_after_frame);
         let mut delivered = vec![frame];
         delivered.extend(self.drain_ordered(&channel_id));
-        delivered
+        Ok(delivered)
     }
 
     fn drain_ordered(&mut self, channel_id: &str) -> Vec<CultNetRudpDeliveredFrame> {

packages/cultnet-rs/tests/cultnet.rs

@@ -2417,3 +2417,222 @@ fn schema_discovery_round_trips_over_legacy_gamecult_contract_when_inline_schema
     );
     Ok(())
 }
+
+#[test]
+fn rudp_sequenced_channel_tracking_is_bounded() -> Result<()> {
+    let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 401,
+        initial_sequence: 1,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 401,
+        initial_sequence: 100,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let connect = sender.create_connect(0, Vec::new())?;
+    let accept = receiver.accept_connect(&connect, 1, Vec::new())?;
+    sender.receive(&accept, 2)?;
+
+    // A peer picks its own channel ids, so the sequenced map is attacker-sized.
+    let mut error = None;
+    for channel in 0..256u32 {
+        let packet = sender.send(
+            &format!("channel-{channel}"),
+            b"state".to_vec(),
+            CultNetRudpSendOptions {
+                sequenced: true,
+                ..CultNetRudpSendOptions::default()
+            },
+        )?;
+        if let Err(refused) = receiver.receive(&packet, 3) {
+            error = Some(refused);
+            break;
+        }
+    }
+    let error = error.expect("unbounded sequenced channel tracking");
+    assert!(
+        error.to_string().contains("too many sequenced channels"),
+        "unexpected refusal: {error}"
+    );
+    Ok(())
+}
+
+#[test]
+fn rudp_ordered_channel_tracking_is_bounded() -> Result<()> {
+    let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 402,
+        initial_sequence: 1,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 402,
+        initial_sequence: 100,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let connect = sender.create_connect(0, Vec::new())?;
+    let accept = receiver.accept_connect(&connect, 1, Vec::new())?;
+    sender.receive(&accept, 2)?;
+
+    let mut error = None;
+    for channel in 0..256u32 {
+        let packet = sender.send(
+            &format!("ordered-{channel}"),
+            b"state".to_vec(),
+            CultNetRudpSendOptions {
+                reliable: true,
+                ordered: true,
+                ..CultNetRudpSendOptions::default()
+            },
+        )?;
+        match receiver.receive(&packet, 3) {
+            Ok(_) => {
+                sender.receive(&receiver.create_ack_for(packet.sequence), 3)?;
+            }
+            Err(refused) => {
+                error = Some(refused);
+                break;
+            }
+        }
+    }
+    let error = error.expect("unbounded ordered channel tracking");
+    assert!(
+        error.to_string().contains("too many ordered channels"),
+        "unexpected refusal: {error}"
+    );
+    Ok(())
+}
+
+#[test]
+fn rudp_ordered_hold_buffer_is_bounded() -> Result<()> {
+    let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 403,
+        initial_sequence: 1,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 403,
+        initial_sequence: 100,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let connect = sender.create_connect(0, Vec::new())?;
+    let accept = receiver.accept_connect(&connect, 1, Vec::new())?;
+    sender.receive(&accept, 2)?;
+
+    let options = CultNetRudpSendOptions {
+        reliable: true,
+        ordered: true,
+        ..CultNetRudpSendOptions::default()
+    };
+    // Withhold the first frame so every later one has to be held.
+    let withheld = sender.send("schema", b"first".to_vec(), options.clone())?;
+    sender.receive(&receiver.create_ack_for(withheld.sequence), 3)?;
+
+    let mut error = None;
+    for step in 0..4_096u32 {
+        let packet = sender.send("schema", vec![(step % 251) as u8], options.clone())?;
+        match receiver.receive(&packet, 4) {
+            Ok(result) => {
+                assert!(result.delivered.is_empty(), "gap should hold every frame");
+                sender.receive(&receiver.create_ack_for(packet.sequence), 4)?;
+            }
+            Err(refused) => {
+                error = Some(refused);
+                break;
+            }
+        }
+    }
+    let error = error.expect("unbounded ordered hold buffer");
+    assert!(
+        error.to_string().contains("ordered hold buffer is full"),
+        "unexpected refusal: {error}"
+    );
+    Ok(())
+}
+
+#[test]
+fn rudp_far_ahead_sequence_cannot_starve_the_session() -> Result<()> {
+    let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 404,
+        initial_sequence: 1,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 404,
+        initial_sequence: 100,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let connect = sender.create_connect(0, Vec::new())?;
+    let accept = receiver.accept_connect(&connect, 1, Vec::new())?;
+    sender.receive(&accept, 2)?;
+
+    let options = CultNetRudpSendOptions {
+        reliable: true,
+        ..CultNetRudpSendOptions::default()
+    };
+    let first = sender.send("schema", b"before".to_vec(), options.clone())?;
+    assert_eq!(receiver.receive(&first, 3)?.delivered.len(), 1);
+    sender.receive(&receiver.create_ack_for(first.sequence), 3)?;
+
+    // A forged packet far past the current highest. Recording it would push the
+    // receive state beyond every sequence still in flight.
+    let forged = CultNetRudpPacket {
+        packet_type: CultNetRudpPacketType::Data,
+        connection_id: 404,
+        sequence: first.sequence.saturating_add(500_000),
+        ack: 0,
+        ack_mask: 0,
+        channel_id: "schema".to_string(),
+        reliable: true,
+        ordered: false,
+        sequenced: false,
+        fragment_id: 0,
+        fragment_index: 0,
+        fragment_count: 0,
+        payload: b"hostile".to_vec(),
+    };
+    assert!(receiver.receive(&forged, 4)?.delivered.is_empty());
+
+    // The session must still accept the next legitimate packet.
+    let second = sender.send("schema", b"after".to_vec(), options)?;
+    assert_eq!(
+        receiver.receive(&second, 5)?.delivered[0].payload,
+        b"after".to_vec()
+    );
+    Ok(())
+}
+
+#[test]
+fn rudp_reliable_sequence_exhaustion_is_reported_not_fatal() -> Result<()> {
+    let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 405,
+        // One sequence left: the handshake takes it, so the next send is the
+        // first allocation with nowhere to go.
+        initial_sequence: u32::MAX - 1,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
+        connection_id: 405,
+        initial_sequence: 100,
+        ..CultNetRudpSessionOptions::default()
+    });
+    let connect = sender.create_connect(0, Vec::new())?;
+    let accept = receiver.accept_connect(&connect, 1, Vec::new())?;
+    sender.receive(&accept, 2)?;
+
+    let error = sender
+        .send(
+            "schema",
+            b"one past the end".to_vec(),
+            CultNetRudpSendOptions {
+                reliable: true,
+                ..CultNetRudpSendOptions::default()
+            },
+        )
+        .unwrap_err();
+    assert!(
+        error.to_string().contains("sequence space is exhausted"),
+        "unexpected error: {error}"
+    );
+    Ok(())
+}