armdb 0.12.0

use super::tests::{CountingHook, open_fixed_tree};
use crate::fixed::slot::{NO_SLOT, STATUS_OCCUPIED, pack_meta, version_of};
use crate::txlog::record::{OpKind, VersionTag};
use crate::txn::participant::{PlannedOp, TxParticipant};

/// The value is on disk at exactly the logged target → nothing to do.
#[test]
fn tx_fixed_replay_skips_an_operation_that_already_landed() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let sid = tree.shard_of(&[1u8; 8]).unwrap();

    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &[1u8; 8],
            delete: false,
        }])
        .unwrap();
    locked.apply_put(&[1u8; 8], &[5u8; 16], tags[0]).unwrap();
    drop(
        locked
            .finish(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap(),
    );

    let mut again = tree.lock_shards(&[sid]).unwrap();
    assert!(
        !again.needs_replay(&[1u8; 8], OpKind::Put, tags[0]).unwrap(),
        "the slot already holds the committed state"
    );
    again.abort();
}

/// Nothing was applied: the slot is still free at an older version.
#[test]
fn tx_fixed_replay_replays_an_operation_that_never_landed() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let sid = tree.shard_of(&[2u8; 8]).unwrap();
    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &[2u8; 8],
            delete: false,
        }])
        .unwrap();
    assert!(
        locked
            .needs_replay(&[2u8; 8], OpKind::Put, tags[0])
            .unwrap()
    );
    locked.abort();
}

/// Target version equals the slot's version, but the value is torn: dirty
/// recovery freed the slot while preserving the version, so equality alone
/// would wrongly skip a write that is not there.
#[test]
fn tx_fixed_replay_replays_a_torn_slot_at_the_target_version() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let sid = tree.shard_of(&[3u8; 8]).unwrap();

    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &[3u8; 8],
            delete: false,
        }])
        .unwrap();
    locked.apply_put(&[3u8; 8], &[6u8; 16], tags[0]).unwrap();
    drop(
        locked
            .finish(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap(),
    );

    let VersionTag::Fixed {
        slot_id,
        target_meta,
    } = tags[0]
    else {
        unreachable!("Fixed participant")
    };
    // Recovery's verdict for a CRC-invalid OCCUPIED slot: FREE, same version.
    tree.tx_debug_force_slot_meta(
        sid,
        slot_id,
        pack_meta(crate::fixed::slot::STATUS_FREE, version_of(target_meta)),
    );

    let mut again = tree.lock_shards(&[sid]).unwrap();
    assert!(
        again.needs_replay(&[3u8; 8], OpKind::Put, tags[0]).unwrap(),
        "a torn slot at the target version must be replayed"
    );
    again.abort();
}

/// The slot was deleted and handed to another key, which ran the version on.
#[test]
fn tx_fixed_replay_skips_a_slot_reused_by_another_key() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let sid = tree.shard_of(&[4u8; 8]).unwrap();

    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &[4u8; 8],
            delete: false,
        }])
        .unwrap();
    locked.apply_put(&[4u8; 8], &[1u8; 16], tags[0]).unwrap();
    drop(
        locked
            .finish(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap(),
    );

    // Delete it, then let another key take the slot back.
    tree.delete(&[4u8; 8]).unwrap();
    tree.put(&[5u8; 8], &[2u8; 16]).unwrap();

    let mut again = tree.lock_shards(&[sid]).unwrap();
    assert!(
        !again.needs_replay(&[4u8; 8], OpKind::Put, tags[0]).unwrap(),
        "the slot has moved on; replaying would clobber another key"
    );
    again.abort();
}

/// A delete whose target status is already on the slot.
#[test]
fn tx_fixed_replay_skips_a_delete_that_already_landed() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    tree.put(&[6u8; 8], &[1u8; 16]).unwrap();
    let sid = tree.shard_of(&[6u8; 8]).unwrap();

    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &[6u8; 8],
            delete: true,
        }])
        .unwrap();
    locked.apply_delete(&[6u8; 8], tags[0]).unwrap();
    drop(
        locked
            .finish(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap(),
    );

    let mut again = tree.lock_shards(&[sid]).unwrap();
    assert!(
        !again
            .needs_replay(&[6u8; 8], OpKind::Delete, tags[0])
            .unwrap()
    );
    again.abort();
}

