// Copyright 2020 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use derivative::Derivative;
use fidl_fuchsia_diagnostics::StreamMode;
use fuchsia_sync::Mutex;
use futures::prelude::*;
use std::{
    collections::VecDeque,
    default::Default,
    fmt::Debug,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll, Waker},
};
use tracing::{trace, warn};

/// A list that can be iterated despite concurrent insertions and deletions.
pub struct ArcList<T> {
    inner: Arc<Mutex<InnerArcList<T>>>,
}

impl<T> Default for ArcList<T> {
    fn default() -> Self {
        Self { inner: Arc::new(Mutex::new(Default::default())) }
    }
}

impl<T> ArcList<T> {
    pub fn is_empty(&self) -> bool {
        self.inner.lock().items.is_empty()
    }

    pub fn push_back(&self, item: T) {
        self.inner.lock().push_back(item);
    }

    pub fn pop_front(&self) -> Option<Arc<T>> {
        self.inner.lock().pop_front()
    }

    pub fn peek_front(&self) -> Option<Arc<T>> {
        self.inner.lock().peek_front().map(|item| item.value)
    }

    pub fn cursor(&self, mode: StreamMode) -> Cursor<T> {
        let id = self.inner.lock().new_cursor_id();
        Cursor::new(id, self.clone(), mode)
    }

    /// End the stream, ignoring new values and causing Cursors to return None after the current ID.
    pub fn terminate(&self) {
        self.inner.lock().terminate();
    }

    #[cfg(test)]
    pub fn final_entry(&self) -> u64 {
        self.inner.lock().final_entry
    }
}

impl<T> Clone for ArcList<T> {
    fn clone(&self) -> Self {
        Self { inner: Arc::clone(&self.inner) }
    }
}

struct ArcListItem<T> {
    id: u64,
    value: Arc<T>,
}

impl<T> Clone for ArcListItem<T> {
    fn clone(&self) -> Self {
        Self { id: self.id, value: Arc::clone(&self.value) }
    }
}

struct InnerArcList<T> {
    /// Map from entries_seen at the point that the entry is added to the list, to the value with
    /// that id.
    items: VecDeque<ArcListItem<T>>,
    /// The number of entries ever inserted into the list.
    entries_seen: u64,
    /// The number of entries ever removed from the list.
    entries_popped: u64,
    /// The last entry this list will yield.
    final_entry: u64,
    /// The next cursor this list will yield.
    next_cursor_id: CursorId,
    /// Wakers from subscribed cursors blocked on their next message.
    pending_cursors: Vec<(CursorId, Waker)>,
}

impl<T> Default for InnerArcList<T> {
    fn default() -> Self {
        Self {
            items: VecDeque::new(),
            entries_seen: 0,
            entries_popped: 0,
            final_entry: u64::MAX,
            next_cursor_id: CursorId::default(),
            pending_cursors: Vec::new(),
        }
    }
}

impl<T> InnerArcList<T> {
    fn push_back(&mut self, item: T) {
        self.entries_seen += 1;
        if self.entries_seen > self.final_entry {
            warn!("push_back() must not be called after terminate()");
            return;
        }

        let id = self.entries_seen;
        self.items.push_back(ArcListItem { id, value: Arc::new(item) });

        self.wake_pending();
    }

    fn pop_front(&mut self) -> Option<Arc<T>> {
        self.items.pop_front().map(|item| {
            self.entries_popped += 1;
            item.value
        })
    }

    fn peek_front(&self) -> Option<ArcListItem<T>> {
        self.items.front().cloned()
    }

    fn first_starting_at(&self, id: u64) -> Option<ArcListItem<T>> {
        self.items.front().and_then(|front_item| {
            if front_item.id < id {
                let index = id - front_item.id;
                self.items.get(index as usize).cloned()
            } else {
                Some(front_item.clone())
            }
        })
    }

    fn new_cursor_id(&mut self) -> CursorId {
        let new_next = CursorId(self.next_cursor_id.0 + 1);
        std::mem::replace(&mut self.next_cursor_id, new_next)
    }

    fn wake_pending(&mut self) {
        for (id, waker) in self.pending_cursors.drain(..) {
            trace!("Waking {:?} for entry {}.", id, self.entries_seen);
            waker.wake();
        }
    }

