reliar-store-postgres 0.9.0

//! `FaultyStore<S>` — a delegating decorator over a real [`OutboxStore`], scripted per method
//! (ADR 0043 §4). Postgres cannot be told to fail its next `complete`, and coupling a dispatcher
//! **control-flow** trial to whichever real database error happens to classify transient would
//! test error classification instead of the control flow this crate's dispatcher trials actually
//! want to prove. `FaultyStore` closes that gap without becoming a second implementation:
//!
//! - **It never fabricates success.** [`StoreStep::Pass`] and [`StoreStep::DelayBefore`] delegate
//!   to the wrapped store; [`StoreStep::Transient`]/[`StoreStep::Permanent`] return
//!   [`FaultyStoreError::Injected`] **without calling through** — there is no path that returns
//!   `Ok` with invented rows, counts or ids.
//! - **It owns no domain state** — only a per-method script and call counter. Every assertion in
//!   a faulted trial still reads Postgres through the wrapped store or a `sqlx` query, never
//!   through this decorator (ADR 0043 §2's stimulus/oracle line).
//! - **`Classify` forwards**: [`FaultyStoreError::Inner`] delegates to the wrapped error's own
//!   `kind()`; [`FaultyStoreError::Injected`] reports the scripted kind directly.
//! - **It is scripted per method** (`on_acquire`, `on_complete`), so a trial names exactly which
//!   call it is perturbing.
//! - **It lives here, in the test harness, only** — never in a published crate, never behind a
//!   feature, so nothing outside these trials can mistake it for an implementation of the
//!   contract (ADR 0043 Amendment A.1).
//!
//! [`StoreStep::DetachOnSignal`] (ADR 0046 P-21) is a fourth shape, for `complete`/`fail` only:
//! unlike [`StoreStep::DelayBefore`], it never delegates on its own — a dropped
//! `store_timeout`-bounded future would just cancel a pending delegation, never let it "land"
//! later. It captures the call's owned arguments instead, returns
//! [`FaultyStoreError::Injected`] at once, and stays armed (every further call is captured and
//! injected the same way, overwriting nothing — the args are identical across retries of the
//! same claim) until a test calls [`FaultyStore::release_detached_complete`]/
//! [`FaultyStore::release_detached_fail`], which disarms it and runs the captured call against
//! the real store — simulating a "lost ack" statement that finally reaches Postgres.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use reliar_core::{Classify, FailureKind};
use reliar_outbox::{
    AcquireRequest, AcquiredBatch, FailedRecord, OutboxStats, OutboxStore, PurgeReport,
    PurgeRequest, RecordRef, WorkerId,
};

/// One scripted response for a single call to a `FaultyStore`-wrapped method.
#[derive(Clone, Copy, Debug)]
pub(crate) enum StoreStep {
    /// Delegate to the wrapped store, unmodified.
    Pass,

    /// Return an injected transient error; the wrapped store is **not** called.
    Transient,

    /// Return an injected permanent error; the wrapped store is **not** called.
    Permanent,

    /// Sleep for the given duration, then delegate.
    DelayBefore(Duration),

    /// Capture the call and return an injected transient error at once — see the module docs.
    /// `complete`/`fail` only.
    DetachOnSignal,
}

/// [`OutboxStore::Error`] for a [`FaultyStore`]: either the wrapped store's own error, or a
/// failure this decorator injected instead of ever calling through.
#[derive(Debug)]
pub(crate) enum FaultyStoreError<E> {
    /// A failure this decorator invented, never the wrapped store's.
    Injected(FailureKind),

    /// The wrapped store's own error, passed through unchanged.
    Inner(E),
}

impl<E: std::fmt::Display> std::fmt::Display for FaultyStoreError<E> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Injected(kind) => write!(f, "faulty store: injected {kind:?} failure"),
            Self::Inner(inner) => write!(f, "faulty store: {inner}"),
        }
    }
}

impl<E: std::error::Error + 'static> std::error::Error for FaultyStoreError<E> {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Injected(_) => None,
            Self::Inner(inner) => Some(inner),
        }
    }
}

impl<E: Classify> Classify for FaultyStoreError<E> {
    fn kind(&self) -> FailureKind {
        match self {
            Self::Injected(kind) => *kind,
            Self::Inner(inner) => inner.kind(),
        }
    }
}

/// Every method [`FaultyStore`] can script, one queue each. An empty queue always answers
/// [`StoreStep::Pass`] — a trial only ever scripts the calls it cares about perturbing.
#[derive(Default)]
struct Script {
    acquire: VecDeque<StoreStep>,

    complete: VecDeque<StoreStep>,

    fail: VecDeque<StoreStep>,

    release: VecDeque<StoreStep>,

