reliar-store-postgres 0.9.0

#![allow(dead_code)]
//! Delegating `OutboxStore` decorators for the dispatcher's claim-cadence and concurrency trials
//! (ADR 0043 §2) — stimuli, never oracles: each wraps a real store (usually
//! [`reliar_store_postgres::PostgresOutboxStore`]) and perturbs only the shape of the request or
//! counts its own calls; every assertion in a trial that uses one still reads Postgres or the
//! publisher/metrics doubles, never these wrappers' own state beyond a call count. Ported from
//! `reliar-outbox`'s former `tests/common/mod.rs` (pre-ADR-0043), generic over `S: OutboxStore` so
//! they now wrap the real store instead of the deleted in-memory fake.

use std::future::Future;
use std::time::Duration;

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

/// Ignores the caller's requested `batch_size` and always claims up to `over_claim_batch_size`
/// instead — stands in for a third-party store that does not honor the request, so a trial can
/// make `outstanding` exceed `max_in_flight` deterministically and prove the dispatcher's
/// `Semaphore` (not just the claim gate) bounds concurrency (architect ruling, RELIAR-15).
#[derive(Clone)]
pub(crate) struct OverDeliveringStore<S> {
    inner: S,

    over_claim_batch_size: u32,
}

impl<S> OverDeliveringStore<S> {
    pub(crate) fn new(inner: S, over_claim_batch_size: u32) -> Self {
        Self {
            inner,
            over_claim_batch_size,
        }
    }
}

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

    fn acquire(
        &self,
        request: AcquireRequest,
    ) -> impl Future<Output = Result<AcquiredBatch, Self::Error>> + Send {
        self.inner
            .acquire(request.batch_size(self.over_claim_batch_size))
    }

    fn complete(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.complete(worker, items)
    }

    fn fail(
        &self,
        worker: &WorkerId,
        items: &[FailedRecord],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.fail(worker, items)
    }

    fn release(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.release(worker, items)
    }

    fn extend_lease(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
        lease: Duration,
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.extend_lease(worker, items, lease)
    }

    fn purge(
        &self,
        request: PurgeRequest,
    ) -> impl Future<Output = Result<PurgeReport, Self::Error>> + Send {
        self.inner.purge(request)
    }

    fn stats(&self) -> impl Future<Output = Result<OutboxStats, Self::Error>> + Send {
        self.inner.stats()
    }
}

/// Wraps a store's `acquire`, counting every call made to it — the only way to tell an immediate
/// (wrong) re-claim apart from a correctly-backed-off one when a partial or fully poisoned claim
/// empties the store, so a premature re-claim returns the same row count (zero) as no re-claim at
/// all (ADR 0039). Mirrors [`PoisoningStore`]'s wrap-and-delegate shape.
#[derive(Clone, Default)]
pub(crate) struct CountingStore<S> {
    inner: S,

    calls: std::sync::Arc<std::sync::atomic::AtomicUsize>,
}

impl<S> CountingStore<S> {
    pub(crate) fn new(inner: S) -> Self {
        Self {
            inner,
            calls: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
        }
    }

    /// The number of `acquire` calls made so far, across every clone sharing this store —
    /// stimulus bookkeeping, never a stand-in for a Postgres-read assertion.
    pub(crate) fn claim_calls(&self) -> usize {
        self.calls.load(std::sync::atomic::Ordering::SeqCst)
    }
}

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

    fn acquire(
        &self,
        request: AcquireRequest,
    ) -> impl Future<Output = Result<AcquiredBatch, Self::Error>> + Send {
        self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

        self.inner.acquire(request)
    }

    fn complete(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.complete(worker, items)
    }

    fn fail(
        &self,
        worker: &WorkerId,
        items: &[FailedRecord],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.fail(worker, items)
    }

    fn release(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.release(worker, items)
    }

    fn extend_lease(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
        lease: Duration,
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.extend_lease(worker, items, lease)
    }

    fn purge(
        &self,
        request: PurgeRequest,
    ) -> impl Future<Output = Result<PurgeReport, Self::Error>> + Send {
        self.inner.purge(request)
    }

    fn stats(&self) -> impl Future<Output = Result<OutboxStats, Self::Error>> + Send {
        self.inner.stats()
    }
}

/// Wraps a store's `acquire`, reclassifying every claimed row beyond the first `n_healthy` as
/// poisoned instead of a real record — stands in for a provider's undecodable-row sweep without
/// needing real corrupt bytes. The underlying rows are genuinely claimed in Postgres; only the
/// `AcquiredBatch` this decorator hands back to the dispatcher relabels them (ADR 0039 §4).
#[derive(Clone)]
pub(crate) struct PoisoningStore<S> {
    inner: S,

    n_healthy: usize,
}

impl<S> PoisoningStore<S> {
    pub(crate) fn new(inner: S, n_healthy: usize) -> Self {
        Self { inner, n_healthy }
    }
}

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

    fn acquire(
        &self,
        request: AcquireRequest,
    ) -> impl Future<Output = Result<AcquiredBatch, Self::Error>> + Send {
        let n_healthy = self.n_healthy;
        let claim = self.inner.acquire(request);

        async move {
            let batch = claim.await?;
            let mut records = batch.records;
            let to_poison = records.split_off(n_healthy.min(records.len()));
            let mut poisoned = batch.poisoned;

            poisoned.extend(to_poison.iter().map(|record| {
                PoisonedRow::new(
                    record.id,
                    record.envelope.id,
                    "harness fixture: poisoned by PoisoningStore",
                )
            }));

            Ok(AcquiredBatch::new(records, poisoned))
        }
    }

    fn complete(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.complete(worker, items)
    }

    fn fail(
        &self,
        worker: &WorkerId,
        items: &[FailedRecord],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.fail(worker, items)
    }

    fn release(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.release(worker, items)
    }

    fn extend_lease(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
        lease: Duration,
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        self.inner.extend_lease(worker, items, lease)
    }

    fn purge(
        &self,
        request: PurgeRequest,
    ) -> impl Future<Output = Result<PurgeReport, Self::Error>> + Send {
        self.inner.purge(request)
    }

    fn stats(&self) -> impl Future<Output = Result<OutboxStats, Self::Error>> + Send {
        self.inner.stats()
    }
}