armdb 0.11.2

//! The post-log-point failure contract, exercised the way a consumer does it:
//! only through the crate's public surface.
//!
//! Needs `armour` + `test-faults` — `armdb::txn` (and therefore `failpoints`)
//! is declared under both.
#![cfg(all(feature = "armour", feature = "test-faults"))]

use std::sync::Arc;

use armdb::armour::Db;
use armdb::txn::failpoints::{ApplyFailGuard, LogSyncFailGuard};
use armdb::{CollectionMeta, Config, DbError, NoHook, RapiraCodec, TtlHook, TxFailStage, TypedMap};
use armour_core::GetType;
use rapira::Rapira;

type Id = [u8; 8];

fn id(n: u64) -> Id {
    n.to_be_bytes()
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct User {
    active: bool,
}
impl CollectionMeta for User {
    type SelfId = Id;
    const NAME: &'static str = "users";
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct Counter {
    n: u32,
}
impl CollectionMeta for Counter {
    type SelfId = Id;
    const NAME: &'static str = "counters";
}

type Users = Arc<TypedMap<Id, User, RapiraCodec, TtlHook<User, NoHook>>>;
type Counters = Arc<TypedMap<Id, Counter, RapiraCodec, TtlHook<Counter, NoHook>>>;

fn open_pair(db: &Arc<Db>) -> (Users, Counters) {
    let users = db
        .open_typed_map::<User, RapiraCodec, _>(Config::test(), NoHook, &[])
        .unwrap();
    let counters = db
        .open_typed_map::<Counter, RapiraCodec, _>(Config::test(), NoHook, &[])
        .unwrap();
    (users, counters)
}

/// The classifier a consumer writes: stage decides the reaction, and only
/// `Apply` with a non-empty `fenced` means "reopen the database".
#[test]
fn an_apply_failure_is_classifiable_from_outside_the_crate() {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let (users, counters) = open_pair(&db);

    let _fail = ApplyFailGuard::at(0);
    let err = db
        .tx()
        .put(&users, &id(1), &User { active: true })
        .unwrap()
        .put(&counters, &id(1), &Counter { n: 1 })
        .unwrap()
        .commit()
        .unwrap_err();
    drop(_fail);

    let DbError::TxFailedAfterLog {
        stage,
        fenced,
        source,
        tx_id,
    } = &err
    else {
        panic!("expected TxFailedAfterLog, got {err}");
    };
    assert_eq!(*stage, TxFailStage::Apply);
    assert!(*tx_id > 0);
    assert!(!fenced.is_empty(), "an apply failure fences its shard");
    assert_eq!(fenced[0].collection, "users");
    assert!(
        format!("{source}").contains("injected apply failure"),
        "{source}"
    );
    assert_eq!(err.status_code(), 500);

    // The engine keeps its word: that shard now refuses writes with the very
    // gsn the report carried. The same key, because `Config::test()` shards the
    // collection two ways and another key could miss the fenced shard.
    //
    // Deviation from the brief's verbatim code: `TypedMap::put` takes its
    // value by value (not `&T`), and its `Ok` payload is `TypedRef<'_, T>`,
    // which does not implement `Debug` — so `.unwrap_err()` (which bounds
    // `T: Debug`) does not compile here. Matched on the `Result` directly
    // instead; the assertion is unchanged.
    match users.put(&id(1), User { active: false }) {
        Err(DbError::ShardFenced { gsn, .. }) => assert_eq!(gsn, fenced[0].gsn),
        Err(other) => panic!("expected ShardFenced, got {other}"),
        Ok(_) => panic!("expected ShardFenced, got Ok"),
    }
}

/// A `durable()` commit whose log fsync fails: applied, wrapped, nothing
/// fenced. Reopening the database would not change anything here.
#[test]
fn a_sync_failure_is_classifiable_from_outside_the_crate() {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let (users, _counters) = open_pair(&db);

    let _fail = LogSyncFailGuard::arm();
    let err = db
        .tx()
        .put(&users, &id(2), &User { active: true })
        .unwrap()
        .durable()
        .commit()
        .unwrap_err();
    drop(_fail);

    let DbError::TxFailedAfterLog { stage, fenced, .. } = &err else {
        panic!("expected TxFailedAfterLog, got {err}");
    };
    assert_eq!(*stage, TxFailStage::Sync);
    assert!(fenced.is_empty(), "{fenced:?}");
    assert_eq!(
        users.get(&id(2)).map(|u| u.active),
        Some(true),
        "apply continues past a sync failure — this is why a retry is wrong"
    );
}

/// The other half of the contract: a failure before the append is a plain
/// error, and the transaction can be retried.
#[test]
fn a_pre_log_failure_is_not_wrapped() {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let (users, _counters) = open_pair(&db);

    let err = db
        .tx()
        .put(&users, &id(3), &User { active: true })
        .unwrap()
        .guard(&users, &id(3), |_cur: Option<&User>| {
            Err(DbError::Client("nope"))
        })
        .unwrap()
        .commit()
        .unwrap_err();

    assert!(!matches!(err, DbError::TxFailedAfterLog { .. }), "{err}");
    assert!(users.get(&id(3)).is_none(), "nothing was written");
}

/// The guards must survive a failed assertion, or one broken test takes the
/// rest of the binary with it.
#[test]
fn failpoint_guards_disarm_on_unwind() {
    let caught = std::panic::catch_unwind(|| {
        let _a = ApplyFailGuard::at(0);
        let _s = LogSyncFailGuard::arm();
        panic!("boom");
    });
    assert!(caught.is_err());

    // Proof by behaviour: a plain commit now succeeds.
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let (users, _counters) = open_pair(&db);
    db.tx()
        .put(&users, &id(4), &User { active: true })
        .unwrap()
        .durable()
        .commit()
        .expect("both failpoints disarmed");
}