datum/

queue.rs

//! Bounded and unbounded queues that bridge external producers into a Datum stream.
//!
//! [`SourceQueue`] / [`BoundedSourceQueue`] let external code feed elements into a running stream,
//! while [`SinkQueue`] lets external code pull materialized results out on demand.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering, fence};
use std::sync::{Arc, Condvar, Mutex, OnceLock};
use std::time::Duration;

use crossbeam_queue::ArrayQueue;

use crate::stream::{BoxStream, NotUsed, OverflowStrategy, Sink, Source, StreamCompletion};
use crate::{StreamError, StreamResult};
use futures::channel::oneshot;

/// Outcome of offering one element to a source queue.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum QueueOfferResult {
    /// The element was accepted into the buffer.
    Enqueued,
    /// The element was dropped (buffer full under a drop overflow strategy).
    Dropped,
    /// The queue is completed/failed and no longer accepts elements.
    QueueClosed,
    /// The offer failed the stream (the `Fail` overflow strategy on overflow).
    Failure(StreamError),
}

#[derive(Clone)]
enum TerminalSignal {
    Complete,
    Error(StreamError),
}

// ── BoundedSourceQueue (synchronous, bounded, no overflow strategy) ─────────
//
// The element buffer is a lock-free bounded MPMC queue (`crossbeam_queue::
// ArrayQueue`), so producers and the draining consumer never contend on a
// shared mutex on the hot path and no per-element allocation occurs (the
// backing array is allocated once). Terminal state (complete/fail) is a small
// atomic checked with a single load per offer. The consumer parks on a condvar
// only when it observes the buffer empty; producers wake it via a `parked` flag
// guarded by a `SeqCst` fence that closes the classic store/load (Dekker) race,
// and each park is additionally bounded by `PARK_BACKSTOP` so any
// theoretically-missed wakeup self-heals rather than deadlocking.

const TERM_NONE: u8 = 0;
const TERM_COMPLETE: u8 = 1;
const TERM_ERROR: u8 = 2;

/// Upper bound on a single consumer park. Wakeups are normally delivered
/// promptly by `wake_consumer`; this bound is a defensive backstop so that a
/// lost wakeup (should one ever slip past the fence protocol) costs at most this
/// much latency instead of deadlocking the stream.
const PARK_BACKSTOP: Duration = Duration::from_millis(10);

struct BoundedQueueShared<T> {
    buffer: ArrayQueue<T>,
    /// `TERM_NONE` / `TERM_COMPLETE` / `TERM_ERROR`; sticky once non-`NONE`.
    terminal: AtomicU8,
    /// Failure payload, published before `terminal` flips to `TERM_ERROR`.
    error: OnceLock<StreamError>,
    /// Guards only the condvar wait; touched solely on the slow park/wake path.
    park: Mutex<()>,
    available: Condvar,
    /// True while the consumer is (about to be) blocked in `wait_timeout`.
    parked: AtomicBool,
}

impl<T> BoundedQueueShared<T> {
    fn new(capacity: usize) -> Arc<Self> {
        Arc::new(Self {
            buffer: ArrayQueue::new(capacity),
            terminal: AtomicU8::new(TERM_NONE),
            error: OnceLock::new(),
            park: Mutex::new(()),
            available: Condvar::new(),
            parked: AtomicBool::new(false),
        })
    }

    /// Mark the queue completed (sticky: the first terminal signal wins).
    fn set_complete(&self) {
        let _ = self.terminal.compare_exchange(
            TERM_NONE,
            TERM_COMPLETE,
            Ordering::AcqRel,
            Ordering::Relaxed,
        );
        self.wake_consumer();
    }

    /// Mark the queue failed (sticky: the first terminal signal wins).
    fn set_error(&self, error: StreamError) {
        // Publish the payload before flipping the state so a consumer that
        // observes `TERM_ERROR` can always read it back.
        let _ = self.error.set(error);
        let _ = self.terminal.compare_exchange(
            TERM_NONE,
            TERM_ERROR,
            Ordering::AcqRel,
            Ordering::Relaxed,
        );
        self.wake_consumer();
    }

    /// Wake the consumer if it is (or is about to be) parked. In steady state —
    /// the consumer actively draining — this is a single fenced atomic load and
    /// touches no lock.
    fn wake_consumer(&self) {
        // Ordered after the producer's enqueue / terminal store. Pairs with the
        // fence on the consumer's park path: the two `SeqCst` fences cannot be
        // mutually reordered, so the producer's enqueue and the consumer's
        // `parked` store cannot both be missed.
        fence(Ordering::SeqCst);
        if self.parked.load(Ordering::Relaxed) {
            let _guard = self
                .park
                .lock()
                .unwrap_or_else(|poison| poison.into_inner());
            self.available.notify_one();
        }
    }
}

/// Materialized handle of [`Source::queue_bounded`]: a synchronous, bounded, lock-free queue with
/// no overflow strategy (a full buffer drops). `Clone`; the stream completes when the last handle
/// is dropped.
#[derive(Clone)]
pub struct BoundedSourceQueue<T> {
    shared: Arc<BoundedQueueShared<T>>,
}

