alkstore/alkstore-mem/tests/contract_suite.rs

//! The suite-driving test target: the mem engine's third backlog
//! column (ADR-022's factory reused verbatim — the rows live in
//! `alkstore-contract-suite`, this target is the executor that runs
//! them against the mem factory; the born-pinned posture per
//! ADR-024 §5 — the engine lands against the finished suite).
//!
//! The factory: a fresh `MemStore::new()` per `open` — distinct
//! engine instances sharing nothing (ADR-024 §4's isolation rule,
//! exactly the factory's isolation contract). Teardown is drop — the
//! store handle itself *is* the backing store, so the holder's drop
//! is the whole teardown; a teardown call has nothing external to
//! dispose of and is trivially idempotent.
//!
//! Flavor: every test runs under the **current-thread** runtime —
//! the current-thread-clean posture of the engine's internals
//! (no `spawn_blocking` anywhere) proven by the flavor the rows
//! actually run under ([ADR-024](../../docs/architecture/decisions/024-mem-engine.md)
//! §5). A blocking call inside a row or inside the engine would
//! stall this flavor into a hang, not pass.

use alkstore::{BoxedFuture, Result, Store};
use alkstore_contract_suite::StoreFactory;
use alkstore_mem::MemStore;

/// The mem suite factory: a fresh isolated engine instance per
/// `open`, teardown = drop (the consumer holds the handle; an
/// explicit teardown call is the trivial no-op — nothing exists
/// outside the instance to dispose of).
struct MemFactory;

impl StoreFactory for MemFactory {
    fn open(&self) -> BoxedFuture<'_, Result<Box<dyn Store>>> {
        Box::pin(async move { Ok(Box::new(MemStore::new()) as Box<dyn Store>) })
    }

    fn teardown(&self) -> BoxedFuture<'_, Result<()>> {
        Box::pin(async move { Ok(()) })
    }
}

mod rows {
    use alkstore_contract_suite::properties::{
        backoff_curve_equivalence, concurrent_try_lock_loser_is_a_value,
        drop_rollback_leaves_no_ghosts, duration_refusal_on_non_positive_ttl,
        enqueue_opts_resolution, extent_clamp_semantics, in_tx_reads_see_own_writes,
        job_handle_validity_predicate, lock_ttl_expiry_and_reacquisition,
        name_validation_rejects_empty_and_reserved, outbox_enqueue_tx_commit_atomicity,
        payload_round_trip_stores_exact_encoding, payload_too_large_never_produced_on_mem,
        publish_with_key_tx_commit_atomicity, queue_depth_reclaim_and_dead_letter,
        receiver_close_and_save_arms, scheduler_boundary_fires, scheduler_bounded_catchup,
        scheduler_leadership_discipline, stream_ordering_equivalence, sweep_no_stranded_rows,
        trim_to_semantics, wake_receiver_shapes, with_tx_panicking_closure_rolls_back,
    };

    use super::MemFactory;

    fn factory() -> MemFactory {
        MemFactory
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_name_validation_rejects_empty_and_reserved() {
        name_validation_rejects_empty_and_reserved(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_extent_clamp_semantics() {
        extent_clamp_semantics(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_duration_refusal_on_non_positive_ttl() {
        duration_refusal_on_non_positive_ttl(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_lock_ttl_expiry_and_reacquisition() {
        lock_ttl_expiry_and_reacquisition(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_concurrent_try_lock_loser_is_a_value() {
        concurrent_try_lock_loser_is_a_value(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_payload_round_trip_stores_exact_encoding() {
        payload_round_trip_stores_exact_encoding(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_payload_too_large_never_produced_on_mem() {
        payload_too_large_never_produced_on_mem(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_drop_rollback_leaves_no_ghosts() {
        drop_rollback_leaves_no_ghosts(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_in_tx_reads_see_own_writes() {
        in_tx_reads_see_own_writes(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_enqueue_opts_resolution() {
        enqueue_opts_resolution(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_backoff_curve_equivalence() {
        backoff_curve_equivalence(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_receiver_close_and_save_arms() {
        receiver_close_and_save_arms(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_job_handle_validity_predicate() {
        job_handle_validity_predicate(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_queue_depth_reclaim_and_dead_letter() {
        queue_depth_reclaim_and_dead_letter(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_sweep_no_stranded_rows() {
        sweep_no_stranded_rows(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_scheduler_boundary_fires() {
        scheduler_boundary_fires(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_scheduler_bounded_catchup() {
        scheduler_bounded_catchup(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_scheduler_leadership_discipline() {
        scheduler_leadership_discipline(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_outbox_enqueue_tx_commit_atomicity() {
        outbox_enqueue_tx_commit_atomicity(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_publish_with_key_tx_commit_atomicity() {
        publish_with_key_tx_commit_atomicity(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_stream_ordering_equivalence() {
        stream_ordering_equivalence(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_trim_to_semantics() {
        trim_to_semantics(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_wake_receiver_shapes() {
        wake_receiver_shapes(&factory()).await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn row_with_tx_panicking_closure_rolls_back() {
        with_tx_panicking_closure_rolls_back(&factory()).await;
    }
}

mod factory_shape {
    use serde_json::json;

    use alkstore_contract_suite::StoreFactory;

    use super::MemFactory;

    /// ADR-022's factory contract, verbatim on the mem engine: every
    /// open yield is isolated (distinct `MemStore::new()` instances
    /// sharing nothing — proved behaviorally, a row one store lands
    /// being invisible to the other) and teardown is idempotent per
    /// instance; a failing assertion early-exit (the panic path)
    /// leaves teardown safe against the abandoned store — for mem the
    /// abandoned store simply drops with the test's frame.
    #[tokio::test(flavor = "current_thread")]
    async fn opens_are_isolated_and_teardown_is_idempotent() {
        let factory = MemFactory;

        let a = factory.open().await.unwrap();
        let b = factory.open().await.unwrap();

        let sa = a.stream("iso").await.unwrap();
        let event_offset = sa.publish(json!({"iso": true})).await.unwrap();
        assert!(event_offset > 0);
        let sb = b.stream("iso").await.unwrap();
        let seen = sb.read_since(0, 100).await.unwrap();
        assert!(
            seen.is_empty(),
            "two opens must never share rows — b saw {seen:?}"
        );
        drop(sa);
        drop(sb);
        drop(a);
        drop(b);

        factory.teardown().await.unwrap();
        factory.teardown().await.unwrap();
    }
}