fdomain_uart_driver_lib/
sender.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//! Target-side ResendSP Go-Back-N sender task and retransmission management.
6
7use crate::error::{Result, TargetDriverError};
8use futures::channel::mpsc;
9use futures::prelude::*;
10use std::pin::Pin;
11use uart_fpl::{
12    AckOutcome, DEFAULT_RETRANSMISSION_TIMEOUT, FrameType, MAX_PAYLOAD_SIZE, ResendSender,
13};
14
15/// Outgoing channel messages queued by the coordinator for [`sender_task`].
16#[derive(Clone, Debug, PartialEq, Eq)]
17pub enum SenderMessage {
18    /// Outgoing payload bytes for a multiplexed FDomain channel.
19    Data { channel_id: u16, payload: Vec<u8> },
20    /// Local close notification for a multiplexed FDomain channel, carrying an
21    /// optional diagnostic error reason when the channel closed abnormally.
22    Close { channel_id: u16, reason: Option<String> },
23}
24
25/// Truncates `s` to at most `max_bytes` on a valid UTF-8 character boundary.
26fn truncate_utf8_bytes(s: &str, max_bytes: usize) -> Vec<u8> {
27    if s.len() <= max_bytes {
28        return s.as_bytes().to_vec();
29    }
30    let end = s.floor_char_boundary(max_bytes);
31    s.as_bytes()[..end].to_vec()
32}
33
34/// Processes a cumulative ACK from `receiver_task` and updates the retransmission timer.
35///
36/// Clears `active_timer` if all in-flight frames are now acknowledged, or resets
37/// the timer deadline if the sliding window advanced with frames still in flight.
38fn handle_sender_ack(
39    sender: &mut ResendSender,
40    ack_event: Option<u8>,
41    active_timer: &mut Option<Pin<Box<fuchsia_async::Timer>>>,
42) -> Result<()> {
43    let Some(ack_seq) = ack_event else {
44        return Err(TargetDriverError::AckChannelClosed);
45    };
46    if let AckOutcome::Advanced { all_acked, .. } = sender.handle_ack(ack_seq) {
47        if all_acked {
48            *active_timer = None;
49        } else {
50            *active_timer =
51                Some(Box::pin(fuchsia_async::Timer::new(DEFAULT_RETRANSMISSION_TIMEOUT)));
52        }
53    }
54    Ok(())
55}
56
57/// Sends an encoded frame to `data_tx` while concurrently draining `ack_rx`.
58///
59/// Polling `ack_rx` alongside `data_tx.send(frame)` prevents `sender_task` from
60/// ignoring incoming cumulative ACKs when `data_tx` is temporarily full.
61async fn send_data_frame(
62    frame: Vec<u8>,
63    sender: &mut ResendSender,
64    ack_rx: &mut mpsc::Receiver<u8>,
65    data_tx: &mut mpsc::Sender<Vec<u8>>,
66    active_timer: &mut Option<Pin<Box<fuchsia_async::Timer>>>,
67) -> Result<()> {
68    let mut send_fut = std::pin::pin!(data_tx.send(frame).fuse());
69    loop {
70        futures::select_biased! {
71            ack_event = ack_rx.next().fuse() => {
72                handle_sender_ack(sender, ack_event, active_timer)?;
73            }
74            res = send_fut => {
75                return res.map_err(|e| {
76                    TargetDriverError::ChannelSend(format!(
77                        "Failed to send frame to writer: {e:?}"
78                    ))
79                });
80            }
81        }
82    }
83}
84
85/// Retransmits all unacknowledged frames in `[base, next_seq)` upon timer expiry
86/// and rearms `active_timer`.
87async fn retransmit_window(
88    sender: &mut ResendSender,
89    ack_rx: &mut mpsc::Receiver<u8>,
90    data_tx: &mut mpsc::Sender<Vec<u8>>,
91    active_timer: &mut Option<Pin<Box<fuchsia_async::Timer>>>,
92) -> Result<()> {
93    let retransmit_frames = sender.handle_timeout()?;
94    for frame in retransmit_frames {
95        if sender.is_idle() {
96            break;
97        }
98        send_data_frame(frame, sender, ack_rx, data_tx, active_timer).await?;
99    }
100    if !sender.is_idle() {
101        *active_timer = Some(Box::pin(fuchsia_async::Timer::new(DEFAULT_RETRANSMISSION_TIMEOUT)));
102    }
103    Ok(())
104}
105
106/// Encodes and transmits `in_flight_msg` if the Go-Back-N sliding window has capacity.
107///
108/// If the window is full (`!sender.can_send()`), `in_flight_msg` remains staged
109/// until a subsequent ACK frees a sequence slot. Arms `active_timer` when the
110/// first unacknowledged frame enters the window.
111async fn send_next_in_flight(
112    sender: &mut ResendSender,
113    session_id: u32,
114    in_flight_msg: &mut Option<SenderMessage>,
115    ack_rx: &mut mpsc::Receiver<u8>,
116    data_tx: &mut mpsc::Sender<Vec<u8>>,
117    active_timer: &mut Option<Pin<Box<fuchsia_async::Timer>>>,
118) -> Result<()> {
119    if !sender.can_send() {
120        return Ok(());
121    }
122    let Some(msg) = in_flight_msg.take() else {
123        return Ok(());
124    };
125    let (frame_type, channel_id, payload) = match msg {
126        SenderMessage::Data { channel_id, payload } => (FrameType::Data, channel_id, payload),
127        SenderMessage::Close { channel_id, reason } => {
128            let payload =
129                reason.map(|r| truncate_utf8_bytes(&r, MAX_PAYLOAD_SIZE)).unwrap_or_default();
130            (FrameType::Close, channel_id, payload)
131        }
132    };
133    let (_seq, frame) = sender.enqueue_frame(session_id, channel_id, frame_type, &payload)?;
134    if sender.in_flight() == 1 {
135        *active_timer = Some(Box::pin(fuchsia_async::Timer::new(DEFAULT_RETRANSMISSION_TIMEOUT)));
136    }
137    send_data_frame(frame, sender, ack_rx, data_tx, active_timer).await
138}
139
140/// Awaits `active_timer` if armed, or suspends indefinitely when no frames are in flight.
141async fn wait_active_timer(active_timer: &mut Option<Pin<Box<fuchsia_async::Timer>>>) {
142    if let Some(timer) = active_timer {
143        timer.as_mut().await;
144    } else {
145        futures::future::pending::<()>().await;
146    }
147}
148
149/// Polls `sender_rx` for the next coordinator message only when `enabled` is true,
150/// suspending when a message is already staged in `in_flight_msg` or `sender_rx` has closed.
151async fn next_sender_msg(
152    sender_rx: &mut mpsc::Receiver<SenderMessage>,
153    enabled: bool,
154) -> Option<SenderMessage> {
155    if enabled { sender_rx.next().await } else { futures::future::pending().await }
156}
157
158/// Runs the Go-Back-N sender loop, encoding [`SenderMessage`]s with `session_id`,
159/// tracking unacknowledged frames via [`ResendSender`], and retransmitting on timeout.
160///
161/// # Errors
162///
163/// Returns [`TargetDriverError::AckChannelClosed`] if `ack_rx` closes,
164/// [`TargetDriverError::RetransmissionLimit`] if retransmission attempts are
165/// exhausted, or [`TargetDriverError::ChannelSend`] if sending to `data_tx` fails.
166pub async fn sender_task(
167    mut sender_rx: mpsc::Receiver<SenderMessage>,
168    mut data_tx: mpsc::Sender<Vec<u8>>,
169    mut ack_rx: mpsc::Receiver<u8>,
170    session_id: u32,
171) -> Result<()> {
172    let mut sender = ResendSender::default();
173    let mut active_timer: Option<Pin<Box<fuchsia_async::Timer>>> = None;
174    let mut in_flight_msg: Option<SenderMessage> = None;
175    let mut sender_rx_closed = false;
176    loop {
177        // Stop pulling from `sender_rx` while the Go-Back-N window is full or a
178        // message is already staged, propagating backpressure to the coordinator.
179        let can_recv = sender.can_send() && in_flight_msg.is_none() && !sender_rx_closed;
180        // Prioritize incoming ACKs ahead of timeout retransmissions and new messages
181        // so an ACK that arrived just as the timer fired advances the window first.
182        futures::select_biased! {
183            ack_event = ack_rx.next().fuse() => {
184                handle_sender_ack(&mut sender, ack_event, &mut active_timer)?;
185            }
186            _ = wait_active_timer(&mut active_timer).fuse() => {
187                retransmit_window(&mut sender, &mut ack_rx, &mut data_tx, &mut active_timer).await?;
188            }
189            msg = next_sender_msg(&mut sender_rx, can_recv).fuse() => match msg {
190                Some(message) => in_flight_msg = Some(message),
191                None => sender_rx_closed = true,
192            },
193        }
194        // Only exit once the coordinator has closed `sender_rx`, any staged message has
195        // been sent, and all in-flight frames have been acknowledged.
196        if in_flight_msg.is_none() && sender_rx_closed && sender.is_idle() {
197            return Ok(());
198        }
199        send_next_in_flight(
200            &mut sender,
201            session_id,
202            &mut in_flight_msg,
203            &mut ack_rx,
204            &mut data_tx,
205            &mut active_timer,
206        )
207        .await?;
208    }
209}
210
211#[cfg(test)]
212mod tests {
213    use super::*;
214    use uart_fpl::encode_frame;
215
216    #[fuchsia::test]
217    async fn test_sender_task_happy_path() {
218        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(10);
219        let (mut ack_tx, ack_rx) = mpsc::channel(10);
220        let (mut sender_tx, sender_rx) = mpsc::channel(10);
221        let session_id = 12345u32;
222        let tx_handle = fuchsia_async::Task::spawn(async move {
223            sender_task(sender_rx, outgoing_tx, ack_rx, session_id).await
224        });
225        let payload = b"data to serial";
226        let channel_id = 42u16;
227        sender_tx
228            .send(SenderMessage::Data { channel_id, payload: payload.to_vec() })
229            .await
230            .unwrap();
231        let frame = outgoing_rx.next().await.unwrap();
232        let expected_frame =
233            encode_frame(session_id, channel_id, 0, FrameType::Data, payload).unwrap();
234        assert_eq!(frame, expected_frame);
235        ack_tx.send(0).await.unwrap();
236        drop(sender_tx);
237        let res = tx_handle.await;
238        assert!(res.is_ok());
239    }
240
241    #[fuchsia::test]
242    async fn test_sender_task_close_message() {
243        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(10);
244        let (mut ack_tx, ack_rx) = mpsc::channel(10);
245        let (mut sender_tx, sender_rx) = mpsc::channel(10);
246        let session_id = 67890u32;
247        let tx_handle = fuchsia_async::Task::spawn(async move {
248            sender_task(sender_rx, outgoing_tx, ack_rx, session_id).await
249        });
250        sender_tx.send(SenderMessage::Close { channel_id: 7, reason: None }).await.unwrap();
251        let frame = outgoing_rx.next().await.unwrap();
252        let expected_frame = encode_frame(session_id, 7, 0, FrameType::Close, &[]).unwrap();
253        assert_eq!(frame, expected_frame);
254        ack_tx.send(0).await.unwrap();
255
256        sender_tx
257            .send(SenderMessage::Close {
258                channel_id: 8,
259                reason: Some("RCS socket read error: PEER_CLOSED".to_string()),
260            })
261            .await
262            .unwrap();
263        let err_frame = outgoing_rx.next().await.unwrap();
264        let expected_err_frame =
265            encode_frame(session_id, 8, 1, FrameType::Close, b"RCS socket read error: PEER_CLOSED")
266                .unwrap();
267        assert_eq!(err_frame, expected_err_frame);
268        ack_tx.send(1).await.unwrap();
269
270        drop(sender_tx);
271        assert!(tx_handle.await.is_ok());
272    }
273
274    #[fuchsia::test]
275    async fn test_sender_task_ack_channel_closed_error() {
276        let (outgoing_tx, _outgoing_rx) = mpsc::channel(10);
277        let (ack_tx, ack_rx) = mpsc::channel(10);
278        let (_sender_tx, sender_rx) = mpsc::channel(10);
279        drop(ack_tx);
280        let res = sender_task(sender_rx, outgoing_tx, ack_rx, 12345).await;
281        assert!(matches!(res, Err(TargetDriverError::AckChannelClosed)));
282    }
283
284    #[fuchsia::test]
285    async fn test_sender_task_retransmits_on_timeout() {
286        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(10);
287        let (mut ack_tx, ack_rx) = mpsc::channel(10);
288        let (mut sender_tx, sender_rx) = mpsc::channel(10);
289        let tx_handle = fuchsia_async::Task::spawn(async move {
290            sender_task(sender_rx, outgoing_tx, ack_rx, 12345).await
291        });
292        sender_tx
293            .send(SenderMessage::Data { channel_id: 1, payload: b"retransmit".to_vec() })
294            .await
295            .unwrap();
296        let first_frame = outgoing_rx.next().await.unwrap();
297        let retransmitted_frame = outgoing_rx.next().await.unwrap();
298        assert_eq!(first_frame, retransmitted_frame);
299        ack_tx.send(0).await.unwrap();
300        drop(sender_tx);
301        assert!(tx_handle.await.is_ok());
302    }
303
304    #[fuchsia::test]
305    async fn test_sender_task_drains_acks_while_data_tx_backpressured() {
306        // `SinkExt::send` calls `start_send` followed by `poll_flush`, and `mpsc::Sender`'s
307        // `poll_flush` waits for `poll_ready` (an available slot for the *next* message).
308        // Using `mpsc::channel(1)` (1 shared slot + 1 per-sender slot = 2 total slots) allows
309        // `frame0` to complete `send` while `frame1` fills the second slot and suspends inside
310        // `send_data_frame` with both `frame0` (seq 0) and `frame1` (seq 1) in the GBN window.
311        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(1);
312        let (mut ack_tx, ack_rx) = mpsc::channel(10);
313        let (mut sender_tx, sender_rx) = mpsc::channel(10);
314        let session_id = 12345u32;
315        let tx_handle = fuchsia_async::Task::spawn(async move {
316            sender_task(sender_rx, outgoing_tx, ack_rx, session_id).await
317        });
318        sender_tx
319            .send(SenderMessage::Data { channel_id: 1, payload: b"frame0".to_vec() })
320            .await
321            .unwrap();
322        sender_tx
323            .send(SenderMessage::Data { channel_id: 1, payload: b"frame1".to_vec() })
324            .await
325            .unwrap();
326        drop(sender_tx);
327        // Because `sender_tx` has buffer capacity, the `.send(...).await` calls above
328        // resolve synchronously without yielding to the single-threaded test executor.
329        // We MUST yield here before sending `ACK 1`: otherwise `sender_task`'s outer
330        // `select_biased!` (which polls `ack_rx` before `sender_rx`) would consume and
331        // ignore `ACK 1` on its very first poll while the Go-Back-N window is still
332        // empty, before `frame0` and `frame1` have even been read from `sender_rx`.
333        fuchsia_async::yield_now().await;
334        // Send cumulative ACK for both frames while `sender_task` is suspended inside
335        // `send_data_frame` waiting on `outgoing_tx.send` for Frame 1.
336        ack_tx.send(1).await.unwrap();
337        // Yield again while `outgoing_tx` is still full so `send_data_frame`'s inner
338        // `select_biased!` drains `ack_rx` during backpressure (before `outgoing_rx` is read).
339        fuchsia_async::yield_now().await;
340        // Now drain `outgoing_rx`; because ACK 1 was already processed during `send_data_frame`,
341        // `sender_task` should exit cleanly immediately after `frame1` is read.
342        let _f0 = outgoing_rx.next().await.unwrap();
343        let _f1 = outgoing_rx.next().await.unwrap();
344        assert!(tx_handle.await.is_ok());
345    }
346}
fdomain_uart_driver_lib/sender.rs

fdomain_uart_driver_lib/
sender.rs
