armdb 0.11.2

//! Public vocabulary for engine-owned durable queues.

pub(crate) mod fixed;
pub(crate) mod retention;
pub(crate) mod tx;
#[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
pub(crate) mod var;
pub(crate) mod wait;

use std::any::Any;
use std::cmp::Ordering;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::time::Duration;

use armour_core::KeyScheme;

use crate::Key;
use crate::error::{DbError, DbResult};

pub const LIVE_CLASS: u8 = 0;
pub const DEAD_CLASS: u8 = 1;
pub const ID_BLOCK: u64 = 1024;

static NEXT_QUEUE_INSTANCE_ID: AtomicU64 = AtomicU64::new(1);

/// Allocate an opaque process-local capability namespace for one queue handle.
/// A lease token is not a durable authorization: after a process restart the
/// worker must lease the row again before it can acknowledge or retry it.
pub(crate) fn next_queue_instance_id() -> u64 {
    NEXT_QUEUE_INSTANCE_ID
        .fetch_update(AtomicOrdering::Relaxed, AtomicOrdering::Relaxed, |id| {
            id.checked_add(1)
        })
        .expect("queue instance id exhausted")
}

/// Pending queue additions admitted by one uncommitted transaction.
/// The live counter remains a soft cap for concurrent producers, but builder
/// calls belonging to the same transaction must account for each other.
#[derive(Default)]
pub(crate) struct QueueDepthAdmissions {
    pending: HashMap<u64, u64>,
}

/// One stream-produced queue addition awaiting transaction-wide admission.
pub(crate) struct QueueDepthAdmission {
    queue_id: u64,
    live_count: Arc<AtomicU64>,
    max_depth: Option<u64>,
}

impl QueueDepthAdmission {
    fn new(queue_id: u64, live_count: Arc<AtomicU64>, max_depth: Option<u64>) -> Self {
        Self {
            queue_id,
            live_count,
            max_depth,
        }
    }

    pub(crate) fn reserve(self, admissions: &mut QueueDepthAdmissions) -> DbResult<()> {
        admissions.reserve(self.queue_id, &self.live_count, self.max_depth)
    }
}

impl QueueDepthAdmissions {
    pub(crate) fn reserve(
        &mut self,
        queue_id: u64,
        live_count: &AtomicU64,
        max_depth: Option<u64>,
    ) -> DbResult<()> {
        let Some(limit) = max_depth else {
            return Ok(());
        };
        let pending = self.pending.entry(queue_id).or_default();
        let admitted = live_count
            .load(AtomicOrdering::Acquire)
            .saturating_add(*pending);
        if admitted >= limit {
            return Err(DbError::QueueFull);
        }
        *pending = pending.saturating_add(1);
        Ok(())
    }
}

pub(crate) use fixed::FixedQueueInner;
#[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
pub(crate) use var::VarQueueInner;
pub use wait::QueueDue;
use wait::{QueueScan, QueueWait};

#[doc(hidden)]
pub trait QueueHandle<N: QueueSpec> {
    fn push(
        queue: &Queue<N>,
        payload: N::Payload,
        not_before: Option<std::time::SystemTime>,
    ) -> DbResult<u64>;
    fn len(queue: &Queue<N>) -> u64;
    fn dead_len(queue: &Queue<N>) -> u64;
    fn lease(
        queue: &Queue<N>,
        n: usize,
        timeout: Option<Duration>,
    ) -> DbResult<Vec<LeasedEntry<N::Payload>>>;
    fn ack(queue: &Queue<N>, token: LeaseToken) -> DbResult<()>;
    fn nack(queue: &Queue<N>, token: LeaseToken) -> DbResult<()>;
    fn scan_due(queue: &Queue<N>) -> DbResult<QueueScan>;
    fn wait(queue: &Queue<N>) -> &QueueWait;
    fn stats(queue: &Queue<N>) -> QueueStats;
}

/// Read-only operational counters for one queue process lifetime.
///
/// `compaction_bytes` is always zero for a fixed-payload queue because its
/// rows live in FixedStore, which updates slots in place. `payload_writes`
/// counts body writes for a variable-payload queue; leasing, retrying, and
/// acknowledging do not rewrite the body. The counters reset when the queue is
/// opened and are intended for diagnostics and benchmark assertions.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct QueueStats {
    /// Fixed-queue compaction bytes. Fixed queues always report zero; this is
    /// not a measurement of a variable queue's separate payload backend.
    pub compaction_bytes: u64,
    /// Variable-payload body writes performed since this queue was opened.
    pub payload_writes: u64,
}

/// Physical key for a queue header row, ordered by class, visibility, then id.
///
/// The fields retain their big-endian representation in memory, so the
/// zerocopy byte view is exactly the on-disk key on every host endianness.
#[repr(C)]
#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    PartialEq,
    Hash,
    zerocopy::FromBytes,
    zerocopy::IntoBytes,
    zerocopy::Immutable,
    zerocopy::KnownLayout,
)]
pub struct QueueKey {
    sort_be: u64,
    id_be: u64,
}