    extend_lease: VecDeque<StoreStep>,

    purge: VecDeque<StoreStep>,

    stats: VecDeque<StoreStep>,
}

impl Script {
    fn take(queue: &mut VecDeque<StoreStep>) -> StoreStep {
        queue.pop_front().unwrap_or(StoreStep::Pass)
    }

    /// As [`Self::take`], except [`StoreStep::DetachOnSignal`] is **peeked, not popped** — it
    /// stays at the front of `queue` (module docs) until a test explicitly disarms it via
    /// [`FaultyStore::release_detached_complete`]/[`FaultyStore::release_detached_fail`], so every
    /// call in between (including the dispatcher's own automatic outcome-write retries) is
    /// captured and injected the same way. `complete`/`fail` only.
    fn take_detachable(queue: &mut VecDeque<StoreStep>) -> StoreStep {
        if matches!(queue.front(), Some(StoreStep::DetachOnSignal)) {
            StoreStep::DetachOnSignal
        } else {
            Self::take(queue)
        }
    }

    /// Pops the sticky [`StoreStep::DetachOnSignal`] scripted by
    /// [`FaultyStore::on_complete`]/[`FaultyStore::on_fail`] — called only by
    /// [`FaultyStore::release_detached_complete`]/[`FaultyStore::release_detached_fail`], once the
    /// captured call is about to run: every call after this one falls through to whatever is
    /// scripted next (default [`StoreStep::Pass`]).
    fn disarm_detach(queue: &mut VecDeque<StoreStep>) {
        if matches!(queue.front(), Some(StoreStep::DetachOnSignal)) {
            queue.pop_front();
        }
    }
}

/// Per-method call counts — stimulus bookkeeping only (ADR 0043 §2), never read by an assertion
/// as a stand-in for store state.
#[derive(Default)]
struct Counters {
    acquire: AtomicU64,

    complete: AtomicU64,

    fail: AtomicU64,

    release: AtomicU64,

    extend_lease: AtomicU64,

    purge: AtomicU64,

    stats: AtomicU64,
}

/// One call captured by [`StoreStep::DetachOnSignal`] (module docs), owned so it can outlive the
/// borrowed `complete`/`fail` call that captured it.
struct DetachedCall<T> {
    worker: WorkerId,

    items: Vec<T>,
}

/// A delegating decorator over a real [`OutboxStore`] `S`, scripted per method (module docs).
/// Cloning a `FaultyStore` shares its script and counters — the same scripted answers apply
/// through every clone, exactly as a real store's state is shared through its own clones.
pub(crate) struct FaultyStore<S> {
    inner: S,

    script: Arc<Mutex<Script>>,

    counters: Arc<Counters>,

    detached_complete: Arc<Mutex<Option<DetachedCall<RecordRef>>>>,

    detached_fail: Arc<Mutex<Option<DetachedCall<FailedRecord>>>>,
}

impl<S> FaultyStore<S> {
    /// Wraps `inner` with an empty script — every call passes through until a trial scripts one.
    pub(crate) fn new(inner: S) -> Self {
        Self {
            inner,
            script: Arc::new(Mutex::new(Script::default())),
            counters: Arc::new(Counters::default()),
            detached_complete: Arc::new(Mutex::new(None)),
            detached_fail: Arc::new(Mutex::new(None)),
        }
    }

    /// Scripts `steps` for the next calls to [`OutboxStore::acquire`], in order; calls past the
    /// end of `steps` fall back to [`StoreStep::Pass`].
    pub(crate) fn on_acquire(&self, steps: impl IntoIterator<Item = StoreStep>) {
        self.script.lock().unwrap().acquire.extend(steps);
    }

    /// Scripts `steps` for [`OutboxStore::complete`], as [`Self::on_acquire`]. A
    /// [`StoreStep::DetachOnSignal`] entry is sticky (module docs) rather than one-shot — script
    /// exactly one, never several, per capture window.
    pub(crate) fn on_complete(&self, steps: impl IntoIterator<Item = StoreStep>) {
        self.script.lock().unwrap().complete.extend(steps);
    }

    /// Scripts `steps` for [`OutboxStore::fail`], as [`Self::on_complete`].
    pub(crate) fn on_fail(&self, steps: impl IntoIterator<Item = StoreStep>) {
        self.script.lock().unwrap().fail.extend(steps);
    }

    /// How many times [`OutboxStore::acquire`] has been called on this store (or any clone of
    /// it) so far — stimulus bookkeeping, never a stand-in for a Postgres-read assertion.
    pub(crate) fn acquire_calls(&self) -> u64 {
        self.counters.acquire.load(Ordering::Relaxed)
    }