impl<T> BoundedSourceQueue<T> {
    /// Offer one element. Returns immediately (never blocks): [`QueueOfferResult::Enqueued`] on
    /// success, [`QueueOfferResult::Dropped`] if the buffer is full, or
    /// [`QueueOfferResult::QueueClosed`] after completion/failure.
    pub fn offer(&self, elem: T) -> QueueOfferResult {
        if self.shared.terminal.load(Ordering::Acquire) != TERM_NONE {
            return QueueOfferResult::QueueClosed;
        }
        match self.shared.buffer.push(elem) {
            Ok(()) => {
                self.shared.wake_consumer();
                QueueOfferResult::Enqueued
            }
            Err(_full) => QueueOfferResult::Dropped,
        }
    }

    /// Complete the stream after buffered elements drain (idempotent; first terminal signal wins).
    pub fn complete(&self) {
        self.shared.set_complete();
    }

    /// Fail the stream; buffered elements drain before the error surfaces downstream.
    pub fn fail(&self, error: StreamError) {
        self.shared.set_error(error);
    }
}

impl<T> Drop for BoundedSourceQueue<T> {
    fn drop(&mut self) {
        // Only the last live producer handle (handle + stream == 2) completes
        // the stream; earlier clones leave it running.
        if Arc::strong_count(&self.shared) != 2 {
            return;
        }
        self.shared.set_complete();
    }
}

struct BoundedQueueStream<T> {
    shared: Arc<BoundedQueueShared<T>>,
}

impl<T> Iterator for BoundedQueueStream<T> {
    type Item = StreamResult<T>;

    fn next(&mut self) -> Option<Self::Item> {
        let shared = &*self.shared;
        loop {
            // Drain buffered elements before honoring any terminal signal.
            if let Some(item) = shared.buffer.pop() {
                return Some(Ok(item));
            }
            match shared.terminal.load(Ordering::Acquire) {
                TERM_COMPLETE => {
                    // Synchronizing with `complete()` can make a producer's
                    // prior push visible only after the first empty pop above.
                    // Re-drain once before honoring the terminal signal so a
                    // last accepted element is not lost at the boundary.
                    if let Some(item) = shared.buffer.pop() {
                        return Some(Ok(item));
                    }
                    return None;
                }
                TERM_ERROR => {
                    if let Some(item) = shared.buffer.pop() {
                        return Some(Ok(item));
                    }
                    let error = shared
                        .error
                        .get()
                        .cloned()
                        .unwrap_or_else(|| StreamError::Failed("queue failed".into()));
                    return Some(Err(error));
                }
                _ => {}
            }

            // Empty and live: park until a producer enqueues or signals.
            let guard = shared
                .park
                .lock()
                .unwrap_or_else(|poison| poison.into_inner());
            shared.parked.store(true, Ordering::Relaxed);
            // Re-check after publishing `parked`. The fence pairs with the one
            // in `wake_consumer`, so an enqueue that landed just before we set
            // `parked` is observed here rather than slept through.
            fence(Ordering::SeqCst);
            if !shared.buffer.is_empty() || shared.terminal.load(Ordering::Acquire) != TERM_NONE {
                shared.parked.store(false, Ordering::Relaxed);
                drop(guard);
                continue;
            }
            let (guard, _timeout) = shared
                .available
                .wait_timeout(guard, PARK_BACKSTOP)
                .unwrap_or_else(|poison| poison.into_inner());
            shared.parked.store(false, Ordering::Relaxed);
            drop(guard);
        }
    }
}

impl<T> Drop for BoundedQueueStream<T> {
    fn drop(&mut self) {
        // Stream gone: close the queue so later offers see `QueueClosed`.
        self.shared.set_complete();
    }
}

// ── SourceQueue (with overflow strategy, maxConcurrentOffers = 1) ──────────

struct SourceQueueShared<T> {
    state: Mutex<SourceQueueState<T>>,
    available: Condvar,
    capacity: usize,
    strategy: OverflowStrategy,
}

struct SourceQueueState<T> {
    buffer: VecDeque<T>,
    terminal: Option<TerminalSignal>,
    terminating: bool,
    pending_count: usize,
}

impl<T> SourceQueueShared<T> {
    fn new(capacity: usize, strategy: OverflowStrategy) -> Arc<Self> {
        Arc::new(Self {
            state: Mutex::new(SourceQueueState {
                buffer: VecDeque::with_capacity(capacity),
                terminal: None,
                terminating: false,
                pending_count: 0,
            }),
            available: Condvar::new(),
            capacity,
            strategy,
        })
    }
}

/// Materialized handle of [`Source::queue`]: a bounded queue with a configurable
/// [`OverflowStrategy`]. Under `Backpressure`, `maxConcurrentOffers` is 1 — `offer` blocks until
/// buffer space is free and a second concurrent offer errors.
pub struct SourceQueue<T> {
    shared: Arc<SourceQueueShared<T>>,
    completion: Option<StreamCompletion<NotUsed>>,
}

impl<T> Clone for SourceQueue<T> {
    fn clone(&self) -> Self {
        Self {
            shared: Arc::clone(&self.shared),
            completion: Some(StreamCompletion::ready(Err(StreamError::Failed(
                "cannot clone queue completion handle; use watch_completion() on the original handle"
                    .into(),
            )))),
        }
    }
}