    fn terminate(&mut self) {
        tracing::debug!("terminating buffer");
        self.final_entry = self.entries_seen;
        self.wake_pending();
    }
}

/// A unique identifier for each cursor used to manage wakers.
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, PartialOrd, Ord)]
struct CursorId(u64);

/// A weak pointer into the buffer which is being concurrently iterated and modified.
///
/// A cursor iterates over nodes in the list, holding the id of the last node seen.
/// If that id is less than the first id found in the deque, the cursor starts at the beginning of
/// the list and first returns a count of items rolled out.
///
/// The count is maintained by giving each successive item in a list a monotonically increasing ID
/// and tracking the "high-water mark" of the largest/last ID seen.
///
/// # Modes
///
/// These IDs are also how we express snapshotting vs. subscribing. The mode determines the minimum
/// and maximum IDs the cursor will yield.
///
/// # Wraparound
///
/// IDs are stored as `u64` and the list would need to receive a new entry once every nanosecond
/// for ~580 consecutive years before ID allocations will wrap around. This eventuality is *not*
/// accounted for in the implementation. If you're reading this after observing a bug due to that,
/// congratulations.
#[derive(Derivative)]
#[derivative(Debug)]
pub struct Cursor<T> {
    id: CursorId,
    last_id_seen: Option<u64>,
    until_id: u64,
    #[derivative(Debug = "ignore")]
    list: ArcList<T>,
    mode: StreamMode,
}

impl<T> Cursor<T> {
    /// Construct a new cursor into the logs buffer. The `mode` passed determines the range over
    /// which the cursor operates:
    ///
    /// | mode      | first ID yielded        | last ID yielded         |
    /// |-----------|-------------------------|-------------------------|
    /// | snapshot  | 0                       | max at time of snapshot |
    /// | subscribe | max at time of snapshot | max ID possible         |
    /// | both      | 0                       | max ID possible         |
    fn new(id: CursorId, list: ArcList<T>, mode: StreamMode) -> Self {
        let from = match mode {
            StreamMode::Snapshot | StreamMode::SnapshotThenSubscribe => None,
            StreamMode::Subscribe => {
                let inner = list.inner.lock();
                Some(inner.entries_seen)
            }
        };

        let to = match mode {
            StreamMode::Snapshot => list.inner.lock().entries_seen,
            StreamMode::SnapshotThenSubscribe | StreamMode::Subscribe => u64::MAX,
        };

        Self { id, list, last_id_seen: from, until_id: to, mode }
    }

    fn maybe_register_for_wakeup(&mut self, cx: &mut Context<'_>) -> Poll<Option<LazyItem<T>>> {
        let mut root = self.list.inner.lock();
        let cursor_at_end = self.last_id_seen.unwrap_or(0) == root.final_entry;
        let list_fully_drained = root.final_entry == root.entries_popped;

        if root.entries_popped > self.last_id_seen.unwrap_or(0) {
            // This happens when entries were popped before they could be returned,
            // but there is not currently anything left to return.  We need
            // to update our position and return the number of entries that
            // were popped.
            let entries_missing = root.entries_popped - self.last_id_seen.unwrap_or(0);
            self.last_id_seen = Some(root.entries_popped);
            Poll::Ready(Some(LazyItem::ItemsRolledOut(entries_missing)))
        } else if cursor_at_end || list_fully_drained {
            trace!("{:?} has reached the end of the terminated stream.", self.id);
            Poll::Ready(None)
        } else if self.mode == StreamMode::Snapshot {
            // There are no further entries to return, and we are in snapshot mode. Report that we
            // are at the end of the stream rather than registering for a wakeup.
            Poll::Ready(None)
        } else {
            trace!("Registering {:?} for wakeup.", self.id);
            root.pending_cursors.push((self.id, cx.waker().clone()));
            root.pending_cursors.sort_by_key(|&(id, _)| id);
            root.pending_cursors.dedup_by_key(|&mut (id, _)| id);
            Poll::Pending
        }
    }
}

impl<T> Stream for Cursor<T> {
    type Item = LazyItem<T>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        trace!("{:?} polled.", self.id);
        if self.last_id_seen.unwrap_or(0) >= self.until_id {
            return Poll::Ready(None);
        }