/// A logged no-op carries no slot and is never replayed.
#[test]
fn tx_fixed_replay_skips_the_no_slot_sentinel() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let sid = tree.shard_of(&[7u8; 8]).unwrap();
    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tag = VersionTag::Fixed {
        slot_id: NO_SLOT,
        target_meta: 0,
    };
    assert!(!locked.needs_replay(&[7u8; 8], OpKind::Delete, tag).unwrap());
    locked.abort();
}

/// Beyond a lag of 2^29 the ring cannot order the pair. Guessing could
/// resurrect a deleted row or clobber a live one, so the open fails.
#[test]
fn tx_fixed_replay_fails_closed_on_an_unorderable_version() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    tree.put(&[8u8; 8], &[1u8; 16]).unwrap();
    let sid = tree.shard_of(&[8u8; 8]).unwrap();

    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &[8u8; 8],
            delete: false,
        }])
        .unwrap();
    let VersionTag::Fixed {
        slot_id,
        target_meta,
    } = tags[0]
    else {
        unreachable!("Fixed participant")
    };
    locked.abort();

    // Push the slot's version exactly 2^29 ahead of the reservation target:
    // `is_newer` is strictly below the half-ring, so neither direction orders.
    let ahead = pack_meta(STATUS_OCCUPIED, version_of(target_meta) + (1 << 29));
    tree.tx_debug_force_slot_meta(sid, slot_id, ahead);

    let mut again = tree.lock_shards(&[sid]).unwrap();
    let err = again
        .needs_replay(&[8u8; 8], OpKind::Put, tags[0])
        .unwrap_err();
    assert!(matches!(err, crate::DbError::TxFixedSlot { .. }), "{err}");
    again.abort();
}

/// The reservation's slot was recycled and now holds a different, intact row at
/// exactly the logged target version. Replaying would overwrite a live row that
/// belongs to another key, so the open fails rather than guessing.
#[test]
fn tx_fixed_replay_refuses_a_slot_owned_by_another_key() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let mine = [10u8; 8];
    let theirs = [11u8; 8];
    let sid = tree.shard_of(&mine).unwrap();

    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &mine,
            delete: false,
        }])
        .unwrap();
    let VersionTag::Fixed {
        slot_id,
        target_meta,
    } = tags[0]
    else {
        unreachable!("Fixed participant")
    };
    // Hand the slot back — what a panic between the log point and apply used to
    // do, and what an abort before the log point legitimately does.
    locked.abort();

    // Another key takes the recycled slot and lands on the same version.
    tree.put(&theirs, &[3u8; 16]).unwrap();
    let state = tree.tx_debug_slot_state(sid, slot_id).unwrap();
    assert_eq!(
        state.meta, target_meta,
        "the recycled slot reached our exact target version"
    );
    assert!(state.crc_valid, "the other key's row is intact");

    let mut again = tree.lock_shards(&[sid]).unwrap();
    let err = again.needs_replay(&mine, OpKind::Put, tags[0]).unwrap_err();
    assert!(matches!(err, crate::DbError::TxFixedSlot { .. }), "{err}");
    again.abort();
    assert_eq!(
        tree.get(&theirs),
        Some([3u8; 16]),
        "the other key's row must be untouched"
    );
}

/// `superseded_elsewhere` takes precedence over the unorderable case: when the
/// key demonstrably lives in another slot, the orphaned reservation is skipped
/// rather than failing the open. Characterises the short-circuit added with the
/// index-slot check, which the replay table now documents.
#[test]
fn tx_fixed_replay_skips_an_unorderable_version_when_the_key_moved_on() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let key = [12u8; 8];
    let sid = tree.shard_of(&key).unwrap();

    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &key,
            delete: false,
        }])
        .unwrap();
    let VersionTag::Fixed {
        slot_id: orphaned,
        target_meta,
    } = tags[0]
    else {
        unreachable!("Fixed participant")
    };
    // Durable but unapplied: the reservation stays, nothing is written.
    drop(
        locked
            .finish(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap(),
    );

    // The key's live incarnation lands in a different slot.
    tree.put(&key, &[2u8; 16]).unwrap();
    // And the orphaned slot drifts beyond the orderable half-ring.
    let ahead = pack_meta(STATUS_OCCUPIED, version_of(target_meta) + (1 << 29));
    tree.tx_debug_force_slot_meta(sid, orphaned, ahead);

    let mut again = tree.lock_shards(&[sid]).unwrap();
    assert!(
        !again.needs_replay(&key, OpKind::Put, tags[0]).unwrap(),
        "a superseded reservation is skipped, not failed closed"
    );
    again.abort();
    assert_eq!(tree.get(&key), Some([2u8; 16]));
}

