Expose exact RUDP acknowledgement receipts
Ported from GameCult/cultnet-rs 9dcb88d into CultLib, where the wire contract belongs. Ghostlight's committed idunn_health.rs imports CultNetRudpReliableSendStatus to hold an Idunn process write lease current, and the symbol existed only in the standalone lineage, so nothing could consume this contract from its canonical home. Three reconciliations were needed because both lineages reworked RUDP independently since the 014f998 copy-in: - reset_peer_state: kept this lineage's version, which also resets the sequence and fragment counters and clears queued_reliable, and folded in only the imported change that matters -- rotating session_scope so a peer reset invalidates outstanding receipts. The imported body also cleared highest_delivered_sequenced_by_channel, a field the sequence-domain work renamed away here. - channel_send_options is a free function in this lineage, not a method. - The imported guard rejected reliable channels with an expiry. This lineage has no reliable-expiry concept at all -- the only mentions of expiry in the file arrived with this patch -- so every reliable channel is already non-expiring and the guard reduces to the reliability check. A comment marks where to restore the expiry half if expiry is ever added. 94 tests pass. The one rudp_server_hub failure is pre-existing and reproduces on clean origin/main. The other two standalone RUDP commits are deliberately not ported. Bounding session memory and rejecting hostile sequence rollover 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. Merging them changes wire-observable drop and reassembly behavior, which needs the cross-runtime interop harness to validate, not a local build. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
67cbb4ea36
commit
9f822b7578
1 changed files with 107 additions and 2 deletions
|
|
@ -6,7 +6,9 @@ use std::collections::VecDeque;
|
|||
use std::io::ErrorKind;
|
||||
use std::net::SocketAddr;
|
||||
use std::net::UdpSocket;
|
||||
use std::thread;
|
||||
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::CultNetMessage;
|
||||
use crate::CultNetReconnectController;
|
||||
|
|
@ -105,6 +107,28 @@ pub struct CultNetRudpSendOptions {
|
|||
pub now_ms: u64,
|
||||
}
|
||||
|
||||
/// Identifies every transport packet belonging to one non-expiring reliable
|
||||
/// send. Receipts are local to the session that issued them and cannot survive
|
||||
/// a peer reset.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct CultNetRudpReliableSendReceipt {
|
||||
session_scope: Uuid,
|
||||
sequences: Vec<u32>,
|
||||
}
|
||||
|
||||
impl CultNetRudpReliableSendReceipt {
|
||||
pub fn packet_sequences(&self) -> &[u32] {
|
||||
&self.sequences
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum CultNetRudpReliableSendStatus {
|
||||
Pending,
|
||||
Acknowledged,
|
||||
Invalidated,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
struct PendingReliablePacket {
|
||||
packet: CultNetRudpPacket,
|
||||
|
|
@ -127,6 +151,7 @@ struct FragmentBuffer {
|
|||
}
|
||||
|
||||
pub struct CultNetRudpSession {
|
||||
session_scope: Uuid,
|
||||
connection_id: u32,
|
||||
resend_delay_ms: u64,
|
||||
max_pending_reliable_packets: Option<usize>,
|
||||
|
|
@ -149,6 +174,7 @@ pub struct CultNetRudpSession {
|
|||
impl CultNetRudpSession {
|
||||
pub fn new(options: CultNetRudpSessionOptions) -> Self {
|
||||
Self {
|
||||
session_scope: Uuid::new_v4(),
|
||||
connection_id: options.connection_id,
|
||||
resend_delay_ms: options.resend_delay_ms,
|
||||
max_pending_reliable_packets: options.max_pending_reliable_packets,
|
||||
|
|
@ -181,6 +207,11 @@ impl CultNetRudpSession {
|
|||
self.connected
|
||||
}
|
||||
|
||||
pub fn assume_connected(&mut self, now_ms: u64) {
|
||||
self.connected = true;
|
||||
self.last_received_at_ms = Some(now_ms);
|
||||
}
|
||||
|
||||
pub fn pending_reliable_sequences(&self) -> Vec<u32> {
|
||||
self.pending_reliable.keys().copied().collect()
|
||||
}
|
||||
|
|
@ -202,6 +233,39 @@ impl CultNetRudpSession {
|
|||
Some(pending.packet.clone())
|
||||
}
|
||||
|
||||
fn reliable_send_status(
|
||||
&self,
|
||||
receipt: &CultNetRudpReliableSendReceipt,
|
||||
) -> CultNetRudpReliableSendStatus {
|
||||
if receipt.session_scope != self.session_scope {
|
||||
return CultNetRudpReliableSendStatus::Invalidated;
|
||||
}
|
||||
if receipt
|
||||
.sequences
|
||||
.iter()
|
||||
.any(|sequence| self.pending_reliable.contains_key(sequence))
|
||||
{
|
||||
CultNetRudpReliableSendStatus::Pending
|
||||
} else {
|
||||
CultNetRudpReliableSendStatus::Acknowledged
|
||||
}
|
||||
}
|
||||
|
||||
fn non_expiring_reliable_receipt(
|
||||
&self,
|
||||
packets: &[CultNetRudpPacket],
|
||||
) -> Result<CultNetRudpReliableSendReceipt> {
|
||||
if packets.is_empty() || packets.iter().any(|packet| !packet.reliable) {
|
||||
return Err(anyhow!(
|
||||
"RUDP acknowledgement receipts require a non-empty reliable packet set"
|
||||
));
|
||||
}
|
||||
Ok(CultNetRudpReliableSendReceipt {
|
||||
session_scope: self.session_scope,
|
||||
sequences: packets.iter().map(|packet| packet.sequence).collect(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn last_received_at_ms(&self) -> Option<u64> {
|
||||
self.last_received_at_ms
|
||||
}
|
||||
|
|
@ -224,6 +288,7 @@ impl CultNetRudpSession {
|
|||
}
|
||||
|
||||
pub fn reset_peer_state(&mut self) {
|
||||
self.session_scope = Uuid::new_v4();
|
||||
self.next_sequence = self.initial_sequence;
|
||||
self.next_fragment_id = 1;
|
||||
self.connected = false;
|
||||
|
|
@ -1508,13 +1573,53 @@ impl CultNetRudpSocketTransportConnection {
|
|||
let packets =
|
||||
self.session
|
||||
.send_many(channel_id, payload, options, self.max_fragment_bytes)?;
|
||||
for packet in packets {
|
||||
self.send_packet(&packet)?;
|
||||
}
|
||||
self.send_data_packets(&packets)?;
|
||||
self.stats.frames_sent += 1;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Sends one frame on a reliable channel whose packets cannot expire and
|
||||
/// returns the transport receipt used to observe its exact acknowledgement.
|
||||
pub fn send_reliable(
|
||||
&mut self,
|
||||
channel_id: &str,
|
||||
payload: Vec<u8>,
|
||||
) -> Result<CultNetRudpReliableSendReceipt> {
|
||||
let options = channel_send_options(channel_id, now_ms());
|
||||
// This lineage has no reliable-expiry concept, so every reliable channel is
|
||||
// already non-expiring. Restore the expiry half of this guard if one is added.
|
||||
if !options.reliable {
|
||||
return Err(anyhow!(
|
||||
"RUDP acknowledgement receipts require a non-expiring reliable channel"
|
||||
));
|
||||
}
|
||||
let packets =
|
||||
self.session
|
||||
.send_many(channel_id, payload, options, self.max_fragment_bytes)?;
|
||||
self.send_data_packets(&packets)?;
|
||||
let receipt = self.session.non_expiring_reliable_receipt(&packets)?;
|
||||
self.stats.frames_sent += 1;
|
||||
Ok(receipt)
|
||||
}
|
||||
|
||||
pub fn reliable_send_status(
|
||||
&self,
|
||||
receipt: &CultNetRudpReliableSendReceipt,
|
||||
) -> CultNetRudpReliableSendStatus {
|
||||
self.session.reliable_send_status(receipt)
|
||||
}
|
||||
|
||||
fn send_data_packets(&mut self, packets: &[CultNetRudpPacket]) -> Result<()> {
|
||||
let fragmented = packets.len() > 1;
|
||||
for (index, packet) in packets.iter().enumerate() {
|
||||
self.send_packet(packet)?;
|
||||
if fragmented && (index + 1) % 16 == 0 {
|
||||
thread::sleep(Duration::from_millis(1));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn send_schema_message(&mut self, message: &CultNetMessage) -> Result<()> {
|
||||
let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
|
||||
self.send("schema", payload)
|
||||
|
|
|
|||
Loading…
Reference in a new issue