impl QueueKey {
    pub const fn live(visible_at_ms: u64, id: u64) -> Self {
        Self::new(LIVE_CLASS, visible_at_ms, id)
    }

    pub const fn dead(visible_at_ms: u64, id: u64) -> Self {
        Self::new(DEAD_CLASS, visible_at_ms, id)
    }

    /// Whether this key names a live or retained dead-letter row.
    pub const fn class(self) -> u8 {
        (u64::from_be(self.sort_be) >> 63) as u8
    }

    /// The wall-clock visibility timestamp in milliseconds.
    pub const fn visible_at_ms(self) -> u64 {
        u64::from_be(self.sort_be) & ((1 << 63) - 1)
    }

    /// The durable queue entry identifier.
    pub const fn id(self) -> u64 {
        u64::from_be(self.id_be)
    }

    const fn new(class: u8, visible_at_ms: u64, id: u64) -> Self {
        assert!(
            visible_at_ms < (1 << 63),
            "queue visibility exceeds 63-bit domain"
        );
        Self {
            sort_be: (((class as u64) << 63) | visible_at_ms).to_be(),
            id_be: id.to_be(),
        }
    }

    /// The fixed, big-endian physical key encoding.
    pub fn as_bytes(&self) -> &[u8] {
        zerocopy::IntoBytes::as_bytes(self)
    }
}

impl Ord for QueueKey {
    fn cmp(&self, other: &Self) -> Ordering {
        (self.class(), self.visible_at_ms(), self.id()).cmp(&(
            other.class(),
            other.visible_at_ms(),
            other.id(),
        ))
    }
}

impl PartialOrd for QueueKey {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Key for QueueKey {
    const KEY_SCHEME: KeyScheme = KeyScheme::Bytes;

    fn as_bytes(&self) -> &[u8] {
        self.as_bytes()
    }

    fn as_bytes_mut(&mut self) -> &mut [u8] {
        zerocopy::IntoBytes::as_mut_bytes(self)
    }

    fn from_bytes(bytes: &[u8]) -> Self {
        zerocopy::FromBytes::read_from_bytes(bytes).expect("16-byte queue key")
    }

    fn zeroed() -> Self {
        Self::new(0, 0, 0)
    }
}

/// Advance a row's lease generation.
///
/// The epoch is only ever compared for equality — a token matches its row or it
/// does not — so wrapping is sound: aliasing a stale token would take 2^32
/// intervening leases of that same row. `checked_add` here would be worse than
/// useless: a saturated row stays due and un-leasable at the front of the scan,
/// so failing the call would make `lease` return an error for the entire queue,
/// permanently, with no way to reach the row that caused it.
pub(crate) const fn next_lease_epoch(epoch: u32) -> u32 {
    epoch.wrapping_add(1)
}

/// Resolve a per-row failure raised part-way through a lease batch.
///
/// Rows claimed before the failure already carry a bumped epoch and a written
/// deadline on disk. Discarding them would strand the work: no token ever
/// reaches the caller, so nobody can ack or nack those rows until their lease
/// expires. A partial batch is an ordinary documented result, so the batch is
/// returned and the scan stops; a failure before the first claim is reported.
pub(crate) fn partial_lease_batch<P>(
    leased: Vec<LeasedEntry<P>>,
    queue: &'static str,
    key: QueueKey,
    error: DbError,
) -> DbResult<Vec<LeasedEntry<P>>> {
    if leased.is_empty() {
        return Err(error);
    }
    tracing::warn!(
        queue,
        id = key.id(),
        claimed = leased.len(),
        %error,
        "queue lease returning a partial batch after a row-level failure"
    );
    Ok(leased)
}

/// Selects the storage representation for a [`QueueSpec`].
pub trait QueueMode<N: QueueSpec>: Send + Sync + 'static {
    /// Crate-private storage owned by the opaque queue handle.
    type Inner;

    /// Number of physical queue rows one stream emission will materialize.
    const STREAM_ROWS: usize;

    /// Whether the queue owns a distinct payload collection as well as its
    /// header collection.
    const HAS_PAYLOAD_ROW: bool;
}

#[doc(hidden)]
pub trait QueueOpen<N: QueueSpec>: QueueMode<N> {
    fn open(db: &crate::armour::Db, config: QueueConfig) -> DbResult<Arc<Queue<N>>>;
}

/// Marker selected by the default [`queue!`](crate::queue) declaration.
pub struct FixedQueueMode;

/// Marker selected by the `var` [`queue!`](crate::queue) declaration.
#[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
pub struct VarQueueMode;

impl<N> QueueMode<N> for FixedQueueMode
where
    N: QueueSpec,
    N::Payload: armour_core::GetType,
{
    type Inner = FixedQueueInner<N>;
    const STREAM_ROWS: usize = 1;
    const HAS_PAYLOAD_ROW: bool = false;
}