        let next = match self.last_id_seen {
            Some(id) => self.list.inner.lock().first_starting_at(id + 1),
            None => {
                // otherwise start again at the head
                trace!("{:?} starting from the head of the list.", self.id);
                self.list.inner.lock().peek_front()
            }
        };

        if let Some(to_return) = next {
            if to_return.id > self.until_id {
                self.last_id_seen = Some(to_return.id);
                // we're past the end of this cursor's valid range
                trace!("{:?} is done.", self.id);
                return Poll::Ready(None);
            }

            // the number we missed is equal to the difference between
            // the last ID we saw and the ID *just before* the current value
            let num_missed = (to_return.id - 1) - self.last_id_seen.unwrap_or(0);
            let item = if num_missed > 0 {
                // advance the cursor's high-water mark by the number we missed
                // so we only report each rolled out item once
                trace!("{:?} reporting {} missed items.", self.id, num_missed);
                self.last_id_seen = Some(self.last_id_seen.unwrap_or(0) + num_missed);
                LazyItem::ItemsRolledOut(num_missed)
            } else {
                // we haven't missed anything, proceed normally
                trace!("{:?} yielding item {}.", self.id, to_return.id);
                self.last_id_seen = Some(to_return.id);
                LazyItem::Next(to_return.value)
            };

            Poll::Ready(Some(item))
        } else {
            // No further data is waiting in the stream. Depending on the stream mode, we may
            // register for a wakeup.
            self.maybe_register_for_wakeup(cx)
        }
    }
}

/// The next element in the stream or a marker of the number of items rolled out since last polled.
#[derive(Debug, PartialEq)]
pub enum LazyItem<T> {
    /// The next item in the stream.
    Next(Arc<T>),
    /// A count of the items dropped between the last call to poll_next and this one.
    ItemsRolledOut(u64),
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures::poll;

    impl<T: Debug> LazyItem<T> {
        #[track_caller]
        fn unwrap(self) -> Arc<T> {
            match self {
                LazyItem::Next(i) => i,
                LazyItem::ItemsRolledOut(n) => panic!("{n} unexpected rolled out items in test"),
            }
        }

        #[track_caller]
        fn expect_rolled_out(self, expected: u64) {
            match self {
                LazyItem::Next(i) => {
                    panic!("expected {expected} rolled out items, found Next({i:#?})")
                }
                LazyItem::ItemsRolledOut(n) => {
                    assert_eq!(n, expected, "wrong number of rolled out items")
                }
            }
        }
    }

    #[fuchsia::test]
    fn list_is_empty() {
        let list = ArcList::default();
        assert!(list.is_empty());
        list.push_back(1);
        assert!(!list.is_empty());
        list.pop_front();
        assert!(list.is_empty());
    }

    #[fuchsia::test]
    fn list_peek_front() {
        let list = ArcList::default();
        assert!(list.peek_front().is_none());
        list.push_back(1);
        list.push_back(2);
        assert_eq!(*list.peek_front().unwrap(), 1);
        list.pop_front();
        assert_eq!(*list.peek_front().unwrap(), 2);
        list.pop_front();
        assert!(list.peek_front().is_none());
    }

