fdomain_uart_driver_lib/
receiver.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 frame receiver task and session validation.
6
7use crate::error::{Result, TargetDriverError};
8use crate::serial::{SerialRead, SerialReader};
9use futures::channel::mpsc;
10use log::{info, warn};
11use uart_fpl::{
12    AckTracker, CONTROL_CHANNEL_ID, FrameStatus, FrameType, HandshakeResponse, HandshakeStatus,
13    ProtocolId, ResendReceiver, encode_frame,
14};
15
16/// In-order serial channel events delivered from [`receiver_task`] to the coordinator.
17#[derive(Clone, Debug, PartialEq, Eq)]
18pub enum SerialEvent {
19    /// Incoming payload bytes for a multiplexed FDomain channel.
20    SerialData { channel_id: u16, data: Vec<u8> },
21    /// Remote close notification for a multiplexed FDomain channel.
22    SerialClose { channel_id: u16 },
23}
24
25const NEGOTIATE_RESP_SEQ: u8 = 1;
26
27#[derive(Clone, Copy, Debug, PartialEq, Eq)]
28enum SessionCheck {
29    Valid,
30    Discard,
31}
32
33fn validate_frame_session(
34    frame: &uart_fpl::Frame,
35    current_session_id: u32,
36) -> Result<SessionCheck> {
37    if frame.session_id == 0 {
38        warn!("Received {:?} frame with invalid session_id 0; discarding", frame.frame_type);
39        return Ok(SessionCheck::Discard);
40    }
41    if frame.session_id == current_session_id {
42        if frame.frame_type == FrameType::Reset
43            || (frame.frame_type == FrameType::Close && frame.channel_id == CONTROL_CHANNEL_ID)
44        {
45            info!(
46                "Received {:?} frame on control channel from host. Restarting bridge...",
47                frame.frame_type
48            );
49            return Err(TargetDriverError::ResetRequested);
50        }
51        return Ok(SessionCheck::Valid);
52    }
53    if frame.frame_type == FrameType::NegotiateReq && frame.channel_id == CONTROL_CHANNEL_ID {
54        info!("Session ID changed from {} to {}, restarting", current_session_id, frame.session_id);
55        return Err(TargetDriverError::SessionIdChanged);
56    }
57    warn!(
58        "Received frame with mismatched session ID (got {}, expected {}) type {:?} channel {}; discarding",
59        frame.session_id, current_session_id, frame.frame_type, frame.channel_id
60    );
61    Ok(SessionCheck::Discard)
62}
63
64fn handle_duplicate_negotiate_req(rx_session_id: u32, data_tx: &mut mpsc::Sender<Vec<u8>>) {
65    info!(
66        "Received duplicate NegotiateReq for active session {}, re-transmitting NegotiateResp",
67        rx_session_id
68    );
69    let resp = HandshakeResponse {
70        status: HandshakeStatus::Success,
71        selected: Some(ProtocolId::ResendSP),
72    };
73    if let Ok(serialized) = resp.serialize()
74        && let Ok(resp_frame) = encode_frame(
75            rx_session_id,
76            CONTROL_CHANNEL_ID,
77            NEGOTIATE_RESP_SEQ,
78            FrameType::NegotiateResp,
79            &serialized,
80        )
81        && let Err(e) = data_tx.try_send(resp_frame)
82    {
83        warn!("Failed to send duplicate NegotiateResp: {:?}", e);
84    }
85}
86
87/// Validates the Go-Back-N sequence number of a `DATA` or `CLOSE` frame, forwards
88/// in-order events to the coordinator without blocking, and updates `ack_tracker`.
89fn handle_sequenced_serial_event(
90    receiver: &mut ResendReceiver,
91    rx_session_id: u32,
92    seq: u8,
93    event: SerialEvent,
94    serial_tx: &mut mpsc::Sender<SerialEvent>,
95    ack_tracker: &AckTracker,
96) -> Result<()> {
97    // Inspect `seq` before advancing `receiver` so we only commit sequence progress
98    // if `serial_tx` has capacity to accept the event.
99    match receiver.inspect(seq) {
100        FrameStatus::InOrder { .. } => match serial_tx.try_send(event) {
101            Ok(()) => ack_tracker.set_ack(rx_session_id, receiver.advance_in_order(seq)?),
102            Err(e) if e.is_full() => {
103                // Downstream coordinator queue is full: drop the frame without advancing
104                // `receiver` and re-assert the last cumulative ACK. This applies Go-Back-N
105                // backpressure (the host will retransmit `seq`) while keeping `receiver_task`
106                // non-blocking so it can still process incoming `Ack` and `Reset` frames.
107                if let Some(ack_seq) = receiver.current_ack_seq() {
108                    ack_tracker.set_ack(rx_session_id, ack_seq);
109                }
110            }
111            Err(e) => {
112                let msg = format!("Failed to send event to coordinator: {e:?}");
113                return Err(TargetDriverError::ChannelSend(msg));
114            }
115        },
116        FrameStatus::OutOfOrder { ack_seq, .. } => {
117            // Discard out-of-order or duplicate frames and re-emit the last contiguous
118            // cumulative ACK so the host knows which sequence number the target expects.
119            if let Some(ack_seq) = ack_seq {
120                ack_tracker.set_ack(rx_session_id, ack_seq);
121            }
122        }
123    }
124    Ok(())
125}
126
127fn handle_ack_frame(seq: u8, ack_tx: &mut mpsc::Sender<u8>) -> Result<()> {
128    if let Err(e) = ack_tx.try_send(seq) {
129        if e.is_full() {
130            warn!("ack_tx full, dropping ACK event seq={}", seq);
131        } else {
132            return Err(TargetDriverError::ChannelSend(format!(
133                "Failed to enqueue ACK event: {:?}",
134                e
135            )));
136        }
137    }
138    Ok(())
139}
140
141fn dispatch_receiver_frame(
142    frame: uart_fpl::Frame,
143    receiver: &mut ResendReceiver,
144    serial_tx: &mut mpsc::Sender<SerialEvent>,
145    ack_tracker: &AckTracker,
146    ack_tx: &mut mpsc::Sender<u8>,
147    data_tx: &mut mpsc::Sender<Vec<u8>>,
148) -> Result<()> {
149    let (session_id, channel_id, seq, payload) =
150        (frame.session_id, frame.channel_id, frame.seq, frame.payload);
151    match frame.frame_type {
152        FrameType::NegotiateReq if channel_id == CONTROL_CHANNEL_ID => {
153            handle_duplicate_negotiate_req(session_id, data_tx);
154        }
155        FrameType::NegotiateReq => {
156            warn!(
157                "Received unexpected NegotiateReq on non-control channel {}; discarding",
158                channel_id
159            );
160        }
161        FrameType::Data if channel_id == CONTROL_CHANNEL_ID => {
162            warn!("Received unexpected DATA frame on CONTROL_CHANNEL_ID; discarding");
163        }
164        FrameType::Data => {
165            let event = SerialEvent::SerialData { channel_id, data: payload };
166            handle_sequenced_serial_event(
167                receiver,
168                session_id,
169                seq,
170                event,
171                serial_tx,
172                ack_tracker,
173            )?;
174        }
175        FrameType::Ack => handle_ack_frame(seq, ack_tx)?,
176        FrameType::Close => {
177            let event = SerialEvent::SerialClose { channel_id };
178            handle_sequenced_serial_event(
179                receiver,
180                session_id,
181                seq,
182                event,
183                serial_tx,
184                ack_tracker,
185            )?;
186        }
187        other => warn!("Unknown frame type: {:?}", other),
188    }
189    Ok(())
190}
191
192/// Reads frames from `reader`, validates `session_id`, tracks ResendSP sequence
193/// numbers via `ack_tracker`, and dispatches in-order [`SerialEvent`]s and ACKs.
194///
195/// # Errors
196///
197/// Returns [`TargetDriverError::ResetRequested`] if the host sends a reset frame,
198/// [`TargetDriverError::SessionIdChanged`] if a new handshake arrives with a
199/// different session ID, or propagates serial read and channel send errors.
200pub async fn receiver_task<R: SerialRead>(
201    reader: &mut SerialReader<R>,
202    mut serial_tx: mpsc::Sender<SerialEvent>,
203    ack_tracker: AckTracker,
204    mut ack_tx: mpsc::Sender<u8>,
205    session_id: u32,
206    mut data_tx: mpsc::Sender<Vec<u8>>,
207) -> Result<()> {
208    let mut receiver = ResendReceiver::new();
209    loop {
210        // `reader.next_frame().await` resolves synchronously without yielding to the
211        // single-threaded executor whenever multiple frames are already buffered in
212        // `FrameParser` from a single bulk `Device.Read`. Yield at the top of every
213        // iteration (including after `SessionCheck::Discard`) so `coordinator_task`,
214        // `sender_task`, and `writer_task` can make progress.
215        fuchsia_async::yield_now().await;
216        let frame = reader.next_frame().await?;
217        match validate_frame_session(&frame, session_id) {
218            Ok(SessionCheck::Valid) => {}
219            Ok(SessionCheck::Discard) => continue,
220            Err(TargetDriverError::SessionIdChanged) => {
221                // `next_frame()` already consumed the new session's `NegotiateReq` from `reader`.
222                // Put it back at the front of `reader` so `run_device_sessions` can pass it into
223                // the next `negotiate_session` call via `take_unconsumed()`, avoiding a host
224                // handshake timeout.
225                reader.requeue_frame(&frame);
226                return Err(TargetDriverError::SessionIdChanged);
227            }
228            Err(e) => return Err(e),
229        }
230        dispatch_receiver_frame(
231            frame,
232            &mut receiver,
233            &mut serial_tx,
234            &ack_tracker,
235            &mut ack_tx,
236            &mut data_tx,
237        )?;
238    }
239}
240
241#[cfg(test)]
242mod tests {
243    use super::*;
244    use crate::serial::MockSerial;
245    use futures::prelude::*;
246    use uart_fpl::{FrameParser, HostHandshake};
247
248    struct MockSerialPending {
249        read_data: Vec<Vec<u8>>,
250    }
251
252    impl SerialRead for MockSerialPending {
253        async fn serial_read(&mut self) -> Result<Vec<u8>> {
254            if self.read_data.is_empty() {
255                futures::future::pending().await
256            } else {
257                Ok(self.read_data.remove(0))
258            }
259        }
260    }
261
262    struct TestReceiverHarness {
263        ack_tracker: AckTracker,
264        serial_rx: mpsc::Receiver<SerialEvent>,
265        data_rx: mpsc::Receiver<Vec<u8>>,
266        _ack_rx: mpsc::Receiver<u8>,
267        task: fuchsia_async::Task<Result<()>>,
268    }
269
270    fn spawn_test_receiver<R: SerialRead + 'static>(
271        client: R,
272        session_id: u32,
273    ) -> TestReceiverHarness {
274        let mut reader = SerialReader::new(client);
275        let ack_tracker = AckTracker::new();
276        let (ack_tx, _ack_rx) = mpsc::channel(10);
277        let (serial_tx, serial_rx) = mpsc::channel(10);
278        let (data_tx, data_rx) = mpsc::channel(10);
279        let ack_clone = ack_tracker.clone();
280        let task = fuchsia_async::Task::local(async move {
281            receiver_task(&mut reader, serial_tx, ack_clone, ack_tx, session_id, data_tx).await
282        });
283        TestReceiverHarness { ack_tracker, serial_rx, data_rx, _ack_rx, task }
284    }
285
286    async fn run_receiver_once(frames: Vec<Vec<u8>>, session_id: u32) -> (Result<()>, Vec<u8>) {
287        let mock_serial = MockSerial { read_data: frames, write_data: Vec::new() };
288        let mut reader = SerialReader::new(mock_serial);
289        let ack_tracker = AckTracker::new();
290        let (ack_tx, _ack_rx) = mpsc::channel(10);
291        let (serial_tx, _serial_rx) = mpsc::channel(10);
292        let (data_tx, _data_rx) = mpsc::channel(10);
293        let res =
294            receiver_task(&mut reader, serial_tx, ack_tracker, ack_tx, session_id, data_tx).await;
295        (res, reader.take_unconsumed())
296    }
297
298    #[fuchsia::test]
299    async fn test_receiver_task_happy_path() {
300        let payload = b"hello rcs";
301        let (session_id, channel_id) = (12345u32, 42u16);
302        let data_frame = encode_frame(session_id, channel_id, 0, FrameType::Data, payload).unwrap();
303        let mock_serial = MockSerial { read_data: vec![data_frame], write_data: Vec::new() };
304        let mut harness = spawn_test_receiver(mock_serial, session_id);
305        let event = harness.serial_rx.next().await.unwrap();
306        assert_eq!(event, SerialEvent::SerialData { channel_id, data: payload.to_vec() });
307        assert_eq!(harness.ack_tracker.take_ack(), Some((session_id, 0)));
308    }
309
310    #[fuchsia::test]
311    async fn test_receiver_task_discards_zero_session_id() {
312        let (valid_sid, channel_id) = (12345u32, 42u16);
313        let zero_frame = encode_frame(0, channel_id, 0, FrameType::Data, b"bad").unwrap();
314        let valid_frame = encode_frame(valid_sid, channel_id, 0, FrameType::Data, b"ok").unwrap();
315        let mock_serial =
316            MockSerial { read_data: vec![zero_frame, valid_frame], write_data: Vec::new() };
317        let mut harness = spawn_test_receiver(mock_serial, valid_sid);
318        let event = harness.serial_rx.next().await.unwrap();
319        assert_eq!(event, SerialEvent::SerialData { channel_id, data: b"ok".to_vec() });
320    }
321
322    #[fuchsia::test]
323    async fn test_receiver_task_out_of_order_reacks() {
324        let (session_id, channel_id) = (12345u32, 42u16);
325        let frame_0 = encode_frame(session_id, channel_id, 0, FrameType::Data, b"first").unwrap();
326        let frame_2 = encode_frame(session_id, channel_id, 2, FrameType::Data, b"third").unwrap();
327        let frame_1 = encode_frame(session_id, channel_id, 1, FrameType::Data, b"second").unwrap();
328        let mock_serial =
329            MockSerial { read_data: vec![frame_0, frame_2, frame_1], write_data: Vec::new() };
330        let mut harness = spawn_test_receiver(mock_serial, session_id);
331        let ev0 = harness.serial_rx.next().await.unwrap();
332        assert_eq!(ev0, SerialEvent::SerialData { channel_id, data: b"first".to_vec() });
333        let ev1 = harness.serial_rx.next().await.unwrap();
334        assert_eq!(ev1, SerialEvent::SerialData { channel_id, data: b"second".to_vec() });
335        assert_eq!(harness.ack_tracker.take_ack(), Some((session_id, 1)));
336    }
337
338    #[fuchsia::test]
339    async fn test_receiver_task_session_change_discard() {
340        let (session_id_1, session_id_2, channel_id) = (12345u32, 67890u32, 42u16);
341        let frame_1 =
342            encode_frame(session_id_2, channel_id, 0, FrameType::Data, b"discard me").unwrap();
343        let frame_2 =
344            encode_frame(session_id_1, channel_id, 0, FrameType::Data, b"accept me").unwrap();
345        let mock_serial = MockSerial { read_data: vec![frame_1, frame_2], write_data: Vec::new() };
346        let mut harness = spawn_test_receiver(mock_serial, session_id_1);
347        let event = harness.serial_rx.next().await.unwrap();
348        assert_eq!(event, SerialEvent::SerialData { channel_id, data: b"accept me".to_vec() });
349    }
350
351    #[fuchsia::test]
352    async fn test_receiver_task_session_change_negotiate_reboot() {
353        let (session_id_1, session_id_2) = (12345u32, 67890u32);
354        let frame_1 =
355            encode_frame(session_id_2, CONTROL_CHANNEL_ID, 0, FrameType::NegotiateReq, &[1, 1])
356                .unwrap();
357        let (result, unconsumed) = run_receiver_once(vec![frame_1.clone()], session_id_1).await;
358        assert!(matches!(result, Err(TargetDriverError::SessionIdChanged)));
359        assert_eq!(unconsumed, frame_1);
360    }
361
362    #[fuchsia::test]
363    async fn test_session_id_resiliency_adversarial() {
364        let (sid_current, sid_forged) = (100u32, 200u32);
365        let frame_bad = encode_frame(sid_forged, 42, 0, FrameType::Data, b"discard").unwrap();
366        let frame_ok = encode_frame(sid_current, 42, 0, FrameType::Data, b"accept").unwrap();
367        let mock_serial =
368            MockSerial { read_data: vec![frame_bad, frame_ok], write_data: Vec::new() };
369        let mut harness = spawn_test_receiver(mock_serial, sid_current);
370        let event = harness.serial_rx.next().await.unwrap();
371        assert_eq!(event, SerialEvent::SerialData { channel_id: 42, data: b"accept".to_vec() });
372    }
373
374    #[fuchsia::test]
375    async fn test_crc_noise_collision_resiliency() {
376        let (sid_current, sid_noise) = (0xABCDEFu32, 0x99999999u32);
377        let noise_frame = encode_frame(sid_noise, 1, 0, FrameType::Data, b"noisebytes").unwrap();
378        let valid_frame = encode_frame(sid_current, 1, 0, FrameType::Data, b"validbytes").unwrap();
379        let mock_serial = MockSerialPending { read_data: vec![noise_frame, valid_frame] };
380        let mut harness = spawn_test_receiver(mock_serial, sid_current);
381        let event = harness.serial_rx.next().await.unwrap();
382        assert_eq!(event, SerialEvent::SerialData { channel_id: 1, data: b"validbytes".to_vec() });
383        assert!(futures::poll!(&mut harness.task).is_pending());
384    }
385
386    #[fuchsia::test]
387    async fn test_receiver_task_duplicate_negotiate_req_retransmits_resp() {
388        let session_id = 99999u32;
389        let host = HostHandshake::new(vec![ProtocolId::ResendSP]);
390        let (frame_type, payload) = host.start().unwrap();
391        let dup_req =
392            encode_frame(session_id, CONTROL_CHANNEL_ID, 0, frame_type, &payload).unwrap();
393        let mock_serial = MockSerial { read_data: vec![dup_req], write_data: Vec::new() };
394        let mut harness = spawn_test_receiver(mock_serial, session_id);
395        let resp_bytes = harness.data_rx.next().await.expect("Expected NegotiateResp");
396        let mut parser = FrameParser::new();
397        parser.feed(&resp_bytes);
398        let resp_frame = parser.next_frame().expect("Should parse NegotiateResp frame");
399        assert_eq!(resp_frame.session_id, session_id);
400        assert_eq!(resp_frame.channel_id, CONTROL_CHANNEL_ID);
401        assert_eq!(resp_frame.frame_type, FrameType::NegotiateResp);
402        let resp = HandshakeResponse::try_from(resp_frame.payload.as_slice()).unwrap();
403        assert_eq!(resp.status, HandshakeStatus::Success);
404        assert_eq!(resp.selected, Some(ProtocolId::ResendSP));
405    }
406
407    #[fuchsia::test]
408    async fn test_receiver_task_handles_reset_frame() {
409        let session_id = 12345u32;
410        let zero_reset = encode_frame(0, CONTROL_CHANNEL_ID, 0, FrameType::Reset, &[]).unwrap();
411        let stale_reset =
412            encode_frame(99999, CONTROL_CHANNEL_ID, 0, FrameType::Reset, &[]).unwrap();
413        let valid_reset =
414            encode_frame(session_id, CONTROL_CHANNEL_ID, 0, FrameType::Reset, &[]).unwrap();
415        let (result, _) =
416            run_receiver_once(vec![zero_reset, stale_reset, valid_reset], session_id).await;
417        assert!(matches!(result, Err(TargetDriverError::ResetRequested)));
418    }
419
420    #[fuchsia::test]
421    async fn test_receiver_task_discards_control_channel_data() {
422        let session_id = 12345u32;
423        let bad_frame = encode_frame(
424            session_id,
425            CONTROL_CHANNEL_ID,
426            0,
427            FrameType::Data,
428            b"spurious control data",
429        )
430        .unwrap();
431        let good_frame =
432            encode_frame(session_id, 42, 0, FrameType::Data, b"valid user data").unwrap();
433        let mock_serial =
434            MockSerial { read_data: vec![bad_frame, good_frame], write_data: Vec::new() };
435        let mut harness = spawn_test_receiver(mock_serial, session_id);
436        let event = harness.serial_rx.next().await.unwrap();
437        assert_eq!(
438            event,
439            SerialEvent::SerialData { channel_id: 42, data: b"valid user data".to_vec() }
440        );
441    }
442}
fdomain_uart_driver_lib/receiver.rs

fdomain_uart_driver_lib/
receiver.rs
