archivist_lib/logs/shared_buffer/
cursor.rs

1// Copyright 2025 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 super::{ContainerId, Inner, InnerGuard, SharedBuffer};
6use crate::identity::ComponentIdentity;
7use diagnostics_data::{LogError, LogsData};
8use diagnostics_log_encoding::parse::ParseError;
9use diagnostics_log_encoding::{Header, TRACING_FORMAT_LOG_RECORD_TYPE};
10use diagnostics_message::error::MessageError;
11use fidl_fuchsia_diagnostics::ComponentSelector;
12use fuchsia_async::condition::{ConditionGuard, WakerEntry};
13use futures::Stream;
14use futures::stream::FusedStream;
15use pin_project::pin_project;
16use ring_buffer::ring_buffer_record_len;
17use selectors::matches_selectors;
18use std::cmp::Ordering;
19use std::collections::BinaryHeap;
20use std::marker::PhantomData;
21use std::pin::Pin;
22use std::sync::Arc;
23use std::task::{Context, Poll, ready};
24use zerocopy::FromBytes;
25
26/// FilterCursor is a cursor that returns all logs optionally filtered by component selectors.
27#[pin_project]
28pub struct FilterCursor {
29    buffer: Arc<SharedBuffer>,
30    index: u64,
31    message_id: u64,
32    end: Option<u64>,
33    selectors: Vec<ComponentSelector>,
34    messages: BinaryHeap<MessageRef>,
35    flush_sockets_for_snapshot: bool,
36    #[pin]
37    waker_entry: WakerEntry<Inner>,
38}
39
40impl FilterCursor {
41    pub fn new(
42        buffer: Arc<SharedBuffer>,
43        index: u64,
44        message_id: u64,
45        snapshot: bool,
46        selectors: Vec<ComponentSelector>,
47    ) -> Self {
48        let waker_entry = buffer.inner.waker_entry();
49        Self {
50            buffer,
51            index,
52            message_id,
53            end: None,
54            selectors,
55            messages: BinaryHeap::new(),
56            waker_entry,
57            flush_sockets_for_snapshot: snapshot,
58        }
59    }
60
61    /// Polls for the next message.  We can't use the Stream trait because this is a lending
62    /// iterator; it returns a reference to a message (and will hold a lock).  Depending on the use,
63    /// this can be more efficient than using a Stream which will involve copying the message.
64    pub fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Message<'_>>> {
65        let this = self.project();
66
67        if *this.flush_sockets_for_snapshot {
68            let mut inner = InnerGuard::new(this.buffer);
69            *this.end = Some(this.buffer.flush_sockets(&mut inner));
70            *this.flush_sockets_for_snapshot = false;
71        }
72
73        let mut inner = this.buffer.inner.lock();
74
75        // Comparing monikers can be expensive, so memoize the results.
76        struct ContainerIds {
77            ids: [ContainerId; 8],
78            result: u8,
79            pos: usize,
80        }
81
82        impl ContainerIds {
83            /// Memoizes the result of `predicate` for a given `id`.
84            fn memoize(&mut self, id: ContainerId, predicate: impl FnOnce() -> bool) -> bool {
85                if let Some(pos) = self.ids.iter().position(|i| i == &id) {
86                    self.result & (1 << pos) != 0
87                } else {
88                    let result = predicate();
89                    self.ids[self.pos] = id;
90                    if result {
91                        self.result |= 1 << self.pos;
92                    } else {
93                        self.result &= !(1 << self.pos);
94                    }
95                    self.pos = (self.pos + 1) % self.ids.len();
96                    result
97                }
98            }
99        }
100
101        let mut container_ids = ContainerIds { ids: [ContainerId(0xffff); 8], result: 0, pos: 0 };
102
103        // NOTE: If messages are dropped the dropped count won't account for filtering: it will
104        // include messages that have been dropped that don't match the filter.  Fixing this is
105        // difficult and not worth the effort.  Dropped messages should be rare and the common case
106        // is that there is no filtering.
107        let mut dropped = 0;
108        if *this.index < inner.tail {
109            *this.index = inner.tail;
110            dropped += inner.tail_message_id - *this.message_id;
111            *this.message_id = inner.tail_message_id;
112        }
113        while *this.index < inner.last_scanned && this.end.is_none_or(|e| *this.index < e) {
114            // SAFETY: We've checked index >= inner.tail, so the range must be valid.  `msg`
115            // will remain valid whilst we're holding the lock.
116            let (component, msg, timestamp) =
117                unsafe { inner.parse_message(*this.index..inner.last_scanned) };
118
119            if let Some(timestamp) = timestamp
120                && let Some(container) = inner.containers.get(component)
121                && (this.selectors.is_empty()
122                    || container_ids.memoize(component, || {
123                        matches_selectors(&container.identity.moniker, this.selectors)
124                    }))
125            {
126                this.messages.push(MessageRef { index: *this.index, timestamp });
127            }
128
129            *this.index += ring_buffer_record_len(msg.len()) as u64;
130            *this.message_id += 1;
131        }
132        while let Some(message) = this.messages.pop() {
133            if message.index < inner.tail {
134                // See note above regarding filtering.
135                dropped += 1;
136                continue;
137            }
138            return Poll::Ready(Some(Message { inner, message, dropped }));
139        }
140        if this.end.is_some_and(|e| *this.index >= e) || inner.terminated {
141            *this.index = u64::MAX;
142            Poll::Ready(None)
143        } else {
144            inner.add_waker(this.waker_entry, cx.waker().clone());
145            Poll::Pending
146        }
147    }
148
149    fn is_terminated(&self) -> bool {
150        self.index == u64::MAX
151    }
152}
153
154#[derive(Eq)]
155struct MessageRef {
156    timestamp: zx::BootInstant,
157    index: u64,
158}
159
160impl PartialEq for MessageRef {
161    fn eq(&self, other: &Self) -> bool {
162        self.timestamp == other.timestamp && self.index == other.index
163    }
164}
165
166impl PartialOrd for MessageRef {
167    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
168        Some(self.cmp(other))
169    }
170}
171
172impl Ord for MessageRef {
173    fn cmp(&self, other: &Self) -> Ordering {
174        // BinaryHeap is a max-heap, but we want min, hence this ordering.
175        other.timestamp.cmp(&self.timestamp).then_with(|| other.index.cmp(&self.index))
176    }
177}
178
179pub struct Message<'a> {
180    inner: ConditionGuard<'a, Inner>,
181    message: MessageRef,
182    dropped: u64,
183}
184
185impl Message<'_> {
186    /// Returns the component and FXT bytes.  The FXT record is validated to be the correct length
187    /// and type.
188    fn parse(&self) -> Result<(&Arc<ComponentIdentity>, &[u8]), MessageError> {
189        // SAFETY: We hold a lock which prevents the buffer from being drained so
190        // it should be safe to read this range.
191        let (container, data, _) =
192            unsafe { self.inner.parse_message(self.message.index..self.inner.last_scanned) };
193        let (header, _) = Header::read_from_prefix(data)
194            .map_err(|_| MessageError::from(ParseError::InvalidHeader))?;
195        let msg_len = header.size_words() as usize * 8;
196        if msg_len > data.len() || msg_len < 16 {
197            return Err(ParseError::ValueOutOfValidRange.into());
198        }
199        if header.raw_type() != TRACING_FORMAT_LOG_RECORD_TYPE {
200            return Err(ParseError::InvalidRecordType.into());
201        }
202        let container = self.inner.containers.get(container).unwrap();
203        Ok((&container.identity, &data[..msg_len]))
204    }
205}
206
207impl TryFrom<Message<'_>> for LogsData {
208    type Error = MessageError;
209
210    fn try_from(value: Message<'_>) -> Result<Self, Self::Error> {
211        let (container, data) = value.parse()?;
212        let mut data = diagnostics_message::from_structured(container.as_ref().into(), data)?;
213        if value.dropped > 0 {
214            data.metadata
215                .errors
216                .get_or_insert(vec![])
217                .push(LogError::RolledOutLogs { count: value.dropped });
218        }
219        Ok(data)
220    }
221}
222
223/// A copy of the raw message.
224#[derive(Debug)]
225pub struct FxtMessage {
226    data: Box<[u8]>,
227    dropped: u64,
228    component_identity: Arc<ComponentIdentity>,
229}
230
231impl FxtMessage {
232    pub fn data(&self) -> &[u8] {
233        &self.data
234    }
235
236    pub fn dropped(&self) -> u64 {
237        self.dropped
238    }
239
240    pub fn component_identity(&self) -> &ComponentIdentity {
241        &self.component_identity
242    }
243
244    pub fn timestamp(&self) -> zx::BootInstant {
245        zx::BootInstant::from_nanos(i64::read_from_bytes(&self.data[8..16]).unwrap())
246    }
247}
248
249impl TryFrom<Message<'_>> for FxtMessage {
250    type Error = MessageError;
251
252    fn try_from(value: Message<'_>) -> Result<Self, Self::Error> {
253        let (component, data) = value.parse()?;
254        Ok(FxtMessage {
255            data: data.into(),
256            dropped: value.dropped,
257            component_identity: Arc::clone(component),
258        })
259    }
260}
261
262/// Like FilterCursor but returns a stream of LogsData.
263#[pin_project]
264pub struct FilterCursorStream<T> {
265    #[pin]
266    cursor: FilterCursor,
267    phantom: PhantomData<T>,
268}
269
270impl<T> Stream for FilterCursorStream<T>
271where
272    T: for<'a> TryFrom<Message<'a>>,
273{
274    type Item = T;
275
276    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
277        let mut this = self.project();
278        loop {
279            match ready!(this.cursor.as_mut().poll_next(cx)) {
280                Some(item) => {
281                    if let Ok(data) = T::try_from(item) {
282                        return Poll::Ready(Some(data));
283                    }
284                    // The message is bad, just ignore it.
285                }
286                None => return Poll::Ready(None),
287            }
288        }
289    }
290}
291
292impl<T> FusedStream for FilterCursorStream<T>
293where
294    T: for<'a> TryFrom<Message<'a>>,
295{
296    fn is_terminated(&self) -> bool {
297        self.cursor.is_terminated()
298    }
299}
300
301impl<T> From<FilterCursor> for FilterCursorStream<T> {
302    fn from(cursor: FilterCursor) -> Self {
303        Self { cursor, phantom: PhantomData }
304    }
305}
306
307#[cfg(test)]
308mod tests {
309    use super::*;
310    use crate::logs::shared_buffer::{
311        InnerGuard, SharedBuffer, SharedBufferOptions, create_ring_buffer,
312    };
313    use crate::logs::testing::make_message;
314    use assert_matches::assert_matches;
315    use fidl_fuchsia_diagnostics::StreamMode;
316    use futures::StreamExt;
317    use selectors::{FastError, parse_component_selector};
318    use std::pin::pin;
319    use std::time::Duration;
320
321    #[fuchsia::test]
322    async fn cursor_basic() {
323        let buffer =
324            SharedBuffer::new(create_ring_buffer(65536), Box::new(|_| {}), Default::default());
325        let container = buffer.new_container_buffer(Arc::new(vec!["a"].into()), Arc::default());
326        let msg = make_message("a", None, zx::BootInstant::from_nanos(1));
327        container.push_back(msg.bytes());
328
329        let cursor = buffer.cursor(StreamMode::Snapshot, vec![]);
330        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
331
332        let item = stream.next().await.unwrap();
333        assert_eq!(item.msg().unwrap(), "a");
334        assert!(stream.next().await.is_none());
335    }
336
337    #[fuchsia::test]
338    async fn cursor_filter() {
339        let buffer =
340            SharedBuffer::new(create_ring_buffer(65536), Box::new(|_| {}), Default::default());
341        let container_a = buffer.new_container_buffer(Arc::new(vec!["a"].into()), Arc::default());
342        let container_b = buffer.new_container_buffer(Arc::new(vec!["b"].into()), Arc::default());
343
344        container_a.push_back(make_message("msg_a", None, zx::BootInstant::from_nanos(1)).bytes());
345        container_b.push_back(make_message("msg_b", None, zx::BootInstant::from_nanos(2)).bytes());
346
347        let selector = parse_component_selector::<FastError>("a").unwrap();
348        let cursor = buffer.cursor(StreamMode::Snapshot, vec![selector]);
349        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
350
351        let item = stream.next().await.unwrap();
352        assert_eq!(item.msg().unwrap(), "msg_a");
353        assert!(stream.next().await.is_none());
354    }
355
356    #[fuchsia::test]
357    async fn cursor_subscribe() {
358        let buffer =
359            SharedBuffer::new(create_ring_buffer(65536), Box::new(|_| {}), Default::default());
360        let container = buffer.new_container_buffer(Arc::new(vec!["a"].into()), Arc::default());
361
362        let cursor = buffer.cursor(StreamMode::Subscribe, vec![]);
363        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
364
365        assert!(futures::poll!(stream.next()).is_pending());
366
367        container.push_back(make_message("msg", None, zx::BootInstant::from_nanos(1)).bytes());
368
369        let item = stream.next().await.unwrap();
370        assert_eq!(item.msg().unwrap(), "msg");
371    }
372
373    #[fuchsia::test]
374    async fn cursor_snapshot_then_subscribe() {
375        let buffer =
376            SharedBuffer::new(create_ring_buffer(65536), Box::new(|_| {}), Default::default());
377        let container = buffer.new_container_buffer(Arc::new(vec!["a"].into()), Arc::default());
378
379        container.push_back(make_message("msg1", None, zx::BootInstant::from_nanos(1)).bytes());
380
381        let cursor = buffer.cursor(StreamMode::SnapshotThenSubscribe, vec![]);
382        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
383
384        let item = stream.next().await.unwrap();
385        assert_eq!(item.msg().unwrap(), "msg1");
386
387        assert!(futures::poll!(stream.next()).is_pending());
388
389        container.push_back(make_message("msg2", None, zx::BootInstant::from_nanos(2)).bytes());
390
391        let item = stream.next().await.unwrap();
392        assert_eq!(item.msg().unwrap(), "msg2");
393    }
394
395    #[fuchsia::test]
396    async fn cursor_dropped_logs() {
397        let buffer = SharedBuffer::new(
398            create_ring_buffer(65536),
399            Box::new(|_| {}),
400            SharedBufferOptions { sleep_time: Duration::ZERO },
401        );
402        let container_a = buffer.new_container_buffer(Arc::new(vec!["a"].into()), Arc::default());
403        let container_b = buffer.new_container_buffer(Arc::new(vec!["b"].into()), Arc::default());
404
405        let msg = make_message("msg", None, zx::BootInstant::from_nanos(1));
406
407        container_a.push_back(msg.bytes()); // 1
408        container_a.push_back(msg.bytes()); // 2
409        container_a.push_back(msg.bytes()); // 3
410
411        let cursor = buffer.cursor(StreamMode::SnapshotThenSubscribe, vec![]);
412        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
413
414        // Consume one
415        let item = stream.next().await.unwrap();
416        assert_eq!(item.msg().unwrap(), "msg");
417
418        // Force roll out of remaining A messages
419        while pin!(FilterCursorStream::<FxtMessage>::from(buffer.cursor(
420            StreamMode::Snapshot,
421            vec![parse_component_selector::<FastError>("a").unwrap()]
422        )))
423        .next()
424        .await
425        .is_some()
426        {
427            container_b.push_back(msg.bytes());
428            // Force the buffer to pop old messages.
429            {
430                let mut inner = InnerGuard::new(&buffer);
431                inner.check_space(inner.ring_buffer.head());
432            }
433        }
434
435        // Next item should indicate the two dropped messages for container A.
436        let item = stream.next().await.unwrap();
437        assert_eq!(item.metadata.errors.as_ref().unwrap().len(), 1);
438        assert_matches!(
439            item.metadata.errors.as_ref().unwrap()[0],
440            LogError::RolledOutLogs { count } if count == 2
441        );
442    }
443
444    #[fuchsia::test]
445    async fn terminate_terminates_cursor() {
446        let buffer = SharedBuffer::new(
447            create_ring_buffer(65536),
448            Box::new(|_| {}),
449            SharedBufferOptions { sleep_time: Duration::from_secs(10) },
450        );
451        let cursor = pin!(FilterCursorStream::<LogsData>::from(
452            buffer.cursor(StreamMode::SnapshotThenSubscribe, vec![])
453        ));
454        let container = buffer.new_container_buffer(Arc::new(vec!["a"].into()), Arc::default());
455        container.push_back(make_message("msg", None, zx::BootInstant::from_nanos(1)).bytes());
456        drop(buffer.terminate());
457        assert_eq!(cursor.count().await, 1);
458    }
459}
archivist_lib/logs/shared_buffer/cursor.rs

archivist_lib/logs/shared_buffer/
cursor.rs