    #[fuchsia::test]
    async fn subscription_delivered_without_drops() {
        let list = ArcList::default();
        let mut early_sub = list.cursor(StreamMode::Subscribe);

        let mut first_from_early_sub = early_sub.next();
        assert_eq!(poll!(&mut first_from_early_sub), Poll::Pending);

        list.push_back(1);
        assert_eq!(*first_from_early_sub.await.unwrap().unwrap(), 1);

        // this subscription starts after the 1 we just pushed
        let mut middle_sub = list.cursor(StreamMode::Subscribe);

        let (mut second_from_early_sub, mut second_from_middle_sub) =
            (early_sub.next(), middle_sub.next());
        assert_eq!(poll!(&mut second_from_early_sub), Poll::Pending);
        assert_eq!(poll!(&mut second_from_middle_sub), Poll::Pending);

        list.push_back(2);
        assert_eq!(*second_from_early_sub.await.unwrap().unwrap(), 2);
        assert_eq!(*second_from_middle_sub.await.unwrap().unwrap(), 2);

        let mut late_sub = list.cursor(StreamMode::Subscribe);

        let (mut third_from_early_sub, mut third_from_middle_sub, mut third_from_late_sub) =
            (early_sub.next(), middle_sub.next(), late_sub.next());
        assert_eq!(poll!(&mut third_from_early_sub), Poll::Pending);
        assert_eq!(poll!(&mut third_from_middle_sub), Poll::Pending);
        assert_eq!(poll!(&mut third_from_late_sub), Poll::Pending);

        list.push_back(3);
        assert_eq!(*third_from_early_sub.await.unwrap().unwrap(), 3);
        assert_eq!(*third_from_middle_sub.await.unwrap().unwrap(), 3);
        assert_eq!(*third_from_late_sub.await.unwrap().unwrap(), 3);

        let mut nop_sub = list.cursor(StreamMode::Subscribe);

        let (
            mut fourth_from_early_sub,
            mut fourth_from_middle_sub,
            mut fourth_from_late_sub,
            mut fourth_from_nop_sub,
        ) = (early_sub.next(), middle_sub.next(), late_sub.next(), nop_sub.next());
        assert_eq!(poll!(&mut fourth_from_early_sub), Poll::Pending);
        assert_eq!(poll!(&mut fourth_from_middle_sub), Poll::Pending);
        assert_eq!(poll!(&mut fourth_from_late_sub), Poll::Pending);
        assert_eq!(poll!(&mut fourth_from_nop_sub), Poll::Pending);

        list.terminate();
        assert_eq!(fourth_from_early_sub.await, None);
        assert_eq!(fourth_from_middle_sub.await, None);
        assert_eq!(fourth_from_late_sub.await, None);
        assert_eq!(fourth_from_nop_sub.await, None);
    }