impl<T: Send + 'static> SourceQueue<T> {
    /// Take the one-shot completion handle for this queue (resolves when the stream ends). Only the
    /// original handle holds it; a cloned `SourceQueue` returns an error completion.
    pub fn watch_completion(mut self) -> StreamCompletion<NotUsed> {
        self.completion.take().unwrap_or_else(|| {
            StreamCompletion::ready(Err(StreamError::Failed(
                "queue completion handle already taken".into(),
            )))
        })
    }

    /// Offer one element, applying the configured [`OverflowStrategy`]. Returns `Err` only when a
    /// second concurrent offer races a pending `Backpressure` offer; otherwise the outcome is in
    /// the [`QueueOfferResult`].
    pub fn offer(&self, elem: T) -> StreamResult<QueueOfferResult> {
        let strategy = self.shared.strategy;
        let capacity = self.shared.capacity;
        let mut state = self
            .shared
            .state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());

        if state.terminal.is_some() {
            return Ok(QueueOfferResult::QueueClosed);
        }

        if state.terminating {
            return Ok(QueueOfferResult::QueueClosed);
        }

        if state.buffer.len() < capacity {
            state.buffer.push_back(elem);
            drop(state);
            self.shared.available.notify_all();
            return Ok(QueueOfferResult::Enqueued);
        }

        match strategy {
            OverflowStrategy::DropHead => {
                let _ = state.buffer.pop_front();
                state.buffer.push_back(elem);
                drop(state);
                self.shared.available.notify_all();
                Ok(QueueOfferResult::Enqueued)
            }
            OverflowStrategy::DropTail => {
                let _ = state.buffer.pop_back();
                state.buffer.push_back(elem);
                drop(state);
                self.shared.available.notify_all();
                Ok(QueueOfferResult::Enqueued)
            }
            OverflowStrategy::DropBuffer => {
                state.buffer.clear();
                state.buffer.push_back(elem);
                drop(state);
                self.shared.available.notify_all();
                Ok(QueueOfferResult::Enqueued)
            }
            OverflowStrategy::DropNew => Ok(QueueOfferResult::Dropped),
            OverflowStrategy::Fail => {
                state.buffer.clear();
                let error =
                    StreamError::Failed(format!("Buffer overflow (max capacity was: {capacity})!"));
                state.terminal = Some(TerminalSignal::Error(error.clone()));
                drop(state);
                self.shared.available.notify_all();
                Ok(QueueOfferResult::Failure(error))
            }
            OverflowStrategy::Backpressure => {
                if state.pending_count >= 1 {
                    return Err(StreamError::Failed(
                        "Too many concurrent offers. Specified maximum is 1. \
                         You have to wait for the previous offer to resolve to send another request"
                            .into(),
                    ));
                }
                state.pending_count += 1;
                loop {
                    if state.terminal.is_some() || state.terminating {
                        state.pending_count -= 1;
                        return Ok(QueueOfferResult::QueueClosed);
                    }
                    if state.buffer.len() < capacity {
                        state.pending_count -= 1;
                        state.buffer.push_back(elem);
                        drop(state);
                        self.shared.available.notify_all();
                        return Ok(QueueOfferResult::Enqueued);
                    }
                    state = self
                        .shared
                        .available
                        .wait(state)
                        .unwrap_or_else(|poison| poison.into_inner());
                }
            }
        }
    }

    /// Complete the stream once buffered/pending elements drain (idempotent).
    pub fn complete(&self) {
        let mut state = self
            .shared
            .state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        if state.buffer.is_empty() && state.pending_count == 0 {
            if state.terminal.is_none() {
                state.terminal = Some(TerminalSignal::Complete);
            }
        } else {
            state.terminating = true;
        }
        drop(state);
        self.shared.available.notify_all();
    }

    /// Fail the stream with `error` (first terminal signal wins).
    pub fn fail(&self, error: StreamError) {
        let mut state = self
            .shared
            .state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        if state.terminal.is_none() {
            state.terminal = Some(TerminalSignal::Error(error));
        }
        drop(state);
        self.shared.available.notify_all();
    }
}

impl<T> Drop for SourceQueue<T> {
    fn drop(&mut self) {
        if Arc::strong_count(&self.shared) != 2 {
            return;
        }

        let mut state = self
            .shared
            .state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        if state.terminal.is_none() && !state.terminating {
            state.terminal = Some(TerminalSignal::Complete);
        }
        drop(state);
        self.shared.available.notify_all();
    }
}

struct SourceQueueStream<T: Send + 'static> {
    shared: Arc<SourceQueueShared<T>>,
    completion_sender: Option<oneshot::Sender<StreamResult<NotUsed>>>,
}