#[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
impl<N> QueueMode<N> for VarQueueMode
where
    N: QueueSpec,
    N::Payload: armour_core::GetType,
{
    type Inner = VarQueueInner<N>;
    const STREAM_ROWS: usize = 2;
    const HAS_PAYLOAD_ROW: bool = true;
}

/// Erased queue identity retained by a change-stream declaration.
///
/// This is intentionally crate-private: users declare targets through
/// [`QueueTargets`] and emit through [`crate::EmitSink`].
pub(crate) trait QueueTarget: Send + Sync {
    fn physical_rows(&self) -> usize;
    fn owns_physical_name(&self, name: &str) -> bool;
    fn name(&self) -> &'static str;
    fn check_managed(&self, db: &crate::armour::Db) -> DbResult<()>;
    fn stream_bytes(&self, payload: &dyn Any) -> DbResult<usize>;
    fn tx_participants(&self) -> Vec<Arc<dyn crate::txn::participant::TxParticipant>>;
    fn prepare_stream(&self) -> DbResult<QueueStreamPreparation>;
    fn tx_stream_writes(
        &self,
        payload: &dyn Any,
        visible_at: std::time::SystemTime,
    ) -> DbResult<crate::txn::participant::TxStreamWrites>;
}

/// Opaque transaction participant for a queue physical row.
///
/// This exists solely so [`QueueStreamPayload`] can remain a usable public
/// bound without exposing the crate-private transaction participant seam.
#[doc(hidden)]
pub struct QueueStreamParticipant {
    inner: Arc<dyn crate::txn::participant::TxParticipant>,
}

/// Opaque result of lowering one buffered stream emission.
#[doc(hidden)]
pub struct QueueStreamPlan {
    inner: crate::txn::participant::TxStreamWrites,
    admission: QueueDepthAdmission,
}

/// Opaque preflight state retained from before queue shards are locked.
#[doc(hidden)]
pub struct QueueStreamPreparation {
    permits: Vec<crate::queue::retention::ProducerPermit>,
}

impl QueueStreamPreparation {
    fn empty() -> Self {
        Self {
            permits: Vec::new(),
        }
    }
    pub(crate) fn into_permits(self) -> Vec<crate::queue::retention::ProducerPermit> {
        self.permits
    }
}

/// Mode-specific payload encoding needed by the stream carrier before Task 9
/// turns one buffered emission into its physical transaction writes.
#[doc(hidden)]
pub trait QueueStreamPayload<N: QueueSpec>: QueueMode<N> {
    fn stream_bytes(payload: &N::Payload) -> usize;
    fn check_managed(inner: &Self::Inner, db: &crate::armour::Db) -> DbResult<()>;
    #[doc(hidden)]
    fn stream_participants(inner: &Self::Inner) -> Vec<QueueStreamParticipant>;
    fn prepare_stream(inner: &Self::Inner) -> DbResult<QueueStreamPreparation>;
    #[doc(hidden)]
    fn lower_stream(
        inner: &Self::Inner,
        payload: &dyn Any,
        visible_at: std::time::SystemTime,
    ) -> DbResult<QueueStreamPlan>;
}

impl<N> QueueStreamPayload<N> for FixedQueueMode
where
    N: QueueSpec<Mode = FixedQueueMode>,
    N::Payload: Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + armour_core::GetType,
{
    fn stream_bytes(_: &N::Payload) -> usize {
        // QueueKey plus the fixed QueueHeader/4 KiB payload slot.
        16 + 32 + 4096
    }
    fn check_managed(inner: &Self::Inner, db: &crate::armour::Db) -> DbResult<()> {
        db.tx_collection_id(Arc::clone(&inner.headers).as_ref())
            .map(|_| ())
    }
    fn stream_participants(inner: &Self::Inner) -> Vec<QueueStreamParticipant> {
        vec![QueueStreamParticipant {
            inner: Arc::clone(&inner.headers) as Arc<dyn crate::txn::participant::TxParticipant>,
        }]
    }
    fn prepare_stream(_: &Self::Inner) -> DbResult<QueueStreamPreparation> {
        Ok(QueueStreamPreparation::empty())
    }
    fn lower_stream(
        inner: &Self::Inner,
        payload: &dyn Any,
        visible_at: std::time::SystemTime,
    ) -> DbResult<QueueStreamPlan> {
        let payload = payload
            .downcast_ref::<N::Payload>()
            .ok_or(DbError::Internal("queue stream payload type mismatch"))?;
        let now = std::time::SystemTime::now();
        let visible_at = fixed::FixedQueueInner::<N>::now_ms(visible_at)?;
        if visible_at >= (1 << 63) {
            return Err(DbError::Client("queue visibility exceeds 63-bit domain"));
        }
        let id = inner.next_id_in_tx()?;
        let key = QueueKey::live(visible_at, id);
        let row =
            fixed::QueueFixedRow::<N>::new(*payload, fixed::FixedQueueInner::<N>::now_ms(now)?)?;
        let count = Arc::clone(&inner.live_count);
        let wait = Arc::clone(&inner.wait);
        Ok(QueueStreamPlan {
            inner: crate::txn::participant::TxStreamWrites {
                writes: vec![crate::txn::participant::TxDerivedWrite {
                    participant: Arc::clone(&inner.headers)
                        as Arc<dyn crate::txn::participant::TxParticipant>,
                    key: crate::Key::as_bytes(&key).to_vec(),
                    value: Some(zerocopy::IntoBytes::as_bytes(&row).to_vec()),
                }],
                after_success: vec![Box::new(move || {
                    count.fetch_add(1, std::sync::atomic::Ordering::Release);
                    wait.notify_visible_at(visible_at);
                })],
                queue_depth_admissions: Vec::new(),
            },
            admission: QueueDepthAdmission::new(
                inner.instance_id,
                Arc::clone(&inner.live_count),
                inner.config.max_depth,
            ),
        })
    }
}

