pagedb 0.1.0-beta.6

//! Integration tests for incremental compaction: watermark persistence,
//! progress reporting, batched compaction, crash-resume semantics, and
//! interleaving of writes between compaction steps.

use pagedb::vfs::memory::MemVfs;
use pagedb::{CompactBudget, Db, OpenOptions, RealmId};

const PAGE: usize = 4096;
const REALM: RealmId = RealmId::new([0x55; 16]);
const KEK: [u8; 32] = [0x22; 32];

async fn fresh_db() -> Db<MemVfs> {
    Db::open(MemVfs::new(), KEK, PAGE, REALM, OpenOptions::default())
        .await
        .unwrap()
}

// ─── Test 1: compact_step returns more_work=false on empty db ────────────────

#[tokio::test(flavor = "current_thread")]
async fn compact_step_empty_db() {
    let db = fresh_db().await;
    let prog = db.compact_step(CompactBudget::default()).await.unwrap();
    assert!(!prog.more_work, "empty db should have no more work");
    assert_eq!(prog.pages_relocated, 0);
    assert!(prog.watermark.is_none());
}

// ─── Test 2: compact_now finishes a db with no deleted keys ──────────────────

#[tokio::test(flavor = "current_thread")]
async fn compact_now_no_free_pages() {
    let db = fresh_db().await;

    // Write 20 keys.
    {
        let mut txn = db.begin_write().await.unwrap();
        for i in 0u32..20 {
            let key = format!("key-{i:04}");
            txn.put(key.as_bytes(), b"value").await.unwrap();
        }
        txn.commit().await.unwrap();
    }

    let stats = db.compact_now().await.unwrap();
    // No pages should be reclaimed (no deletions).
    let _ = stats; // stats are informational; just verify it completes
}

// ─── Test 3: compact_step reclaims freed space and reports completion ─────────

#[tokio::test(flavor = "current_thread")]
async fn compact_step_reclaims_after_deletes() {
    let db = fresh_db().await;

    // Write 100 keys and delete half to free pages.
    {
        let mut txn = db.begin_write().await.unwrap();
        for i in 0u32..100 {
            let key = format!("wm-{i:04}");
            txn.put(key.as_bytes(), b"reclaim-test-value")
                .await
                .unwrap();
        }
        txn.commit().await.unwrap();
    }
    {
        let mut txn = db.begin_write().await.unwrap();
        for i in (0u32..100).step_by(2) {
            let key = format!("wm-{i:04}");
            txn.delete(key.as_bytes()).await.unwrap();
        }
        txn.commit().await.unwrap();
    }

    // Compaction runs to completion in one call (a full rewrite cannot be safely
    // chunked across lock releases) and reclaims the freed pages.
    let prog = db.compact_step(CompactBudget::default()).await.unwrap();
    assert!(!prog.more_work, "compaction completes in a single call");
    assert!(
        prog.pages_relocated > 0,
        "expected reclaimed pages after deleting half the keys"
    );

    // A second call has nothing to reclaim.
    let prog2 = db.compact_step(CompactBudget::default()).await.unwrap();
    assert!(!prog2.more_work);
    assert_eq!(prog2.pages_relocated, 0, "already compact: nothing to do");

    // Surviving keys remain readable.
    let r = db.begin_read().await.unwrap();
    for i in (1u32..100).step_by(2) {
        let key = format!("wm-{i:04}");
        assert_eq!(
            r.get(key.as_bytes()).await.unwrap().as_deref(),
            Some(b"reclaim-test-value".as_slice()),
            "{key} must survive compaction"
        );
    }
}

// ─── Test 4: Writes between steps do not corrupt data ────────────────────────

#[tokio::test(flavor = "current_thread")]
async fn writes_between_steps_do_not_corrupt() {
    let db = fresh_db().await;

    // Write 60 keys.
    {
        let mut txn = db.begin_write().await.unwrap();
        for i in 0u32..60 {
            let key = format!("interleave-{i:04}");
            txn.put(key.as_bytes(), b"initial-value").await.unwrap();
        }
        txn.commit().await.unwrap();
    }
    // Delete half.
    {
        let mut txn = db.begin_write().await.unwrap();
        for i in (0u32..60).step_by(2) {
            let key = format!("interleave-{i:04}");
            txn.delete(key.as_bytes()).await.unwrap();
        }
        txn.commit().await.unwrap();
    }

    let budget = CompactBudget::new(3, 10_000);

    // Alternate: one compaction step, then a normal write.
    let mut extra_writes = 0u32;
    let mut iters = 0usize;
    loop {
        let prog = db.compact_step(budget).await.unwrap();
        iters += 1;

        // Write a new key between compaction steps.
        {
            let mut txn = db.begin_write().await.unwrap();
            let key = format!("new-{extra_writes:04}");
            txn.put(key.as_bytes(), b"new-value").await.unwrap();
            txn.commit().await.unwrap();
        }
        extra_writes += 1;

        if !prog.more_work {
            break;
        }
        assert!(iters < 10_000, "compaction did not converge");
    }

    // Verify all surviving original keys are intact.
    let txn = db.begin_read().await.unwrap();
    for i in 0u32..60 {
        let key = format!("interleave-{i:04}");
        let got = txn.get(key.as_bytes()).await.unwrap();
        if i % 2 == 0 {
            assert!(got.is_none(), "{key} should be deleted");
        } else {
            assert_eq!(
                got.unwrap().as_ref(),
                b"initial-value",
                "{key} value mismatch"
            );
        }
    }
    // Verify newly written keys are readable.
    for j in 0..extra_writes {
        let key = format!("new-{j:04}");
        let got = txn.get(key.as_bytes()).await.unwrap();
        assert_eq!(got.unwrap().as_ref(), b"new-value", "{key} missing");
    }
}

