uart_fpl/
resend_sp.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! ResendSP Go-Back-N sliding-window transport state machines.
6//!
7//! Provides [`ResendSender`] and [`ResendReceiver`] to encapsulate sliding-window
8//! sequence allocation, in-flight frame buffering, cumulative ACK processing,
9//! round-trip time (RTT) tracking, and retransmission timeout recovery for
10//! reliable transport over noisy or high-latency serial links.
11
12use std::time::{Duration, Instant};
13
14use crate::error::{FrameError, RetransmissionLimitExceeded, UnexpectedSeqError};
15use crate::frame::{FrameType, encode_frame};
16use crate::window::is_seq_in_window;
17
18/// Maximum allowable sliding window size (half the 8-bit sequence space to avoid ambiguity).
19pub const MAX_WINDOW_SIZE: u8 = 128;
20
21/// Total 8-bit sequence space size.
22const SEQUENCE_SPACE: usize = (u8::MAX as usize) + 1;
23
24/// Default sliding window size for ResendSP (number of frames in flight).
25pub const DEFAULT_WINDOW_SIZE: u8 = 64;
26
27/// Default retransmission timeout duration for Go-Back-N recovery (1,000 ms).
28pub const DEFAULT_RETRANSMISSION_TIMEOUT: Duration = Duration::from_millis(1000);
29
30/// Maximum consecutive retransmission timeouts before the transport session fails.
31pub const DEFAULT_MAX_RETRANSMISSION_ATTEMPTS: usize = 60;
32
33/// Outcome of processing an acknowledgment sequence number in the sender.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub enum AckOutcome {
36    /// The ACK fell within the active window `[base, next_seq)` and advanced `base`.
37    Advanced {
38        /// The new base sequence number (one past the acknowledged sequence).
39        new_base: u8,
40        /// The round-trip time measured for the acknowledged frame, if available.
41        rtt: Option<Duration>,
42        /// Whether all outstanding frames in the window have now been acknowledged.
43        all_acked: bool,
44    },
45    /// The ACK sequence fell outside the active window and was ignored.
46    OutOfWindow {
47        /// Current base sequence number.
48        base: u8,
49        /// Current next sequence number.
50        next_seq: u8,
51    },
52}
53
54#[derive(Debug, Clone)]
55struct Slot {
56    frame: Vec<u8>,
57    tx_time: Instant,
58    retransmitted: bool,
59}
60
61/// Go-Back-N sliding-window sender state machine.
62///
63/// Encapsulates sequence allocation, in-flight frame buffering, sliding-window
64/// bounds checks, cumulative ACK processing, round-trip time measurement, and
65/// timeout retransmission gathering.
66#[derive(Debug)]
67pub struct ResendSender {
68    base: u8,
69    next_seq: u8,
70    window_size: u8,
71    max_retransmission_attempts: usize,
72    retransmission_attempts: usize,
73    slots: [Option<Slot>; SEQUENCE_SPACE],
74}
75
76impl Default for ResendSender {
77    fn default() -> Self {
78        Self::new(DEFAULT_WINDOW_SIZE, DEFAULT_MAX_RETRANSMISSION_ATTEMPTS)
79    }
80}
81
82impl ResendSender {
83    /// Creates a new sender state machine with the specified window size and retransmission limit.
84    ///
85    /// # Panics
86    ///
87    /// Panics if `window_size == 0` or `window_size > 128`, or if `max_retransmission_attempts == 0`.
88    pub fn new(window_size: u8, max_retransmission_attempts: usize) -> Self {
89        assert!(
90            window_size > 0 && window_size <= MAX_WINDOW_SIZE,
91            "window_size must be in range 1..=128, got {window_size}"
92        );
93        assert!(
94            max_retransmission_attempts > 0,
95            "max_retransmission_attempts must be greater than 0, got {max_retransmission_attempts}"
96        );
97        const INIT_SLOT: Option<Slot> = None;
98        Self {
99            base: 0,
100            next_seq: 0,
101            window_size,
102            max_retransmission_attempts,
103            retransmission_attempts: 0,
104            slots: [INIT_SLOT; SEQUENCE_SPACE],
105        }
106    }
107
108    /// Returns the current base sequence number (oldest unacknowledged frame).
109    pub fn base(&self) -> u8 {
110        self.base
111    }
112
113    /// Returns the next sequence number to be assigned.
114    pub fn next_seq(&self) -> u8 {
115        self.next_seq
116    }
117
118    /// Returns the number of frames currently in flight.
119    pub fn in_flight(&self) -> u8 {
120        self.next_seq.wrapping_sub(self.base)
121    }
122
123    /// Returns true if another frame can be transmitted within the sliding window.
124    pub fn can_send(&self) -> bool {
125        self.in_flight() < self.window_size
126    }
127
128    /// Returns true if there are no in-flight unacknowledged frames.
129    pub fn is_idle(&self) -> bool {
130        self.base == self.next_seq
131    }
132
133    /// Returns the number of consecutive retransmission attempts made for the current window.
134    pub fn retransmission_attempts(&self) -> usize {
135        self.retransmission_attempts
136    }
137
138    /// Enqueues and encodes a new frame to be transmitted.
139    ///
140    /// # Note
141    /// This method is intended exclusively for sequenced outbound frames (such as
142    /// [`FrameType::Data`]). Pure cumulative acknowledgments ([`FrameType::Ack`]) must
143    /// not be enqueued into the sliding window; emit them directly using [`encode_frame`]
144    /// or coordinate them via [`AckTracker`].
145    ///
146    /// # Errors
147    /// Returns [`FrameError::WindowFull`] if `in_flight() >= window_size`.
148    /// Returns [`FrameError::PayloadTooLarge`] if `payload.len() > MAX_PAYLOAD_SIZE`.
149    pub fn enqueue_frame(
150        &mut self,
151        session_id: u32,
152        channel_id: u16,
153        frame_type: FrameType,
154        payload: &[u8],
155    ) -> Result<(u8, Vec<u8>), FrameError> {
156        if !self.can_send() {
157            return Err(FrameError::WindowFull {
158                window_size: self.window_size,
159                in_flight: self.in_flight(),
160            });
161        }
162        let seq = self.next_seq;
163        let wire_bytes = encode_frame(session_id, channel_id, seq, frame_type, payload)?;
164        self.slots[seq as usize] =
165            Some(Slot { frame: wire_bytes.clone(), tx_time: Instant::now(), retransmitted: false });
166        self.next_seq = self.next_seq.wrapping_add(1);
167        Ok((seq, wire_bytes))
168    }
169
170    /// Processes an incoming cumulative acknowledgment.
171    ///
172    /// If `ack_seq` is within `[base, next_seq)`, all frames from `base` through
173    /// `ack_seq` are removed from the buffer, `base` is advanced to `ack_seq + 1`,
174    /// retransmission attempts are reset to 0, and the elapsed RTT for `ack_seq`
175    /// is returned in `AckOutcome::Advanced`.
176    pub fn handle_ack(&mut self, ack_seq: u8) -> AckOutcome {
177        if is_seq_in_window(self.base, self.next_seq, ack_seq) {
178            let mut curr = self.base;
179            loop {
180                if curr == ack_seq {
181                    break;
182                }
183                self.slots[curr as usize] = None;
184                curr = curr.wrapping_add(1);
185            }
186
187            let rtt = self.slots[ack_seq as usize]
188                .take()
189                .and_then(|s| if !s.retransmitted { Some(s.tx_time.elapsed()) } else { None });
190            self.base = ack_seq.wrapping_add(1);
191            self.retransmission_attempts = 0;
192
193            AckOutcome::Advanced { new_base: self.base, rtt, all_acked: self.base == self.next_seq }
194        } else {
195            AckOutcome::OutOfWindow { base: self.base, next_seq: self.next_seq }
196        }
197    }
198
199    /// Handles a retransmission timer timeout.
200    ///
201    /// Increments `retransmission_attempts`. If the limit is reached, returns
202    /// `Err(RetransmissionLimitExceeded)`. Otherwise, gathers and returns clones of all
203    /// unacknowledged frames from `base` to `next_seq` and updates their transmission
204    /// timestamps.
205    ///
206    /// # Errors
207    /// Returns [`RetransmissionLimitExceeded`] if `retransmission_attempts >= max_retransmission_attempts`.
208    /// Callers should call [`reset()`](Self::reset) before reusing the sender after exceeding the limit.
209    pub fn handle_timeout(&mut self) -> Result<Vec<Vec<u8>>, RetransmissionLimitExceeded> {
210        if self.is_idle() {
211            return Ok(Vec::new());
212        }
213        self.retransmission_attempts += 1;
214        if self.retransmission_attempts >= self.max_retransmission_attempts {
215            return Err(RetransmissionLimitExceeded {
216                base: self.base,
217                next_seq: self.next_seq,
218                attempts: self.retransmission_attempts,
219            });
220        }
221
222        let mut frames = Vec::new();
223        let mut curr = self.base;
224        let now = Instant::now();
225        while curr != self.next_seq {
226            if let Some(slot) = &mut self.slots[curr as usize] {
227                slot.tx_time = now;
228                slot.retransmitted = true;
229                frames.push(slot.frame.clone());
230            }
231            curr = curr.wrapping_add(1);
232        }
233        Ok(frames)
234    }
235
236    /// Resets the sender to an empty initial state.
237    pub fn reset(&mut self) {
238        self.base = 0;
239        self.next_seq = 0;
240        self.retransmission_attempts = 0;
241        for slot in &mut self.slots {
242            *slot = None;
243        }
244    }
245}
246
247/// Status of an incoming frame evaluated against the receiver window.
248#[derive(Debug, Clone, Copy, PartialEq, Eq)]
249pub enum FrameStatus {
250    /// The frame is in order (`seq == expected_seq`).
251    InOrder {
252        /// Sequence number of the received in-order frame.
253        seq: u8,
254    },
255    /// The frame is a duplicate or arrived out of order.
256    OutOfOrder {
257        /// Sequence number of the unexpected frame.
258        seq: u8,
259        /// Sequence number currently expected by the receiver.
260        expected_seq: u8,
261        /// Cumulative ACK to send in response, if any.
262        ack_seq: Option<u8>,
263    },
264}
265
266/// Go-Back-N receiver state machine.
267///
268/// In Go-Back-N ARQ, frames arriving out of order are not queued or reordered; they are
269/// immediately dropped and prompt a cumulative ACK of the latest contiguous in-order frame.
270///
271/// This state machine provides two-phase sequence progression ([`inspect`](Self::inspect) and
272/// [`advance_in_order`](Self::advance_in_order)) to allow transport drivers to apply backpressure.
273#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
274pub struct ResendReceiver {
275    expected_seq: u8,
276    last_acked: Option<u8>,
277}
278
279impl ResendReceiver {
280    /// Creates a new receiver state machine with `expected_seq = 0`.
281    pub fn new() -> Self {
282        Self::default()
283    }
284
285    /// Returns the sequence number currently expected by the receiver.
286    pub fn expected_seq(&self) -> u8 {
287        self.expected_seq
288    }
289
290    /// Evaluates an incoming frame sequence number without advancing state.
291    ///
292    /// Returns [`FrameStatus::InOrder`] if `seq == expected_seq`. Otherwise, returns
293    /// [`FrameStatus::OutOfOrder`] with the cumulative ACK sequence to transmit.
294    pub fn inspect(&self, seq: u8) -> FrameStatus {
295        if seq == self.expected_seq {
296            FrameStatus::InOrder { seq }
297        } else {
298            FrameStatus::OutOfOrder {
299                seq,
300                expected_seq: self.expected_seq,
301                ack_seq: self.current_ack_seq(),
302            }
303        }
304    }
305
306    /// Advances `expected_seq` by 1 and returns the cumulative ACK sequence to transmit.
307    ///
308    /// # Errors
309    /// Returns [`UnexpectedSeqError`] if `seq != expected_seq`.
310    pub fn advance_in_order(&mut self, seq: u8) -> Result<u8, UnexpectedSeqError> {
311        if seq != self.expected_seq {
312            return Err(UnexpectedSeqError { expected: self.expected_seq, received: seq });
313        }
314        let ack = self.expected_seq;
315        self.last_acked = Some(ack);
316        self.expected_seq = self.expected_seq.wrapping_add(1);
317        Ok(ack)
318    }
319
320    /// Processes an incoming frame sequence number in a single step.
321    ///
322    /// If `seq == expected_seq`, advances the sequence and returns [`FrameStatus::InOrder`].
323    /// Otherwise, returns [`FrameStatus::OutOfOrder`] with the cumulative ACK to send in response.
324    ///
325    /// Callers implementing two-phase flow control (e.g. attempting non-blocking writes to bounded
326    /// channels before committing sequence progression) should use [`inspect`](Self::inspect)
327    /// followed by [`advance_in_order`](Self::advance_in_order) instead.
328    pub fn accept_seq(&mut self, seq: u8) -> FrameStatus {
329        match self.inspect(seq) {
330            FrameStatus::InOrder { seq } => {
331                let _ = self.advance_in_order(seq);
332                FrameStatus::InOrder { seq }
333            }
334            out_of_order => out_of_order,
335        }
336    }
337
338    /// Returns the current cumulative ACK sequence for out-of-order/duplicate frames.
339    ///
340    /// If no frames have been received yet, returns `None` to suppress spurious `ACK 255` emission
341    /// on startup packet loss (`BUG-PROT-02`).
342    pub fn current_ack_seq(&self) -> Option<u8> {
343        self.last_acked
344    }
345
346    /// Resets the receiver to expect sequence number 0.
347    pub fn reset(&mut self) {
348        self.expected_seq = 0;
349        self.last_acked = None;
350    }
351}
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356    use crate::frame::FrameType;
357
358    #[fuchsia::test]
359    fn test_sender_window_full_and_advance() {
360        let mut sender = ResendSender::new(4, 10);
361        assert!(sender.can_send());
362        assert!(sender.is_idle());
363        assert_eq!(sender.in_flight(), 0);
364
365        for i in 0..4 {
366            let (seq, _) = sender
367                .enqueue_frame(1, 10, FrameType::Data, format!("msg{}", i).as_bytes())
368                .unwrap();
369            assert_eq!(seq, i);
370        }
371        assert_eq!(sender.in_flight(), 4);
372        assert!(!sender.can_send());
373        assert!(!sender.is_idle());
374
375        // Attempting to send when window is full fails
376        let err = sender.enqueue_frame(1, 10, FrameType::Data, b"overflow").unwrap_err();
377        assert_eq!(err, FrameError::WindowFull { window_size: 4, in_flight: 4 });
378
379        // Cumulative ACK for packet 1 acknowledges 0 and 1
380        let outcome = sender.handle_ack(1);
381        assert!(matches!(outcome, AckOutcome::Advanced { new_base: 2, all_acked: false, .. }));
382        assert_eq!(sender.base(), 2);
383        assert_eq!(sender.next_seq(), 4);
384        assert_eq!(sender.in_flight(), 2);
385        assert!(sender.can_send());
386
387        // Acknowledge remaining 2 and 3
388        let outcome = sender.handle_ack(3);
389        assert!(matches!(outcome, AckOutcome::Advanced { new_base: 4, all_acked: true, .. }));
390        assert_eq!(sender.in_flight(), 0);
391        assert!(sender.is_idle());
392    }
393
394    #[fuchsia::test]
395    fn test_sender_out_of_window_ack_ignored() {
396        let mut sender = ResendSender::new(10, 10);
397        sender.enqueue_frame(1, 1, FrameType::Data, b"test").unwrap();
398        assert_eq!(sender.base(), 0);
399        assert_eq!(sender.next_seq(), 1);
400
401        // ACK for frame 5 is out of window
402        let outcome = sender.handle_ack(5);
403        assert_eq!(outcome, AckOutcome::OutOfWindow { base: 0, next_seq: 1 });
404        assert_eq!(sender.base(), 0);
405    }
406
407    #[fuchsia::test]
408    fn test_sender_timeout_and_max_attempts() {
409        let mut sender = ResendSender::new(10, 3);
410        sender.enqueue_frame(1, 1, FrameType::Data, b"pkt0").unwrap();
411        sender.enqueue_frame(1, 1, FrameType::Data, b"pkt1").unwrap();
412
413        // Timeout 1
414        let retrans = sender.handle_timeout().unwrap();
415        assert_eq!(retrans.len(), 2);
416        assert_eq!(sender.retransmission_attempts(), 1);
417
418        // Timeout 2
419        let retrans = sender.handle_timeout().unwrap();
420        assert_eq!(retrans.len(), 2);
421        assert_eq!(sender.retransmission_attempts(), 2);
422
423        // Timeout 3 -> exceeds max_attempts (3)
424        let err = sender.handle_timeout().unwrap_err();
425        assert_eq!(err, RetransmissionLimitExceeded { base: 0, next_seq: 2, attempts: 3 });
426    }
427
428    #[fuchsia::test]
429    fn test_sender_sequence_wrap() {
430        let mut sender = ResendSender::new(10, 10);
431        // Advance sender near sequence boundary
432        sender.base = 254;
433        sender.next_seq = 254;
434
435        let (seq0, _) = sender.enqueue_frame(1, 1, FrameType::Data, b"a").unwrap();
436        assert_eq!(seq0, 254);
437        let (seq1, _) = sender.enqueue_frame(1, 1, FrameType::Data, b"b").unwrap();
438        assert_eq!(seq1, 255);
439        let (seq2, _) = sender.enqueue_frame(1, 1, FrameType::Data, b"c").unwrap();
440        assert_eq!(seq2, 0);
441        let (seq3, _) = sender.enqueue_frame(1, 1, FrameType::Data, b"d").unwrap();
442        assert_eq!(seq3, 1);
443
444        assert_eq!(sender.in_flight(), 4);
445
446        // Cumulative ACK for 0 acknowledges 254, 255, and 0
447        let outcome = sender.handle_ack(0);
448        assert!(matches!(outcome, AckOutcome::Advanced { new_base: 1, all_acked: false, .. }));
449        assert_eq!(sender.base(), 1);
450        assert_eq!(sender.in_flight(), 1);
451
452        let outcome = sender.handle_ack(1);
453        assert!(matches!(outcome, AckOutcome::Advanced { new_base: 2, all_acked: true, .. }));
454        assert_eq!(sender.in_flight(), 0);
455        assert!(sender.is_idle());
456    }
457
458    #[fuchsia::test]
459    fn test_receiver_in_order_and_duplicates() {
460        let mut receiver = ResendReceiver::new();
461        assert_eq!(receiver.expected_seq(), 0);
462
463        // Packet 0 arrives in order
464        let status = receiver.inspect(0);
465        assert_eq!(status, FrameStatus::InOrder { seq: 0 });
466        let ack = receiver.advance_in_order(0).unwrap();
467        assert_eq!(ack, 0);
468        assert_eq!(receiver.expected_seq(), 1);
469
470        // Advance with wrong sequence fails
471        assert_eq!(
472            receiver.advance_in_order(99),
473            Err(UnexpectedSeqError { expected: 1, received: 99 })
474        );
475
476        // Packet 1 arrives in order
477        let status = receiver.inspect(1);
478        assert_eq!(status, FrameStatus::InOrder { seq: 1 });
479        let ack = receiver.advance_in_order(1).unwrap();
480        assert_eq!(ack, 1);
481        assert_eq!(receiver.expected_seq(), 2);
482
483        // Duplicate packet 1 arrives -> returns cumulative ACK for 1
484        let status = receiver.inspect(1);
485        assert_eq!(status, FrameStatus::OutOfOrder { seq: 1, expected_seq: 2, ack_seq: Some(1) });
486        assert_eq!(receiver.expected_seq(), 2);
487
488        // Out-of-order packet 5 arrives -> returns cumulative ACK for 1
489        let status = receiver.inspect(5);
490        assert_eq!(status, FrameStatus::OutOfOrder { seq: 5, expected_seq: 2, ack_seq: Some(1) });
491    }
492
493    #[fuchsia::test]
494    fn test_receiver_accept_seq() {
495        let mut receiver = ResendReceiver::new();
496        assert_eq!(receiver.accept_seq(0), FrameStatus::InOrder { seq: 0 });
497        assert_eq!(receiver.expected_seq(), 1);
498        assert_eq!(receiver.current_ack_seq(), Some(0));
499
500        // Out of order
501        assert_eq!(
502            receiver.accept_seq(5),
503            FrameStatus::OutOfOrder { seq: 5, expected_seq: 1, ack_seq: Some(0) }
504        );
505        assert_eq!(receiver.expected_seq(), 1);
506    }
507
508    #[fuchsia::test]
509    fn test_receiver_suppresses_ack_255_on_startup_packet_loss() {
510        let receiver = ResendReceiver::new();
511        // Packet 0 was dropped, packet 1 arrives out of order
512        let status = receiver.inspect(1);
513        assert_eq!(status, FrameStatus::OutOfOrder { seq: 1, expected_seq: 0, ack_seq: None });
514        // ack_seq is None, suppressing spurious ACK 255 emission
515        assert_eq!(receiver.current_ack_seq(), None);
516    }
517
518    #[fuchsia::test]
519    fn test_receiver_sequence_wrap_duplicate_ack() {
520        let mut receiver = ResendReceiver::new();
521        for i in 0..256 {
522            receiver.advance_in_order(i as u8).unwrap();
523        }
524        // After 256 in-order packets, expected_seq wraps to 0
525        assert_eq!(receiver.expected_seq(), 0);
526        assert_eq!(receiver.current_ack_seq(), Some(255));
527
528        // A duplicate of packet 255 arrives -> must return cumulative ACK Some(255), not None
529        let status = receiver.inspect(255);
530        assert_eq!(
531            status,
532            FrameStatus::OutOfOrder { seq: 255, expected_seq: 0, ack_seq: Some(255) }
533        );
534    }
535
536    #[fuchsia::test]
537    fn test_sender_timeout_idle_ignored() {
538        let mut sender = ResendSender::new(10, 3);
539        assert!(sender.is_idle());
540        // Spurious timeout while idle should not increment attempts or error
541        let retrans = sender.handle_timeout().unwrap();
542        assert!(retrans.is_empty());
543        assert_eq!(sender.retransmission_attempts(), 0);
544    }
545
546    #[fuchsia::test]
547    fn test_karns_algorithm_rtt_ignored_on_retransmission() {
548        let mut sender = ResendSender::new(10, 5);
549        sender.enqueue_frame(1, 1, FrameType::Data, b"pkt0").unwrap();
550
551        // Timeout retransmits pkt0
552        let retrans = sender.handle_timeout().unwrap();
553        assert_eq!(retrans.len(), 1);
554
555        // ACK arrives for retransmitted pkt0: rtt must be None under Karn's algorithm
556        let outcome = sender.handle_ack(0);
557        match outcome {
558            AckOutcome::Advanced { rtt, .. } => {
559                assert_eq!(rtt, None, "Karn's algorithm must ignore RTT for retransmitted frame");
560            }
561            other => panic!("Unexpected outcome: {other:?}"),
562        }
563
564        // Send a fresh frame without timeout
565        sender.enqueue_frame(1, 1, FrameType::Data, b"pkt1").unwrap();
566        let outcome = sender.handle_ack(1);
567        match outcome {
568            AckOutcome::Advanced { rtt, .. } => {
569                assert!(rtt.is_some(), "RTT must be present for non-retransmitted frame");
570            }
571            other => panic!("Unexpected outcome: {other:?}"),
572        }
573    }
574
575    #[fuchsia::test]
576    #[should_panic(expected = "window_size must be in range 1..=128")]
577    fn test_sender_zero_window_panics() {
578        let _ = ResendSender::new(0, 10);
579    }
580
581    #[fuchsia::test]
582    #[should_panic(expected = "window_size must be in range 1..=128")]
583    fn test_sender_excessive_window_panics() {
584        let _ = ResendSender::new(129, 10);
585    }
586
587    #[fuchsia::test]
588    #[should_panic(expected = "max_retransmission_attempts must be greater than 0")]
589    fn test_sender_zero_max_attempts_panics() {
590        let _ = ResendSender::new(10, 0);
591    }
592}
uart_fpl/resend_sp.rs

uart_fpl/
resend_sp.rs