#[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
impl<N> QueueStreamPayload<N> for VarQueueMode
where
    N: QueueSpec<Mode = VarQueueMode>,
    N::Payload: rapira::Rapira + armour_core::GetType + Send + Sync,
{
    fn stream_bytes(payload: &N::Payload) -> usize {
        // Header key/value plus the payload id and the exact bytes written by
        // QueuePayloadTree. The outer QueueVarPayload is Rapira-encoded too:
        // counting only its inner bytes would omit the Vec framing.
        let payload = var::QueueVarPayload::<N> {
            bytes: rapira::serialize(payload),
            _queue: std::marker::PhantomData,
        };
        16 + 32 + 8 + rapira::serialize(&payload).len()
    }
    fn check_managed(inner: &Self::Inner, db: &crate::armour::Db) -> DbResult<()> {
        db.tx_collection_id(Arc::clone(&inner.headers).as_ref())?;
        db.tx_collection_id(Arc::clone(&inner.payloads).as_ref())
            .map(|_| ())
    }
    fn stream_participants(inner: &Self::Inner) -> Vec<QueueStreamParticipant> {
        vec![
            QueueStreamParticipant {
                inner: Arc::clone(&inner.headers)
                    as Arc<dyn crate::txn::participant::TxParticipant>,
            },
            QueueStreamParticipant {
                inner: Arc::clone(&inner.payloads)
                    as Arc<dyn crate::txn::participant::TxParticipant>,
            },
        ]
    }
    fn prepare_stream(inner: &Self::Inner) -> DbResult<QueueStreamPreparation> {
        inner.retention.process_before_queue_operation();
        Ok(QueueStreamPreparation {
            permits: vec![inner.retention.producer_permit()],
        })
    }
    fn lower_stream(
        inner: &Self::Inner,
        payload: &dyn Any,
        visible_at: std::time::SystemTime,
    ) -> DbResult<QueueStreamPlan> {
        let payload = payload
            .downcast_ref::<N::Payload>()
            .ok_or(DbError::Internal("queue stream payload type mismatch"))?;
        let now = std::time::SystemTime::now();
        let visible_at = fixed::FixedQueueInner::<N>::now_ms(visible_at)?;
        if visible_at >= (1 << 63) {
            return Err(DbError::Client("queue visibility exceeds 63-bit domain"));
        }
        let id = inner.next_id_in_tx()?;
        let key = QueueKey::live(visible_at, id);
        let payload_key = rend::u64_be::from_native(id);
        let payload = var::QueueVarPayload::<N> {
            bytes: rapira::serialize(payload),
            _queue: std::marker::PhantomData,
        };
        let mut payload_bytes = Vec::new();
        crate::Codec::encode_to(&crate::RapiraCodec, &payload, &mut payload_bytes)?;
        let header = var::QueueVarHeaderRow::<N>::new(fixed::FixedQueueInner::<N>::now_ms(now)?);
        let count = Arc::clone(&inner.live_count);
        let wait = Arc::clone(&inner.wait);
        let payload_writes = Arc::clone(&inner.payload_writes);
        Ok(QueueStreamPlan {
            inner: crate::txn::participant::TxStreamWrites {
                writes: vec![
                    crate::txn::participant::TxDerivedWrite {
                        participant: Arc::clone(&inner.payloads)
                            as Arc<dyn crate::txn::participant::TxParticipant>,
                        key: crate::Key::as_bytes(&payload_key).to_vec(),
                        value: Some(payload_bytes),
                    },
                    crate::txn::participant::TxDerivedWrite {
                        participant: Arc::clone(&inner.headers)
                            as Arc<dyn crate::txn::participant::TxParticipant>,
                        key: crate::Key::as_bytes(&key).to_vec(),
                        value: Some(zerocopy::IntoBytes::as_bytes(&header).to_vec()),
                    },
                ],
                after_success: vec![Box::new(move || {
                    count.fetch_add(1, std::sync::atomic::Ordering::Release);
                    payload_writes.fetch_add(1, std::sync::atomic::Ordering::Release);
                    wait.notify_visible_at(visible_at);
                })],
                queue_depth_admissions: Vec::new(),
            },
            admission: QueueDepthAdmission::new(
                inner.instance_id,
                Arc::clone(&inner.live_count),
                inner.config.max_depth,
            ),
        })
    }
}