    /// As [`Self::acquire_calls`], for [`OutboxStore::complete`].
    pub(crate) fn complete_calls(&self) -> u64 {
        self.counters.complete.load(Ordering::Relaxed)
    }

    /// `true` once a `complete` call has been captured by an armed
    /// [`StoreStep::DetachOnSignal`] — a trial polls this (bounded, never a sleep) before moving
    /// on to the next step of a P-21-shaped scenario.
    pub(crate) fn has_detached_complete(&self) -> bool {
        self.detached_complete.lock().unwrap().is_some()
    }

    /// As [`Self::has_detached_complete`], for `fail`.
    pub(crate) fn has_detached_fail(&self) -> bool {
        self.detached_fail.lock().unwrap().is_some()
    }
}

impl<S: OutboxStore> FaultyStore<S> {
    /// Disarms [`StoreStep::DetachOnSignal`] for `complete` and executes the captured call
    /// against the real store — the "lost" statement finally reaching Postgres. Panics if no
    /// call was captured: a test that reaches this point without one has a scenario bug, not a
    /// store fault to report.
    pub(crate) async fn release_detached_complete(&self) -> Result<u64, S::Error> {
        Script::disarm_detach(&mut self.script.lock().unwrap().complete);

        let DetachedCall { worker, items } = self
            .detached_complete
            .lock()
            .unwrap()
            .take()
            .expect("release_detached_complete called with no call captured");

        self.inner.complete(&worker, &items).await
    }

    /// As [`Self::release_detached_complete`], for `fail`.
    pub(crate) async fn release_detached_fail(&self) -> Result<u64, S::Error> {
        Script::disarm_detach(&mut self.script.lock().unwrap().fail);

        let DetachedCall { worker, items } = self
            .detached_fail
            .lock()
            .unwrap()
            .take()
            .expect("release_detached_fail called with no call captured");

        self.inner.fail(&worker, &items).await
    }
}

impl<S: Clone> Clone for FaultyStore<S> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            script: Arc::clone(&self.script),
            counters: Arc::clone(&self.counters),
            detached_complete: Arc::clone(&self.detached_complete),
            detached_fail: Arc::clone(&self.detached_fail),
        }
    }
}

impl<S: OutboxStore> OutboxStore for FaultyStore<S> {
    type Error = FaultyStoreError<S::Error>;

    async fn acquire(&self, request: AcquireRequest) -> Result<AcquiredBatch, Self::Error> {
        self.counters.acquire.fetch_add(1, Ordering::Relaxed);
        let step = Script::take(&mut self.script.lock().unwrap().acquire);

        match step {
            StoreStep::Pass => self
                .inner
                .acquire(request)
                .await
                .map_err(FaultyStoreError::Inner),

            StoreStep::DelayBefore(delay) => {
                tokio::time::sleep(delay).await;

                self.inner
                    .acquire(request)
                    .await
                    .map_err(FaultyStoreError::Inner)
            }

            StoreStep::Transient => Err(FaultyStoreError::Injected(FailureKind::Transient)),

            StoreStep::Permanent => Err(FaultyStoreError::Injected(FailureKind::Permanent)),

            StoreStep::DetachOnSignal => {
                unimplemented!("StoreStep::DetachOnSignal is only supported for complete/fail")
            }
        }
    }

    async fn complete(&self, worker: &WorkerId, items: &[RecordRef]) -> Result<u64, Self::Error> {
        self.counters.complete.fetch_add(1, Ordering::Relaxed);
        let step = Script::take_detachable(&mut self.script.lock().unwrap().complete);

        match step {
            StoreStep::Pass => self
                .inner
                .complete(worker, items)
                .await
                .map_err(FaultyStoreError::Inner),

            StoreStep::DelayBefore(delay) => {
                tokio::time::sleep(delay).await;

                self.inner
                    .complete(worker, items)
                    .await
                    .map_err(FaultyStoreError::Inner)
            }

            StoreStep::Transient => Err(FaultyStoreError::Injected(FailureKind::Transient)),

            StoreStep::Permanent => Err(FaultyStoreError::Injected(FailureKind::Permanent)),

            StoreStep::DetachOnSignal => {
                let mut slot = self.detached_complete.lock().unwrap();

                if slot.is_none() {
                    *slot = Some(DetachedCall {
                        worker: worker.clone(),
                        items: items.to_vec(),
                    });
                }

                Err(FaultyStoreError::Injected(FailureKind::Transient))
            }
        }
    }

