reliar-store-postgres 0.9.0

#![allow(dead_code)]
//! `StubTransport` — a harness-local `Publisher` stimulus (ADR 0043 A.1) standing in for this
//! binary's dispatcher trials' transport. It can only succeed, stall or panic; its error type is
//! uninhabited, so no trial in this binary can assert the dispatcher's reaction to a *fabricated*
//! transport error — a real publish failure (transient or permanent) is provable only against a
//! real broker, in `tests/system`'s Toxiproxy/NATS trials (E13–E16).
//!
//! [`PublishStep::Stall`] is what a slow broker looks like from inside the dispatcher, and it is
//! the only way to hold one named row in flight while others complete; [`PublishStep::Panic`] is a
//! defensive property about *any* caller-supplied `Publisher`, and no broker produces one.

use std::collections::HashMap;
use std::future::Future;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;

use reliar_core::{Classify, FailureKind, MessageId, Publisher, SerializedEnvelope};

/// Uninhabited on purpose: `StubTransport` has no way to report a publish failure, so "transient"
/// and "permanent" publish outcomes are unreachable from this binary (ADR 0043 A.1).
#[derive(Debug)]
pub(crate) enum Never {}

impl std::fmt::Display for Never {
    fn fmt(&self, _f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match *self {}
    }
}

impl std::error::Error for Never {}

impl Classify for Never {
    fn kind(&self) -> FailureKind {
        match *self {}
    }
}

/// One scripted publish outcome — never a failure (module docs).
#[derive(Clone, Copy, Debug)]
pub(crate) enum PublishStep {
    /// Resolves `Ok` on the future's first poll.
    Ok,

    /// Resolves `Ok` only after `Duration` has elapsed.
    Stall(Duration),

    /// Panics instead of returning.
    Panic,
}

enum Script {
    /// The same outcome for every publish, regardless of id.
    Always(PublishStep),

    /// Per-message outcomes; an id absent from the map publishes [`PublishStep::Ok`].
    Keyed(HashMap<MessageId, PublishStep>),
}

impl Default for Script {
    fn default() -> Self {
        Self::Always(PublishStep::Ok)
    }
}

#[derive(Default)]
struct Inner {
    published: Vec<MessageId>,

    in_flight: usize,

    in_flight_peak: usize,

    script: Script,
}

/// A `Publisher` stimulus that can only succeed, stall or panic (ADR 0043 A.1). Records every
/// publish, in order, with duplicates — duplicates are the assertion, not a bug: they are what a
/// crash-after-publish or a reclaimed lease produces.
#[derive(Clone, Default)]
pub(crate) struct StubTransport {
    inner: Arc<Mutex<Inner>>,
}

impl StubTransport {
    /// Every publish succeeds immediately. Never sleeps.
    pub(crate) fn ok() -> Self {
        Self::default()
    }

    /// Every publish succeeds only after `delay` — long enough for several concurrently spawned
    /// publishes to genuinely overlap, so a caller can observe [`Self::in_flight_peak`] past `1`.
    pub(crate) fn with_concurrency_probe(delay: Duration) -> Self {
        Self::always(PublishStep::Stall(delay))
    }

    /// Per-message outcomes, order-independent and safe at any concurrency.
    pub(crate) fn keyed(steps: impl IntoIterator<Item = (MessageId, PublishStep)>) -> Self {
        Self {
            inner: Arc::new(Mutex::new(Inner {
                script: Script::Keyed(steps.into_iter().collect()),
                ..Inner::default()
            })),
        }
    }

    /// The same outcome for every publish, at any concurrency.
    pub(crate) fn always(step: PublishStep) -> Self {
        Self {
            inner: Arc::new(Mutex::new(Inner {
                script: Script::Always(step),
                ..Inner::default()
            })),
        }
    }

    fn lock(&self) -> MutexGuard<'_, Inner> {
        self.inner.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Every published envelope's id, in call order, duplicates included. Recorded on the
    /// future's **first poll**, not when `publish` is called and not when it completes.
    pub(crate) fn published(&self) -> Vec<MessageId> {
        self.lock().published.clone()
    }

    /// How many times `id` was published — `2` proves the duplicate window.
    pub(crate) fn count(&self, id: MessageId) -> usize {
        self.lock()
            .published
            .iter()
            .filter(|&&seen| seen == id)
            .count()
    }

    /// The high-water mark of concurrently in-flight `publish` calls.
    pub(crate) fn in_flight_peak(&self) -> usize {
        self.lock().in_flight_peak
    }

    /// Records the call and picks this call's step. Synchronous: the lock never crosses an
    /// `.await`.
    fn step_for(&self, id: MessageId) -> PublishStep {
        let mut guard = self.lock();

        guard.published.push(id);
        guard.in_flight += 1;
        guard.in_flight_peak = guard.in_flight_peak.max(guard.in_flight);

        match &guard.script {
            Script::Always(step) => *step,
            Script::Keyed(steps) => steps.get(&id).copied().unwrap_or(PublishStep::Ok),
        }
    }
}

/// Decrements `in_flight` on drop, not on a normal-return code path — a future dropped mid-poll
/// (aborted at shutdown, or dropped by `tokio::time::timeout` when `publish_timeout` elapses on a
/// [`PublishStep::Stall`]) or unwinding through a [`PublishStep::Panic`] must still release its
/// slot, or `in_flight`/`in_flight_peak` stay permanently inflated for the rest of the test.
struct InFlightGuard(StubTransport);

impl Drop for InFlightGuard {
    fn drop(&mut self) {
        self.0.lock().in_flight -= 1;
    }
}

impl Publisher for StubTransport {
    type Error = Never;

    #[allow(
        clippy::panic,
        reason = "PublishStep::Panic exists to simulate a publish task crashing mid-flight — the \
                  panic is the stub's whole purpose here, not an accident"
    )]
    fn publish(
        &self,
        envelope: &SerializedEnvelope,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        let stub = self.clone();
        let id = envelope.id;

        async move {
            // `async move` bodies are lazy: everything here runs on the future's first poll, not
            // when `publish` is called — `published` and `in_flight` are recorded here, so a
            // stalled or never-polled publish still counts as attempted.
            let step = stub.step_for(id);
            let _in_flight = InFlightGuard(stub.clone());

            match step {
                PublishStep::Ok => {}
                PublishStep::Stall(duration) => tokio::time::sleep(duration).await,
                PublishStep::Panic => panic!("StubTransport: scripted publish panic"),
            }

            Ok(())
        }
    }
}