impl<N> QueueTarget for Queue<N>
where
    N: QueueSpec,
    <N::Mode as QueueMode<N>>::Inner: Send + Sync,
    N::Mode: QueueStreamPayload<N>,
{
    fn physical_rows(&self) -> usize {
        N::Mode::STREAM_ROWS
    }

    fn owns_physical_name(&self, name: &str) -> bool {
        N::NAME == name || (N::Mode::HAS_PAYLOAD_ROW && N::PAYLOAD_NAME == name)
    }

    fn name(&self) -> &'static str {
        N::NAME
    }

    fn check_managed(&self, db: &crate::armour::Db) -> DbResult<()> {
        N::Mode::check_managed(&self.inner, db)
    }

    fn stream_bytes(&self, payload: &dyn Any) -> DbResult<usize> {
        payload
            .downcast_ref::<N::Payload>()
            .map(N::Mode::stream_bytes)
            .ok_or(DbError::Internal("queue stream payload type mismatch"))
    }

    fn tx_participants(&self) -> Vec<Arc<dyn crate::txn::participant::TxParticipant>> {
        N::Mode::stream_participants(&self.inner)
            .into_iter()
            .map(|token| token.inner)
            .collect()
    }
    fn prepare_stream(&self) -> DbResult<QueueStreamPreparation> {
        N::Mode::prepare_stream(&self.inner)
    }

    fn tx_stream_writes(
        &self,
        payload: &dyn Any,
        visible_at: std::time::SystemTime,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        let QueueStreamPlan {
            mut inner,
            admission,
        } = N::Mode::lower_stream(&self.inner, payload, visible_at)?;
        inner.queue_depth_admissions.push(admission);
        Ok(inner)
    }
}

/// Declared destination queues for one source's change stream.
#[derive(Default, Clone)]
pub struct QueueTargets {
    handles: Vec<Arc<dyn QueueTarget>>,
}

impl QueueTargets {
    pub fn new() -> Self {
        Self::default()
    }

    /// Add a queue destination. A physical header or payload collection may
    /// belong to only one declared queue.
    #[must_use]
    pub fn with<N>(self, queue: &Arc<Queue<N>>) -> Self
    where
        N: QueueSpec,
        <N::Mode as QueueMode<N>>::Inner: Send + Sync,
        N::Mode: QueueStreamPayload<N>,
    {
        assert!(
            !self
                .handles
                .iter()
                .any(|target| target.owns_physical_name(N::NAME)
                    || (N::Mode::HAS_PAYLOAD_ROW && target.owns_physical_name(N::PAYLOAD_NAME))),
            "QueueTargets: duplicate physical queue name"
        );
        let mut handles = self.handles;
        handles.push(Arc::clone(queue) as Arc<dyn QueueTarget>);
        Self { handles }
    }

    pub fn is_empty(&self) -> bool {
        self.handles.is_empty()
    }

    pub fn len(&self) -> usize {
        self.handles.len()
    }

    /// The first declared queue's name, for diagnostics.
    ///
    /// `and_events` rejects empty targets, but `TypedWriteHook` is a public
    /// trait: a hand-written impl can report `EMITS` with an empty target set,
    /// and a panic on the write path is not the way to tell it so.
    pub(crate) fn first_name(&self) -> &'static str {
        self.handles.first().map_or("", |target| target.name())
    }

    /// Queue header names, in declaration order, for schema graph edges.
    pub(crate) fn names(&self) -> impl Iterator<Item = &'static str> + '_ {
        self.handles.iter().map(|target| target.name())
    }

    pub(crate) fn check_managed(&self, db: &crate::armour::Db) -> DbResult<()> {
        for target in &self.handles {
            target.check_managed(db)?;
        }
        Ok(())
    }

    pub(crate) fn tx_participants(&self) -> Vec<Arc<dyn crate::txn::participant::TxParticipant>> {
        self.handles
            .iter()
            .flat_map(|target| target.tx_participants())
            .collect()
    }
    pub(crate) fn prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        let mut permits = Vec::new();
        for target in &self.handles {
            permits.extend(target.prepare_stream()?.into_permits());
        }
        Ok(permits)
    }

    pub(crate) fn lower(
        &self,
        sink: &EmitSink<'_>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        let mut out = crate::txn::participant::TxStreamWrites::empty();
        for emission in sink.emissions() {
            let mut plan = emission
                .target
                .tx_stream_writes(emission.payload.as_ref(), emission.visible_at)?;
            out.writes.append(&mut plan.writes);
            out.after_success.append(&mut plan.after_success);
            out.queue_depth_admissions
                .append(&mut plan.queue_depth_admissions);
        }
        Ok(out)
    }

    pub(crate) fn contains<N>(&self, queue: &Arc<Queue<N>>) -> bool
    where
        N: QueueSpec,
        <N::Mode as QueueMode<N>>::Inner: Send + Sync,
        N::Mode: QueueStreamPayload<N>,
    {
        let queue = Arc::clone(queue) as Arc<dyn QueueTarget>;
        self.handles
            .iter()
            .any(|target| Arc::ptr_eq(target, &queue))
    }
}