impl<T: Send + 'static> Iterator for SourceQueueStream<T> {
    type Item = StreamResult<T>;

    fn next(&mut self) -> Option<Self::Item> {
        let mut state = self
            .shared
            .state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        loop {
            if let Some(TerminalSignal::Error(error)) = &state.terminal {
                let error = error.clone();
                drop(state);
                self.signal_completion(Err(error.clone()));
                self.shared.available.notify_all();
                return Some(Err(error));
            }

            if let Some(item) = state.buffer.pop_front() {
                drop(state);
                self.shared.available.notify_all();
                return Some(Ok(item));
            }

            if let Some(terminal) = state.terminal.clone() {
                if state.terminating {
                    state.terminating = false;
                }
                drop(state);
                self.signal_completion(match &terminal {
                    TerminalSignal::Complete => Ok(NotUsed),
                    TerminalSignal::Error(error) => Err(error.clone()),
                });
                self.shared.available.notify_all();
                return match terminal {
                    TerminalSignal::Complete => None,
                    TerminalSignal::Error(error) => Some(Err(error)),
                };
            }

            if state.terminating && state.buffer.is_empty() && state.pending_count == 0 {
                state.terminal = Some(TerminalSignal::Complete);
                state.terminating = false;
                drop(state);
                self.signal_completion(Ok(NotUsed));
                self.shared.available.notify_all();
                return None;
            }

            state = self
                .shared
                .available
                .wait(state)
                .unwrap_or_else(|poison| poison.into_inner());
        }
    }
}

impl<T: Send + 'static> SourceQueueStream<T> {
    fn signal_completion(&mut self, result: StreamResult<NotUsed>) {
        if let Some(sender) = self.completion_sender.take() {
            let _ = sender.send(result);
        }
    }
}

impl<T: Send + 'static> Drop for SourceQueueStream<T> {
    fn drop(&mut self) {
        let mut state = self
            .shared
            .state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        if state.terminal.is_none() {
            state.terminal = Some(TerminalSignal::Complete);
        }
        state.terminating = false;
        drop(state);
        self.signal_completion(Ok(NotUsed));
        self.shared.available.notify_all();
    }
}

// ── SinkQueue (blocking pull from sink) ────────────────────────────────────

struct SinkQueueShared<T> {
    state: Mutex<SinkQueueState<T>>,
    available: Condvar,
}

struct SinkQueueState<T> {
    buffer: VecDeque<T>,
    error: Option<StreamError>,
    completed: bool,
}

impl<T> SinkQueueShared<T> {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            state: Mutex::new(SinkQueueState {
                buffer: VecDeque::new(),
                error: None,
                completed: false,
            }),
            available: Condvar::new(),
        })
    }
}

/// Materialized handle of [`Sink::queue`]: lets external code pull materialized elements out of a
/// running stream on demand.
pub struct SinkQueue<T> {
    shared: Arc<SinkQueueShared<T>>,
    _completion: StreamCompletion<NotUsed>,
}

impl<T> SinkQueue<T> {
    /// Block until the next element is available: `Ok(Some(elem))` for an element, `Ok(None)` at
    /// completion, or `Err` if the upstream failed.
    pub fn pull(&self) -> StreamResult<Option<T>> {
        let mut state = self
            .shared
            .state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        loop {
            if let Some(item) = state.buffer.pop_front() {
                return Ok(Some(item));
            }
            if let Some(error) = state.error.clone() {
                return Err(error);
            }
            if state.completed {
                return Ok(None);
            }
            state = self
                .shared
                .available
                .wait(state)
                .unwrap_or_else(|poison| poison.into_inner());
        }
    }
}

// ── Source factory methods ──────────────────────────────────────────────────

impl<T: Send + 'static> Source<T, NotUsed> {
    /// A source fed by an external [`BoundedSourceQueue`] handle (lock-free, no overflow strategy;
    /// a full buffer drops). Panics if `capacity == 0`.
    #[must_use]
    pub fn queue_bounded(capacity: usize) -> Source<T, BoundedSourceQueue<T>> {
        assert!(capacity > 0, "queue capacity must be greater than zero");
        Source::from_materialized_factory(move |_materializer| {
            let shared = BoundedQueueShared::new(capacity);
            let stream: BoxStream<T> = Box::new(BoundedQueueStream {
                shared: Arc::clone(&shared),
            });
            let handle = BoundedSourceQueue { shared };
            Ok((stream, handle))
        })
    }

    /// A source fed by an external [`SourceQueue`] handle that applies `strategy` on overflow.
    /// Panics if `capacity == 0`.
    #[must_use]
    pub fn queue(capacity: usize, strategy: OverflowStrategy) -> Source<T, SourceQueue<T>> {
        assert!(capacity > 0, "queue capacity must be greater than zero");
        Source::from_materialized_factory(move |_materializer| {
            let shared = SourceQueueShared::new(capacity, strategy);
            let (completion_sender, completion_receiver) = oneshot::channel();
            let stream: BoxStream<T> = Box::new(SourceQueueStream {
                shared: Arc::clone(&shared),
                completion_sender: Some(completion_sender),
            });
            let handle = SourceQueue {
                shared,
                completion: Some(StreamCompletion::from_receiver(completion_receiver, None)),
            };
            Ok((stream, handle))
        })
    }
}

// ── Sink factory methods ────────────────────────────────────────────────────