// ─── Test 5: compact_now loops until more_work = false ───────────────────────

#[tokio::test(flavor = "current_thread")]
async fn compact_now_completes_fully() {
    let db = fresh_db().await;

    {
        let mut txn = db.begin_write().await.unwrap();
        for i in 0u32..80 {
            let key = format!("full-{i:04}");
            txn.put(key.as_bytes(), b"full-value").await.unwrap();
        }
        txn.commit().await.unwrap();
    }
    {
        let mut txn = db.begin_write().await.unwrap();
        for i in 0u32..40 {
            let key = format!("full-{i:04}");
            txn.delete(key.as_bytes()).await.unwrap();
        }
        txn.commit().await.unwrap();
    }

    db.compact_now().await.unwrap();

    // After full compaction, a single compact_step should report more_work=false.
    let prog = db.compact_step(CompactBudget::default()).await.unwrap();
    assert!(
        !prog.more_work,
        "after compact_now, compact_step should have no more work"
    );
    assert!(prog.watermark.is_none());

    // All surviving keys must still be readable.
    let txn = db.begin_read().await.unwrap();
    for i in 40u32..80 {
        let key = format!("full-{i:04}");
        let got = txn.get(key.as_bytes()).await.unwrap();
        assert_eq!(
            got.unwrap().as_ref(),
            b"full-value",
            "{key} missing after compact_now"
        );
    }
}

// ─── Test: history is consistently discarded across compact_step (no dangling
//     commit-history root in the durable header) ────────────────────────────
#[tokio::test(flavor = "current_thread")]
async fn compact_step_reopen_history_consistent() {
    let vfs = MemVfs::new();
    {
        // Default options retain commit history (Count(1024)). Build several
        // commits so a history tree exists, then incrementally compact through
        // intermediate steps and a final dense repack.
        let db = Db::open(vfs.clone(), KEK, PAGE, REALM, OpenOptions::default())
            .await
            .unwrap();
        for round in 0u32..15 {
            let mut w = db.begin_write().await.unwrap();
            for i in 0u32..30 {
                w.put(format!("k{round:02}_{i:04}").as_bytes(), &[round as u8; 64])
                    .await
                    .unwrap();
            }
            w.commit().await.unwrap();
        }
        let budget = CompactBudget::new(5, 10_000); // small → intermediate steps + final
        let mut iters = 0;
        loop {
            let p = db.compact_step(budget).await.unwrap();
            iters += 1;
            assert!(iters < 10_000, "compaction did not converge");
            if !p.more_work {
                break;
            }
        }
        // Db drops here.
    }

    // Reopen. Before the fix, the header still pointed at the pre-repack
    // commit-history root (overwritten/truncated by the dense repack), so the
    // next write — which opens the history tree at that root — corrupted or
    // errored. It must now be a clean reset (root = 0).
    let db = Db::open(vfs.clone(), KEK, PAGE, REALM, OpenOptions::default())
        .await
        .unwrap();
    {
        let mut w = db.begin_write().await.unwrap();
        w.put(b"after-reopen", b"v").await.unwrap();
        w.commit().await.unwrap();
    }
    let r = db.begin_read().await.unwrap();
    assert_eq!(
        r.get(b"after-reopen").await.unwrap().as_deref(),
        Some(&b"v"[..])
    );
    // Pre-compaction data survived the repack.
    assert!(
        r.get(b"k00_0000").await.unwrap().is_some(),
        "data written before compaction must survive"
    );
}

/// `compact_step` runs the same atomic repack as `compact_now`, so it must
/// preserve overflow (large) values rather than failing or corrupting the store.
#[tokio::test(flavor = "current_thread")]
async fn compact_step_preserves_large_overflow_values() {
    let vfs = MemVfs::new();
    let db = Db::open(vfs.clone(), KEK, PAGE, REALM, OpenOptions::default())
        .await
        .unwrap();

    let big = vec![0x3Cu8; 4096]; // > PAGE/4 → overflow chain
    let n = 60u32;
    {
        let mut w = db.begin_write().await.unwrap();
        for i in 0..n {
            w.put(format!("k-{i:05}").as_bytes(), &big).await.unwrap();
        }
        w.commit().await.unwrap();
    }

    let prog = db.compact_step(CompactBudget::default()).await.unwrap();
    assert!(!prog.more_work, "compaction completes in a single call");

    let r = db.begin_read().await.unwrap();
    for i in 0..n {
        let key = format!("k-{i:05}");
        assert_eq!(
            r.get(key.as_bytes()).await.unwrap().as_deref(),
            Some(big.as_slice()),
            "large value {key} lost after compact_step repack"
        );
    }
}