/// An erased, buffered stream emission. Task 9 lowers these records into
/// concrete rank-1 queue writes before the transaction log is appended.
pub(crate) struct BufferedEmission {
    #[allow(dead_code)] // Consumed by Task 9's transaction lowering.
    pub(crate) target: Arc<dyn QueueTarget>,
    #[allow(dead_code)] // Consumed by Task 9's transaction lowering.
    pub(crate) payload: Box<dyn Any + Send>,
    #[allow(dead_code)] // Consumed by Task 9's transaction lowering.
    pub(crate) visible_at: std::time::SystemTime,
}

/// Admission and buffering sink for one source transition.
pub struct EmitSink<'a> {
    targets: &'a QueueTargets,
    max_rows: usize,
    max_bytes: usize,
    row_count: usize,
    byte_count: usize,
    emissions: Vec<BufferedEmission>,
}

impl<'a> EmitSink<'a> {
    /// Construct a sink with an explicit budget, for budget-accounting tests.
    ///
    /// The production constructor is crate-private: a sink's budget comes from
    /// `min(carrier, Config::max_projected_*)` on the commit path and is not a
    /// public knob. This mirrors `ProjectionSink`/`ClaimSink`, which are
    /// crate-private for the same reason.
    #[cfg(feature = "queue-test-support")]
    #[doc(hidden)]
    pub fn debug_new(targets: &'a QueueTargets, max_rows: usize, max_bytes: usize) -> Self {
        Self::new(targets, max_rows, max_bytes)
    }

    pub(crate) fn new(targets: &'a QueueTargets, max_rows: usize, max_bytes: usize) -> Self {
        Self {
            targets,
            max_rows,
            max_bytes,
            row_count: 0,
            byte_count: 0,
            emissions: Vec::new(),
        }
    }

    /// Schedule an emission for immediate visibility.
    pub fn emit<N>(&mut self, queue: &Arc<Queue<N>>, payload: N::Payload) -> DbResult<()>
    where
        N: QueueSpec,
        N::Payload: Send + 'static,
        <N::Mode as QueueMode<N>>::Inner: Send + Sync,
        N::Mode: QueueStreamPayload<N>,
    {
        self.emit_at(queue, payload, std::time::SystemTime::now())
    }

    /// Schedule an emission for the supplied wall-clock visibility time.
    pub fn emit_at<N>(
        &mut self,
        queue: &Arc<Queue<N>>,
        payload: N::Payload,
        visible_at: std::time::SystemTime,
    ) -> DbResult<()>
    where
        N: QueueSpec,
        N::Payload: Send + 'static,
        <N::Mode as QueueMode<N>>::Inner: Send + Sync,
        N::Mode: QueueStreamPayload<N>,
    {
        if !self.targets.contains(queue) {
            return Err(DbError::Client("undeclared queue target"));
        }
        let target = Arc::clone(queue) as Arc<dyn QueueTarget>;
        let added_rows = target.physical_rows();
        let added_bytes = target.stream_bytes(&payload)?;
        let next_rows = self.row_count + added_rows;
        let next_bytes = self.byte_count + added_bytes;
        if next_rows > self.max_rows || next_bytes > self.max_bytes {
            return Err(DbError::ProjectionBudgetExceeded {
                rows: next_rows,
                bytes: next_bytes,
            });
        }
        self.row_count = next_rows;
        self.byte_count = next_bytes;
        self.emissions.push(BufferedEmission {
            target,
            payload: Box::new(payload),
            visible_at,
        });
        Ok(())
    }

    /// Number of accepted emissions, in source callback order.
    pub fn len(&self) -> usize {
        self.emissions.len()
    }

    pub fn is_empty(&self) -> bool {
        self.emissions.is_empty()
    }

    #[allow(dead_code)] // Consumed by Task 9's transaction lowering.
    pub(crate) fn emissions(&self) -> &[BufferedEmission] {
        &self.emissions
    }
}

/// Compile-time declaration of a durable queue.
pub trait QueueSpec: Sized + Send + Sync + 'static {
    type Payload;
    type Mode: QueueMode<Self>;

    const NAME: &'static str;
    const PAYLOAD_NAME: &'static str;
}

/// Opaque handle to an engine-owned durable queue.
///
/// A queue is intentionally not a collection declaration or a derived-data
/// target. Downstream crates cannot implement armdb's `CollectionMeta`,
/// `ProjectableIndex`, counter, or unique-namespace traits for this opaque
/// armdb type under Rust's orphan rules, and armdb deliberately provides none
/// of those implementations. This closes queue-as-source and queue-as-target
/// stream topologies at the type boundary.
#[allow(dead_code)]
pub struct Queue<N: QueueSpec> {
    pub(crate) inner: <N::Mode as QueueMode<N>>::Inner,
}