impl<T: Send + 'static> Sink<T, SinkQueue<T>> {
    /// A sink whose materialized [`SinkQueue`] handle lets external code `pull` elements on demand.
    #[must_use]
    pub fn queue() -> Self {
        Sink::from_runner(move |mut input, materializer| {
            let shared = SinkQueueShared::new();
            let worker_shared = Arc::clone(&shared);

            let completion = materializer.spawn_stream(move |stream_cancelled| {
                loop {
                    if stream_cancelled.load(Ordering::SeqCst) {
                        return Ok(NotUsed);
                    }
                    match input.next() {
                        Some(Ok(item)) => {
                            let mut state = worker_shared
                                .state
                                .lock()
                                .unwrap_or_else(|poison| poison.into_inner());
                            state.buffer.push_back(item);
                            drop(state);
                            worker_shared.available.notify_all();
                        }
                        Some(Err(error)) => {
                            let mut state = worker_shared
                                .state
                                .lock()
                                .unwrap_or_else(|poison| poison.into_inner());
                            if state.error.is_none() {
                                state.error = Some(error);
                            }
                            drop(state);
                            worker_shared.available.notify_all();
                            return Ok(NotUsed);
                        }
                        None => {
                            let mut state = worker_shared
                                .state
                                .lock()
                                .unwrap_or_else(|poison| poison.into_inner());
                            state.completed = true;
                            drop(state);
                            worker_shared.available.notify_all();
                            return Ok(NotUsed);
                        }
                    }
                }
            });

            Ok(SinkQueue {
                shared: Arc::clone(&shared),
                _completion: completion,
            })
        })
    }
}

