Skip to main content

rust_raknet/
arq.rs

1use crate::sequence::{self, ReliableWindow};
2
3use std::{
4    collections::{HashMap, VecDeque},
5    net::SocketAddr,
6    sync::{
7        Arc,
8        atomic::{AtomicUsize, Ordering},
9    },
10};
11
12use crate::{datatype::*, error::*, fragment::FragmentQ, raknet_log_debug, utils::*};
13
14pub(crate) struct OutgoingFrames {
15    frames: Vec<FrameSetPacket>,
16    pub(crate) coalesced: bool,
17}
18
19impl OutgoingFrames {
20    #[cfg(test)]
21    pub(crate) fn new() -> Self {
22        Self::with_capacity(0)
23    }
24
25    fn with_capacity(capacity: usize) -> Self {
26        Self {
27            frames: Vec::with_capacity(capacity),
28            coalesced: false,
29        }
30    }
31}
32
33impl std::ops::Deref for OutgoingFrames {
34    type Target = Vec<FrameSetPacket>;
35    fn deref(&self) -> &Self::Target {
36        &self.frames
37    }
38}
39
40impl std::ops::DerefMut for OutgoingFrames {
41    fn deref_mut(&mut self) -> &mut Self::Target {
42        &mut self.frames
43    }
44}
45
46impl IntoIterator for OutgoingFrames {
47    type Item = FrameSetPacket;
48    type IntoIter = std::vec::IntoIter<FrameSetPacket>;
49    fn into_iter(self) -> Self::IntoIter {
50        self.frames.into_iter()
51    }
52}
53
54impl<'a> IntoIterator for &'a OutgoingFrames {
55    type Item = &'a FrameSetPacket;
56    type IntoIter = std::slice::Iter<'a, FrameSetPacket>;
57    fn into_iter(self) -> Self::IntoIter {
58        self.frames.iter()
59    }
60}
61
62/// Delivery and ordering guarantees for a RakNet frame.
63#[derive(Clone, Copy, Debug, PartialEq, Eq)]
64#[repr(u8)]
65pub enum Reliability {
66    /// The frame may be lost or arrive out of order.
67    Unreliable = 0x00,
68    /// Only the newest frame in the ordering stream is delivered.
69    UnreliableSequenced = 0x01,
70    /// The frame is retransmitted until acknowledged, without ordering guarantees.
71    Reliable = 0x02,
72    /// The frame is retransmitted and delivered in order within its channel.
73    ReliableOrdered = 0x03,
74    /// The frame is reliable, but older frames in its ordering stream may be skipped.
75    ReliableSequenced = 0x04,
76}
77
78impl Reliability {
79    pub fn to_u8(&self) -> u8 {
80        *self as u8
81    }
82
83    pub fn from(flags: u8) -> Result<Self> {
84        Self::try_from(flags)
85    }
86}
87
88impl TryFrom<u8> for Reliability {
89    type Error = RaknetError;
90
91    fn try_from(flags: u8) -> Result<Self> {
92        match flags {
93            0x00 => Ok(Self::Unreliable),
94            0x01 => Ok(Self::UnreliableSequenced),
95            0x02 => Ok(Self::Reliable),
96            0x03 => Ok(Self::ReliableOrdered),
97            0x04 => Ok(Self::ReliableSequenced),
98            _ => Err(RaknetError::IncorrectReliability),
99        }
100    }
101}
102
103const NEEDS_B_AND_AS_FLAG: u8 = 0x4;
104const CONTINUOUS_SEND_FLAG: u8 = 0x8;
105
106// Received payloads move into the application without another allocation.
107// Reliable sends share their backing allocation with retransmission frames.
108#[derive(Clone, Debug)]
109pub(crate) enum FramePayload {
110    Owned(Vec<u8>),
111    Bytes(bytes::Bytes),
112    Shared {
113        owner: Arc<[u8]>,
114        start: u32,
115        length: u16,
116    },
117}
118
119impl Default for FramePayload {
120    fn default() -> Self {
121        Self::Owned(Vec::new())
122    }
123}
124
125impl FramePayload {
126    fn copy_shared(data: &[u8]) -> Self {
127        Self::Shared {
128            owner: Arc::from(data),
129            start: 0,
130            length: data.len() as u16,
131        }
132    }
133}
134
135impl std::ops::Deref for FramePayload {
136    type Target = [u8];
137    fn deref(&self) -> &[u8] {
138        match self {
139            Self::Owned(data) => data,
140            Self::Bytes(data) => data,
141            Self::Shared {
142                owner,
143                start,
144                length,
145            } => &owner[*start as usize..*start as usize + usize::from(*length)],
146        }
147    }
148}
149
150impl AsRef<[u8]> for FramePayload {
151    fn as_ref(&self) -> &[u8] {
152        self
153    }
154}
155
156impl From<Vec<u8>> for FramePayload {
157    fn from(data: Vec<u8>) -> Self {
158        Self::Owned(data)
159    }
160}
161
162impl From<FramePayload> for Vec<u8> {
163    fn from(data: FramePayload) -> Self {
164        match data {
165            FramePayload::Owned(data) => data,
166            shared => shared.to_vec(),
167        }
168    }
169}
170
171impl PartialEq for FramePayload {
172    fn eq(&self, other: &Self) -> bool {
173        self.as_ref() == other.as_ref()
174    }
175}
176
177// A fragmented send retains its complete allocation until the final fragment
178// is acknowledged. Charge that allocation once, independently of frame clones.
179struct SharedPayload {
180    remaining: AtomicUsize,
181    bytes: usize,
182}
183
184#[derive(Clone)]
185pub struct FrameSetPacket {
186    pub sequence_number: u32,
187    pub flags: u8,
188    pub length_in_bytes: u16,
189    pub reliable_frame_index: u32,
190    pub sequenced_frame_index: u32,
191    pub ordered_frame_index: u32,
192    pub order_channel: u8,
193    pub compound_size: u32,
194    pub compound_id: u16,
195    pub fragment_index: u32,
196    pub(crate) data: FramePayload,
197    shared_payload: Option<Arc<SharedPayload>>,
198}
199
200impl FrameSetPacket {
201    pub fn new(r: Reliability, data: Vec<u8>) -> Self {
202        Self::from_payload(r, data.into())
203    }
204
205    fn from_payload(r: Reliability, data: FramePayload) -> Self {
206        let flag = r.to_u8() << 5;
207
208        Self {
209            sequence_number: 0,
210            flags: flag,
211            length_in_bytes: data.len() as u16,
212            reliable_frame_index: 0,
213            sequenced_frame_index: 0,
214            ordered_frame_index: 0,
215            order_channel: 0,
216            compound_size: 0,
217            compound_id: 0,
218            fragment_index: 0,
219            data,
220            shared_payload: None,
221        }
222    }
223
224    #[cfg(test)]
225    pub fn deserialize(buf: &[u8]) -> Result<(Self, bool)> {
226        let mut reader = RaknetReader::new(buf);
227        let input_len = buf.len() as u64;
228
229        let mut ret = Self {
230            sequence_number: 0,
231            flags: 0,
232            length_in_bytes: 0,
233            reliable_frame_index: 0,
234            sequenced_frame_index: 0,
235            ordered_frame_index: 0,
236            order_channel: 0,
237            compound_size: 0,
238            compound_id: 0,
239            fragment_index: 0,
240            data: FramePayload::default(),
241            shared_payload: None,
242        };
243
244        let packet_id = reader.read_u8()?;
245        if !(0x80..=0x8d).contains(&packet_id) {
246            return Err(RaknetError::PacketHeaderError);
247        }
248        ret.sequence_number = reader.read_u24(Endian::Little)?;
249        ret.flags = reader.read_u8()?;
250
251        let length_in_bits = reader.read_u16(Endian::Big)?;
252        if length_in_bits % 8 != 0 {
253            return Err(RaknetError::PacketParseError);
254        }
255        ret.length_in_bytes = length_in_bits / 8;
256
257        if ret.is_reliable()? {
258            ret.reliable_frame_index = reader.read_u24(Endian::Little)?;
259        }
260
261        if ret.is_sequenced()? {
262            ret.sequenced_frame_index = reader.read_u24(Endian::Little)?;
263        }
264        if ret.is_ordered()? {
265            ret.ordered_frame_index = reader.read_u24(Endian::Little)?;
266            ret.order_channel = reader.read_u8()?;
267        }
268
269        if ret.is_fragment() {
270            ret.compound_size = reader.read_u32(Endian::Big)?;
271            ret.compound_id = reader.read_u16(Endian::Big)?;
272            ret.fragment_index = reader.read_u32(Endian::Big)?;
273        }
274
275        let mut data = vec![0; usize::from(ret.length_in_bytes)];
276        reader.read(&mut data)?;
277        ret.data = data.into();
278
279        if ret.is_fragment() && (ret.compound_size == 0 || ret.fragment_index >= ret.compound_size)
280        {
281            return Err(RaknetError::PacketParseError);
282        }
283
284        Ok((ret, reader.pos() == input_len))
285    }
286
287    pub fn serialize(&self) -> Result<Vec<u8>> {
288        let mut buffer = Vec::new();
289        self.serialize_into(&mut buffer)?;
290        Ok(buffer)
291    }
292
293    pub(crate) fn serialize_into(&self, buffer: &mut Vec<u8>) -> Result<()> {
294        self.validate()?;
295        let size = self._size()?;
296        buffer.clear();
297        buffer.reserve(size);
298        let mut writer = RaknetWriter::from_buffer(std::mem::take(buffer));
299        let result = self
300            .write_header(&mut writer)
301            .and_then(|()| writer.write(self.data.as_ref()));
302        *buffer = writer.get_raw_payload();
303        result
304    }
305
306    pub(crate) fn serialize_group_into(frames: &[Self], output: &mut Vec<u8>) -> Result<()> {
307        let first = frames.first().ok_or(RaknetError::PacketParseError)?;
308        output.clear();
309        for (index, frame) in frames.iter().enumerate() {
310            if frame.sequence_number != first.sequence_number {
311                return Err(RaknetError::PacketParseError);
312            }
313            let mut header = [0; 32];
314            let length = frame.encode_header(&mut header)?;
315            output.extend_from_slice(&header[if index == 0 { 0 } else { 4 }..length]);
316            output.extend_from_slice(frame.data.as_ref());
317        }
318        Ok(())
319    }
320
321    pub(crate) fn validate(&self) -> Result<()> {
322        if self.data.len() != usize::from(self.length_in_bytes)
323            || self.data.len() > usize::from(u16::MAX / 8)
324            || (self.is_fragment()
325                && (self.compound_size == 0 || self.fragment_index >= self.compound_size))
326        {
327            return Err(RaknetError::PacketParseError);
328        }
329
330        self._size()?;
331        Ok(())
332    }
333
334    pub(crate) fn encode_header(&self, header: &mut [u8; 32]) -> Result<usize> {
335        self.validate()?;
336        let mut writer = RaknetWriter::from_buffer(header.as_mut_slice());
337        self.write_header(&mut writer)?;
338        Ok(32 - writer.remaining_mut())
339    }
340
341    fn write_header<B: bytes::BufMut>(&self, writer: &mut RaknetWriter<B>) -> Result<()> {
342        let mut id = 0x80 | NEEDS_B_AND_AS_FLAG;
343
344        // Set the continuation bit for all fragments after the first.
345        if (self.flags & 16) != 0 && self.fragment_index != 0 {
346            id |= CONTINUOUS_SEND_FLAG;
347        }
348
349        writer.write_u8(id)?;
350        writer.write_u24(self.sequence_number, Endian::Little)?;
351
352        // The top three bits encode the reliability mode.
353        writer.write_u8(self.flags)?;
354        writer.write_u16(self.length_in_bytes * 8, Endian::Big)?;
355
356        if self.is_reliable()? {
357            writer.write_u24(self.reliable_frame_index, Endian::Little)?;
358        }
359
360        if self.is_sequenced()? {
361            writer.write_u24(self.sequenced_frame_index, Endian::Little)?;
362        }
363        if self.is_ordered()? {
364            writer.write_u24(self.ordered_frame_index, Endian::Little)?;
365            writer.write_u8(self.order_channel)?;
366        }
367
368        // Bit 4 marks a fragmented frame.
369        if (self.flags & 16) != 0 {
370            writer.write_u32(self.compound_size, Endian::Big)?;
371            writer.write_u16(self.compound_id, Endian::Big)?;
372            writer.write_u32(self.fragment_index, Endian::Big)?;
373        }
374        Ok(())
375    }
376
377    pub fn is_fragment(&self) -> bool {
378        (self.flags & 16) != 0
379    }
380
381    pub fn is_reliable(&self) -> Result<bool> {
382        let r = Reliability::from((self.flags & 224) >> 5)?;
383        Ok(matches!(
384            r,
385            Reliability::Reliable | Reliability::ReliableOrdered | Reliability::ReliableSequenced
386        ))
387    }
388
389    pub fn is_ordered(&self) -> Result<bool> {
390        let r = Reliability::from((self.flags & 224) >> 5)?;
391        Ok(matches!(
392            r,
393            Reliability::UnreliableSequenced
394                | Reliability::ReliableOrdered
395                | Reliability::ReliableSequenced
396        ))
397    }
398
399    pub fn is_sequenced(&self) -> Result<bool> {
400        let r = Reliability::from((self.flags & 224) >> 5)?;
401        Ok(matches!(
402            r,
403            Reliability::UnreliableSequenced | Reliability::ReliableSequenced
404        ))
405    }
406    pub fn reliability(&self) -> Result<Reliability> {
407        Reliability::from((self.flags & 224) >> 5)
408    }
409
410    pub fn _size(&self) -> Result<usize> {
411        let mut ret = 0;
412        // id
413        ret += 1;
414        // sequence number
415        ret += 3;
416        // flags
417        ret += 1;
418        // length_in_bits
419        ret += 2;
420
421        if self.is_reliable()? {
422            // reliable frame index
423            ret += 3;
424        }
425        if self.is_sequenced()? {
426            // sequenced frame index
427            ret += 3;
428        }
429        if self.is_ordered()? {
430            // Ordered frame index and channel.
431            ret += 4;
432        }
433        if (self.flags & 16) != 0 {
434            // Compound size, compound ID, and fragment index.
435            ret += 10;
436        }
437        //body
438        ret += self.data.len();
439        Ok(ret)
440    }
441}
442
443pub struct FrameVec {
444    #[cfg(test)]
445    pub frames: Vec<FrameSetPacket>,
446}
447
448impl FrameVec {
449    #[cfg(test)]
450    pub fn new(buf: &[u8]) -> Result<Self> {
451        let mut frames = Vec::new();
452        Self::decode_into(buf, &mut frames)?;
453        Ok(Self { frames })
454    }
455
456    // Reuse the frame metadata allocation across incoming datagrams.
457    pub(crate) fn decode_into(buf: &[u8], frames: &mut Vec<FrameSetPacket>) -> Result<()> {
458        Self::decode(buf, frames, false)
459    }
460
461    pub(crate) fn decode_owned_into(
462        mut buf: Vec<u8>,
463        frames: &mut Vec<FrameSetPacket>,
464    ) -> Result<()> {
465        Self::decode(&buf, frames, true)?;
466        if let [frame] = frames.as_mut_slice() {
467            // A single-frame datagram can reuse its allocation for the payload.
468            // Compact the body in place without a shared slice owner.
469            let length = usize::from(frame.length_in_bytes);
470            let offset = buf.len() - length;
471            buf.copy_within(offset.., 0);
472            buf.truncate(length);
473            frame.data = buf.into();
474        }
475        Ok(())
476    }
477
478    fn decode(buf: &[u8], frames: &mut Vec<FrameSetPacket>, owned: bool) -> Result<()> {
479        frames.clear();
480        let mut reader = RaknetReader::new(buf);
481        let id = reader.read_u8()?;
482        if !(0x80..=0x8d).contains(&id) {
483            return Err(RaknetError::PacketHeaderError);
484        }
485        let sequence_number = reader.read_u24(Endian::Little)?;
486        if reader.pos() == buf.len() as u64 {
487            return Err(RaknetError::PacketParseError);
488        }
489        while reader.pos() < buf.len() as u64 {
490            let mut frame = FrameSetPacket {
491                sequence_number,
492                flags: 0,
493                length_in_bytes: 0,
494                reliable_frame_index: 0,
495                sequenced_frame_index: 0,
496                ordered_frame_index: 0,
497                order_channel: 0,
498                compound_size: 0,
499                compound_id: 0,
500                fragment_index: 0,
501                data: FramePayload::default(),
502                shared_payload: None,
503            };
504
505            frame.flags = reader.read_u8()?;
506            let length_in_bits = reader.read_u16(Endian::Big)?;
507            if length_in_bits % 8 != 0 {
508                return Err(RaknetError::PacketParseError);
509            }
510            frame.length_in_bytes = length_in_bits / 8;
511
512            if frame.is_reliable()? {
513                frame.reliable_frame_index = reader.read_u24(Endian::Little)?;
514            }
515            if frame.is_sequenced()? {
516                frame.sequenced_frame_index = reader.read_u24(Endian::Little)?;
517            }
518            if frame.is_ordered()? {
519                frame.ordered_frame_index = reader.read_u24(Endian::Little)?;
520                frame.order_channel = reader.read_u8()?;
521            }
522            if frame.is_fragment() {
523                frame.compound_size = reader.read_u32(Endian::Big)?;
524                frame.compound_id = reader.read_u16(Endian::Big)?;
525                frame.fragment_index = reader.read_u32(Endian::Big)?;
526                if frame.compound_size == 0 || frame.fragment_index >= frame.compound_size {
527                    return Err(RaknetError::PacketParseError);
528                }
529            }
530
531            let data = reader.read_slice(usize::from(frame.length_in_bytes))?;
532            // Multi-frame datagrams retain independent payload allocations.
533            // The owned single-frame path moves the original allocation below.
534            if !owned || !frames.is_empty() || reader.pos() != buf.len() as u64 {
535                frame.data = data.to_vec().into();
536            }
537            frames.push(frame);
538        }
539
540        Ok(())
541    }
542}
543
544#[derive(Default)]
545pub struct ACKSet {
546    ack: Vec<(u32, u32)>,
547    nack: Vec<(u32, u32)>,
548    next_expected: u32,
549}
550
551impl ACKSet {
552    pub fn insert(&mut self, s: u32) {
553        // Remove a hole that arrived before the next NACK was emitted.
554        let mut split = None;
555        self.nack.retain_mut(|(start, end)| {
556            if s < *start || s > *end {
557                return true;
558            }
559            if *start == *end {
560                return false;
561            }
562            if s == *start {
563                *start += 1;
564            } else if s == *end {
565                *end -= 1;
566            } else {
567                split = Some((s + 1, *end));
568                *end = s - 1;
569            }
570            true
571        });
572        if let Some(range) = split {
573            self.nack.push(range);
574        }
575        if s == self.next_expected || sequence::newer(s, self.next_expected) {
576            if s != self.next_expected {
577                if s > self.next_expected {
578                    self.nack.push((self.next_expected, s - 1));
579                } else {
580                    self.nack.push((self.next_expected, sequence::MASK));
581                    if s != 0 {
582                        self.nack.push((0, s - 1));
583                    }
584                }
585            }
586            self.next_expected = sequence::next(s);
587        }
588
589        for i in 0..self.ack.len() {
590            let a = self.ack[i];
591            if (a.0..=a.1).contains(&s) {
592                return;
593            }
594            if a.0 != 0 && s == a.0 - 1 {
595                self.ack[i].0 = s;
596                return;
597            }
598            if s == a.1.saturating_add(1) {
599                self.ack[i].1 = s;
600                return;
601            }
602        }
603        self.ack.push((s, s));
604    }
605
606    #[cfg(test)]
607    pub fn get_ack(&mut self) -> Vec<(u32, u32)> {
608        std::mem::take(&mut self.ack)
609    }
610
611    fn take_ack_into(&mut self, output: &mut Vec<(u32, u32)>) {
612        output.clear();
613        std::mem::swap(output, &mut self.ack);
614    }
615
616    fn take_nack_into(&mut self, output: &mut Vec<(u32, u32)>) {
617        output.clear();
618        std::mem::swap(output, &mut self.nack);
619    }
620
621    #[cfg(test)]
622    pub fn get_nack(&mut self) -> Vec<(u32, u32)> {
623        std::mem::take(&mut self.nack)
624    }
625}
626
627/// Channel zero is the common path; other channels allocate only when used.
628#[derive(Default)]
629struct ChannelIndexes {
630    primary: u32,
631    others: HashMap<u8, u32>,
632}
633
634impl ChannelIndexes {
635    fn get(&self, channel: u8) -> u32 {
636        if channel == 0 {
637            self.primary
638        } else {
639            self.others.get(&channel).copied().unwrap_or(0)
640        }
641    }
642
643    fn get_mut(&mut self, channel: u8) -> &mut u32 {
644        if channel == 0 {
645            &mut self.primary
646        } else {
647            self.others.entry(channel).or_default()
648        }
649    }
650
651    #[cfg(test)]
652    fn insert(&mut self, channel: u8, index: u32) {
653        *self.get_mut(channel) = index;
654    }
655}
656
657#[derive(Default)]
658pub struct RecvQ {
659    sequenced_frame_indexes: ChannelIndexes,
660    last_ordered_indexes: ChannelIndexes,
661    sequence_number_ackset: ACKSet,
662    packets: HashMap<u32, FrameSetPacket>,
663    ordered_packets: HashMap<(u8, u32), FrameSetPacket>,
664    fragment_queue: FragmentQ,
665    reliable_window: ReliableWindow,
666    ordered_bytes: usize,
667    ready_ordered: Vec<FrameSetPacket>,
668    ready_bytes: usize,
669}
670
671impl RecvQ {
672    pub fn new() -> Self {
673        Self::default()
674    }
675
676    pub fn insert(&mut self, frame: FrameSetPacket) -> Result<()> {
677        let reliability = frame.reliability()?;
678        self.sequence_number_ackset.insert(frame.sequence_number);
679        if frame.is_reliable()? && !self.reliable_window.accept(frame.reliable_frame_index)? {
680            return Ok(());
681        }
682
683        match reliability {
684            Reliability::Unreliable => {
685                self.packets.entry(frame.sequence_number).or_insert(frame);
686            }
687            Reliability::UnreliableSequenced => {
688                let channel = frame.order_channel;
689                let sequence_index = frame.sequenced_frame_index;
690                let sequence_number = frame.sequence_number;
691                let last_index = self.sequenced_frame_indexes.get_mut(channel);
692                if sequence_index == *last_index || sequence::newer(sequence_index, *last_index) {
693                    self.packets.entry(sequence_number).or_insert(frame);
694                    *last_index = sequence::next(sequence_index);
695                }
696            }
697            Reliability::Reliable => {
698                self.packets.insert(frame.sequence_number, frame);
699            }
700            Reliability::ReliableOrdered => {
701                let expected_index = self.last_ordered_indexes.get(frame.order_channel);
702                if sequence::newer(expected_index, frame.ordered_frame_index) {
703                    return Ok(());
704                }
705                if sequence::distance(frame.ordered_frame_index, expected_index)
706                    >= sequence::RECEIVE_WINDOW
707                {
708                    return Err(RaknetError::PacketParseError);
709                }
710                let complete = if frame.is_fragment() {
711                    self.fragment_queue.insert_and_take_completed(frame)?
712                } else {
713                    Some(frame)
714                };
715                if let Some(frame) = complete {
716                    let key = (frame.order_channel, frame.ordered_frame_index);
717                    if self.ordered_packets.contains_key(&key) {
718                        return Ok(());
719                    }
720                    if self.ordered_packets.len() + self.ready_ordered.len()
721                        >= sequence::RECEIVE_WINDOW as usize
722                        || self.ordered_bytes + self.ready_bytes + frame.data.len()
723                            > 64 * 1024 * 1024
724                    {
725                        return Err(RaknetError::PacketParseError);
726                    }
727                    if frame.ordered_frame_index == expected_index {
728                        let channel = frame.order_channel;
729                        self.ready_bytes += frame.data.len();
730                        self.ready_ordered.push(frame);
731                        let expected = self.last_ordered_indexes.get_mut(channel);
732                        *expected = sequence::next(*expected);
733                        // Contiguous traffic bypasses the reordering hash table.
734                        while !self.ordered_packets.is_empty() {
735                            let Some(frame) = self.ordered_packets.remove(&(channel, *expected))
736                            else {
737                                break;
738                            };
739                            self.ordered_bytes -= frame.data.len();
740                            self.ready_bytes += frame.data.len();
741                            self.ready_ordered.push(frame);
742                            *expected = sequence::next(*expected);
743                        }
744                    } else {
745                        self.ordered_bytes += frame.data.len();
746                        self.ordered_packets.insert(key, frame);
747                    }
748                }
749            }
750
751            Reliability::ReliableSequenced => {
752                let channel = frame.order_channel;
753                let sequence_index = frame.sequenced_frame_index;
754                let sequence_number = frame.sequence_number;
755                let last_index = self.sequenced_frame_indexes.get_mut(channel);
756                if sequence_index == *last_index || sequence::newer(sequence_index, *last_index) {
757                    self.packets.entry(sequence_number).or_insert(frame);
758                    *last_index = sequence::next(sequence_index);
759                }
760            }
761        }
762        Ok(())
763    }
764
765    #[cfg(test)]
766    pub fn get_ack(&mut self) -> Vec<(u32, u32)> {
767        self.sequence_number_ackset.get_ack()
768    }
769
770    #[cfg(test)]
771    pub fn get_nack(&mut self) -> Vec<(u32, u32)> {
772        self.sequence_number_ackset.get_nack()
773    }
774
775    pub(crate) fn take_ack_into(&mut self, output: &mut Vec<(u32, u32)>) {
776        self.sequence_number_ackset.take_ack_into(output);
777    }
778
779    pub(crate) fn take_nack_into(&mut self, output: &mut Vec<(u32, u32)>) {
780        self.sequence_number_ackset.take_nack_into(output);
781    }
782
783    #[cfg(test)]
784    pub fn flush(&mut self, _peer_addr: &SocketAddr) -> Vec<FrameSetPacket> {
785        let mut frames = Vec::new();
786        self.flush_into(&mut frames);
787        frames
788    }
789
790    pub(crate) fn flush_into(&mut self, frames: &mut Vec<FrameSetPacket>) {
791        frames.append(&mut self.ready_ordered);
792        self.ready_bytes = 0;
793        if self.packets.is_empty() {
794            return;
795        }
796        let mut packets: Vec<_> = self.packets.drain().map(|(_, frame)| frame).collect();
797        packets.sort_unstable_by_key(|frame| frame.sequence_number);
798        frames.extend(packets);
799    }
800    pub(crate) fn needs_maintenance(&self) -> bool {
801        !self.sequence_number_ackset.nack.is_empty() || self.fragment_queue.size() != 0
802    }
803
804    pub fn fragments_expired(&self) -> bool {
805        self.fragment_queue.expired()
806    }
807
808    pub fn get_ordered_packet(&self) -> usize {
809        self.ordered_packets.len() + self.ready_ordered.len()
810    }
811
812    pub fn get_fragment_queue_size(&self) -> usize {
813        self.fragment_queue.size()
814    }
815
816    pub fn get_ordered_keys(&self) -> Vec<(u8, u32)> {
817        self.ordered_packets.keys().copied().collect()
818    }
819
820    pub fn get_size(&self) -> usize {
821        self.packets.len()
822    }
823}
824
825#[cfg(feature = "send-policy")]
826struct SendScheduler {
827    capacity_notifier: std::sync::Weak<tokio::sync::Notify>,
828    options: crate::SendOptions,
829    unreliable: VecDeque<FrameSetPacket>,
830    unreliable_bytes: usize,
831    use_legacy: bool,
832    persistent: bool,
833    next_lane: usize,
834}
835
836pub struct SendQ {
837    mtu: u16,
838    ack_sequence_number: u32,
839    sequence_number: u32,
840    reliable_frame_index: u32,
841    sequenced_frame_indexes: ChannelIndexes,
842    ordered_frame_indexes: ChannelIndexes,
843    compound_id: u16,
844    // Each entry stores a frame, send state, last send time, retry count, and prior sequence IDs.
845    packets: VecDeque<FrameSetPacket>,
846    queued_unreliable: usize,
847    buffered_bytes: usize,
848    rto: i64,
849    srtt: i64,
850    sent_packet: Vec<(FrameSetPacket, bool, i64, u32, Vec<u32>)>,
851    flight_order_dirty: bool,
852    next_retry: i64,
853    retries_pending: bool,
854    coalesce: bool,
855    coalescing_allowed: bool,
856    grouped_acks: bool,
857    #[cfg(feature = "send-policy")]
858    scheduler: Option<Box<SendScheduler>>,
859}
860
861impl SendQ {
862    pub const DEFAULT_TIMEOUT_MILLS: i64 = 50;
863    const MAX_IN_FLIGHT_PACKETS: usize = 64;
864    const MAX_COALESCED_FRAMES: usize = 8;
865    const MAX_BUFFERED_BYTES: usize = 64 * 1024 * 1024;
866    const SEND_HIGH_WATER: usize = 256 * 1024;
867    const FRAME_BUDGET: usize = 128;
868
869    const RTO_UBOUND: i64 = 12000;
870    const RTO_LBOUND: i64 = 50;
871
872    pub fn new(mtu: u16) -> Self {
873        Self {
874            mtu,
875            ack_sequence_number: sequence::MASK,
876            sequence_number: 0,
877            packets: VecDeque::new(),
878            #[cfg(feature = "send-policy")]
879            scheduler: None,
880            queued_unreliable: 0,
881            buffered_bytes: 0,
882            sent_packet: vec![],
883            flight_order_dirty: false,
884            next_retry: i64::MAX,
885            retries_pending: false,
886            coalesce: false,
887            coalescing_allowed: true,
888            grouped_acks: false,
889            reliable_frame_index: 0,
890            sequenced_frame_indexes: ChannelIndexes::default(),
891            ordered_frame_indexes: ChannelIndexes::default(),
892            compound_id: 0,
893
894            rto: SendQ::DEFAULT_TIMEOUT_MILLS,
895            srtt: SendQ::DEFAULT_TIMEOUT_MILLS,
896        }
897    }
898
899    fn required_bytes(&self, reliability: Reliability, len: usize) -> Result<usize> {
900        let payload = usize::from(self.mtu)
901            .checked_sub(60)
902            .filter(|n| *n > 0)
903            .ok_or(RaknetError::PacketSizeExceedMTU)?;
904        let frames = if reliability == Reliability::ReliableOrdered && len > payload {
905            let frames = len.div_ceil(payload);
906            if frames > crate::fragment::MAX_FRAGMENTS {
907                return Err(RaknetError::PacketSizeExceedMTU);
908            }
909            frames
910        } else {
911            1
912        };
913        let bytes = frames
914            .checked_mul(Self::FRAME_BUDGET)
915            .and_then(|n| n.checked_add(len))
916            .filter(|n| *n <= Self::MAX_BUFFERED_BYTES)
917            .ok_or(RaknetError::PacketSizeExceedMTU)?;
918        Ok(bytes)
919    }
920
921    pub fn has_capacity(&self, reliability: Reliability, len: usize) -> Result<bool> {
922        let required = self.required_bytes(reliability, len)?;
923        // Admit a single large message while applying backpressure to bursts.
924        #[cfg(feature = "send-policy")]
925        {
926            self.has_reserved_capacity(reliability, required)
927        }
928        #[cfg(not(feature = "send-policy"))]
929        {
930            Ok(self.buffered_bytes.saturating_add(required) <= Self::SEND_HIGH_WATER.max(required))
931        }
932    }
933
934    pub(crate) fn enable_coalescing(&mut self) {
935        self.coalesce = self.coalescing_allowed;
936    }
937
938    pub(crate) fn allows_coalescing(&self) -> bool {
939        self.coalescing_allowed
940    }
941
942    pub(crate) fn has_batch_capacity(
943        &self,
944        reliability: Reliability,
945        mut lengths: impl Iterator<Item = usize>,
946    ) -> Result<bool> {
947        let required = lengths.try_fold(0usize, |total, length| {
948            total
949                .checked_add(self.required_bytes(reliability, length)?)
950                .filter(|&bytes| bytes <= Self::MAX_BUFFERED_BYTES)
951                .ok_or(RaknetError::PacketSizeExceedMTU)
952        })?;
953        #[cfg(feature = "send-policy")]
954        {
955            self.has_batch_required_capacity(reliability, required)
956        }
957        #[cfg(not(feature = "send-policy"))]
958        {
959            Ok(self.buffered_bytes.saturating_add(required) <= Self::SEND_HIGH_WATER.max(required))
960        }
961    }
962
963    pub fn insert(&mut self, reliability: Reliability, buf: &[u8]) -> Result<()> {
964        self.insert_with_order_channel(reliability, buf, 0)
965    }
966
967    pub fn insert_with_order_channel(
968        &mut self,
969        reliability: Reliability,
970        buf: &[u8],
971        order_channel: u8,
972    ) -> Result<()> {
973        self.insert_payload(reliability, buf, order_channel, None)
974    }
975
976    pub(crate) fn insert_bytes(
977        &mut self,
978        reliability: Reliability,
979        data: &bytes::Bytes,
980        order_channel: u8,
981    ) -> Result<()> {
982        self.insert_payload(reliability, data, order_channel, Some(data))
983    }
984
985    fn insert_payload(
986        &mut self,
987        reliability: Reliability,
988        buf: &[u8],
989        order_channel: u8,
990        owned: Option<&bytes::Bytes>,
991    ) -> Result<()> {
992        let reserved = self.required_bytes(reliability, buf.len())?;
993        if reserved > Self::MAX_BUFFERED_BYTES - self.buffered_bytes {
994            return Err(RaknetError::PacketSizeExceedMTU);
995        }
996        let max_payload = self
997            .mtu
998            .checked_sub(60)
999            .map(usize::from)
1000            .filter(|size| *size > 0)
1001            .ok_or(RaknetError::PacketSizeExceedMTU)?;
1002
1003        match reliability {
1004            Reliability::Unreliable => {
1005                if buf.len() > max_payload {
1006                    return Err(RaknetError::PacketSizeExceedMTU);
1007                }
1008
1009                let frame = FrameSetPacket::from_payload(
1010                    reliability,
1011                    owned.map_or_else(
1012                        || FramePayload::Owned(buf.to_vec()),
1013                        |data| FramePayload::Bytes(data.clone()),
1014                    ),
1015                );
1016                #[cfg(feature = "send-policy")]
1017                self.push_unreliable(frame);
1018                #[cfg(not(feature = "send-policy"))]
1019                self.packets.push_back(frame);
1020            }
1021            Reliability::UnreliableSequenced => {
1022                if buf.len() > max_payload {
1023                    return Err(RaknetError::PacketSizeExceedMTU);
1024                }
1025
1026                let sequenced_frame_index = {
1027                    let index = self.sequenced_frame_indexes.get_mut(order_channel);
1028                    let current = *index;
1029                    *index = sequence::next(*index);
1030                    current
1031                };
1032                let ordered_frame_index = self.ordered_frame_indexes.get(order_channel);
1033                let mut frame = FrameSetPacket::from_payload(
1034                    reliability,
1035                    owned.map_or_else(
1036                        || FramePayload::Owned(buf.to_vec()),
1037                        |data| FramePayload::Bytes(data.clone()),
1038                    ),
1039                );
1040                frame.order_channel = order_channel;
1041                frame.ordered_frame_index = ordered_frame_index;
1042                frame.sequenced_frame_index = sequenced_frame_index;
1043                #[cfg(feature = "send-policy")]
1044                self.push_unreliable(frame);
1045                #[cfg(not(feature = "send-policy"))]
1046                self.packets.push_back(frame);
1047            }
1048            Reliability::Reliable => {
1049                if buf.len() > max_payload {
1050                    return Err(RaknetError::PacketSizeExceedMTU);
1051                }
1052
1053                let mut frame = FrameSetPacket::from_payload(
1054                    reliability,
1055                    owned.map_or_else(
1056                        || FramePayload::copy_shared(buf),
1057                        |data| FramePayload::Bytes(data.clone()),
1058                    ),
1059                );
1060                frame.reliable_frame_index = self.reliable_frame_index;
1061                self.packets.push_back(frame);
1062                self.reliable_frame_index = sequence::next(self.reliable_frame_index);
1063            }
1064            Reliability::ReliableOrdered => {
1065                if buf.len() <= max_payload {
1066                    let mut frame = FrameSetPacket::from_payload(
1067                        reliability,
1068                        owned.map_or_else(
1069                            || FramePayload::copy_shared(buf),
1070                            |data| FramePayload::Bytes(data.clone()),
1071                        ),
1072                    );
1073                    frame.order_channel = order_channel;
1074                    frame.reliable_frame_index = self.reliable_frame_index;
1075                    frame.ordered_frame_index = *self.ordered_frame_indexes.get_mut(order_channel);
1076                    self.packets.push_back(frame);
1077                    self.reliable_frame_index = sequence::next(self.reliable_frame_index);
1078                } else {
1079                    let compound_size = buf.len().div_ceil(max_payload);
1080                    let compound_size = u32::try_from(compound_size)
1081                        .map_err(|_| RaknetError::PacketSizeExceedMTU)?;
1082                    let ordered_frame_index = *self.ordered_frame_indexes.get_mut(order_channel);
1083
1084                    let data: Option<Arc<[u8]>> = owned.is_none().then(|| Arc::from(buf));
1085                    let shared_payload = Arc::new(SharedPayload {
1086                        remaining: AtomicUsize::new(compound_size as usize),
1087                        bytes: buf.len(),
1088                    });
1089                    for (fragment_index, chunk) in buf.chunks(max_payload).enumerate() {
1090                        let start = fragment_index * max_payload;
1091                        let mut frame = FrameSetPacket::from_payload(
1092                            reliability,
1093                            match owned {
1094                                Some(data) => {
1095                                    FramePayload::Bytes(data.slice(start..start + chunk.len()))
1096                                }
1097                                None => FramePayload::Shared {
1098                                    owner: data.as_ref().unwrap().clone(),
1099                                    start: start as u32,
1100                                    length: chunk.len() as u16,
1101                                },
1102                            },
1103                        );
1104                        frame.shared_payload = Some(shared_payload.clone());
1105                        frame.flags |= 16;
1106                        frame.compound_size = compound_size;
1107                        frame.compound_id = self.compound_id;
1108                        frame.fragment_index = fragment_index as u32;
1109                        frame.order_channel = order_channel;
1110                        frame.reliable_frame_index = self.reliable_frame_index;
1111                        frame.ordered_frame_index = ordered_frame_index;
1112                        self.packets.push_back(frame);
1113                        self.reliable_frame_index = sequence::next(self.reliable_frame_index);
1114                    }
1115                    self.compound_id = self.compound_id.wrapping_add(1);
1116                }
1117                let index = self.ordered_frame_indexes.get_mut(order_channel);
1118                *index = sequence::next(*index);
1119            }
1120            Reliability::ReliableSequenced => {
1121                if buf.len() > max_payload {
1122                    return Err(RaknetError::PacketSizeExceedMTU);
1123                }
1124
1125                let sequenced_frame_index = {
1126                    let index = self.sequenced_frame_indexes.get_mut(order_channel);
1127                    let current = *index;
1128                    *index = sequence::next(*index);
1129                    current
1130                };
1131                let ordered_frame_index = self.ordered_frame_indexes.get(order_channel);
1132                let mut frame = FrameSetPacket::from_payload(
1133                    reliability,
1134                    owned.map_or_else(
1135                        || FramePayload::copy_shared(buf),
1136                        |data| FramePayload::Bytes(data.clone()),
1137                    ),
1138                );
1139                frame.order_channel = order_channel;
1140                frame.reliable_frame_index = self.reliable_frame_index;
1141                frame.sequenced_frame_index = sequenced_frame_index;
1142                frame.ordered_frame_index = ordered_frame_index;
1143                self.packets.push_back(frame);
1144                self.reliable_frame_index = sequence::next(self.reliable_frame_index);
1145            }
1146        };
1147        self.buffered_bytes += reserved;
1148        #[cfg(not(feature = "send-policy"))]
1149        if matches!(
1150            reliability,
1151            Reliability::Unreliable | Reliability::UnreliableSequenced
1152        ) {
1153            self.queued_unreliable += 1;
1154        }
1155        Ok(())
1156    }
1157
1158    fn update_rto(&mut self, rtt: i64) {
1159        let previous = self.rto;
1160        (self.srtt, self.rto) = Self::latency_sample(self.srtt, rtt);
1161        if self.rto != previous {
1162            self.next_retry = 0;
1163        }
1164    }
1165
1166    fn latency_sample(srtt: i64, rtt: i64) -> (i64, i64) {
1167        let srtt = (srtt * 4 + rtt) / 5;
1168        (
1169            srtt,
1170            (srtt * 3 / 2).clamp(Self::RTO_LBOUND, Self::RTO_UBOUND),
1171        )
1172    }
1173
1174    pub(crate) fn mtu(&self) -> u16 {
1175        self.mtu
1176    }
1177
1178    pub fn get_rto(&self) -> i64 {
1179        self.rto
1180    }
1181
1182    pub fn nack(&mut self, sequence: u32, tick: i64) {
1183        for i in 0..self.sent_packet.len() {
1184            let item = &mut self.sent_packet[i];
1185            if item.1 && item.0.sequence_number == sequence {
1186                // Preserve ordinary pacing once this connection has observed
1187                // loss, including feedback received before batching is enabled.
1188                self.coalescing_allowed = false;
1189                self.coalesce = false;
1190                raknet_log_debug!(
1191                    "packet {}-{}-{} nack {} times",
1192                    item.0.sequence_number,
1193                    item.0.reliable_frame_index,
1194                    item.0.ordered_frame_index,
1195                    item.3 + 1
1196                );
1197                let previous = item.0.sequence_number;
1198                self.flight_order_dirty = true;
1199                item.0.sequence_number = self.sequence_number;
1200                self.sequence_number = sequence::next(self.sequence_number);
1201                item.1 = false;
1202                self.retries_pending = true;
1203                item.2 = tick;
1204
1205                if item.4.len() == 64 {
1206                    item.4.remove(0);
1207                }
1208                item.4.push(previous);
1209            }
1210        }
1211    }
1212
1213    pub fn ack(&mut self, sequence: u32, tick: i64) {
1214        if self.grouped_acks && self.sent_packet.len() > 1 {
1215            self.ack_ranges(&[(sequence, sequence)], tick);
1216            return;
1217        }
1218        // Ignore ACKs outside the actual send history before advancing any state.
1219        let Some(index) = self
1220            .sent_packet
1221            .iter()
1222            .position(|item| item.0.sequence_number == sequence || item.4.contains(&sequence))
1223        else {
1224            return;
1225        };
1226        let item = self.sent_packet.remove(index);
1227        if self.grouped_acks && self.sent_packet.is_empty() {
1228            self.grouped_acks = false;
1229        }
1230        let released_payload = match &item.0.shared_payload {
1231            Some(payload) if payload.remaining.fetch_sub(1, Ordering::Relaxed) == 1 => {
1232                payload.bytes
1233            }
1234            Some(_) => 0,
1235            None => item.0.data.len(),
1236        };
1237        self.buffered_bytes -= released_payload + Self::FRAME_BUDGET;
1238        if sequence::newer(sequence, self.ack_sequence_number) {
1239            let span = sequence::distance(sequence, self.ack_sequence_number);
1240            let missing: Vec<_> = self
1241                .sent_packet
1242                .iter()
1243                .filter_map(|item| {
1244                    let distance =
1245                        sequence::distance(item.0.sequence_number, self.ack_sequence_number);
1246                    (distance > 0 && distance < span).then_some(item.0.sequence_number)
1247                })
1248                .collect();
1249            for id in missing {
1250                self.nack(id, tick);
1251            }
1252            self.ack_sequence_number = sequence;
1253        }
1254        // A retransmitted frame has an ambiguous RTT sample (Karn's algorithm).
1255        if item.3 == 0 {
1256            self.update_rto(tick.saturating_sub(item.2).max(0));
1257        }
1258    }
1259
1260    #[inline]
1261    pub fn ack_ranges(&mut self, ranges: &[(u32, u32)], tick: i64) {
1262        if self.grouped_acks {
1263            self.ack_ranges_inner::<true>(ranges, tick);
1264        } else {
1265            self.ack_ranges_inner::<false>(ranges, tick);
1266        }
1267    }
1268
1269    fn ack_ranges_inner<const GROUPED: bool>(&mut self, ranges: &[(u32, u32)], tick: i64) {
1270        if let [(start, end)] = ranges {
1271            if start == end && (!GROUPED || self.sent_packet.len() <= 1) {
1272                self.ack(*start, tick);
1273                return;
1274            }
1275        }
1276        // Compact the bounded flight window once. Removing each acknowledged
1277        // frame separately repeatedly shifts the remaining frame metadata.
1278        let previous = self.ack_sequence_number;
1279        let mut newest = previous;
1280        let mut latency = (self.srtt, self.rto);
1281        let mut released_bytes = 0;
1282        let mut previous_sample = None;
1283        self.sent_packet.retain_mut(|item| {
1284            let acknowledged_id = std::iter::once(item.0.sequence_number)
1285                .chain(item.4.iter().copied())
1286                .find(|id| {
1287                    ranges
1288                        .iter()
1289                        .any(|&(start, end)| start <= *id && *id <= end)
1290                });
1291            let Some(id) = acknowledged_id else {
1292                return true;
1293            };
1294            let payload_bytes = match &item.0.shared_payload {
1295                Some(payload) if payload.remaining.fetch_sub(1, Ordering::Relaxed) == 1 => {
1296                    payload.bytes
1297                }
1298                Some(_) => 0,
1299                None => item.0.data.len(),
1300            };
1301            released_bytes += payload_bytes + Self::FRAME_BUDGET;
1302            if sequence::newer(id, newest) {
1303                newest = id;
1304            }
1305            // Initial members of a frame set are contiguous. Retransmitted
1306            // frames never contribute RTT samples (Karn's algorithm).
1307            if item.3 == 0 && (!GROUPED || previous_sample != Some(id)) {
1308                if GROUPED {
1309                    previous_sample = Some(id);
1310                }
1311                latency = Self::latency_sample(latency.0, tick.saturating_sub(item.2).max(0));
1312            }
1313            false
1314        });
1315        self.buffered_bytes -= released_bytes;
1316        if GROUPED && self.sent_packet.is_empty() {
1317            self.grouped_acks = false;
1318        }
1319
1320        if self.rto != latency.1 {
1321            self.next_retry = 0;
1322        }
1323        (self.srtt, self.rto) = latency;
1324        if newest != previous {
1325            let span = sequence::distance(newest, previous);
1326            let missing: Vec<_> = self
1327                .sent_packet
1328                .iter()
1329                .filter_map(|item| {
1330                    let distance = sequence::distance(item.0.sequence_number, previous);
1331                    (distance > 0 && distance < span).then_some(item.0.sequence_number)
1332                })
1333                .collect();
1334            for id in missing {
1335                self.nack(id, tick);
1336            }
1337            self.ack_sequence_number = newest;
1338        }
1339    }
1340
1341    pub fn nack_ranges(&mut self, ranges: &[(u32, u32)], tick: i64) {
1342        if let [(start, end)] = ranges {
1343            if start == end {
1344                self.nack(*start, tick);
1345                return;
1346            }
1347        }
1348        let missing: Vec<_> = self
1349            .sent_packet
1350            .iter()
1351            .filter_map(|item| {
1352                let id = item.0.sequence_number;
1353                ranges
1354                    .iter()
1355                    .any(|&(start, end)| start <= id && id <= end)
1356                    .then_some(id)
1357            })
1358            .collect();
1359        for id in missing {
1360            self.nack(id, tick);
1361        }
1362    }
1363
1364    fn retry_timeout(rto: i64, retries: u32) -> i64 {
1365        let mut timeout = rto;
1366        for _ in 0..retries.min(32) {
1367            timeout = ((timeout as f64 * 1.5) as i64).min(Self::RTO_UBOUND);
1368        }
1369        timeout
1370    }
1371
1372    fn tick(&mut self, tick: i64) {
1373        if tick < self.next_retry {
1374            return;
1375        }
1376        self.next_retry = i64::MAX;
1377        for i in 0..self.sent_packet.len() {
1378            let p = &mut self.sent_packet[i];
1379
1380            let cur_rto = Self::retry_timeout(self.rto, p.3);
1381
1382            if p.1 && tick - p.2 >= cur_rto {
1383                self.coalescing_allowed = false;
1384                self.coalesce = false;
1385                let previous = p.0.sequence_number;
1386                self.flight_order_dirty = true;
1387                p.0.sequence_number = self.sequence_number;
1388                self.sequence_number = sequence::next(self.sequence_number);
1389                p.1 = false;
1390                self.retries_pending = true;
1391                if p.4.len() == 64 {
1392                    p.4.remove(0);
1393                }
1394                p.4.push(previous);
1395            } else if p.1 {
1396                self.next_retry = self.next_retry.min(p.2.saturating_add(cur_rto));
1397            }
1398        }
1399    }
1400
1401    pub fn flush(&mut self, tick: i64, peer_addr: &SocketAddr) -> OutgoingFrames {
1402        #[cfg(feature = "send-policy")]
1403        if self
1404            .scheduler
1405            .as_ref()
1406            .is_some_and(|s| !s.use_legacy || !s.unreliable.is_empty())
1407        {
1408            return self.flush_scheduled(tick);
1409        }
1410        self.tick(tick);
1411
1412        // Reserve queued batches once; a full reliable flight window reserves
1413        // nothing. The compact vector keeps async frame transfers cheap.
1414        let capacity = self.packets.len().min(
1415            Self::MAX_IN_FLIGHT_PACKETS.saturating_sub(self.sent_packet.len())
1416                + self.queued_unreliable,
1417        );
1418        let mut ret = OutgoingFrames::with_capacity(capacity);
1419        // New sends preserve numeric order until the 24-bit sequence wraps.
1420        // ACK removal preserves it too; only resequencing requires sorting.
1421        if self.flight_order_dirty {
1422            self.sent_packet
1423                .sort_unstable_by_key(|packet| packet.0.sequence_number);
1424            self.flight_order_dirty = false;
1425        }
1426
1427        if self.retries_pending {
1428            for packet in &mut self.sent_packet {
1429                if !packet.1 {
1430                    raknet_log_debug!(
1431                        "{} , packet {}-{}-{} resend {} times",
1432                        peer_addr,
1433                        packet.0.sequence_number,
1434                        packet.0.reliable_frame_index,
1435                        packet.0.ordered_frame_index,
1436                        packet.3 + 1
1437                    );
1438                    ret.push(packet.0.clone());
1439                    packet.1 = true;
1440                    packet.2 = tick;
1441                    packet.3 = packet.3.saturating_add(1);
1442                    self.next_retry = self
1443                        .next_retry
1444                        .min(tick.saturating_add(Self::retry_timeout(self.rto, packet.3)));
1445                }
1446            }
1447            self.retries_pending = false;
1448        }
1449
1450        let mut group_sequence = None;
1451        let mut group_bytes = 0;
1452        let mut group_frames = 0;
1453        let datagram_budget = usize::from(self.mtu).saturating_sub(28);
1454        let queued_count = self.packets.len();
1455        for _ in 0..queued_count {
1456            if self.sent_packet.len() >= Self::MAX_IN_FLIGHT_PACKETS && self.queued_unreliable == 0
1457            {
1458                break;
1459            }
1460            let Some(mut packet) = self.packets.pop_front() else {
1461                break;
1462            };
1463            let reliable = packet.is_reliable().unwrap_or(false);
1464            if reliable && self.sent_packet.len() >= Self::MAX_IN_FLIGHT_PACKETS {
1465                self.packets.push_back(packet);
1466                continue;
1467            }
1468
1469            if self.coalesce {
1470                let packable = matches!(packet.reliability(), Ok(Reliability::ReliableOrdered))
1471                    && !packet.is_fragment();
1472                let frame_bytes = packet._size().expect("queued frames were validated") - 4;
1473                if packable
1474                    && group_sequence.is_some()
1475                    && group_frames < Self::MAX_COALESCED_FRAMES
1476                    && group_bytes + frame_bytes <= datagram_budget
1477                {
1478                    packet.sequence_number = group_sequence.unwrap();
1479                    group_bytes += frame_bytes;
1480                    group_frames += 1;
1481                    self.grouped_acks = true;
1482                    ret.coalesced = true;
1483                } else {
1484                    packet.sequence_number = self.sequence_number;
1485                    self.sequence_number = sequence::next(self.sequence_number);
1486                    group_sequence = packable.then_some(packet.sequence_number);
1487                    group_bytes = 4 + frame_bytes;
1488                    // Keep a 16-message application window spread across at
1489                    // least two datagrams so later packets can expose a loss.
1490                    group_frames = 1;
1491                }
1492            } else {
1493                packet.sequence_number = self.sequence_number;
1494                self.sequence_number = sequence::next(self.sequence_number);
1495            }
1496
1497            if reliable {
1498                ret.push(packet.clone());
1499                if self
1500                    .sent_packet
1501                    .last()
1502                    .is_some_and(|previous| previous.0.sequence_number > packet.sequence_number)
1503                {
1504                    self.flight_order_dirty = true;
1505                }
1506                self.sent_packet.push((packet, true, tick, 0, Vec::new()));
1507                self.next_retry = self.next_retry.min(tick.saturating_add(self.rto));
1508            } else {
1509                self.queued_unreliable -= 1;
1510                self.buffered_bytes -= packet.data.len() + Self::FRAME_BUDGET;
1511                ret.push(packet);
1512            }
1513        }
1514
1515        ret
1516    }
1517
1518    pub fn is_empty(&self) -> bool {
1519        #[cfg(feature = "send-policy")]
1520        {
1521            self.packets.is_empty()
1522                && self.sent_packet.is_empty()
1523                && self
1524                    .scheduler
1525                    .as_ref()
1526                    .is_none_or(|s| s.unreliable.is_empty())
1527        }
1528        #[cfg(not(feature = "send-policy"))]
1529        {
1530            self.packets.is_empty() && self.sent_packet.is_empty()
1531        }
1532    }
1533
1534    pub fn get_reliable_queue_size(&self) -> usize {
1535        self.packets.len()
1536    }
1537
1538    pub fn get_sent_queue_size(&self) -> usize {
1539        self.sent_packet.len()
1540    }
1541}
1542
1543#[test]
1544fn malformed_datagrams_are_rejected() {
1545    assert!(FrameVec::new(&[0x80, 0, 0]).is_err());
1546    assert!(FrameVec::new(&[0x10, 0, 0, 0]).is_err());
1547}
1548
1549#[test]
1550fn send_queue_rejects_an_invalid_mtu() {
1551    let mut send_queue = SendQ::new(59);
1552    assert!(send_queue.insert(Reliability::Reliable, &[0xfe]).is_err());
1553}
1554
1555#[cfg(feature = "send-policy")]
1556impl SendQ {
1557    /// Validate batch sizes and determine packing eligibility in one pass.
1558    /// Oversized messages still validate their fragmentation budget, but cannot
1559    /// be packed. Avoid rescanning an entire batch that will use ordinary sends.
1560    pub(crate) fn batch_can_coalesce(
1561        &self,
1562        reliability: Reliability,
1563        mut lengths: impl Iterator<Item = usize>,
1564    ) -> Result<bool> {
1565        let allowed = self.allows_coalescing() && reliability == Reliability::ReliableOrdered;
1566        if reliability == Reliability::ReliableOrdered && !allowed {
1567            self.has_batch_capacity(reliability, lengths)?;
1568            return Ok(false);
1569        }
1570        let max_payload = usize::from(self.mtu).saturating_sub(60);
1571        let datagram_budget = usize::from(self.mtu).saturating_sub(28);
1572        let mut previous = None;
1573        let mut packable = false;
1574        let required = lengths.try_fold(0usize, |total, length| {
1575            if reliability != Reliability::ReliableOrdered && length > max_payload {
1576                return Err(RaknetError::PacketSizeExceedMTU);
1577            }
1578            if allowed && !packable && length <= max_payload {
1579                packable = previous.is_some_and(|last: usize| {
1580                    last <= max_payload
1581                        && last.saturating_add(length).saturating_add(24) <= datagram_budget
1582                });
1583            }
1584            previous = Some(length);
1585            total
1586                .checked_add(self.required_bytes(reliability, length)?)
1587                .filter(|&bytes| bytes <= Self::MAX_BUFFERED_BYTES)
1588                .ok_or(RaknetError::PacketSizeExceedMTU)
1589        })?;
1590        self.has_batch_required_capacity(reliability, required)?;
1591        Ok(allowed && packable)
1592    }
1593
1594    fn has_batch_required_capacity(
1595        &self,
1596        reliability: Reliability,
1597        required: usize,
1598    ) -> Result<bool> {
1599        // Unreliable batches use the streaming path. Their aggregate may
1600        // exceed the pending reserve even though every member fits and drains.
1601        if matches!(
1602            reliability,
1603            Reliability::Unreliable | Reliability::UnreliableSequenced
1604        ) && self
1605            .send_options()
1606            .is_some_and(|options| required > options.unreliable_queue_bytes)
1607        {
1608            return Ok(false);
1609        }
1610        self.has_reserved_capacity(reliability, required)
1611    }
1612
1613    fn ensure_scheduler(&mut self) -> &mut SendScheduler {
1614        if self.scheduler.is_none() {
1615            let mut unreliable = VecDeque::with_capacity(self.queued_unreliable);
1616            let mut unreliable_bytes = 0;
1617            for _ in 0..self.packets.len() {
1618                let frame = self.packets.pop_front().unwrap();
1619                if frame.is_reliable().unwrap_or(false) {
1620                    self.packets.push_back(frame);
1621                } else {
1622                    unreliable_bytes += frame.data.len() + Self::FRAME_BUDGET;
1623                    unreliable.push_back(frame);
1624                }
1625            }
1626            self.queued_unreliable = 0;
1627            self.scheduler = Some(Box::new(SendScheduler {
1628                capacity_notifier: std::sync::Weak::new(),
1629                options: crate::SendOptions::default(),
1630                unreliable,
1631                unreliable_bytes,
1632                use_legacy: usize::from(self.mtu) * Self::MAX_IN_FLIGHT_PACKETS
1633                    <= crate::SendOptions::default().flush_bytes,
1634                persistent: true,
1635                next_lane: 2,
1636            }));
1637        }
1638        self.scheduler.as_mut().unwrap()
1639    }
1640
1641    fn push_unreliable(&mut self, mut frame: FrameSetPacket) {
1642        if let Some(scheduler) = &mut self.scheduler {
1643            scheduler.unreliable_bytes += frame.data.len() + Self::FRAME_BUDGET;
1644            scheduler.unreliable.push_back(frame);
1645        } else {
1646            frame.sequence_number = self.sequence_number;
1647            self.sequence_number = sequence::next(self.sequence_number);
1648            self.queued_unreliable += 1;
1649            self.packets.push_back(frame);
1650        }
1651    }
1652
1653    /// Bind queue-release notifications only after a policy is enabled.
1654    pub(crate) fn register_capacity_notifier(&mut self, notifier: &Arc<tokio::sync::Notify>) {
1655        let Some(scheduler) = &mut self.scheduler else {
1656            return;
1657        };
1658        if !std::ptr::eq(scheduler.capacity_notifier.as_ptr(), Arc::as_ptr(notifier)) {
1659            scheduler.capacity_notifier = Arc::downgrade(notifier);
1660        }
1661    }
1662
1663    pub(crate) fn set_send_options(&mut self, options: crate::SendOptions) -> Result<()> {
1664        let mtu = usize::from(self.mtu);
1665        if options.in_flight_bytes < mtu
1666            || options.flush_bytes < mtu
1667            || options.unreliable_queue_bytes < mtu
1668        {
1669            return Err(RaknetError::PacketSizeExceedMTU);
1670        }
1671        let scheduler = self.ensure_scheduler();
1672        scheduler.options = options;
1673        scheduler.persistent = true;
1674        let defaults = crate::SendOptions::default();
1675        scheduler.use_legacy = options.in_flight_frames == defaults.in_flight_frames
1676            && options.in_flight_bytes == defaults.in_flight_bytes
1677            && options.flush_bytes == defaults.flush_bytes
1678            && mtu * Self::MAX_IN_FLIGHT_PACKETS <= options.flush_bytes;
1679        Ok(())
1680    }
1681
1682    pub(crate) fn send_options(&self) -> Option<crate::SendOptions> {
1683        self.scheduler.as_ref().map(|s| s.options)
1684    }
1685
1686    #[inline]
1687    fn has_reserved_capacity(&self, reliability: Reliability, required: usize) -> Result<bool> {
1688        if self.scheduler.is_none() {
1689            return Ok(
1690                self.buffered_bytes.saturating_add(required) <= Self::SEND_HIGH_WATER.max(required)
1691            );
1692        }
1693        let unreliable = matches!(
1694            reliability,
1695            Reliability::Unreliable | Reliability::UnreliableSequenced
1696        );
1697        if !unreliable {
1698            // The common reliable-only case needs neither the full options
1699            // copy nor reservation arithmetic. Valid budgets leave room for
1700            // the unreliable reserve, including when no unreliable data waits.
1701            let budget = match &self.scheduler {
1702                None if required <= Self::MAX_BUFFERED_BYTES - 64 * 1024 => {
1703                    Some(Self::SEND_HIGH_WATER.max(required))
1704                }
1705                None => None,
1706                Some(s)
1707                    if s.unreliable_bytes == 0 && required <= s.options.reliable_queue_bytes =>
1708                {
1709                    Some(s.options.reliable_queue_bytes)
1710                }
1711                Some(_) => None,
1712            };
1713            if let Some(budget) = budget {
1714                return Ok(self.buffered_bytes.saturating_add(required) <= budget);
1715            }
1716        }
1717        let options = self.send_options().unwrap();
1718        let used_unreliable = self.scheduler.as_ref().map_or(0, |s| s.unreliable_bytes);
1719        let (used, limit) = if unreliable {
1720            if required > options.unreliable_queue_bytes {
1721                return Err(RaknetError::PacketSizeExceedMTU);
1722            }
1723            (used_unreliable, options.unreliable_queue_bytes)
1724        } else {
1725            let hard_limit = Self::MAX_BUFFERED_BYTES - options.unreliable_queue_bytes;
1726            if required > hard_limit {
1727                return Err(RaknetError::PacketSizeExceedMTU);
1728            }
1729            (
1730                self.buffered_bytes - used_unreliable,
1731                options.reliable_queue_bytes.max(required).min(hard_limit),
1732            )
1733        };
1734        Ok(used.saturating_add(required) <= limit
1735            && self.buffered_bytes.saturating_add(required) <= Self::MAX_BUFFERED_BYTES)
1736    }
1737
1738    fn flush_scheduled(&mut self, tick: i64) -> OutgoingFrames {
1739        self.tick(tick);
1740        if self.flight_order_dirty {
1741            self.sent_packet
1742                .sort_unstable_by_key(|packet| packet.0.sequence_number);
1743            self.flight_order_dirty = false;
1744        }
1745        // Taking the sidecar keeps the frame queues and connection indexes under
1746        // one mutable owner. Unconfigured sockets never allocate this state.
1747        let mut scheduler = self.scheduler.take().unwrap();
1748        let options = scheduler.options;
1749        // Compute only in configured scheduling. ACK handling remains the
1750        // original path with no per-frame policy or byte-ledger checks.
1751        let mut flight_bytes: usize = self.sent_packet.iter().map(|p| p.0._size().unwrap()).sum();
1752        let capacity = self.packets.len().min(
1753            options
1754                .in_flight_frames
1755                .saturating_sub(self.sent_packet.len()),
1756        ) + scheduler.unreliable.len();
1757        let mut output = OutgoingFrames::with_capacity(capacity.min(256));
1758        let quantum = usize::from(self.mtu) * 4;
1759        let mut bytes = 0;
1760        let mut unreliable_drained = false;
1761        let mut retry_cursor = 0;
1762        let mut group_sequence = None;
1763        let mut group_bytes = 0;
1764        let mut group_frames = 0;
1765        let datagram_budget = usize::from(self.mtu).saturating_sub(28);
1766        'rounds: loop {
1767            let mut progressed = false;
1768            let start = scheduler.next_lane;
1769            for offset in 0..3 {
1770                let lane = (start + offset) % 3;
1771                let mut lane_bytes = 0;
1772                while lane_bytes < quantum {
1773                    let charge = match lane {
1774                        0 => scheduler
1775                            .unreliable
1776                            .front()
1777                            .map(|frame| frame._size().unwrap()),
1778                        1 => {
1779                            if !self.retries_pending {
1780                                break;
1781                            }
1782                            while retry_cursor < self.sent_packet.len()
1783                                && self.sent_packet[retry_cursor].1
1784                            {
1785                                retry_cursor += 1;
1786                            }
1787                            self.sent_packet
1788                                .get(retry_cursor)
1789                                .map(|item| item.0._size().unwrap())
1790                        }
1791                        _ => self.packets.front().and_then(|frame| {
1792                            let size = frame._size().unwrap();
1793                            (self.sent_packet.len() < options.in_flight_frames
1794                                && flight_bytes + size <= options.in_flight_bytes)
1795                                .then_some(size)
1796                        }),
1797                    };
1798                    let Some(charge) = charge else {
1799                        break;
1800                    };
1801                    // Skip a class that cannot fit the remaining burst; smaller
1802                    // realtime messages in another class can still make progress.
1803                    if bytes + charge > options.flush_bytes {
1804                        break;
1805                    }
1806                    match lane {
1807                        0 => {
1808                            let mut packet = scheduler.unreliable.pop_front().unwrap();
1809                            packet.sequence_number = self.sequence_number;
1810                            self.sequence_number = sequence::next(self.sequence_number);
1811                            let reserved = packet.data.len() + Self::FRAME_BUDGET;
1812                            self.buffered_bytes -= reserved;
1813                            scheduler.unreliable_bytes -= reserved;
1814                            unreliable_drained = true;
1815                            output.push(packet);
1816                            group_sequence = None;
1817                        }
1818                        1 => {
1819                            let item = &mut self.sent_packet[retry_cursor];
1820                            output.push(item.0.clone());
1821                            item.1 = true;
1822                            item.2 = tick;
1823                            item.3 = item.3.saturating_add(1);
1824                            self.next_retry = self
1825                                .next_retry
1826                                .min(tick.saturating_add(Self::retry_timeout(self.rto, item.3)));
1827                            retry_cursor += 1;
1828                            group_sequence = None;
1829                        }
1830                        _ => {
1831                            let mut packet = self.packets.pop_front().unwrap();
1832                            let packable = self.coalesce
1833                                && matches!(packet.reliability(), Ok(Reliability::ReliableOrdered))
1834                                && !packet.is_fragment();
1835                            let frame_bytes = charge - 4;
1836                            if packable
1837                                && group_sequence.is_some()
1838                                && group_frames < Self::MAX_COALESCED_FRAMES
1839                                && group_bytes + frame_bytes <= datagram_budget
1840                            {
1841                                packet.sequence_number = group_sequence.unwrap();
1842                                group_bytes += frame_bytes;
1843                                group_frames += 1;
1844                                self.grouped_acks = true;
1845                                output.coalesced = true;
1846                            } else {
1847                                packet.sequence_number = self.sequence_number;
1848                                self.sequence_number = sequence::next(self.sequence_number);
1849                                group_sequence = packable.then_some(packet.sequence_number);
1850                                group_bytes = charge;
1851                                group_frames = 1;
1852                            }
1853                            output.push(packet.clone());
1854                            if self
1855                                .sent_packet
1856                                .last()
1857                                .is_some_and(|p| p.0.sequence_number > packet.sequence_number)
1858                            {
1859                                self.flight_order_dirty = true;
1860                            }
1861                            self.sent_packet.push((packet, true, tick, 0, Vec::new()));
1862                            flight_bytes += charge;
1863                            self.next_retry = self.next_retry.min(tick.saturating_add(self.rto));
1864                        }
1865                    }
1866                    progressed = true;
1867                    bytes += charge;
1868                    lane_bytes += charge;
1869                    scheduler.next_lane = (lane + 1) % 3;
1870                    if bytes == options.flush_bytes {
1871                        break 'rounds;
1872                    }
1873                }
1874            }
1875            if !progressed {
1876                break;
1877            }
1878        }
1879        self.retries_pending = self.sent_packet.iter().any(|item| !item.1);
1880        if unreliable_drained {
1881            if let Some(capacity) = scheduler.capacity_notifier.upgrade() {
1882                capacity.notify_waiters();
1883            }
1884        }
1885        if scheduler.persistent || !scheduler.use_legacy || !scheduler.unreliable.is_empty() {
1886            self.scheduler = Some(scheduler);
1887        }
1888        output
1889    }
1890}
1891
1892#[tokio::test]
1893async fn test_recvq_orders_channels_independently() {
1894    let mut recvq = RecvQ::new();
1895
1896    let mut later_on_channel_one = FrameSetPacket::new(Reliability::ReliableOrdered, vec![1]);
1897    later_on_channel_one.sequence_number = 0;
1898    later_on_channel_one.order_channel = 1;
1899    later_on_channel_one.ordered_frame_index = 1;
1900    recvq.insert(later_on_channel_one).unwrap();
1901
1902    let mut first_on_channel_two = FrameSetPacket::new(Reliability::ReliableOrdered, vec![2]);
1903    first_on_channel_two.sequence_number = 1;
1904    first_on_channel_two.reliable_frame_index = 1;
1905    first_on_channel_two.order_channel = 2;
1906    first_on_channel_two.ordered_frame_index = 0;
1907    recvq.insert(first_on_channel_two).unwrap();
1908
1909    let ready = recvq.flush(&"127.0.0.1:0".parse().unwrap());
1910    assert_eq!(ready.len(), 1);
1911    assert_eq!(ready[0].order_channel, 2);
1912
1913    let mut first_on_channel_one = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0]);
1914    first_on_channel_one.sequence_number = 2;
1915    first_on_channel_one.reliable_frame_index = 2;
1916    first_on_channel_one.order_channel = 1;
1917    first_on_channel_one.ordered_frame_index = 0;
1918    recvq.insert(first_on_channel_one).unwrap();
1919
1920    let ready = recvq.flush(&"127.0.0.1:0".parse().unwrap());
1921    assert_eq!(ready.len(), 2);
1922    assert_eq!(ready[0].ordered_frame_index, 0);
1923    assert_eq!(ready[1].ordered_frame_index, 1);
1924    assert!(ready.iter().all(|frame| frame.order_channel == 1));
1925}
1926
1927#[tokio::test]
1928async fn test_sendq_maintains_ordered_indexes_per_channel() {
1929    let mut sendq = SendQ::new(1500);
1930    sendq
1931        .insert_with_order_channel(Reliability::ReliableOrdered, &[0xfe, 1], 3)
1932        .unwrap();
1933    sendq
1934        .insert_with_order_channel(Reliability::ReliableOrdered, &[0xfe, 2], 9)
1935        .unwrap();
1936
1937    let sent = sendq.flush(0, &"127.0.0.1:0".parse().unwrap());
1938    assert_eq!(sent.len(), 2);
1939    assert_eq!(sent[0].order_channel, 3);
1940    assert_eq!(sent[0].ordered_frame_index, 0);
1941    assert_eq!(sent[1].order_channel, 9);
1942    assert_eq!(sent[1].ordered_frame_index, 0);
1943}
1944
1945#[tokio::test]
1946async fn test_ackset() {
1947    let mut ackset = ACKSet::default();
1948
1949    ackset.insert(0);
1950    ackset.insert(1);
1951    ackset.insert(2);
1952    ackset.insert(4);
1953
1954    let acks = ackset.get_ack();
1955
1956    assert!(acks == vec![(0, 2), (4, 4)]);
1957
1958    let mut ackset = ACKSet::default();
1959
1960    ackset.insert(0);
1961    ackset.insert(1);
1962    ackset.insert(2);
1963    ackset.insert(6);
1964
1965    let acks = ackset.get_ack();
1966
1967    assert!(acks == vec![(0, 2), (6, 6)]);
1968
1969    let acks = ackset.get_ack();
1970
1971    assert!(acks == vec![]);
1972
1973    ackset.insert(0);
1974    ackset.insert(2);
1975
1976    let acks = ackset.get_ack();
1977
1978    assert!(acks == vec![(0, 0), (2, 2)]);
1979}
1980
1981#[tokio::test]
1982async fn test_frame_serialize_deserialize() {
1983    // Captured first frame datagram from a Bedrock 1.18.12 client.
1984    let p: Vec<u8> = [
1985        132, 0, 0, 0, 64, 0, 144, 0, 0, 0, 9, 146, 33, 7, 47, 57, 18, 128, 111, 0, 0, 0, 0, 20,
1986        200, 47, 41, 0,
1987    ]
1988    .to_vec();
1989
1990    let a = FrameSetPacket::deserialize(&p).unwrap();
1991    assert!(a.0.serialize().unwrap() == p);
1992}
1993
1994#[tokio::test]
1995async fn test_recvq() {
1996    let mut r = RecvQ::new();
1997    let mut p = FrameSetPacket::new(Reliability::Reliable, vec![]);
1998    p.sequence_number = 0;
1999    p.ordered_frame_index = 0;
2000    r.insert(p).unwrap();
2001
2002    let mut p = FrameSetPacket::new(Reliability::Reliable, vec![]);
2003    p.sequence_number = 1;
2004    p.reliable_frame_index = 1;
2005    p.ordered_frame_index = 1;
2006    r.insert(p).unwrap();
2007
2008    let ret = r.flush(&"0.0.0.0:0".parse().unwrap());
2009    assert!(ret.len() == 2);
2010}
2011
2012#[tokio::test]
2013async fn test_recvq_fragment() {
2014    let mut r = RecvQ::new();
2015    let mut p = FrameSetPacket::new(Reliability::ReliableOrdered, vec![1]);
2016    p.flags |= 16;
2017    p.sequence_number = 0;
2018    p.ordered_frame_index = 0;
2019    p.compound_id = 1;
2020    p.compound_size = 3;
2021    p.fragment_index = 0;
2022    p.order_channel = 7;
2023    r.insert(p).unwrap();
2024
2025    let mut p = FrameSetPacket::new(Reliability::ReliableOrdered, vec![2]);
2026    p.flags |= 16;
2027    p.sequence_number = 1;
2028    p.reliable_frame_index = 1;
2029    p.ordered_frame_index = 0;
2030    p.compound_id = 1;
2031    p.compound_size = 3;
2032    p.fragment_index = 1;
2033    p.order_channel = 7;
2034    r.insert(p).unwrap();
2035
2036    let mut p = FrameSetPacket::new(Reliability::ReliableOrdered, vec![3]);
2037    p.flags |= 16;
2038    p.sequence_number = 2;
2039    p.reliable_frame_index = 2;
2040    p.ordered_frame_index = 0;
2041    p.compound_id = 1;
2042    p.compound_size = 3;
2043    p.fragment_index = 2;
2044    p.order_channel = 7;
2045    r.insert(p).unwrap();
2046
2047    let ret = r.flush(&"0.0.0.0:0".parse().unwrap());
2048    assert!(ret.len() == 1);
2049    assert_eq!(ret[0].data.as_ref(), &[1, 2, 3]);
2050    assert_eq!(ret[0].order_channel, 7);
2051}
2052
2053#[tokio::test]
2054async fn test_sendq() {
2055    let mut s = SendQ::new(1500);
2056    let p = FrameSetPacket::new(Reliability::Reliable, vec![]);
2057    s.insert(Reliability::Reliable, &p.serialize().unwrap())
2058        .unwrap();
2059
2060    let p = FrameSetPacket::new(Reliability::Reliable, vec![]);
2061    s.insert(Reliability::Reliable, &p.serialize().unwrap())
2062        .unwrap();
2063
2064    let sockaddr: SocketAddr = "127.0.0.1:8000".parse().unwrap();
2065    let ret = s.flush(0, &sockaddr);
2066    assert!(ret.len() == 2);
2067
2068    s.ack(0, 0);
2069    s.ack(1, 0);
2070
2071    let ret = s.flush(300, &sockaddr);
2072    assert!(ret.is_empty());
2073}
2074
2075#[test]
2076fn send_queue_pipelines_reliable_packets_with_a_bounded_window() {
2077    let mut sendq = SendQ::new(1500);
2078    let peer = "127.0.0.1:8000".parse().unwrap();
2079
2080    for index in 0..SendQ::MAX_IN_FLIGHT_PACKETS + 1 {
2081        sendq
2082            .insert(Reliability::ReliableOrdered, &[0xfe, index as u8])
2083            .unwrap();
2084        let sent = sendq.flush(0, &peer);
2085        if index < SendQ::MAX_IN_FLIGHT_PACKETS {
2086            assert_eq!(sent.len(), 1);
2087            assert_eq!(sent[0].ordered_frame_index, index as u32);
2088        } else {
2089            assert!(sent.is_empty());
2090        }
2091    }
2092
2093    assert_eq!(sendq.get_sent_queue_size(), SendQ::MAX_IN_FLIGHT_PACKETS);
2094    assert_eq!(sendq.get_reliable_queue_size(), 1);
2095
2096    sendq.ack(0, 1);
2097    let sent = sendq.flush(1, &peer);
2098    assert_eq!(sent.len(), 1);
2099    assert_eq!(
2100        sent[0].ordered_frame_index,
2101        SendQ::MAX_IN_FLIGHT_PACKETS as u32
2102    );
2103    assert_eq!(sendq.get_sent_queue_size(), SendQ::MAX_IN_FLIGHT_PACKETS);
2104    assert_eq!(sendq.get_reliable_queue_size(), 0);
2105}
2106
2107#[test]
2108fn send_queue_retransmits_immediately_after_nack() {
2109    let mut sendq = SendQ::new(1500);
2110    let peer = "127.0.0.1:8000".parse().unwrap();
2111    sendq.insert(Reliability::Reliable, &[0xfe]).unwrap();
2112
2113    let first = sendq.flush(0, &peer);
2114    assert_eq!(first.len(), 1);
2115    assert_eq!(first[0].sequence_number, 0);
2116
2117    sendq.nack(0, 1);
2118    let retransmitted = sendq.flush(1, &peer);
2119    assert_eq!(retransmitted.len(), 1);
2120    assert_eq!(retransmitted[0].sequence_number, 1);
2121}
2122
2123#[tokio::test]
2124async fn test_client_packet1() {
2125    let a = [
2126        140, 3, 0, 0, 112, 44, 192, 2, 0, 0, 1, 0, 0, 0, 0, 0, 0, 19, 0, 0, 0, 0, 0, 0, 254, 236,
2127        189, 203, 114, 234, 74, 183, 239, 185, 79, 175, 170, 30, 99, 159, 110, 125, 59, 36, 1, 222,
2128        166, 122, 19, 35, 97, 48, 18, 19, 161, 11, 82, 69, 197, 23, 128, 248, 38, 32, 9, 107, 218,
2129        152, 91, 69, 61, 79, 53, 234, 141, 78, 243, 52, 234, 5, 170, 119, 90, 149, 41, 165, 48, 41,
2130        11, 12, 2, 95, 214, 154, 255, 198, 47, 86, 120, 77, 15, 75, 202, 203, 24, 57, 46, 153, 249,
2131        223, 255, 199, 127, 253, 47, 255, 246, 111, 255, 229, 255, 253, 111, 255, 227, 191, 254,
2132        63, 255, 243, 191, 253, 219, 255, 249, 239, 163, 201, 96, 58, 255, 247, 255, 237, 127, 255,
2133        247, 241, 166, 53, 25, 54, 70, 211, 206, 180, 165, 152, 91, 181, 172, 221, 53, 159, 155,
2134        243, 95, 162, 215, 107, 222, 52, 125, 177, 105, 247, 61, 165, 107, 5, 205, 158, 84, 29,
2135        244, 250, 209, 111, 85, 214, 106, 166, 21, 52, 122, 91, 93, 209, 77, 165, 222, 51, 117,
2136        215, 21, 20, 197, 109, 148, 55, 166, 220, 122, 234, 216, 101, 105, 108, 43, 157, 241, 180,
2137        218, 49, 26, 250, 189, 165, 184, 243, 129, 242, 171, 50, 170, 7, 247, 109, 241, 113, 173,
2138        221, 235, 205, 225, 60, 18, 218, 226, 196, 234, 134, 186, 57, 240, 43, 83, 43, 112, 167,
2139        150, 57, 89, 13, 75, 90, 205, 146, 244, 123, 183, 46, 111, 199, 182, 176, 214, 182, 147,
2140        141, 214, 16, 187, 227, 134, 38, 247, 234, 150, 219, 107, 84, 231, 78, 201, 17, 186, 115,
2141        189, 174, 139, 154, 51, 82, 212, 210, 192, 178, 30, 13, 49, 216, 186, 179, 218, 203, 192,
2142        143, 238, 61, 177, 37, 186, 211, 234, 147, 41, 105, 91, 87, 169, 61, 142, 239, 22, 207,
2143        170, 28, 205, 199, 117, 253, 121, 32, 107, 117, 213, 119, 166, 255, 234, 62, 254, 7, 249,
2144        238, 153, 219, 111, 9, 3, 219, 141, 28, 73, 17, 92, 83, 17, 189, 198, 100, 57, 10, 3, 97,
2145        76, 190, 221, 187, 39, 127, 167, 183, 154, 186, 253, 201, 170, 57, 35, 239, 61, 235, 150,
2146        58, 117, 103, 173, 25, 242, 186, 125, 215, 138, 220, 134, 245, 226, 53, 200, 239, 90, 53,
2147        209, 9, 215, 145, 35, 44, 130, 241, 153, 109, 214, 145, 45, 117, 32, 5, 229, 177, 185, 158,
2148        121, 210, 122, 48, 154, 7, 150, 105, 107, 162, 106, 233, 146, 41, 87, 23, 61, 163, 117,
2149        167, 149, 92, 167, 83, 215, 94, 220, 70, 165, 111, 6, 214, 196, 110, 8, 37, 237, 222, 107,
2150        184, 161, 76, 191, 243, 217, 19, 149, 142, 213, 112, 55, 134, 226, 54, 29, 163, 213, 29,
2151        218, 214, 203, 72, 246, 90, 154, 31, 61, 246, 76, 209, 234, 133, 74, 223, 158, 183, 126,
2152        15, 77, 241, 119, 199, 168, 117, 134, 194, 162, 163, 7, 90, 167, 59, 183, 218, 110, 67, 40,
2153        143, 130, 32, 178, 239, 181, 208, 233, 63, 110, 123, 91, 85, 26, 223, 221, 174, 45, 163,
2154        41, 245, 238, 107, 247, 170, 18, 149, 123, 155, 170, 173, 90, 206, 148, 124, 243, 139, 19,
2155        58, 211, 206, 76, 150, 180, 250, 72, 234, 212, 127, 73, 154, 161, 84, 239, 126, 253, 199,
2156        63, 158, 123, 119, 119, 203, 126, 120, 247, 220, 113, 55, 245, 97, 189, 212, 172, 255, 136,
2157        30, 159, 164, 122, 255, 95, 245, 141, 220, 159, 41, 149, 169, 218, 173, 174, 5, 249, 247,
2158        83, 205, 188, 185, 81, 220, 113, 251, 229, 101, 213, 249, 245, 123, 212, 248, 71, 239, 105,
2159        48, 151, 68, 103, 32, 175, 215, 63, 220, 113, 216, 169, 252, 243, 247, 211, 118, 230, 255,
2160        179, 210, 173, 173, 123, 230, 203, 63, 170, 163, 127, 12, 52, 39, 26, 247, 158, 182, 147,
2161        113, 52, 154, 4, 75, 219, 122, 168, 223, 120, 255, 152, 223, 138, 173, 135, 122, 79, 43,
2162        245, 27, 255, 254, 191, 210, 81, 92, 214, 250, 38, 29, 197, 90, 79, 9, 74, 164, 149, 221,
2163        158, 188, 88, 218, 126, 112, 51, 152, 213, 234, 93, 50, 12, 117, 255, 89, 208, 77, 171,
2164        214, 21, 2, 217, 182, 189, 154, 110, 76, 20, 163, 177, 136, 198, 247, 129, 234, 148, 188,
2165        103, 210, 74, 21, 203, 82, 102, 158, 161, 152, 94, 80, 251, 57, 52, 253, 77, 151, 244, 139,
2166        93, 255, 181, 29, 134, 250, 139, 37, 182, 44, 199, 154, 44, 180, 173, 94, 210, 67, 119,
2167        162, 217, 238, 168, 45, 173, 75, 166, 89, 49, 221, 121, 75, 181, 252, 201, 68, 13, 106, 11,
2168        215, 174, 172, 116, 193, 151, 122, 225, 164, 101, 88, 214, 131, 99, 121, 131, 81, 24, 45,
2169        12, 251, 177, 162, 202, 149, 101, 79, 168, 52, 76, 161, 114, 231, 153, 11, 127, 104, 76,
2170        54, 182, 29, 56, 35, 201, 157, 140, 103, 90, 91, 221, 186, 229, 126, 182, 7, 196, 213, 210,
2171        154, 41, 118, 115, 186, 154, 58, 246, 122, 78, 70, 227, 84, 183, 212, 109, 167, 222, 37,
2172        35, 153, 14, 228, 156, 142, 145, 201, 36, 38, 19, 59, 212, 151, 67, 179, 178, 28, 134, 90,
2173        64, 196, 74, 234, 204, 41, 107, 51, 191, 172, 209, 137, 62, 107, 10, 228, 191, 2, 249, 125,
2174        129, 14, 232, 81, 73, 141, 155, 111, 40, 69, 68, 110, 68, 59, 60, 24, 223, 255, 72, 255,
2175        46, 249, 125, 89, 84, 233, 223, 37, 127, 107, 52, 215, 35, 55, 12, 102, 78, 95, 15, 186,
2176        125, 75, 24, 52, 170, 155, 65, 95, 175, 52, 103, 145, 48, 154, 91, 1, 253, 123, 78, 191,
2177        187, 255, 78, 165, 68, 54, 240, 73, 243, 8, 244, 119, 205, 123, 107, 58, 108, 4, 179, 158,
2178        100, 85, 232, 39, 25, 230, 196, 245, 4, 171, 102, 155, 147, 246, 80, 140, 90, 227, 240,
2179        113, 213, 21, 20, 221, 54, 221, 182, 38, 91, 74, 215, 212, 90, 186, 18, 204, 187, 166, 181,
2180        26, 26, 90, 93, 187, 91, 180, 212, 128, 40, 129, 198, 194, 232, 88, 171, 229, 80, 28, 109,
2181        92, 203, 43, 121, 115, 125, 221, 19, 162, 178, 102, 138, 174, 109, 77, 12, 215, 168, 217,
2182        186, 36, 151, 71, 164, 139, 61, 223, 26, 12, 132, 245, 114, 68, 148, 196, 64, 138, 106,
2183        166, 160, 77, 6, 210, 66, 232, 137, 122, 107, 228, 59, 226, 184, 111, 169, 35, 203, 85, 61,
2184        165, 86, 25, 247, 221, 126, 215, 94, 76, 186, 118, 180, 30, 207, 38, 91, 77, 209, 219, 61,
2185        203, 213, 186, 225, 164, 231, 206, 213, 167, 14, 153, 252, 142, 45, 170, 227, 198, 164, 57,
2186        188, 143, 30, 109, 197, 93, 147, 103, 205, 173, 251, 214, 202, 178, 221, 27, 39, 80, 250,
2187        134, 160, 210, 46, 50, 255, 53, 31, 119, 199, 195, 137, 182, 49, 141, 199, 40, 108, 191,
2188        244, 212, 222, 63, 198, 213, 255, 188, 239, 206, 102, 155, 223, 102, 235, 159, 255, 252,
2189        241, 16, 244, 194, 233, 68, 170, 7, 27, 189, 211, 250, 189, 104, 143, 255, 241, 159, 66,
2190        221, 121, 169, 73, 98, 189, 90, 107, 78, 23, 29, 229, 215, 96, 54, 27, 153, 183, 182, 223,
2191        155, 223, 55, 127, 24, 227, 219, 167, 217, 237, 188, 31, 74, 154, 226, 255, 115, 220, 169,
2192        143, 165, 231, 135, 95, 245, 217, 118, 174, 201, 134, 255, 88, 121, 92, 132, 129, 123, 251,
2193        175, 250, 204, 189, 169, 46, 138, 78, 144, 126, 109, 161, 10, 170, 240, 32, 248, 27, 83,
2194        156, 172, 6, 162, 90, 233, 223, 85, 151, 214, 182, 21, 88, 37, 79, 242, 238, 149, 118, 143,
2195        152, 10, 131, 232, 111, 91, 209, 2, 85, 113, 239, 29, 99, 178, 37, 170, 76, 210, 173, 232,
2196        201, 104, 84, 183, 230, 220, 122, 26, 248, 138, 209, 149, 148, 167, 65, 73, 111, 91, 50,
2197        249, 194, 153, 85, 241, 204, 245, 218, 242, 215, 130, 121, 31, 84, 188, 192, 171, 13, 36,
2198        165, 54, 152, 43, 55, 157, 123, 85, 176, 228, 69, 203, 242, 197, 187, 129, 210, 10, 71,
2199        155, 231, 50, 25, 45, 147, 161, 185, 46, 15, 228, 201, 243, 56, 156, 56, 214, 92, 9, 212,
2200        134, 103, 142, 230, 53, 203, 157, 71, 83, 211, 242, 126, 118, 55, 196,
2201    ];
2202
2203    let b = FrameVec::new(&a).unwrap();
2204    assert!(b.frames.len() == 1);
2205    assert!(b.frames[0].is_fragment());
2206}
2207
2208#[tokio::test]
2209async fn test_client_packet2() {
2210    // Connection Request - reliable [ reliable_frame_index = 0 ]
2211    let p0 = [
2212        132, 0, 0, 0, 64, 0, 144, 0, 0, 0, 9, 162, 70, 235, 28, 218, 182, 26, 192, 0, 0, 0, 0, 16,
2213        151, 43, 113, 0,
2214    ];
2215    // Connection Request - reliable [ reliable_frame_index = 0 ]
2216    let p1 = [
2217        132, 1, 0, 0, 64, 0, 144, 0, 0, 0, 9, 162, 70, 235, 28, 218, 182, 26, 192, 0, 0, 0, 0, 16,
2218        151, 43, 113, 0,
2219    ];
2220    // 2 frames Incompatible Protocol(extract data?) Connected ping - reliable ordered [ reliable_frame_index = 1 ]
2221    let p2 = [
2222        132, 2, 0, 0, 96, 9, 64, 1, 0, 0, 0, 0, 0, 0, 19, 4, 83, 237, 234, 82, 74, 188, 6, 23, 0,
2223        225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 196, 178, 112, 86, 5, 59, 97, 219, 15, 0,
2224        0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 188, 210, 59, 150, 246,
2225        167, 182, 213, 33, 0, 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0,
2226        132, 194, 47, 23, 175, 46, 78, 138, 23, 0, 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128,
2227        0, 0, 0, 0, 0, 0, 136, 219, 85, 240, 191, 125, 172, 233, 10, 0, 0, 0, 6, 23, 0, 225, 138,
2228        0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 80, 211, 212, 44, 191, 227, 124, 40, 13, 0, 0, 0,
2229        6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 77, 13, 19, 149, 102, 140, 134,
2230        77, 16, 0, 0, 0, 4, 83, 237, 239, 254, 225, 138, 4, 63, 87, 214, 254, 225, 138, 4, 83, 236,
2231        159, 254, 225, 138, 4, 63, 87, 46, 254, 225, 138, 4, 63, 87, 56, 128, 225, 138, 4, 63, 87,
2232        56, 123, 225, 138, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255,
2233        255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255,
2234        255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 0, 0, 0, 0, 0, 0, 0,
2235        0, 0, 0, 0, 0, 16, 151, 56, 146, 0, 0, 72, 0, 0, 0, 0, 0, 16, 151, 56, 146,
2236    ];
2237    // 1 frames Connected ping - unreliable
2238    let p3 = [132, 3, 0, 0, 0, 0, 72, 0, 0, 0, 0, 0, 16, 151, 56, 161];
2239    // 1 frames game packet - reliable ordered [is_fragment reliable_frame_index = 2 ordered_frame_index = 1 ]
2240    let p4 = [
2241        140, 4, 0, 0, 112, 44, 192, 2, 0, 0, 1, 0, 0, 0, 0, 0, 0, 19, 0, 0, 0, 0, 0, 0, 254, 236,
2242        189, 203, 114, 234, 202, 214, 239, 249, 157, 94, 85, 61, 198, 119, 186, 181, 79, 72, 2,
2243        188, 77, 245, 140, 145, 48, 24, 137, 137, 208, 5, 169, 162, 98, 7, 32, 246, 4, 36, 97, 77,
2244        155, 201, 173, 162, 158, 231, 188, 81, 53, 171, 89, 47, 80, 189, 211, 170, 148, 148, 194,
2245        164, 156, 96, 16, 248, 178, 214, 252, 55, 126, 177, 194, 107, 122, 88, 82, 94, 198, 200,
2246        113, 201, 204, 255, 231, 127, 252, 215, 255, 242, 31, 255, 241, 95, 254, 223, 255, 251,
2247        127, 252, 215, 255, 254, 63, 255, 199, 127, 252, 159, 255, 57, 154, 12, 166, 243, 255, 252,
2248        223, 254, 247, 255, 28, 111, 90, 147, 97, 99, 52, 237, 76, 91, 138, 185, 85, 203, 218, 125,
2249        243, 165, 57, 255, 41, 122, 189, 230, 77, 211, 23, 155, 118, 223, 83, 186, 86, 208, 236,
2250        73, 213, 65, 175, 31, 253, 82, 101, 173, 102, 90, 65, 163, 183, 213, 21, 221, 84, 234, 61,
2251        83, 119, 93, 65, 81, 172, 70, 107, 233, 25, 138, 210, 179, 149, 23, 79, 18, 205, 142, 41,
2252        8, 163, 134, 179, 177, 77, 87, 182, 131, 86, 121, 80, 106, 213, 135, 125, 179, 162, 139,
2253        122, 203, 241, 131, 138, 109, 4, 93, 51, 168, 173, 135, 74, 75, 29, 214, 29, 97, 104, 4,
2254        55, 222, 182, 86, 26, 205, 189, 242, 192, 127, 217, 90, 219, 174, 48, 40, 233, 115, 171,
2255        49, 105, 141, 109, 189, 63, 182, 91, 63, 70, 141, 167, 237, 120, 166, 148, 213, 173, 92,
2256        30, 110, 149, 78, 91, 12, 66, 93, 241, 5, 93, 208, 45, 45, 156, 56, 143, 130, 18, 90, 194,
2257        147, 104, 55, 116, 209, 8, 163, 103, 211, 94, 151, 134, 91, 189, 50, 174, 235, 191, 135,
2258        138, 213, 237, 218, 21, 211, 182, 252, 233, 191, 187, 79, 255, 141, 124, 247, 204, 237,
2259        183, 132, 129, 237, 70, 142, 164, 8, 174, 169, 136, 94, 99, 178, 28, 133, 129, 48, 38, 223,
2260        238, 61, 180, 68, 183, 183, 154, 186, 253, 201, 170, 57, 123, 90, 107, 179, 110, 169, 83,
2261        119, 214, 154, 33, 175, 219, 247, 173, 200, 109, 88, 191, 189, 6, 249, 93, 171, 38, 58,
2262        225, 58, 114, 132, 69, 48, 62, 179, 205, 58, 178, 165, 14, 164, 160, 60, 54, 215, 51, 79,
2263        90, 15, 70, 243, 192, 50, 109, 77, 84, 45, 93, 50, 229, 234, 162, 103, 180, 238, 181, 146,
2264        235, 116, 234, 218, 111, 183, 81, 233, 155, 129, 53, 177, 27, 66, 73, 123, 240, 26, 110,
2265        40, 139, 238, 180, 250, 226, 137, 74, 199, 106, 184, 27, 67, 113, 155, 142, 209, 234, 14,
2266        109, 235, 247, 72, 246, 90, 154, 31, 61, 245, 76, 209, 234, 133, 74, 223, 158, 183, 126,
2267        13, 77, 241, 87, 199, 168, 117, 134, 194, 162, 163, 7, 90, 167, 59, 183, 218, 110, 67, 40,
2268        143, 130, 32, 178, 31, 180, 208, 233, 63, 109, 123, 91, 85, 26, 223, 223, 174, 45, 163, 41,
2269        245, 30, 106, 15, 170, 18, 149, 123, 155, 170, 173, 90, 206, 148, 124, 243, 111, 39, 116,
2270        166, 157, 153, 44, 105, 245, 145, 212, 169, 255, 148, 52, 67, 169, 222, 255, 252, 111, 237,
2271        167, 127, 74, 191, 234, 254, 100, 169, 140, 42, 171, 201, 92, 173, 142, 252, 127, 77, 86,
2272        247, 77, 251, 95, 254, 175, 127, 215, 127, 223, 246, 42, 255, 26, 253, 235, 165, 108, 133,
2273        255, 246, 151, 203, 202, 211, 250, 151, 175, 12, 203, 193, 124, 221, 155, 255, 67, 126,
2274        190, 109, 253, 99, 242, 243, 95, 211, 167, 219, 187, 237, 157, 107, 254, 43, 24, 107, 81,
2275        79, 22, 171, 225, 38, 252, 183, 240, 24, 4, 255, 114, 31, 21, 213, 235, 11, 195, 81, 239,
2276        159, 253, 161, 97, 169, 222, 77, 215, 41, 173, 164, 231, 254, 68, 157, 151, 199, 131, 31,
2277        15, 51, 229, 63, 255, 215, 120, 20, 151, 181, 190, 25, 143, 98, 173, 167, 4, 37, 210, 202,
2278        110, 79, 94, 44, 109, 63, 184, 25, 204, 106, 245, 174, 165, 184, 186, 255, 34, 232, 166,
2279        85, 235, 10, 129, 108, 219, 94, 77, 55, 38, 138, 209, 88, 68, 227, 135, 64, 117, 74, 222,
2280        11, 105, 165, 138, 101, 41, 51, 50, 138, 77, 47, 168, 253, 24, 154, 254, 166, 75, 250, 197,
2281        174, 255, 220, 14, 67, 253, 183, 37, 182, 44, 199, 154, 44, 180, 173, 94, 210, 67, 119,
2282        162, 217, 238, 168, 45, 173, 75, 166, 89, 49, 221, 121, 75, 181, 252, 201, 68, 13, 106, 11,
2283        215, 174, 172, 116, 193, 151, 122, 225, 164, 101, 88, 214, 163, 99, 121, 131, 81, 24, 45,
2284        12, 251, 169, 162, 202, 149, 101, 79, 168, 52, 76, 161, 114, 239, 153, 11, 127, 104, 76,
2285        54, 182, 29, 56, 35, 201, 157, 140, 103, 90, 91, 221, 186, 229, 126, 190, 7, 196, 213, 210,
2286        154, 41, 118, 115, 186, 154, 58, 246, 122, 78, 70, 227, 84, 183, 212, 109, 167, 222, 37,
2287        35, 57, 30, 200, 156, 142, 145, 201, 36, 38, 19, 59, 212, 151, 67, 179, 178, 28, 134, 90,
2288        64, 196, 74, 234, 204, 41, 107, 51, 191, 172, 197, 19, 125, 214, 20, 200, 127, 5, 242, 251,
2289        66, 60, 160, 71, 37, 53, 105, 190, 161, 20, 17, 185, 81, 220, 225, 193, 248, 225, 46, 251,
2290        187, 228, 247, 101, 81, 141, 255, 46, 249, 91, 163, 185, 30, 185, 97, 48, 115, 250, 122,
2291        208, 237, 91, 194, 160, 81, 221, 12, 250, 122, 165, 57, 139, 132, 209, 220, 10, 226, 191,
2292        231, 244, 187, 251, 239, 84, 74, 101, 3, 159, 52, 143, 16, 255, 174, 249, 96, 77, 135, 141,
2293        96, 214, 147, 172, 74, 252, 73, 134, 57, 113, 61, 193, 170, 217, 230, 164, 61, 20, 163,
2294        214, 56, 124, 90, 117, 5, 69, 39, 10, 164, 173, 201, 150, 210, 53, 181, 150, 174, 4, 243,
2295        174, 105, 173, 134, 134, 86, 215, 238, 23, 45, 53, 208, 156, 81, 99, 97, 116, 172, 213,
2296        114, 40, 142, 54, 174, 229, 149, 188, 185, 190, 238, 9, 81, 89, 51, 69, 215, 182, 38, 134,
2297        107, 212, 108, 93, 146, 203, 35, 210, 197, 158, 111, 13, 6, 194, 122, 57, 18, 91, 226, 64,
2298        138, 106, 166, 160, 77, 6, 210, 66, 232, 17, 165, 52, 242, 29, 113, 220, 183, 212, 145,
2299        229, 170, 158, 82, 171, 140, 251, 110, 191, 107, 47, 38, 93, 59, 90, 143, 103, 147, 173,
2300        166, 232, 237, 158, 229, 106, 221, 112, 210, 115, 231, 234, 115, 135, 76, 126, 199, 22,
2301        213, 113, 99, 210, 28, 62, 68, 79, 182, 226, 174, 201, 179, 230, 214, 67, 107, 101, 217,
2302        238, 141, 19, 40, 125, 67, 80, 227, 46, 50, 255, 61, 31, 119, 199, 195, 137, 182, 49, 141,
2303        167, 40, 108, 255, 238, 169, 189, 127, 140, 171, 255, 124, 232, 206, 102, 155, 95, 102,
2304        235, 95, 255, 186, 123, 12, 122, 225, 116, 34, 213, 131, 141, 222, 105, 253, 90, 180, 199,
2305        255, 248, 167, 80, 119, 126, 215, 36, 177, 94, 173, 53, 167, 139, 142, 242, 115, 48, 155,
2306        141, 204, 91, 219, 239, 205, 31, 154, 119, 198, 248, 246, 121, 118, 59, 239, 135, 146, 166,
2307        248, 255, 26, 119, 234, 99, 233, 229, 241, 103, 125, 182, 157, 107, 178, 225, 63, 85, 158,
2308        22, 97, 224, 222, 254, 187, 62, 115, 111, 170, 139, 162, 19, 164, 95, 91, 168, 130, 42, 60,
2309        10, 254, 198, 20, 39, 171, 129, 168, 86, 250, 247, 213, 165, 181, 109, 5, 86, 201, 147,
2310        188, 7, 165, 221, 35, 166, 194, 176, 2, 215, 86, 180, 64, 85, 220, 7, 199, 152, 108, 137,
2311        42, 147, 116, 43, 122, 54, 26, 213, 173, 57, 183, 158, 7, 190, 98, 116, 37, 229, 153, 168,
2312        243, 182, 37, 147, 47, 156, 89, 21, 207, 92, 175, 45, 127, 45, 152, 15, 65, 197, 11, 188,
2313        218, 64, 82, 106, 131, 185, 114, 211, 121, 80, 5, 75, 94, 180, 44, 95, 188, 31, 40, 173,
2314        112, 180, 121, 41, 147, 209, 50, 25, 154, 235, 242, 64, 158, 188, 140, 137, 154, 183, 230,
2315        74, 160, 54, 60, 115, 52, 175, 89, 238, 60, 154, 154, 150, 247, 163, 187,
2316    ];
2317    // 1 frames Incompatible Protocol(extract data?) - reliable ordered [reliable_frame_index = 1 ordered_frame_index = 0] == p2
2318    let p5 = [
2319        140, 5, 0, 0, 96, 9, 64, 1, 0, 0, 0, 0, 0, 0, 19, 4, 83, 237, 234, 82, 74, 188, 6, 23, 0,
2320        225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 196, 178, 112, 86, 5, 59, 97, 219, 15, 0,
2321        0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 188, 210, 59, 150, 246,
2322        167, 182, 213, 33, 0, 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0,
2323        132, 194, 47, 23, 175, 46, 78, 138, 23, 0, 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128,
2324        0, 0, 0, 0, 0, 0, 136, 219, 85, 240, 191, 125, 172, 233, 10, 0, 0, 0, 6, 23, 0, 225, 138,
2325        0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 80, 211, 212, 44, 191, 227, 124, 40, 13, 0, 0, 0,
2326        6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 77, 13, 19, 149, 102, 140, 134,
2327        77, 16, 0, 0, 0, 4, 83, 237, 239, 254, 225, 138, 4, 63, 87, 214, 254, 225, 138, 4, 83, 236,
2328        159, 254, 225, 138, 4, 63, 87, 46, 254, 225, 138, 4, 63, 87, 56, 128, 225, 138, 4, 63, 87,
2329        56, 123, 225, 138, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255,
2330        255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255,
2331        255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 0, 0, 0, 0, 0, 0, 0,
2332        0, 0, 0, 0, 0, 16, 151, 56, 146,
2333    ];
2334    // 1 frames game packet - reliable ordered [is_fragment reliable_frame_index = 2 ordered_frame_index = 1 ] == p4
2335    let p6 = [
2336        140, 6, 0, 0, 112, 44, 192, 2, 0, 0, 1, 0, 0, 0, 0, 0, 0, 19, 0, 0, 0, 0, 0, 0, 254, 236,
2337        189, 203, 114, 234, 202, 214, 239, 249, 157, 94, 85, 61, 198, 119, 186, 181, 79, 72, 2,
2338        188, 77, 245, 140, 145, 48, 24, 137, 137, 208, 5, 169, 162, 98, 7, 32, 246, 4, 36, 97, 77,
2339        155, 201, 173, 162, 158, 231, 188, 81, 53, 171, 89, 47, 80, 189, 211, 170, 148, 148, 194,
2340        164, 156, 96, 16, 248, 178, 214, 252, 55, 126, 177, 194, 107, 122, 88, 82, 94, 198, 200,
2341        113, 201, 204, 255, 231, 127, 252, 215, 255, 242, 31, 255, 241, 95, 254, 223, 255, 251,
2342        127, 252, 215, 255, 254, 63, 255, 199, 127, 252, 159, 255, 57, 154, 12, 166, 243, 255, 252,
2343        223, 254, 247, 255, 28, 111, 90, 147, 97, 99, 52, 237, 76, 91, 138, 185, 85, 203, 218, 125,
2344        243, 165, 57, 255, 41, 122, 189, 230, 77, 211, 23, 155, 118, 223, 83, 186, 86, 208, 236,
2345        73, 213, 65, 175, 31, 253, 82, 101, 173, 102, 90, 65, 163, 183, 213, 21, 221, 84, 234, 61,
2346        83, 119, 93, 65, 81, 172, 70, 107, 233, 25, 138, 210, 179, 149, 23, 79, 18, 205, 142, 41,
2347        8, 163, 134, 179, 177, 77, 87, 182, 131, 86, 121, 80, 106, 213, 135, 125, 179, 162, 139,
2348        122, 203, 241, 131, 138, 109, 4, 93, 51, 168, 173, 135, 74, 75, 29, 214, 29, 97, 104, 4,
2349        55, 222, 182, 86, 26, 205, 189, 242, 192, 127, 217, 90, 219, 174, 48, 40, 233, 115, 171,
2350        49, 105, 141, 109, 189, 63, 182, 91, 63, 70, 141, 167, 237, 120, 166, 148, 213, 173, 92,
2351        30, 110, 149, 78, 91, 12, 66, 93, 241, 5, 93, 208, 45, 45, 156, 56, 143, 130, 18, 90, 194,
2352        147, 104, 55, 116, 209, 8, 163, 103, 211, 94, 151, 134, 91, 189, 50, 174, 235, 191, 135,
2353        138, 213, 237, 218, 21, 211, 182, 252, 233, 191, 187, 79, 255, 141, 124, 247, 204, 237,
2354        183, 132, 129, 237, 70, 142, 164, 8, 174, 169, 136, 94, 99, 178, 28, 133, 129, 48, 38, 223,
2355        238, 61, 180, 68, 183, 183, 154, 186, 253, 201, 170, 57, 123, 90, 107, 179, 110, 169, 83,
2356        119, 214, 154, 33, 175, 219, 247, 173, 200, 109, 88, 191, 189, 6, 249, 93, 171, 38, 58,
2357        225, 58, 114, 132, 69, 48, 62, 179, 205, 58, 178, 165, 14, 164, 160, 60, 54, 215, 51, 79,
2358        90, 15, 70, 243, 192, 50, 109, 77, 84, 45, 93, 50, 229, 234, 162, 103, 180, 238, 181, 146,
2359        235, 116, 234, 218, 111, 183, 81, 233, 155, 129, 53, 177, 27, 66, 73, 123, 240, 26, 110,
2360        40, 139, 238, 180, 250, 226, 137, 74, 199, 106, 184, 27, 67, 113, 155, 142, 209, 234, 14,
2361        109, 235, 247, 72, 246, 90, 154, 31, 61, 245, 76, 209, 234, 133, 74, 223, 158, 183, 126,
2362        13, 77, 241, 87, 199, 168, 117, 134, 194, 162, 163, 7, 90, 167, 59, 183, 218, 110, 67, 40,
2363        143, 130, 32, 178, 31, 180, 208, 233, 63, 109, 123, 91, 85, 26, 223, 223, 174, 45, 163, 41,
2364        245, 30, 106, 15, 170, 18, 149, 123, 155, 170, 173, 90, 206, 148, 124, 243, 111, 39, 116,
2365        166, 157, 153, 44, 105, 245, 145, 212, 169, 255, 148, 52, 67, 169, 222, 255, 252, 111, 237,
2366        167, 127, 74, 191, 234, 254, 100, 169, 140, 42, 171, 201, 92, 173, 142, 252, 127, 77, 86,
2367        247, 77, 251, 95, 254, 175, 127, 215, 127, 223, 246, 42, 255, 26, 253, 235, 165, 108, 133,
2368        255, 246, 151, 203, 202, 211, 250, 151, 175, 12, 203, 193, 124, 221, 155, 255, 67, 126,
2369        190, 109, 253, 99, 242, 243, 95, 211, 167, 219, 187, 237, 157, 107, 254, 43, 24, 107, 81,
2370        79, 22, 171, 225, 38, 252, 183, 240, 24, 4, 255, 114, 31, 21, 213, 235, 11, 195, 81, 239,
2371        159, 253, 161, 97, 169, 222, 77, 215, 41, 173, 164, 231, 254, 68, 157, 151, 199, 131, 31,
2372        15, 51, 229, 63, 255, 215, 120, 20, 151, 181, 190, 25, 143, 98, 173, 167, 4, 37, 210, 202,
2373        110, 79, 94, 44, 109, 63, 184, 25, 204, 106, 245, 174, 165, 184, 186, 255, 34, 232, 166,
2374        85, 235, 10, 129, 108, 219, 94, 77, 55, 38, 138, 209, 88, 68, 227, 135, 64, 117, 74, 222,
2375        11, 105, 165, 138, 101, 41, 51, 50, 138, 77, 47, 168, 253, 24, 154, 254, 166, 75, 250, 197,
2376        174, 255, 220, 14, 67, 253, 183, 37, 182, 44, 199, 154, 44, 180, 173, 94, 210, 67, 119,
2377        162, 217, 238, 168, 45, 173, 75, 166, 89, 49, 221, 121, 75, 181, 252, 201, 68, 13, 106, 11,
2378        215, 174, 172, 116, 193, 151, 122, 225, 164, 101, 88, 214, 163, 99, 121, 131, 81, 24, 45,
2379        12, 251, 169, 162, 202, 149, 101, 79, 168, 52, 76, 161, 114, 239, 153, 11, 127, 104, 76,
2380        54, 182, 29, 56, 35, 201, 157, 140, 103, 90, 91, 221, 186, 229, 126, 190, 7, 196, 213, 210,
2381        154, 41, 118, 115, 186, 154, 58, 246, 122, 78, 70, 227, 84, 183, 212, 109, 167, 222, 37,
2382        35, 57, 30, 200, 156, 142, 145, 201, 36, 38, 19, 59, 212, 151, 67, 179, 178, 28, 134, 90,
2383        64, 196, 74, 234, 204, 41, 107, 51, 191, 172, 197, 19, 125, 214, 20, 200, 127, 5, 242, 251,
2384        66, 60, 160, 71, 37, 53, 105, 190, 161, 20, 17, 185, 81, 220, 225, 193, 248, 225, 46, 251,
2385        187, 228, 247, 101, 81, 141, 255, 46, 249, 91, 163, 185, 30, 185, 97, 48, 115, 250, 122,
2386        208, 237, 91, 194, 160, 81, 221, 12, 250, 122, 165, 57, 139, 132, 209, 220, 10, 226, 191,
2387        231, 244, 187, 251, 239, 84, 74, 101, 3, 159, 52, 143, 16, 255, 174, 249, 96, 77, 135, 141,
2388        96, 214, 147, 172, 74, 252, 73, 134, 57, 113, 61, 193, 170, 217, 230, 164, 61, 20, 163,
2389        214, 56, 124, 90, 117, 5, 69, 39, 10, 164, 173, 201, 150, 210, 53, 181, 150, 174, 4, 243,
2390        174, 105, 173, 134, 134, 86, 215, 238, 23, 45, 53, 208, 156, 81, 99, 97, 116, 172, 213,
2391        114, 40, 142, 54, 174, 229, 149, 188, 185, 190, 238, 9, 81, 89, 51, 69, 215, 182, 38, 134,
2392        107, 212, 108, 93, 146, 203, 35, 210, 197, 158, 111, 13, 6, 194, 122, 57, 18, 91, 226, 64,
2393        138, 106, 166, 160, 77, 6, 210, 66, 232, 17, 165, 52, 242, 29, 113, 220, 183, 212, 145,
2394        229, 170, 158, 82, 171, 140, 251, 110, 191, 107, 47, 38, 93, 59, 90, 143, 103, 147, 173,
2395        166, 232, 237, 158, 229, 106, 221, 112, 210, 115, 231, 234, 115, 135, 76, 126, 199, 22,
2396        213, 113, 99, 210, 28, 62, 68, 79, 182, 226, 174, 201, 179, 230, 214, 67, 107, 101, 217,
2397        238, 141, 19, 40, 125, 67, 80, 227, 46, 50, 255, 61, 31, 119, 199, 195, 137, 182, 49, 141,
2398        167, 40, 108, 255, 238, 169, 189, 127, 140, 171, 255, 124, 232, 206, 102, 155, 95, 102,
2399        235, 95, 255, 186, 123, 12, 122, 225, 116, 34, 213, 131, 141, 222, 105, 253, 90, 180, 199,
2400        255, 248, 167, 80, 119, 126, 215, 36, 177, 94, 173, 53, 167, 139, 142, 242, 115, 48, 155,
2401        141, 204, 91, 219, 239, 205, 31, 154, 119, 198, 248, 246, 121, 118, 59, 239, 135, 146, 166,
2402        248, 255, 26, 119, 234, 99, 233, 229, 241, 103, 125, 182, 157, 107, 178, 225, 63, 85, 158,
2403        22, 97, 224, 222, 254, 187, 62, 115, 111, 170, 139, 162, 19, 164, 95, 91, 168, 130, 42, 60,
2404        10, 254, 198, 20, 39, 171, 129, 168, 86, 250, 247, 213, 165, 181, 109, 5, 86, 201, 147,
2405        188, 7, 165, 221, 35, 166, 194, 176, 2, 215, 86, 180, 64, 85, 220, 7, 199, 152, 108, 137,
2406        42, 147, 116, 43, 122, 54, 26, 213, 173, 57, 183, 158, 7, 190, 98, 116, 37, 229, 153, 168,
2407        243, 182, 37, 147, 47, 156, 89, 21, 207, 92, 175, 45, 127, 45, 152, 15, 65, 197, 11, 188,
2408        218, 64, 82, 106, 131, 185, 114, 211, 121, 80, 5, 75, 94, 180, 44, 95, 188, 31, 40, 173,
2409        112, 180, 121, 41, 147, 209, 50, 25, 154, 235, 242, 64, 158, 188, 140, 137, 154, 183, 230,
2410        74, 160, 54, 60, 115, 52, 175, 89, 238, 60, 154, 154, 150, 247, 163, 187,
2411    ];
2412    let ps: Vec<Vec<u8>> = vec![
2413        p0.to_vec(),
2414        p1.to_vec(),
2415        p2.to_vec(),
2416        p3.to_vec(),
2417        p4.to_vec(),
2418        p5.to_vec(),
2419        p6.to_vec(),
2420    ];
2421
2422    let mut n = 0;
2423
2424    let mut rq = RecvQ::new();
2425    for i in ps {
2426        let v = FrameVec::new(&i).unwrap();
2427        for i in v.frames {
2428            rq.insert(i).unwrap();
2429            if !rq.flush(&"0.0.0.0:0".parse().unwrap()).is_empty() {
2430                n += 1;
2431            }
2432        }
2433    }
2434
2435    // p1 retransmits p0 with a new datagram ID but the same reliable frame ID.
2436    assert_eq!(n, 4);
2437}
2438
2439#[cfg(test)]
2440#[path = "arq_tests.rs"]
2441mod regressions;
2442
2443#[test]
2444fn rejects_messages_exceeding_the_receivers_fragment_limit() {
2445    let sendq = SendQ::new(576);
2446    let limit = crate::fragment::MAX_FRAGMENTS * (576 - 60);
2447    assert!(
2448        sendq
2449            .required_bytes(Reliability::ReliableOrdered, limit)
2450            .is_ok()
2451    );
2452    assert!(matches!(
2453        sendq.required_bytes(Reliability::ReliableOrdered, limit + 1),
2454        Err(RaknetError::PacketSizeExceedMTU)
2455    ));
2456}
2457
2458#[test]
2459fn exact_mtu_payload_is_not_marked_as_split() {
2460    let mut sendq = SendQ::new(576);
2461    let payload = vec![0xfe; 516];
2462    sendq
2463        .insert(Reliability::ReliableOrdered, &payload)
2464        .unwrap();
2465    let frames = sendq.flush(0, &"127.0.0.1:19132".parse().unwrap());
2466    assert_eq!(frames.len(), 1);
2467    assert_eq!(frames[0].flags & 16, 0);
2468    assert_eq!(frames[0].data.as_ref(), payload);
2469    sendq.ack(frames[0].sequence_number, 1);
2470    assert!(sendq.is_empty());
2471    assert_eq!(sendq.buffered_bytes, 0);
2472}
2473
2474#[cfg(test)]
2475mod allocation_regressions {
2476    use super::*;
2477
2478    #[test]
2479    fn frame_decoder_reuses_storage_and_discards_previous_frames() {
2480        let packet = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe; 800]);
2481        let encoded = packet.serialize().unwrap();
2482        let mut frames = Vec::new();
2483        FrameVec::decode_into(&encoded, &mut frames).unwrap();
2484        let capacity = frames.capacity();
2485        assert_eq!(frames[0].data.as_ref(), &[0xfe; 800]);
2486        frames.clear();
2487        FrameVec::decode_into(&encoded, &mut frames).unwrap();
2488        assert_eq!(frames.len(), 1);
2489        assert_eq!(frames.capacity(), capacity);
2490        assert!(FrameVec::decode_into(&[0x84], &mut frames).is_err());
2491        assert!(frames.is_empty());
2492    }
2493
2494    #[test]
2495    fn ordered_delivery_retains_the_receive_queue_allocation() {
2496        let mut queue = RecvQ::new();
2497        let mut delivered = Vec::new();
2498        for index in 0..16 {
2499            let mut frame = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe]);
2500            frame.sequence_number = index;
2501            frame.reliable_frame_index = index;
2502            frame.ordered_frame_index = index;
2503            queue.insert(frame).unwrap();
2504            queue.flush_into(&mut delivered);
2505            assert_eq!(delivered.len(), 1);
2506            delivered.clear();
2507            assert!(queue.ready_ordered.capacity() > 0);
2508        }
2509    }
2510}
2511
2512#[cfg(test)]
2513mod serialization_reuse_tests {
2514    use super::*;
2515
2516    #[test]
2517    fn reused_fragment_buffer_preserves_the_wire_format() {
2518        let mut frame = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe; 16]);
2519        frame.flags |= 16;
2520        frame.sequence_number = 0xabcdef;
2521        frame.reliable_frame_index = 0x123456;
2522        frame.ordered_frame_index = 0x654321;
2523        frame.order_channel = 7;
2524        frame.compound_size = 3;
2525        frame.compound_id = 0x1234;
2526        frame.fragment_index = 1;
2527        let mut expected = vec![
2528            0x8c, 0xef, 0xcd, 0xab, 0x70, 0, 0x80, 0x56, 0x34, 0x12, 0x21, 0x43, 0x65, 7, 0, 0, 0,
2529            3, 0x12, 0x34, 0, 0, 0, 1,
2530        ];
2531        expected.extend_from_slice(&[0xfe; 16]);
2532        let mut buffer = Vec::with_capacity(128);
2533        buffer.extend_from_slice(&[0xff; 64]);
2534        for _ in 0..100 {
2535            frame.serialize_into(&mut buffer).unwrap();
2536            assert_eq!(buffer, expected);
2537            assert_eq!(buffer.capacity(), 128);
2538        }
2539        frame.compound_size = 0;
2540        assert!(frame.serialize_into(&mut buffer).is_err());
2541        assert_eq!(buffer, expected);
2542    }
2543}
2544
2545#[cfg(test)]
2546mod shared_fragment_tests {
2547    use super::*;
2548
2549    #[test]
2550    fn retransmitted_fragment_accepts_old_ack_without_releasing_the_budget_twice() {
2551        let peer = "127.0.0.1:19132".parse().unwrap();
2552        let mut queue = SendQ::new(1400);
2553        queue
2554            .insert(Reliability::ReliableOrdered, &[0xfe; 4096])
2555            .unwrap();
2556        let frames = queue.flush(0, &peer);
2557        let old_id = frames[3].sequence_number;
2558        queue.nack(old_id, 1);
2559        let retry = queue.flush(1, &peer);
2560        assert_eq!(retry.len(), 1);
2561        let new_id = retry[0].sequence_number;
2562        queue.ack(old_id, 2);
2563        let budget = queue.buffered_bytes;
2564        queue.ack(old_id, 2);
2565        queue.ack(new_id, 2);
2566        assert_eq!(queue.buffered_bytes, budget);
2567        for frame in &frames[..3] {
2568            queue.ack(frame.sequence_number, 2);
2569        }
2570        assert_eq!(queue.buffered_bytes, 0);
2571    }
2572
2573    #[test]
2574    fn fragment_slices_share_one_allocation_and_keep_its_budget_until_final_ack() {
2575        let peer = "127.0.0.1:19132".parse().unwrap();
2576        let mut queue = SendQ::new(1400);
2577        let payload = vec![0xfe; 4096];
2578        queue
2579            .insert(Reliability::ReliableOrdered, &payload)
2580            .unwrap();
2581        let frames = queue.flush(0, &peer);
2582        assert_eq!(frames.len(), 4);
2583        let owner = frames[0].shared_payload.as_ref().unwrap();
2584        for frame in &frames {
2585            assert!(Arc::ptr_eq(owner, frame.shared_payload.as_ref().unwrap()));
2586        }
2587        let restored: Vec<_> = frames
2588            .iter()
2589            .flat_map(|frame| frame.data.iter().copied())
2590            .collect();
2591        assert_eq!(restored, payload);
2592        let initial = queue.buffered_bytes;
2593        for (index, frame) in frames.iter().enumerate() {
2594            queue.ack(frame.sequence_number, 1);
2595            queue.ack(frame.sequence_number, 1);
2596            if index + 1 < frames.len() {
2597                assert_eq!(
2598                    queue.buffered_bytes,
2599                    initial - (index + 1) * SendQ::FRAME_BUDGET
2600                );
2601            }
2602        }
2603        assert_eq!(queue.buffered_bytes, 0);
2604        assert_eq!(owner.remaining.load(Ordering::Relaxed), 0);
2605    }
2606}
2607
2608#[cfg(test)]
2609mod flight_order_tests {
2610    use super::*;
2611
2612    #[test]
2613    fn retries_and_sequence_wrap_keep_the_existing_numeric_order() {
2614        let peer = "127.0.0.1:19132".parse().unwrap();
2615        let mut queue = SendQ::new(1400);
2616        queue.sequence_number = sequence::MASK;
2617        queue.insert(Reliability::ReliableOrdered, &[0xfe]).unwrap();
2618        queue.insert(Reliability::ReliableOrdered, &[0xfe]).unwrap();
2619        let initial = queue.flush(0, &peer);
2620        assert_eq!(
2621            initial
2622                .iter()
2623                .map(|frame| frame.sequence_number)
2624                .collect::<Vec<_>>(),
2625            [sequence::MASK, 0]
2626        );
2627        assert!(queue.flight_order_dirty);
2628        assert!(queue.flush(1, &peer).is_empty());
2629        assert_eq!(
2630            queue
2631                .sent_packet
2632                .iter()
2633                .map(|packet| packet.0.sequence_number)
2634                .collect::<Vec<_>>(),
2635            [0, sequence::MASK]
2636        );
2637        assert!(!queue.flight_order_dirty);
2638        queue.nack(sequence::MASK, 2);
2639        assert!(queue.flight_order_dirty);
2640        let retry = queue.flush(2, &peer);
2641        assert_eq!(retry.len(), 1);
2642        assert_eq!(retry[0].sequence_number, 1);
2643        assert!(!queue.flight_order_dirty);
2644        assert_eq!(
2645            queue
2646                .sent_packet
2647                .iter()
2648                .map(|packet| packet.0.sequence_number)
2649                .collect::<Vec<_>>(),
2650            [0, 1]
2651        );
2652        queue.ack(sequence::MASK, 3);
2653        queue.ack(0, 3);
2654        assert!(queue.is_empty());
2655        assert_eq!(queue.buffered_bytes, 0);
2656    }
2657}
2658
2659#[cfg(test)]
2660mod owned_decode_tests {
2661    use super::*;
2662
2663    #[test]
2664    fn a_single_frame_moves_the_udp_allocation_into_its_payload() {
2665        let frame = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe; 800]);
2666        let packet = frame.serialize().unwrap();
2667        let pointer = packet.as_ptr();
2668        let mut decoded = Vec::new();
2669        FrameVec::decode_owned_into(packet, &mut decoded).unwrap();
2670        assert_eq!(decoded.len(), 1);
2671        assert_eq!(decoded[0].data.as_ptr(), pointer);
2672        assert_eq!(decoded[0].data.as_ref(), frame.data.as_ref());
2673        let data: Vec<u8> = decoded.pop().unwrap().data.into();
2674        assert_eq!(data.as_ptr(), pointer);
2675    }
2676
2677    #[test]
2678    fn multiple_frames_and_invalid_lengths_keep_the_borrowed_decoder_contract() {
2679        let first = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe; 800]);
2680        let second = FrameSetPacket::new(Reliability::Unreliable, vec![0xfe; 64]);
2681        let mut packet = first.serialize().unwrap();
2682        packet.extend_from_slice(&second.serialize().unwrap()[4..]);
2683        let mut expected = Vec::new();
2684        FrameVec::decode_into(&packet, &mut expected).unwrap();
2685        let mut actual = Vec::new();
2686        FrameVec::decode_owned_into(packet.clone(), &mut actual).unwrap();
2687        assert_eq!(actual.len(), 2);
2688        for (actual, expected) in actual.iter().zip(&expected) {
2689            assert_eq!(actual.serialize().unwrap(), expected.serialize().unwrap());
2690        }
2691        for length in 0..packet.len() {
2692            let mut borrowed = Vec::new();
2693            let mut owned = Vec::new();
2694            assert_eq!(
2695                FrameVec::decode_into(&packet[..length], &mut borrowed).is_ok(),
2696                FrameVec::decode_owned_into(packet[..length].to_vec(), &mut owned).is_ok()
2697            );
2698        }
2699    }
2700}
2701
2702#[cfg(test)]
2703mod scheduling_regressions {
2704    use super::*;
2705
2706    #[test]
2707    fn cached_deadlines_match_a_full_scan_across_ack_loss_and_wrap() {
2708        let peer = "127.0.0.1:19132".parse().unwrap();
2709        let mut cached = SendQ::new(1400);
2710        cached.sequence_number = sequence::MASK - 80;
2711        let mut scanned = SendQ::new(1400);
2712        scanned.sequence_number = cached.sequence_number;
2713        let mut rng = 0x12345678u32;
2714        for tick in 0..4000 {
2715            rng = rng.wrapping_mul(1664525).wrapping_add(1013904223);
2716            if rng & 3 == 0 {
2717                for queue in [&mut cached, &mut scanned] {
2718                    queue
2719                        .insert(Reliability::ReliableOrdered, &[0xfe, 42])
2720                        .unwrap();
2721                }
2722            }
2723            if !cached.sent_packet.is_empty() && rng & 7 == 1 {
2724                let index = (rng as usize >> 8) % cached.sent_packet.len();
2725                let sequence = cached.sent_packet[index].0.sequence_number;
2726                for queue in [&mut cached, &mut scanned] {
2727                    if rng & 16 == 0 {
2728                        queue.ack(sequence, tick);
2729                    } else {
2730                        queue.nack(sequence, tick);
2731                    }
2732                }
2733            }
2734            // The oracle deliberately does the old full scan on every flush.
2735            scanned.next_retry = 0;
2736            scanned.retries_pending = true;
2737            let actual: Vec<_> = cached
2738                .flush(tick, &peer)
2739                .iter()
2740                .map(|f| f.serialize().unwrap())
2741                .collect();
2742            let expected: Vec<_> = scanned
2743                .flush(tick, &peer)
2744                .iter()
2745                .map(|f| f.serialize().unwrap())
2746                .collect();
2747            assert_eq!(actual, expected, "tick {tick}");
2748            assert_eq!(cached.buffered_bytes, scanned.buffered_bytes);
2749        }
2750    }
2751
2752    #[test]
2753    fn lower_rto_invalidates_the_deadline_and_retries_keep_the_backoff() {
2754        let peer = "127.0.0.1:19132".parse().unwrap();
2755        let mut queue = SendQ::new(1400);
2756        queue.srtt = 133;
2757        queue.rto = 200;
2758        for _ in 0..2 {
2759            queue.insert(Reliability::ReliableOrdered, &[0xfe]).unwrap();
2760        }
2761        queue.flush(0, &peer);
2762        queue.ack(0, 50);
2763        assert_eq!(queue.rto, 174);
2764        assert!(queue.flush(173, &peer).is_empty());
2765        assert_eq!(queue.flush(174, &peer).len(), 1);
2766        assert!(queue.flush(434, &peer).is_empty());
2767        assert_eq!(queue.flush(435, &peer).len(), 1);
2768        queue.ack(1, 436);
2769        assert!(queue.is_empty());
2770    }
2771
2772    #[test]
2773    fn ack_storage_is_recycled_without_replaying_old_ranges() {
2774        let mut set = ACKSet::default();
2775        let mut output = Vec::new();
2776        for index in 0..100 {
2777            set.insert(index);
2778            set.take_ack_into(&mut output);
2779            assert_eq!(output, [(index, index)]);
2780        }
2781        set.take_ack_into(&mut output);
2782        assert!(output.is_empty());
2783        assert!(set.ack.capacity() > 0);
2784    }
2785}
2786
2787#[cfg(test)]
2788mod owned_payload_tests {
2789    use super::*;
2790
2791    #[test]
2792    fn owned_sends_match_borrowed_wire_bytes_and_share_retry_storage() {
2793        let peer = "127.0.0.1:19132".parse().unwrap();
2794        for mode in [
2795            Reliability::Unreliable,
2796            Reliability::UnreliableSequenced,
2797            Reliability::Reliable,
2798            Reliability::ReliableOrdered,
2799            Reliability::ReliableSequenced,
2800        ] {
2801            for size in [64, 800, 4096] {
2802                let data = bytes::Bytes::from(vec![0xfe; size]);
2803                let mut owned = SendQ::new(1400);
2804                let mut borrowed = SendQ::new(1400);
2805                let a = owned.insert_bytes(mode, &data, 7);
2806                let b = borrowed.insert_with_order_channel(mode, &data, 7);
2807                assert_eq!(a.is_ok(), b.is_ok());
2808                if a.is_err() {
2809                    continue;
2810                }
2811                let frames = owned.flush(0, &peer);
2812                let expected = borrowed.flush(0, &peer);
2813                assert_eq!(
2814                    frames
2815                        .iter()
2816                        .map(|f| f.serialize().unwrap())
2817                        .collect::<Vec<_>>(),
2818                    expected
2819                        .iter()
2820                        .map(|f| f.serialize().unwrap())
2821                        .collect::<Vec<_>>()
2822                );
2823                let mut offset = 0;
2824                for frame in &frames {
2825                    assert_eq!(frame.data.as_ptr(), data[offset..].as_ptr());
2826                    offset += frame.data.len();
2827                }
2828                if mode != Reliability::Unreliable && mode != Reliability::UnreliableSequenced {
2829                    owned.nack(frames[0].sequence_number, 1);
2830                    let retry = owned.flush(1, &peer);
2831                    assert_eq!(retry[0].data.as_ptr(), data.as_ptr());
2832                    // Old ACK IDs still release each payload budget once.
2833                    owned.ack_ranges(&[(0, (frames.len() - 1) as u32)], 2);
2834                    assert!(owned.is_empty());
2835                    assert_eq!(owned.buffered_bytes, 0);
2836                }
2837            }
2838        }
2839    }
2840}