/// A failed apply leaves the reserved slot allocated but the key out of the
/// index. A later commit for the same key then lands in a different slot. Replay
/// of the orphaned reservation must not overwrite that newer incarnation.
#[test]
fn tx_fixed_replay_skips_when_key_lands_in_a_newer_slot() {
    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let key = [9u8; 8];
    let sid = tree.shard_of(&key).unwrap();

    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let tags = locked
        .reserve(&[PlannedOp {
            key: &key,
            delete: false,
        }])
        .unwrap();
    let VersionTag::Fixed {
        slot_id: orphaned, ..
    } = tags[0]
    else {
        unreachable!("Fixed participant")
    };
    // Same as a durable-but-unapplied finish: keep the reservation, write nothing.
    drop(
        locked
            .finish(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap(),
    );

    tree.put(&key, &[2u8; 16]).unwrap();
    let live = {
        let mut locked = tree.lock_shards(&[sid]).unwrap();
        let live_tags = locked
            .reserve(&[PlannedOp {
                key: &key,
                delete: false,
            }])
            .unwrap();
        let VersionTag::Fixed {
            slot_id: live_slot, ..
        } = live_tags[0]
        else {
            unreachable!("Fixed participant")
        };
        locked.abort();
        live_slot
    };
    assert_ne!(
        orphaned, live,
        "the later write must have allocated a different slot"
    );

    let mut again = tree.lock_shards(&[sid]).unwrap();
    assert!(
        !again.needs_replay(&key, OpKind::Put, tags[0]).unwrap(),
        "a newer incarnation of the key supersedes the orphaned reservation"
    );
    again.abort();
    assert_eq!(tree.get(&key), Some([2u8; 16]));
}

/// A reserve that fails part-way must hand back the slots it already took.
///
/// `reserve_in_shard` walks the shard's operations one key at a time, and each
/// absent key takes a slot from the allocator. When a later key fails — a full
/// disk makes `grow()` return `Err` — the tags of the earlier keys never reach
/// the caller, so `TxHeld` has nothing to release on abort and those slots stay
/// marked used until the process restarts and rebuilds the bitmap from disk.
#[test]
fn tx_fixed_reserve_releases_its_slots_when_a_later_key_fails() {
    use crate::fixed::shard::FAIL_RESERVE_IN;
    use std::sync::atomic::Ordering;

    let dir = tempfile::tempdir().unwrap();
    let tree = open_fixed_tree(dir.path(), CountingHook::default());
    let sid = tree.shard_of(&[10u8; 8]).unwrap();
    let before = tree.tx_debug_occupied_slots(sid);

    let keys: [[u8; 8]; 3] = [[10u8; 8], [11u8; 8], [12u8; 8]];
    let planned: Vec<PlannedOp<'_>> = keys
        .iter()
        .map(|k| PlannedOp {
            key: k,
            delete: false,
        })
        .collect();

    // Two keys reserve a slot each, the third fails.
    FAIL_RESERVE_IN.store(2, Ordering::SeqCst);
    let mut locked = tree.lock_shards(&[sid]).unwrap();
    let err = locked.reserve(&planned).unwrap_err();
    FAIL_RESERVE_IN.store(usize::MAX, Ordering::SeqCst);
    assert!(
        format!("{err}").contains("injected reserve failure"),
        "{err}"
    );
    locked.abort();

    assert_eq!(
        tree.tx_debug_occupied_slots(sid),
        before,
        "a reserve that failed part-way must leak no slots"
    );
}