// ── Tests ───────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::stream::Materializer;
    use std::sync::atomic::AtomicBool;
    use std::sync::atomic::Ordering;
    use std::sync::mpsc;
    use std::thread;
    use std::time::{Duration, Instant};

    // ── BoundedSourceQueue tests ────────────────────────────────────────

    #[test]
    fn bounded_offer_accepted_vs_processed_distinct() {
        let (queue, mut stream) = materialize_bounded_queue(2);
        // offer accepted means enqueued into buffer, not processed downstream
        assert_eq!(queue.offer(1), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2), QueueOfferResult::Enqueued);
        // proceed to process one element
        assert_eq!(stream.next(), Some(Ok(1)));
        // accepted is distinct from processed — element 2 still in buffer
        assert_eq!(queue.offer(3), QueueOfferResult::Enqueued);
        assert_eq!(stream.next(), Some(Ok(2)));
        assert_eq!(stream.next(), Some(Ok(3)));
    }

    #[test]
    fn bounded_offer_dropped_when_full() {
        let (queue, mut stream) = materialize_bounded_queue(1);
        assert_eq!(queue.offer(1), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2), QueueOfferResult::Dropped);
        assert_eq!(queue.offer(3), QueueOfferResult::Dropped);
        queue.complete();
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), None);
    }

    #[test]
    fn bounded_queue_closed_after_complete() {
        let (queue, mut stream) = materialize_bounded_queue(2);
        assert_eq!(queue.offer(1), QueueOfferResult::Enqueued);
        queue.complete();
        // element still drainable
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), None);
        // offers after complete return QueueClosed
        assert_eq!(queue.offer(2), QueueOfferResult::QueueClosed);
    }

    #[test]
    fn bounded_queue_closed_after_fail() {
        let (queue, mut stream) = materialize_bounded_queue(2);
        assert_eq!(queue.offer(1), QueueOfferResult::Enqueued);
        queue.fail(StreamError::Failed("boom".into()));
        // offers after fail return QueueClosed
        assert_eq!(queue.offer(2), QueueOfferResult::QueueClosed);
        // buffered elements drain before failure surfaces
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), Some(Err(StreamError::Failed("boom".into()))));
    }

    #[test]
    fn bounded_drop_handle_completes_stream() {
        let queue = {
            let (queue, mut stream) = materialize_bounded_queue::<i32>(2);
            queue.offer(1);
            queue.offer(2);
            assert_eq!(stream.next(), Some(Ok(1)));
            queue
        };
        // stream dropped, handle dropped
        assert_eq!(queue.offer(3), QueueOfferResult::QueueClosed);
    }

    #[test]
    fn bounded_drop_last_producer_completes_stream() {
        let (queue, mut stream) = materialize_bounded_queue::<i32>(1);
        let producer = queue.clone();
        let consumer = thread::spawn(move || {
            assert_eq!(stream.next(), None);
        });
        drop(producer);
        drop(queue);
        consumer.join().unwrap();
    }

    #[test]
    fn bounded_drain_before_complete() {
        let (queue, mut stream) = materialize_bounded_queue(3);
        assert_eq!(queue.offer(1), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(3), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(4), QueueOfferResult::Dropped);
        assert_eq!(queue.offer(5), QueueOfferResult::Dropped);
        queue.complete();
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), Some(Ok(2)));
        assert_eq!(stream.next(), Some(Ok(3)));
        assert_eq!(stream.next(), None);
    }

    #[test]
    fn bounded_terminal_completion_is_sticky() {
        let (queue, mut stream) = materialize_bounded_queue(2);
        queue.offer(1);
        queue.complete();
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), None);
        assert_eq!(stream.next(), None);
        assert_eq!(stream.next(), None);
    }

    #[test]
    fn bounded_terminal_failure_is_sticky() {
        let (queue, mut stream) = materialize_bounded_queue::<i32>(2);
        queue.fail(StreamError::Failed("boom".into()));
        assert_eq!(stream.next(), Some(Err(StreamError::Failed("boom".into()))));
        assert_eq!(stream.next(), Some(Err(StreamError::Failed("boom".into()))));
    }

    #[test]
    fn bounded_producer_consumer_across_threads() {
        let (queue, mut stream) = materialize_bounded_queue::<i32>(16);
        // start consumer first so it drains as producer fills
        let consumer = thread::spawn(move || {
            let mut collected = Vec::new();
            while let Some(Ok(item)) = stream.next() {
                collected.push(item);
            }
            collected
        });
        let producer = thread::spawn({
            let queue = queue.clone();
            move || {
                for i in 0..10 {
                    assert_eq!(queue.offer(i), QueueOfferResult::Enqueued);
                }
                queue.complete();
            }
        });
        producer.join().unwrap();
        let collected = consumer.join().unwrap();
        assert_eq!(collected, (0..10).collect::<Vec<_>>());
    }

    #[test]
    fn bounded_multi_producer_single_consumer() {
        let n = 50_i32;
        let producer_count = 4;
        let total = (producer_count * n) as usize;
        let (queue, mut stream) = materialize_bounded_queue::<i32>(total);

        // start consumer first
        let consumer = thread::spawn(move || {
            let mut collected = Vec::new();
            while let Some(Ok(item)) = stream.next() {
                collected.push(item);
            }
            collected
        });

        let mut handles = Vec::new();
        for p in 0..producer_count {
            let q = queue.clone();
            handles.push(thread::spawn(move || {
                for i in 0..n {
                    assert_eq!(q.offer(p * n + i), QueueOfferResult::Enqueued);
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        queue.complete();
        let mut collected = consumer.join().unwrap();
        collected.sort_unstable();
        assert_eq!(collected.len(), total);
        collected.sort_unstable();
        assert_eq!(collected.len(), (producer_count * n) as usize);
    }

    #[test]
    fn bounded_offer_poll_stress_no_lost_wakeup() {
        // A small capacity makes the buffer oscillate between empty and full, so
        // the consumer repeatedly drains it dry and parks while producers race
        // to refill it. Every element that reports `Enqueued` must be delivered
        // exactly once and the stream must terminate — a lost wakeup or a
        // stranded element would hang the join or break the count.
        const ROUNDS: usize = 40;
        const PRODUCERS: i64 = 8;
        const PER_PRODUCER: i64 = 500;

        for _ in 0..ROUNDS {
            let (queue, stream) = materialize_bounded_queue::<i64>(4);

            let consumer = thread::spawn(move || {
                let mut count = 0_u64;
                for item in stream {
                    assert!(item.is_ok(), "unexpected stream failure: {item:?}");
                    count += 1;
                }
                count
            });

            let mut handles = Vec::new();
            for p in 0..PRODUCERS {
                let q = queue.clone();
                handles.push(thread::spawn(move || {
                    let mut enqueued = 0_u64;
                    for i in 0..PER_PRODUCER {
                        if q.offer(p * PER_PRODUCER + i) == QueueOfferResult::Enqueued {
                            enqueued += 1;
                        }
                    }
                    enqueued
                }));
            }

            let mut total_enqueued = 0_u64;
            for h in handles {
                total_enqueued += h.join().unwrap();
            }
            // Every offer has now happened-before `complete`, so every accepted
            // element is guaranteed to drain before the stream ends.
            queue.complete();
            let delivered = consumer.join().unwrap();
            assert_eq!(
                delivered, total_enqueued,
                "every accepted element must be delivered exactly once"
            );
        }
    }

    #[test]
    fn bounded_pingpong_in_order_stress() {
        // Capacity 1 forces the producer to wait for the consumer to drain (and
        // re-park) before nearly every element, exercising the park/wake
        // handshake on the tightest possible loop. Elements must arrive exactly
        // once, in order; a genuine deadlock trips the deadline rather than
        // hanging the suite forever.
        const ROUNDS: usize = 10;
        const ITEMS: i64 = 1_000;

        for _ in 0..ROUNDS {
            let (queue, stream) = materialize_bounded_queue::<i64>(1);
            let consumer = thread::spawn(move || {
                let mut got = Vec::new();
                for item in stream {
                    got.push(item.expect("no spurious failure"));
                }
                got
            });

            for i in 0..ITEMS {
                let deadline = Instant::now() + Duration::from_secs(5);
                loop {
                    match queue.offer(i) {
                        QueueOfferResult::Enqueued => break,
                        QueueOfferResult::Dropped => {
                            assert!(
                                Instant::now() < deadline,
                                "offer never accepted within deadline — consumer stuck?"
                            );
                            thread::yield_now();
                        }
                        other => panic!("unexpected offer result: {other:?}"),
                    }
                }
            }
            queue.complete();
            let got = consumer.join().unwrap();
            assert_eq!(got, (0..ITEMS).collect::<Vec<_>>());
        }
    }

    // ── SourceQueue tests ───────────────────────────────────────────────

    fn materialize_source_queue<T: Send + 'static>(
        capacity: usize,
        strategy: OverflowStrategy,
    ) -> (SourceQueue<T>, BoxStream<T>) {
        let materializer = Materializer::new();
        let (stream, queue) = Source::<T>::queue(capacity, strategy)
            .factory
            .create(&materializer)
            .unwrap();
        (queue, stream)
    }

    #[test]
    fn source_queue_offer_enqueued() {
        let (queue, mut stream) =
            materialize_source_queue::<i32>(2, OverflowStrategy::Backpressure);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), Some(Ok(2)));
    }

    #[test]
    fn source_queue_drop_head() {
        let (queue, mut stream) = materialize_source_queue::<i32>(2, OverflowStrategy::DropHead);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2).unwrap(), QueueOfferResult::Enqueued);
        // buffer full, drop head
        assert_eq!(queue.offer(3).unwrap(), QueueOfferResult::Enqueued);
        queue.complete();
        assert_eq!(stream.next(), Some(Ok(2)));
        assert_eq!(stream.next(), Some(Ok(3)));
        assert_eq!(stream.next(), None);
    }

    #[test]
    fn source_queue_drop_tail() {
        let (queue, mut stream) = materialize_source_queue::<i32>(2, OverflowStrategy::DropTail);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(3).unwrap(), QueueOfferResult::Enqueued);
        queue.complete();
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), Some(Ok(3)));
        assert_eq!(stream.next(), None);
    }

    #[test]
    fn source_queue_drop_buffer() {
        let (queue, mut stream) = materialize_source_queue::<i32>(2, OverflowStrategy::DropBuffer);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(3).unwrap(), QueueOfferResult::Enqueued);
        queue.complete();
        // DropBuffer clears the buffer, then enqueues the new element
        assert_eq!(stream.next(), Some(Ok(3)));
        assert_eq!(stream.next(), None);
    }

    #[test]
    fn source_queue_drop_new() {
        let (queue, mut stream) = materialize_source_queue::<i32>(2, OverflowStrategy::DropNew);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(3).unwrap(), QueueOfferResult::Dropped);
        queue.complete();
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), Some(Ok(2)));
        assert_eq!(stream.next(), None);
    }

    #[test]
    fn source_queue_fail_strategy() {
        let (queue, _stream) = materialize_source_queue::<i32>(2, OverflowStrategy::Fail);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);
        assert_eq!(queue.offer(2).unwrap(), QueueOfferResult::Enqueued);
        match queue.offer(3).unwrap() {
            QueueOfferResult::Failure(e) => {
                assert!(format!("{e:?}").contains("Buffer overflow"));
            }
            other => panic!("expected Failure, got {other:?}"),
        }
    }

    #[test]
    fn source_queue_backpressure_blocks() {
        let (queue, mut stream) =
            materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);

        // Keep the stream alive until the blocked offer has a chance to observe
        // the newly freed slot.
        let consumed = Arc::new(AtomicBool::new(false));
        let c = Arc::clone(&consumed);
        let (release_tx, release_rx) = mpsc::channel();
        let consumer = thread::spawn(move || {
            assert_eq!(stream.next(), Some(Ok(1)));
            c.store(true, Ordering::SeqCst);
            release_rx.recv().unwrap();
        });

        wait_until(Duration::from_secs(1), || consumed.load(Ordering::SeqCst));
        assert_eq!(queue.offer(2).unwrap(), QueueOfferResult::Enqueued);
        release_tx.send(()).unwrap();
        consumer.join().unwrap();
        assert!(consumed.load(Ordering::SeqCst));
    }

    #[test]
    fn source_queue_concurrent_offer_violation() {
        let (queue, mut stream) =
            materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);

        let q = queue.clone();
        let started = Arc::new(AtomicBool::new(false));
        let s = Arc::clone(&started);
        let blocker = thread::spawn(move || {
            s.store(true, Ordering::SeqCst);
            // this blocks until drained
            let _ = q.offer(2);
        });

        // wait for blocker to start
        while !started.load(Ordering::SeqCst) {
            thread::yield_now();
        }
        wait_until(Duration::from_secs(1), || {
            queue
                .shared
                .state
                .lock()
                .unwrap_or_else(|poison| poison.into_inner())
                .pending_count
                == 1
        });

        // second concurrent offer should fail
        let result = queue.offer(3);
        assert!(result.is_err());
        assert!(format!("{result:?}").contains("Too many concurrent offers"));

        // drain to unblock
        assert_eq!(stream.next(), Some(Ok(1)));
        blocker.join().unwrap();
    }

    #[test]
    fn source_queue_watch_completion() {
        let (queue, mut stream) =
            materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        queue.offer(1).unwrap();
        queue.complete();
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), None);
        assert_eq!(queue.watch_completion().wait(), Ok(NotUsed));
    }

    #[test]
    fn source_queue_watch_completion_on_failure() {
        let (queue, mut stream) =
            materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        queue.fail(StreamError::Failed("boom".into()));
        assert_eq!(stream.next(), Some(Err(StreamError::Failed("boom".into()))));
        assert_eq!(
            queue.watch_completion().wait(),
            Err(StreamError::Failed("boom".into()))
        );
    }

    #[test]
    fn source_queue_terminal_completion_is_sticky() {
        let (queue, mut stream) =
            materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        queue.complete();
        assert_eq!(stream.next(), None);
        assert_eq!(stream.next(), None);
        assert_eq!(stream.next(), None);
    }

    #[test]
    fn source_queue_offer_after_complete_returns_queue_closed() {
        let (queue, _stream) = materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        queue.complete();
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::QueueClosed);
    }

    #[test]
    fn source_queue_drop_stream_closes_queue() {
        let (queue, stream) = materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        queue.offer(1).unwrap();
        drop(stream);
        // handle still alive, stream dropped → terminal set
        assert_eq!(queue.offer(2).unwrap(), QueueOfferResult::QueueClosed);
    }

    #[test]
    fn source_queue_drop_last_producer_completes_stream() {
        let (queue, mut stream) =
            materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        let producer = queue.clone();
        let consumer = thread::spawn(move || {
            assert_eq!(stream.next(), None);
        });
        drop(producer);
        drop(queue);
        consumer.join().unwrap();
    }

    #[test]
    fn source_queue_backpressure_unblocks_on_complete() {
        let (queue, mut stream) =
            materialize_source_queue::<i32>(1, OverflowStrategy::Backpressure);
        assert_eq!(queue.offer(1).unwrap(), QueueOfferResult::Enqueued);

        let q = queue.clone();
        let blocker = thread::spawn(move || {
            // blocks waiting for space
            q.offer(2)
        });

        wait_until(Duration::from_secs(1), || {
            queue
                .shared
                .state
                .lock()
                .unwrap_or_else(|poison| poison.into_inner())
                .pending_count
                == 1
        });
        // complete wakes the backpressured offerer
        queue.complete();
        let result = blocker.join().unwrap();
        assert_eq!(result.unwrap(), QueueOfferResult::QueueClosed);

        // buffer still has item 1
        assert_eq!(stream.next(), Some(Ok(1)));
        assert_eq!(stream.next(), None);
    }

    // ── SinkQueue tests ─────────────────────────────────────────────────

    fn materialize_sink_queue<T: Send + 'static>(source: Source<T>) -> SinkQueue<T> {
        source.run_with(Sink::queue()).unwrap()
    }

    #[test]
    fn sink_queue_pull_elements() {
        let queue = materialize_sink_queue(Source::from_iter([1, 2, 3]));
        assert_eq!(queue.pull().unwrap(), Some(1));
        assert_eq!(queue.pull().unwrap(), Some(2));
        assert_eq!(queue.pull().unwrap(), Some(3));
        assert_eq!(queue.pull().unwrap(), None);
    }

    #[test]
    fn sink_queue_pull_none_after_upstream_completion() {
        let queue = materialize_sink_queue(Source::from_iter([42]));
        assert_eq!(queue.pull().unwrap(), Some(42));
        assert_eq!(queue.pull().unwrap(), None);
        assert_eq!(queue.pull().unwrap(), None);
    }

    #[test]
    fn sink_queue_pull_error_from_upstream_failure() {
        let queue =
            materialize_sink_queue(Source::<i32>::failed(StreamError::Failed("boom".into())));
        assert_eq!(
            queue.pull().unwrap_err(),
            StreamError::Failed("boom".into())
        );
        assert_eq!(
            queue.pull().unwrap_err(),
            StreamError::Failed("boom".into())
        );
    }

    #[test]
    fn sink_queue_terminal_stickiness() {
        let queue = materialize_sink_queue(Source::<i32>::empty());
        assert_eq!(queue.pull().unwrap(), None);
        assert_eq!(queue.pull().unwrap(), None);
        assert_eq!(queue.pull().unwrap(), None);
    }

    #[test]
    fn sink_queue_terminal_failure_stickiness() {
        let queue =
            materialize_sink_queue(Source::<i32>::failed(StreamError::Failed("boom".into())));
        assert_eq!(
            queue.pull().unwrap_err(),
            StreamError::Failed("boom".into())
        );
        assert_eq!(
            queue.pull().unwrap_err(),
            StreamError::Failed("boom".into())
        );
    }

    #[test]
    fn sink_queue_drop_cancels_upstream() {
        let queue = materialize_sink_queue(Source::repeat(42));
        // dropping queue cancels worker
        drop(queue);
        // if this doesn't deadlock, the worker was cancelled
    }

    #[test]
    fn sink_queue_drain_multiple_items() {
        let queue = materialize_sink_queue(Source::from_iter(0..100));
        for i in 0..100 {
            assert_eq!(queue.pull().unwrap(), Some(i));
        }
        assert_eq!(queue.pull().unwrap(), None);
    }

    // ── Helpers ─────────────────────────────────────────────────────────

    fn materialize_bounded_queue<T: Send + 'static>(
        capacity: usize,
    ) -> (BoundedSourceQueue<T>, BoxStream<T>) {
        let materializer = Materializer::new();
        let (stream, queue) = Source::<T>::queue_bounded(capacity)
            .factory
            .create(&materializer)
            .unwrap();
        (queue, stream)
    }

    fn wait_until(timeout: Duration, condition: impl Fn() -> bool) {
        let deadline = Instant::now() + timeout;
        while Instant::now() < deadline {
            if condition() {
                return;
            }
            thread::yield_now();
            thread::sleep(Duration::from_millis(1));
        }
        assert!(condition(), "condition was not met within {timeout:?}");
    }
}