impl<N> Queue<N>
where
    N: QueueSpec,
    N::Mode: QueueHandle<N>,
{
    pub fn push(
        &self,
        payload: N::Payload,
        not_before: Option<std::time::SystemTime>,
    ) -> DbResult<u64> {
        <N::Mode as QueueHandle<N>>::push(self, payload, not_before)
    }
    #[allow(clippy::len_without_is_empty)]
    pub fn len(&self) -> u64 {
        <N::Mode as QueueHandle<N>>::len(self)
    }
    pub fn dead_len(&self) -> u64 {
        <N::Mode as QueueHandle<N>>::dead_len(self)
    }
    pub fn lease(
        &self,
        n: usize,
        timeout: Option<Duration>,
    ) -> DbResult<Vec<LeasedEntry<N::Payload>>> {
        <N::Mode as QueueHandle<N>>::lease(self, n, timeout)
    }
    pub fn ack(&self, token: LeaseToken) -> DbResult<()> {
        <N::Mode as QueueHandle<N>>::ack(self, token)
    }
    pub fn nack(&self, token: LeaseToken) -> DbResult<()> {
        <N::Mode as QueueHandle<N>>::nack(self, token)
    }
    /// Return the bounded delay until work may become leasable.
    ///
    /// `None` means nothing is scheduled — an error is reported, not folded
    /// into that same answer, because a caller using this as a timer source
    /// would otherwise sleep on a default forever.
    pub fn next_delay(&self) -> DbResult<Option<Duration>> {
        Ok(<N::Mode as QueueHandle<N>>::scan_due(self)?.delay)
    }
    /// Block until work is due or `timeout` elapses.
    pub fn wait_due_timeout(&self, timeout: Duration) -> DbResult<bool> {
        wait::wait_due_timeout(self, timeout)
    }
    /// Build a runtime-agnostic future that resolves when work is due.
    pub fn due(&self) -> QueueDue<'_, N> {
        QueueDue::new(self)
    }
    /// Return a read-only snapshot of process-local queue operational counters.
    pub fn stats(&self) -> QueueStats {
        <N::Mode as QueueHandle<N>>::stats(self)
    }
    #[cfg(all(debug_assertions, feature = "queue-test-support"))]
    #[doc(hidden)]
    pub fn debug_wait_registration_count(&self) -> usize {
        <N::Mode as QueueHandle<N>>::wait(self).registration_count()
    }
}

/// Capability authorizing an acknowledgement or retry of one leased row.
///
/// A token is minted exclusively by a successful queue lease; callers cannot
/// manufacture one for an arbitrary row or epoch.
///
/// ```compile_fail
/// use armdb::{LeaseToken, QueueKey};
///
/// let _ = LeaseToken::new(0, QueueKey::live(0, 1), 1);
/// ```
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct LeaseToken {
    queue_instance_id: u64,
    key: QueueKey,
    epoch: u32,
}

impl LeaseToken {
    #[allow(dead_code)] // Used by Queue::lease once queue operations are introduced.
    pub(crate) const fn new(queue_instance_id: u64, key: QueueKey, epoch: u32) -> Self {
        Self {
            queue_instance_id,
            key,
            epoch,
        }
    }

    pub(crate) const fn belongs_to(self, queue_instance_id: u64) -> bool {
        self.queue_instance_id == queue_instance_id
    }

    /// The key of the leased queue row.
    pub const fn key(self) -> QueueKey {
        self.key
    }

    /// The lease epoch that guards acknowledgement and retry.
    pub const fn epoch(self) -> u32 {
        self.epoch
    }
}

/// A payload claimed by a successful lease operation.
#[allow(dead_code)]
pub struct LeasedEntry<P> {
    key: QueueKey,
    id: u64,
    attempts: u16,
    payload: P,
    token: LeaseToken,
}

impl<P> LeasedEntry<P> {
    /// The key of the leased queue row.
    pub const fn key(&self) -> QueueKey {
        self.key
    }

    /// The durable queue entry identifier.
    pub const fn id(&self) -> u64 {
        self.id
    }

    /// The number of failed deliveries recorded before this lease.
    pub const fn attempts(&self) -> u16 {
        self.attempts
    }

    /// Borrow the claimed payload.
    pub fn payload(&self) -> &P {
        &self.payload
    }

    /// The capability required to acknowledge or retry this entry.
    pub const fn token(&self) -> LeaseToken {
        self.token
    }
}

#[cfg(test)]
mod lease_value_tests {
    use super::*;

    #[test]
    fn leased_entry_accessors_expose_the_claimed_value() {
        let key = QueueKey::live(2, 9);
        let token = LeaseToken::new(1, key, 5);
        let entry = LeasedEntry {
            key,
            id: 9,
            attempts: 4,
            payload: 1u64,
            token,
        };
        assert_eq!(entry.key(), key);
        assert_eq!(entry.id(), 9);
        assert_eq!(entry.attempts(), 4);
        assert_eq!(*entry.payload(), 1);
        assert_eq!(entry.token().key(), key);
        assert_eq!(entry.token().epoch(), 5);
    }
}