    async fn fail(&self, worker: &WorkerId, items: &[FailedRecord]) -> Result<u64, Self::Error> {
        self.counters.fail.fetch_add(1, Ordering::Relaxed);
        let step = Script::take_detachable(&mut self.script.lock().unwrap().fail);

        match step {
            StoreStep::Pass => self
                .inner
                .fail(worker, items)
                .await
                .map_err(FaultyStoreError::Inner),

            StoreStep::DelayBefore(delay) => {
                tokio::time::sleep(delay).await;

                self.inner
                    .fail(worker, items)
                    .await
                    .map_err(FaultyStoreError::Inner)
            }

            StoreStep::Transient => Err(FaultyStoreError::Injected(FailureKind::Transient)),

            StoreStep::Permanent => Err(FaultyStoreError::Injected(FailureKind::Permanent)),

            StoreStep::DetachOnSignal => {
                let mut slot = self.detached_fail.lock().unwrap();

                if slot.is_none() {
                    *slot = Some(DetachedCall {
                        worker: worker.clone(),
                        items: items.to_vec(),
                    });
                }

                Err(FaultyStoreError::Injected(FailureKind::Transient))
            }
        }
    }

    async fn release(&self, worker: &WorkerId, items: &[RecordRef]) -> Result<u64, Self::Error> {
        self.counters.release.fetch_add(1, Ordering::Relaxed);
        let step = Script::take(&mut self.script.lock().unwrap().release);

        match step {
            StoreStep::Pass => self
                .inner
                .release(worker, items)
                .await
                .map_err(FaultyStoreError::Inner),

            StoreStep::DelayBefore(delay) => {
                tokio::time::sleep(delay).await;

                self.inner
                    .release(worker, items)
                    .await
                    .map_err(FaultyStoreError::Inner)
            }

            StoreStep::Transient => Err(FaultyStoreError::Injected(FailureKind::Transient)),

            StoreStep::Permanent => Err(FaultyStoreError::Injected(FailureKind::Permanent)),

            StoreStep::DetachOnSignal => {
                unimplemented!("StoreStep::DetachOnSignal is only supported for complete/fail")
            }
        }
    }

    async fn extend_lease(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
        lease: Duration,
    ) -> Result<u64, Self::Error> {
        self.counters.extend_lease.fetch_add(1, Ordering::Relaxed);
        let step = Script::take(&mut self.script.lock().unwrap().extend_lease);

        match step {
            StoreStep::Pass => self
                .inner
                .extend_lease(worker, items, lease)
                .await
                .map_err(FaultyStoreError::Inner),

            StoreStep::DelayBefore(delay) => {
                tokio::time::sleep(delay).await;

                self.inner
                    .extend_lease(worker, items, lease)
                    .await
                    .map_err(FaultyStoreError::Inner)
            }

            StoreStep::Transient => Err(FaultyStoreError::Injected(FailureKind::Transient)),

            StoreStep::Permanent => Err(FaultyStoreError::Injected(FailureKind::Permanent)),

            StoreStep::DetachOnSignal => {
                unimplemented!("StoreStep::DetachOnSignal is only supported for complete/fail")
            }
        }
    }

    async fn purge(&self, request: PurgeRequest) -> Result<PurgeReport, Self::Error> {
        self.counters.purge.fetch_add(1, Ordering::Relaxed);
        let step = Script::take(&mut self.script.lock().unwrap().purge);

        match step {
            StoreStep::Pass => self
                .inner
                .purge(request)
                .await
                .map_err(FaultyStoreError::Inner),

            StoreStep::DelayBefore(delay) => {
                tokio::time::sleep(delay).await;

                self.inner
                    .purge(request)
                    .await
                    .map_err(FaultyStoreError::Inner)
            }

            StoreStep::Transient => Err(FaultyStoreError::Injected(FailureKind::Transient)),

            StoreStep::Permanent => Err(FaultyStoreError::Injected(FailureKind::Permanent)),

            StoreStep::DetachOnSignal => {
                unimplemented!("StoreStep::DetachOnSignal is only supported for complete/fail")
            }
        }
    }

    async fn stats(&self) -> Result<OutboxStats, Self::Error> {
        self.counters.stats.fetch_add(1, Ordering::Relaxed);
        let step = Script::take(&mut self.script.lock().unwrap().stats);

        match step {
            StoreStep::Pass => self.inner.stats().await.map_err(FaultyStoreError::Inner),

            StoreStep::DelayBefore(delay) => {
                tokio::time::sleep(delay).await;

                self.inner.stats().await.map_err(FaultyStoreError::Inner)
            }

            StoreStep::Transient => Err(FaultyStoreError::Injected(FailureKind::Transient)),

            StoreStep::Permanent => Err(FaultyStoreError::Injected(FailureKind::Permanent)),

            StoreStep::DetachOnSignal => {
                unimplemented!("StoreStep::DetachOnSignal is only supported for complete/fail")
            }
        }
    }
}