log_command_ctf/
fxt_streamer.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
5use crate::LogError;
6use diagnostics_data::{ExtendedMoniker, LogsData};
7use diagnostics_log_encoding::{Header, LOG_CONTROL_BIT, MONIKER, URL, Value};
8use diagnostics_message::MonikerWithUrl;
9use flyweights::FlyStr;
10use futures_util::stream::FusedStream;
11use futures_util::{AsyncReadExt, stream};
12use std::collections::HashMap;
13use zerocopy::FromBytes;
14
15pub struct FxtStreamer {
16    socket_reader: SocketReader,
17    tags: HashMap<u32, MonikerWithUrl>,
18}
19
20impl FxtStreamer {
21    pub fn new(socket: flex_client::AsyncSocket) -> Self {
22        Self { socket_reader: SocketReader::new(socket), tags: HashMap::default() }
23    }
24
25    pub fn stream(self) -> impl FusedStream<Item = Result<LogsData, LogError>> {
26        stream::unfold(Some(self), |maybe_this| async {
27            let mut this = maybe_this?;
28            let item = this.next_message().await.transpose()?;
29            let maybe_this = item.is_ok().then_some(this);
30            Some((item, maybe_this))
31        })
32    }
33
34    async fn next_message(&mut self) -> Result<Option<LogsData>, LogError> {
35        loop {
36            // 1. Ensure we have at least 8 bytes for the header.
37            let Some(header_bytes) = self.socket_reader.fill(8).await? else {
38                return Ok(None);
39            };
40
41            // 2. Parse header to get record size.
42            let header = Header::read_from_bytes(header_bytes)
43                .map_err(|_| LogError::UnknownError(anyhow::anyhow!("Failed to parse header")))?;
44            let record_len = header.size_words() as usize * 8;
45            if record_len < 8 {
46                Err(LogError::UnknownError(anyhow::anyhow!("Invalid record size in header")))?;
47            }
48
49            // 3. Ensure we have the full record.
50            let record_bytes = self.socket_reader.fill(record_len).await?.unwrap();
51
52            // 4. Parse the record.
53            let (record, remaining) = diagnostics_log_encoding::parse::parse_record(record_bytes)
54                .map_err(|e| LogError::UnknownError(e.into()))?;
55            debug_assert!(remaining.is_empty());
56
57            // 5. Handle Manifest vs Log
58            let tag = header.tag();
59            let is_manifest = (tag & LOG_CONTROL_BIT) != 0;
60            let actual_tag = tag & !LOG_CONTROL_BIT;
61
62            if is_manifest {
63                let mut moniker = None;
64                let mut url = None;
65                for arg in &record.arguments {
66                    if arg.name() == MONIKER
67                        && let Value::Text(t) = arg.value()
68                    {
69                        moniker = Some(
70                            ExtendedMoniker::parse_str(&t)
71                                .map_err(|e| LogError::UnknownError(e.into()))?,
72                        );
73                    } else if arg.name() == URL
74                        && let Value::Text(t) = arg.value()
75                    {
76                        url = Some(FlyStr::new(t));
77                    }
78                }
79                if let (Some(moniker), Some(url)) = (moniker, url) {
80                    self.tags.insert(actual_tag, MonikerWithUrl { moniker, url });
81                }
82                self.socket_reader.consume(record_len);
83                continue;
84            }
85
86            let source = self.tags.get(&actual_tag).cloned().unwrap_or_else(|| MonikerWithUrl {
87                moniker: ExtendedMoniker::parse_str("/UNKNOWN").unwrap(),
88                url: FlyStr::new(""),
89            });
90            let logs_data = diagnostics_message::from_structured(source, record_bytes)
91                .map_err(|e| LogError::UnknownError(e.into()))?;
92
93            self.socket_reader.consume(record_len);
94            return Ok(Some(logs_data));
95        }
96    }
97}
98
99struct SocketReader {
100    socket: flex_client::AsyncSocket,
101    buffer: Vec<u8>,
102    head: usize,
103}
104
105impl SocketReader {
106    fn new(socket: flex_client::AsyncSocket) -> Self {
107        Self { socket, buffer: Vec::new(), head: 0 }
108    }
109
110    /// Ensures that `count` bytes are present in the buffer and returns them.  Returns
111    /// None, if the socket is closed and there is no partial data.
112    async fn fill(&mut self, count: usize) -> Result<Option<&[u8]>, LogError> {
113        let remaining = self.buffer.len() - self.head;
114        if remaining < count {
115            if self.head > 0 {
116                self.buffer.copy_within(self.head.., 0);
117                self.buffer.truncate(remaining);
118                self.head = 0;
119            }
120            while self.buffer.len() < count {
121                let start_len = self.buffer.len();
122                self.buffer.resize(std::cmp::max(start_len + 4096, count), 0);
123                let n = self
124                    .socket
125                    .read(&mut self.buffer[start_len..])
126                    .await
127                    .map_err(LogError::IOError)?;
128                self.buffer.truncate(start_len + n);
129                if n == 0 {
130                    if !self.buffer.is_empty() {
131                        return Err(LogError::UnknownError(anyhow::anyhow!(
132                            "Truncated record at end of stream"
133                        )));
134                    }
135                    return Ok(None);
136                }
137            }
138        }
139        Ok(Some(&self.buffer[self.head..self.head + count]))
140    }
141
142    /// Consumes `count` bytes from the buffer.
143    fn consume(&mut self, count: usize) {
144        self.head += count;
145        if self.head == self.buffer.len() {
146            self.buffer.clear();
147            self.head = 0;
148        }
149    }
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155    use assert_matches::assert_matches;
156    use diagnostics_log_encoding::encode::{Encoder, EncoderOpts, ResizableBuffer};
157    use diagnostics_log_encoding::{Argument, Record};
158    use fuchsia_async as fasync;
159    use futures_util::StreamExt;
160    use zerocopy::IntoBytes;
161
162    fn fn_encode(record: Record<'_>, tag: u32) -> Vec<u8> {
163        let mut encoder = Encoder::new(
164            std::io::Cursor::new(ResizableBuffer::from(Vec::new())),
165            EncoderOpts::default(),
166        );
167        encoder.write_record(record).unwrap();
168        let mut bytes = encoder.take().into_inner().into_inner();
169        let mut header = Header::read_from_bytes(&bytes[0..8]).unwrap();
170        header.set_tag(tag);
171        bytes[0..8].copy_from_slice(header.as_bytes());
172        bytes
173    }
174
175    // --- 1. Exercise SocketReader directly ---
176    #[fuchsia::test]
177    async fn test_socket_reader_clean_eof() {
178        let (sender, receiver) = zx::Socket::create_stream();
179        let mut reader = SocketReader::new(flex_client::socket_to_async(receiver));
180        drop(sender);
181        assert!(reader.fill(8).await.unwrap().is_none());
182    }
183
184    #[fuchsia::test]
185    async fn test_socket_reader_truncated() {
186        let (sender, receiver) = zx::Socket::create_stream();
187        let mut reader = SocketReader::new(flex_client::socket_to_async(receiver));
188        sender.write(b"1234").unwrap();
189        drop(sender);
190        assert_matches!(
191            reader.fill(8).await,
192            Err(LogError::UnknownError(e))
193                if e.to_string().contains("Truncated record at end of stream")
194        );
195    }
196
197    #[fuchsia::test]
198    async fn test_socket_reader_shift_and_consume() {
199        let (sender, receiver) = zx::Socket::create_stream();
200        let mut reader = SocketReader::new(flex_client::socket_to_async(receiver));
201
202        // Write 16 bytes
203        sender.write(b"abcdefghijklmnop").unwrap();
204        drop(sender);
205
206        // Fill 8 bytes
207        let slice1 = reader.fill(8).await.unwrap().unwrap();
208        assert_eq!(slice1, b"abcdefgh");
209        reader.consume(4); // head becomes 4, remaining is 12
210
211        // Fill 8 bytes starting from head 4 ("efghijkl")
212        // Since remaining 12 is >= 8, this returns directly without shifting!
213        let slice2 = reader.fill(8).await.unwrap().unwrap();
214        assert_eq!(slice2, b"efghijkl");
215        reader.consume(6); // head becomes 10, remaining is 6
216
217        // Fill 8 bytes starting from head 10 ("klmnop")
218        // Remaining is 6, which is < 8! This triggers copy_within shifting to head 0!
219        // But since sender is dropped, reading more returns EOF, so we get Truncated error!
220        assert_matches!(
221            reader.fill(8).await,
222            Err(LogError::UnknownError(e))
223                if e.to_string().contains("Truncated record at end of stream") => {}
224        );
225    }
226
227    #[fuchsia::test]
228    async fn test_socket_reader_partial_reads_loop() {
229        let (sender, receiver) = zx::Socket::create_stream();
230        let mut reader = SocketReader::new(flex_client::socket_to_async(receiver));
231
232        // Write 4 bytes
233        sender.write(b"1234").unwrap();
234
235        // Use a separate task to push more bytes so reader.fill can loop asynchronously
236        let push_task = fasync::Task::spawn(async move {
237            fasync::Timer::new(std::time::Duration::from_millis(10)).await;
238            sender.write(b"5678").unwrap();
239        });
240
241        let slice = reader.fill(8).await.unwrap().unwrap();
242        assert_eq!(slice, b"12345678");
243        push_task.await;
244    }
245
246    // --- 2. Exercise FxtStreamer error branches ---
247    #[fuchsia::test]
248    async fn test_fxt_streamer_invalid_record_size() {
249        let (sender, receiver) = zx::Socket::create_stream();
250        let streamer = FxtStreamer::new(flex_client::socket_to_async(receiver));
251        let mut stream = std::pin::pin!(streamer.stream());
252
253        // Write 8 bytes of zeroes. Header size_words will be 0. record_len = 0 < 8.
254        sender.write(&[0; 8]).unwrap();
255        drop(sender);
256
257        assert_matches!(
258            stream.next().await,
259            Some(Err(LogError::UnknownError(e)))
260                if e.to_string().contains("Invalid record size in header")
261        );
262    }
263
264    #[fuchsia::test]
265    async fn test_fxt_streamer_truncated_record() {
266        let (sender, receiver) = zx::Socket::create_stream();
267        let streamer = FxtStreamer::new(flex_client::socket_to_async(receiver));
268        let mut stream = std::pin::pin!(streamer.stream());
269
270        // Build a valid 16-byte record
271        let valid_16_byte_record = fn_encode(
272            Record {
273                timestamp: zx::BootInstant::from_nanos(0),
274                severity: 0x30,
275                arguments: vec![Argument::message("1234")],
276            },
277            1,
278        );
279
280        // Write only the first 8 bytes!
281        sender.write(&valid_16_byte_record[0..8]).unwrap();
282        drop(sender);
283
284        assert_matches!(
285            stream.next().await,
286            Some(Err(LogError::UnknownError(e)))
287                if e.to_string().contains("Truncated record at end of stream")
288        );
289    }
290
291    #[fuchsia::test]
292    async fn test_fxt_streamer_parse_error() {
293        let (sender, receiver) = zx::Socket::create_stream();
294        let streamer = FxtStreamer::new(flex_client::socket_to_async(receiver));
295        let mut stream = std::pin::pin!(streamer.stream());
296
297        // Build a valid 16-byte record
298        let valid_16_byte_record = fn_encode(
299            Record {
300                timestamp: zx::BootInstant::from_nanos(0),
301                severity: 0x30,
302                arguments: vec![Argument::message("1234")],
303            },
304            1,
305        );
306
307        // Write the 8 byte header claiming 16 bytes, then provide 8 bytes of garbage payload.
308        sender.write(&valid_16_byte_record[0..8]).unwrap();
309        sender.write(&[0xff; 8]).unwrap(); // Invalid record encoding
310        drop(sender);
311
312        assert_matches!(stream.next().await, Some(Err(LogError::UnknownError(_))));
313    }
314
315    // --- 3. Exercise FxtStreamer non-error branches & continuous processing ---
316    #[fuchsia::test]
317    async fn test_fxt_streamer_unknown_tag() {
318        let (sender, receiver) = zx::Socket::create_stream();
319        let streamer = FxtStreamer::new(flex_client::socket_to_async(receiver));
320        let mut stream = std::pin::pin!(streamer.stream());
321
322        let log_bytes = fn_encode(
323            Record {
324                timestamp: zx::BootInstant::from_nanos(123),
325                severity: 0x30,
326                arguments: vec![Argument::message("hello")],
327            },
328            1,
329        );
330        sender.write(&log_bytes).unwrap();
331        drop(sender);
332
333        let item = stream.next().await.unwrap().unwrap();
334        assert_eq!(item.moniker.to_string(), "UNKNOWN");
335        assert_eq!(item.metadata.component_url, Some(FlyStr::new("")));
336        assert_eq!(item.msg().unwrap(), "hello");
337    }
338
339    #[fuchsia::test]
340    async fn test_fxt_streamer_partial_manifest_ignored() {
341        let (sender, receiver) = zx::Socket::create_stream();
342        let streamer = FxtStreamer::new(flex_client::socket_to_async(receiver));
343        let mut stream = std::pin::pin!(streamer.stream());
344
345        // Write a manifest missing the URL argument
346        let manifest_bytes = fn_encode(
347            Record {
348                timestamp: zx::BootInstant::from_nanos(0),
349                severity: 0x30,
350                arguments: vec![Argument::other(MONIKER, "core/partial")],
351            },
352            1 | LOG_CONTROL_BIT,
353        );
354
355        let log_bytes = fn_encode(
356            Record {
357                timestamp: zx::BootInstant::from_nanos(123),
358                severity: 0x30,
359                arguments: vec![Argument::message("hello")],
360            },
361            1,
362        );
363
364        sender.write(&manifest_bytes).unwrap();
365        sender.write(&log_bytes).unwrap();
366        drop(sender);
367
368        // Manifest should be cleanly consumed and ignored. Log record falls back to /UNKNOWN.
369        let item = stream.next().await.unwrap().unwrap();
370        assert_eq!(item.moniker.to_string(), "UNKNOWN");
371        assert_eq!(item.metadata.component_url, Some(FlyStr::new("")));
372        assert_eq!(item.msg().unwrap(), "hello");
373    }
374
375    #[fuchsia::test]
376    async fn test_fxt_streamer_multiple_messages() {
377        let (sender, receiver) = zx::Socket::create_stream();
378        let streamer = FxtStreamer::new(flex_client::socket_to_async(receiver));
379        let mut stream = std::pin::pin!(streamer.stream());
380
381        let manifest_a = fn_encode(
382            Record {
383                timestamp: zx::BootInstant::from_nanos(0),
384                severity: 0x30,
385                arguments: vec![Argument::other(MONIKER, "core/a"), Argument::other(URL, "url_a")],
386            },
387            1 | LOG_CONTROL_BIT,
388        );
389
390        let log_a1 = fn_encode(
391            Record {
392                timestamp: zx::BootInstant::from_nanos(1),
393                severity: 0x30,
394                arguments: vec![Argument::message("msg a1")],
395            },
396            1,
397        );
398
399        let log_a2 = fn_encode(
400            Record {
401                timestamp: zx::BootInstant::from_nanos(2),
402                severity: 0x30,
403                arguments: vec![Argument::message("msg a2")],
404            },
405            1,
406        );
407
408        let log_unknown = fn_encode(
409            Record {
410                timestamp: zx::BootInstant::from_nanos(3),
411                severity: 0x30,
412                arguments: vec![Argument::message("msg unknown")],
413            },
414            99, // Unmapped tag
415        );
416
417        sender.write(&manifest_a).unwrap();
418        sender.write(&log_a1).unwrap();
419        sender.write(&log_a2).unwrap();
420        sender.write(&log_unknown).unwrap();
421        drop(sender);
422
423        let item1 = stream.next().await.unwrap().unwrap();
424        assert_eq!(item1.moniker.to_string(), "core/a");
425        assert_eq!(item1.msg().unwrap(), "msg a1");
426
427        let item2 = stream.next().await.unwrap().unwrap();
428        assert_eq!(item2.moniker.to_string(), "core/a");
429        assert_eq!(item2.msg().unwrap(), "msg a2");
430
431        let item3 = stream.next().await.unwrap().unwrap();
432        assert_eq!(item3.moniker.to_string(), "UNKNOWN");
433        assert_eq!(item3.msg().unwrap(), "msg unknown");
434
435        assert!(stream.next().await.is_none());
436    }
437}
log_command_ctf/fxt_streamer.rs

log_command_ctf/
fxt_streamer.rs