/// Runtime policy for a durable queue.
pub struct QueueConfig {
    pub(crate) max_attempts: u16,
    pub(crate) backoff_secs: Vec<u64>,
    pub(crate) lease_timeout: Duration,
    pub(crate) dead_retention: Option<Duration>,
    pub(crate) max_depth: Option<u64>,
    pub(crate) wakeup_recheck: Duration,
    header_config: Option<crate::FixedConfig>,
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    payload_config: Option<crate::Config>,
}

impl QueueConfig {
    /// Start from the balanced queue policy.
    pub fn balanced() -> Self {
        Self {
            max_attempts: 3,
            backoff_secs: vec![60],
            lease_timeout: Duration::from_secs(60),
            dead_retention: None,
            max_depth: None,
            wakeup_recheck: Duration::from_secs(5),
            header_config: None,
            #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
            payload_config: None,
        }
    }

    /// Set the number of failed deliveries before dead-lettering.
    pub fn max_attempts(mut self, max_attempts: u16) -> Self {
        self.max_attempts = max_attempts;
        self
    }
    /// Set retry delays in seconds; the final delay repeats as needed.
    pub fn backoff_secs(mut self, backoff_secs: &[u64]) -> Self {
        self.backoff_secs = backoff_secs.to_vec();
        self
    }
    /// Set the default lease duration.
    pub fn lease_timeout(mut self, lease_timeout: Duration) -> Self {
        self.lease_timeout = lease_timeout;
        self
    }
    /// Set retention for dead-letter rows, or retain them indefinitely.
    pub fn dead_retention(mut self, dead_retention: Option<Duration>) -> Self {
        self.dead_retention = dead_retention;
        self
    }
    /// Set the optional soft cap for live rows.
    pub fn max_depth(mut self, max_depth: Option<u64>) -> Self {
        self.max_depth = max_depth;
        self
    }
    /// Set the maximum delay before waiters recheck the wall clock.
    pub fn wakeup_recheck(mut self, wakeup_recheck: Duration) -> Self {
        self.wakeup_recheck = wakeup_recheck;
        self
    }

    /// Configure the FixedStore collection that holds queue headers.
    ///
    /// Queue delivery always scans headers in ascending visibility order, so
    /// the supplied [`crate::FixedConfig::reversed`] setting is ignored and
    /// normalized to `false`. All other FixedStore settings, including sizing,
    /// shard routing, and sync policy, are retained.
    ///
    /// **`grow_step` is retained too, and it is the expensive one.** The queue
    /// default is `100_000`; `FixedConfig::balanced()` is `1_000_000`, i.e. ten
    /// times the version-vector RAM baseline and ten times the preallocated
    /// file length per shard — with a 4 KiB payload that is roughly 4 GiB of
    /// logical length per shard. Override the header config only when you have
    /// a reason to, and set `grow_step` explicitly when you do.
    pub fn header_config(mut self, header_config: crate::FixedConfig) -> Self {
        self.header_config = Some(header_config);
        self
    }

    /// Configure the Bitcask collection that holds a variable queue's payloads.
    ///
    /// This is used only by [`VarQueueMode`] queues. The configuration is passed
    /// to the payload collection unchanged, including its encryption key and
    /// shard settings. Fixed-payload queues ignore it.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    pub fn payload_config(mut self, payload_config: crate::Config) -> Self {
        self.payload_config = Some(payload_config);
        self
    }

    /// Validate and finalize this queue policy.
    pub fn build(self) -> DbResult<Self> {
        if self.max_attempts == 0 {
            return Err(DbError::Config(
                "queue max_attempts must be greater than zero",
            ));
        }
        if self.backoff_secs.is_empty() {
            return Err(DbError::Config("queue backoff_secs must not be empty"));
        }
        if self.lease_timeout.as_millis() == 0 {
            return Err(DbError::Config(
                "queue lease_timeout must be at least one millisecond",
            ));
        }
        if self.wakeup_recheck.as_millis() == 0 {
            return Err(DbError::Config(
                "queue wakeup_recheck must be at least one millisecond",
            ));
        }
        if let Some(config) = &self.header_config {
            config.validate()?;
        }
        #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
        if let Some(config) = &self.payload_config {
            config.validate()?;
        }
        Ok(self)
    }

    /// FixedStore settings reserved by the queue implementation.
    pub fn fixed_config(&self) -> crate::FixedConfig {
        let mut config = self.header_config.clone().unwrap_or_else(|| {
            crate::FixedConfig::balanced()
                .grow_step(100_000)
                .reversed(false)
                .build()
        });
        config.reversed = false;
        config
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    pub(crate) fn var_payload_config(&self) -> crate::Config {
        self.payload_config
            .clone()
            .unwrap_or_else(|| crate::Config::balanced().build())
    }
}