    #[fuchsia::test]
    async fn snapshot_delivered_without_drops() {
        let list = ArcList::default();
        let mut dead_cursor = list.cursor(StreamMode::Snapshot);
        assert_eq!(dead_cursor.next().await, None, "no items in the list");

        list.push_back(1);
        list.push_back(2);
        list.push_back(3);

        let mut middle_cursor = list.cursor(StreamMode::Snapshot);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 1);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 2);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 3);
        assert_eq!(dead_cursor.next().await, None, "no items in the list at snapshot");
        assert_eq!(dead_cursor.next().await, None, "no items in list at snapshot");
        assert_eq!(middle_cursor.next().await, None, "no items left in list");

        list.push_back(4);
        list.push_back(5);

        assert_eq!(dead_cursor.next().await, None, "no items in list at snapshot");
        assert_eq!(middle_cursor.next().await, None, "no items left in list at snapshot");

        let mut full_cursor = list.cursor(StreamMode::Snapshot);
        assert_eq!(*full_cursor.next().await.unwrap().unwrap(), 1);
        assert_eq!(*full_cursor.next().await.unwrap().unwrap(), 2);
        assert_eq!(*full_cursor.next().await.unwrap().unwrap(), 3);
        assert_eq!(*full_cursor.next().await.unwrap().unwrap(), 4);
        assert_eq!(*full_cursor.next().await.unwrap().unwrap(), 5);
        assert_eq!(full_cursor.next().await, None, "no items left");
        assert_eq!(dead_cursor.next().await, None, "no items in list at snapshot");
        assert_eq!(middle_cursor.next().await, None, "no items left in list at snapshot");
    }

    #[fuchsia::test]
    async fn snapshot_then_subscribe_gets_before_and_after() {
        let list: ArcList<i32> = ArcList::default();

        list.push_back(1);
        list.push_back(2);
        list.push_back(3);

        let mut middle_cursor = list.cursor(StreamMode::SnapshotThenSubscribe);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 1);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 2);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 3);

        list.push_back(4);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 4);

        list.push_back(5);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 5);

        list.push_back(6);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 6);

        list.terminate();
        assert_eq!(middle_cursor.next().await, None);
    }

    #[fuchsia::test]
    async fn subscription_drops_are_counted() {
        let list: ArcList<i32> = ArcList::default();
        let mut early_cursor = list.cursor(StreamMode::Subscribe);

        list.push_back(1);
        list.push_back(2);
        list.push_back(3);
        assert_eq!(*list.pop_front().unwrap(), 1);

        early_cursor.next().await.unwrap().expect_rolled_out(1);
        assert_eq!(*early_cursor.next().await.unwrap().unwrap(), 2);

        let mut middle_cursor = list.cursor(StreamMode::Subscribe);

        assert_eq!(*list.pop_front().unwrap(), 2);
        assert_eq!(*list.pop_front().unwrap(), 3);
        list.push_back(4);
        list.push_back(5);

        early_cursor.next().await.unwrap().expect_rolled_out(1);
        assert_eq!(*early_cursor.next().await.unwrap().unwrap(), 4);
        assert_eq!(*early_cursor.next().await.unwrap().unwrap(), 5);

        assert_eq!(*list.pop_front().unwrap(), 4);
        middle_cursor.next().await.unwrap().expect_rolled_out(1);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 5);

        list.terminate();
        assert_eq!(middle_cursor.next().await, None);
    }

    #[fuchsia::test]
    async fn snapshot_drops_are_counted() {
        let list: ArcList<i32> = ArcList::default();
        let mut dead_cursor = list.cursor(StreamMode::Snapshot);
        assert!(dead_cursor.next().await.is_none(), "no items in the list");

        list.push_back(1);
        list.push_back(2);
        list.push_back(3);
        list.push_back(4);
        list.push_back(5);

        let mut middle_cursor = list.cursor(StreamMode::Snapshot);
        list.pop_front();
        list.pop_front();

        middle_cursor.next().await.unwrap().expect_rolled_out(2);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 3);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 4);
        assert_eq!(*middle_cursor.next().await.unwrap().unwrap(), 5);
        assert!(dead_cursor.next().await.is_none(), "no items in list at snapshot");
        assert!(middle_cursor.next().await.is_none(), "no items left in list");

        let mut full_cursor = list.cursor(StreamMode::Snapshot);
        full_cursor.next().await.unwrap().expect_rolled_out(2);
        assert_eq!(*full_cursor.next().await.unwrap().unwrap(), 3);
        assert_eq!(*full_cursor.next().await.unwrap().unwrap(), 4);
        assert_eq!(*full_cursor.next().await.unwrap().unwrap(), 5);
        assert!(full_cursor.next().await.is_none(), "no items left");
        assert!(dead_cursor.next().await.is_none(), "no items in list at snapshot");
        assert!(middle_cursor.next().await.is_none(), "no items left in list at snapshot");
    }

    #[fuchsia::test]
    async fn post_termination_cursors() {
        let list: ArcList<i32> = ArcList::default();
        list.push_back(1);
        list.push_back(2);
        list.push_back(3);
        list.push_back(4);
        list.push_back(5);
        list.terminate();

        let snapshot: Vec<_> =
            list.cursor(StreamMode::Snapshot).map(|i| *i.unwrap()).collect().await;
        let subscribe: Vec<_> =
            list.cursor(StreamMode::Subscribe).map(|i| *i.unwrap()).collect().await;
        let both: Vec<_> =
            list.cursor(StreamMode::SnapshotThenSubscribe).map(|i| *i.unwrap()).collect().await;

        assert_eq!(snapshot, vec![1, 2, 3, 4, 5]);
        assert!(subscribe.is_empty());
        assert_eq!(both, vec![1, 2, 3, 4, 5]);
    }

    #[fuchsia::test]
    async fn empty_post_termination_cursors() {
        let list: ArcList<i32> = ArcList::default();
        list.terminate();

        let snapshot: Vec<_> =
            list.cursor(StreamMode::Snapshot).map(|i| *i.unwrap()).collect().await;
        let subscribe: Vec<_> =
            list.cursor(StreamMode::Subscribe).map(|i| *i.unwrap()).collect().await;
        let both: Vec<_> =
            list.cursor(StreamMode::SnapshotThenSubscribe).map(|i| *i.unwrap()).collect().await;

        assert!(snapshot.is_empty());
        assert!(subscribe.is_empty());
        assert!(both.is_empty());
    }

    #[fuchsia::test]
    async fn drained_post_termination_cursors() {
        let list: ArcList<i32> = ArcList::default();
        list.push_back(1);
        list.push_back(2);
        list.push_back(3);
        list.push_back(4);
        list.push_back(5);
        list.terminate();
        list.pop_front();
        list.pop_front();
        list.pop_front();
        list.pop_front();
        list.pop_front();

        let mut snapshot_cursor = list.cursor(StreamMode::Snapshot);
        snapshot_cursor.next().await.unwrap().expect_rolled_out(5);
        let snapshot: Vec<_> = snapshot_cursor.map(|i| *i.unwrap()).collect().await;
        let subscribe: Vec<_> =
            list.cursor(StreamMode::Subscribe).map(|i| *i.unwrap()).collect().await;
        let mut both_cursor = list.cursor(StreamMode::Snapshot);
        both_cursor.next().await.unwrap().expect_rolled_out(5);
        let both: Vec<_> = both_cursor.map(|i| *i.unwrap()).collect().await;

        assert!(snapshot.is_empty());
        assert!(subscribe.is_empty());
        assert!(both.is_empty());
    }

    #[fuchsia::test]
    async fn snapshot_does_not_hang_when_nothing_is_left() {
        let list: ArcList<i32> = ArcList::default();
        list.push_back(1);
        list.push_back(2);
        list.push_back(3);
        list.push_back(4);
        list.push_back(5);
        list.pop_front();
        list.pop_front();
        list.pop_front();
        list.pop_front();
        list.pop_front();

        let mut cursor = list.cursor(StreamMode::Snapshot);

        cursor.next().await.unwrap().expect_rolled_out(5);

        let snapshot: Vec<_> = cursor.map(|i| *i.unwrap()).collect().await;

        assert!(snapshot.is_empty());
    }

    #[fuchsia::test]
    async fn snapshot_does_not_hang_when_values_are_popped_before_start() {
        let list: ArcList<i32> = ArcList::default();
        list.push_back(1);
        list.push_back(2);
        list.push_back(3);
        list.push_back(4);
        list.push_back(5);
        list.pop_front();
        list.pop_front();
        list.pop_front();

        let mut cursor = list.cursor(StreamMode::Snapshot);
        cursor.next().await.expect("initial value exists").expect_rolled_out(3);

        let snapshot: Vec<_> = cursor.map(|i| *i.unwrap()).collect().await;

        assert_eq!(2, snapshot.len());
    }

    #[fuchsia::test]
    async fn popping_more_elements_than_exist_does_not_break_readers() {
        let list: ArcList<i32> = ArcList::default();
        list.push_back(1);
        list.push_back(2);
        list.push_back(3);
        let mut cursor = list.cursor(StreamMode::SnapshotThenSubscribe);

        assert_eq!(*cursor.next().await.unwrap().unwrap(), 1);
        assert_eq!(*cursor.next().await.unwrap().unwrap(), 2);
        assert_eq!(*cursor.next().await.unwrap().unwrap(), 3);
        let mut next = cursor.next();
        assert_eq!(poll!(&mut next), Poll::Pending);

        list.pop_front();
        list.pop_front();
        list.pop_front();
        list.pop_front();
        list.pop_front();

        let mut next = cursor.next();
        assert_eq!(poll!(&mut next), Poll::Pending);
        list.push_back(4);
        assert_eq!(*cursor.next().await.unwrap().unwrap(), 4);
    }

    #[fuchsia::test]
    async fn snapshot_then_subscribe_works_when_only_dropped_notifications_are_returned() {
        let list: ArcList<i32> = ArcList::default();
        list.push_back(1);
        list.push_back(2);
        list.push_back(3);
        list.pop_front();
        list.pop_front();
        list.pop_front();

        let mut cursor = list.cursor(StreamMode::SnapshotThenSubscribe);
        cursor.next().await.unwrap().expect_rolled_out(3);
        let mut next = cursor.next();
        assert_eq!(poll!(&mut next), Poll::Pending);

        list.push_back(4);
        list.pop_front();

        cursor.next().await.unwrap().expect_rolled_out(1);
        let mut next = cursor.next();
        assert_eq!(poll!(&mut next), Poll::Pending);

        list.terminate();
        let snapshot: Vec<_> = cursor.map(|i| *i.unwrap()).collect().await;
        assert!(snapshot.is_empty());
    }
}