writeahead 0.4.1

//! Crash-recovery kill-point matrix: each test crafts the on-disk state a
//! crash would leave behind, then verifies `start()` restores a consistent
//! WAL where every acknowledged record is readable and new writes work.

use futures::StreamExt;
use writeahead::logfile::{
    recover_and_seal, FILE_HEADER_SIZE, FOOTER_SIZE, FORMAT_VERSION, MAGIC_NUMBER,
    RECORD_HEADER_SIZE,
};
use writeahead::write_ahead::WriteAheadError;
use writeahead::{SimpleFile, WriteAhead, WriteAheadOptions};

fn opts(dir: &std::path::Path) -> WriteAheadOptions {
    WriteAheadOptions {
        log_dir: dir.to_path_buf(),
        ..Default::default()
    }
}

fn create_max_id_active(dir: &std::path::Path) -> std::path::PathBuf {
    let mut wal = WriteAhead::<SimpleFile>::with_options(WriteAheadOptions {
        log_dir: dir.to_path_buf(),
        preallocation_chunk_size: None,
        ..Default::default()
    });
    wal.start().unwrap();
    drop(wal);
    let source = dir.join("0000000000.log");
    let target = dir.join(format!("{}.log", u64::MAX));
    std::fs::rename(source, &target).unwrap();
    target
}

async fn write_str(wal: &WriteAhead<SimpleFile>, s: &str) -> writeahead::RecordID {
    wal.write_batch(vec![s.as_bytes().to_vec()])
        .await
        .unwrap()
        .remove(0)
}

async fn collect_stream(wal: &WriteAhead<SimpleFile>) -> Vec<Vec<u8>> {
    let mut out = Vec::new();
    let mut stream = wal.create_stream().unwrap();
    while let Some(r) = stream.next().await {
        out.push(r.unwrap().1);
    }
    out
}

#[tokio::test]
async fn restart_resumes_appending() {
    let dir = tempfile::tempdir().unwrap();
    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();
    let id1 = write_str(&wal, "before restart").await;
    drop(wal);

    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();
    let id2 = write_str(&wal, "after restart").await;

    assert_eq!(
        wal.read(id1.file_id, id1.file_offset).unwrap(),
        b"before restart"
    );
    assert_eq!(
        wal.read(id2.file_id, id2.file_offset).unwrap(),
        b"after restart"
    );
    assert_eq!(collect_stream(&wal).await.len(), 2);
}

#[tokio::test]
async fn crash_after_seal_before_next_file_rotates() {
    // Simulate: last file is sealed but the next file was never created.
    let dir = tempfile::tempdir().unwrap();
    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();
    let id1 = write_str(&wal, "sealed away").await;
    drop(wal);

    // Seal file 0 out-of-band (as rotation would, crashing right after)
    let path = dir.path().join("0000000000.log");
    recover_and_seal::<SimpleFile>(&path, 1).unwrap();

    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();

    // Must NOT have appended to the sealed file: new writes go to file 1
    let id2 = write_str(&wal, "fresh file").await;
    assert_eq!(id2.file_id, 1, "writer must rotate off a sealed file");
    assert_eq!(
        wal.read(id1.file_id, id1.file_offset).unwrap(),
        b"sealed away"
    );
    let all = collect_stream(&wal).await;
    assert_eq!(all, vec![b"sealed away".to_vec(), b"fresh file".to_vec()]);
}

#[tokio::test]
async fn crash_mid_record_truncates_torn_tail() {
    let dir = tempfile::tempdir().unwrap();
    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();
    let id1 = write_str(&wal, "acknowledged").await;
    let id2 = write_str(&wal, "will be torn").await;
    drop(wal);

    // Tear the last record mid-write. The physical EOF is an allocation
    // window, so truncate relative to the record's logical end.
    let path = dir.path().join("0000000000.log");
    let logical_end = id2.file_offset + RECORD_HEADER_SIZE + b"will be torn".len() as u64;
    let f = std::fs::OpenOptions::new().write(true).open(&path).unwrap();
    f.set_len(logical_end - 5).unwrap();

    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();

    // The torn record is gone, the acknowledged prefix survives, and the
    // file accepts new appends.
    assert_eq!(
        wal.read(id1.file_id, id1.file_offset).unwrap(),
        b"acknowledged"
    );
    let id3 = write_str(&wal, "appended after recovery").await;
    assert_eq!(id3.file_id, 0);
    let all = collect_stream(&wal).await;
    assert_eq!(
        all,
        vec![
            b"acknowledged".to_vec(),
            b"appended after recovery".to_vec()
        ]
    );
}

#[tokio::test]
async fn crash_before_header_heals_empty_active_file() {
    // File 0 is sealed and complete; file 1 was created but never got its
    // header (0 bytes). Recovery should adopt file 1 as the active file.
    let dir = tempfile::tempdir().unwrap();
    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();
    let id1 = write_str(&wal, "in file zero").await;
    drop(wal);

    recover_and_seal::<SimpleFile>(&dir.path().join("0000000000.log"), 1).unwrap();
    std::fs::write(dir.path().join("0000000001.log"), b"").unwrap();

    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();

    let id2 = write_str(&wal, "in file one").await;
    assert_eq!(id2.file_id, 1);
    assert_eq!(id2.file_offset, FILE_HEADER_SIZE);
    assert_eq!(
        wal.read(id1.file_id, id1.file_offset).unwrap(),
        b"in file zero"
    );
    assert_eq!(collect_stream(&wal).await.len(), 2);
}

#[tokio::test]
async fn full_reserved_header_without_valid_slot_is_reinitialized() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("0000000000.log");
    let mut torn = vec![0u8; FILE_HEADER_SIZE as usize];
    torn[0..8].copy_from_slice(&MAGIC_NUMBER);
    torn[8] = FORMAT_VERSION;
    std::fs::write(&path, torn).unwrap();

    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();
    let id = write_str(&wal, "after torn header").await;
    assert_eq!(id.file_id, 0);
    assert_eq!(id.file_offset, FILE_HEADER_SIZE);
    assert_eq!(
        wal.read(id.file_id, id.file_offset).unwrap(),
        b"after torn header"
    );
}

#[tokio::test]
async fn max_file_id_stream_exhaustion_and_writer_rotation_are_controlled() {
    let dir = tempfile::tempdir().unwrap();
    create_max_id_active(dir.path());
    let mut wal = WriteAhead::<SimpleFile>::with_options(WriteAheadOptions {
        log_dir: dir.path().to_path_buf(),
        max_file_size: FILE_HEADER_SIZE - 1,
        preallocation_chunk_size: None,
        ..Default::default()
    });
    wal.start().unwrap();

    // Exhausting a stream at u64::MAX must not overflow while looking for a
    // successor file.
    let mut stream = wal.create_stream().unwrap();
    assert!(stream.next().await.is_none());

    // An oversized first group is acknowledged in the empty file. Subsequent
    // rotation cannot allocate max+1 and returns a controlled error.
    let first = write_str(&wal, "fills tiny max").await;
    assert_eq!(first.file_id, u64::MAX);
    let err = wal
        .write_batch(vec![b"needs rotation".to_vec()])
        .await
        .unwrap_err();
    assert!(format!("{err:#}").contains("Log file id space exhausted"));
}

#[test]
fn sealed_max_file_id_fails_start_without_overflow() {
    let dir = tempfile::tempdir().unwrap();
    let path = create_max_id_active(dir.path());
    recover_and_seal::<SimpleFile>(&path, 1).unwrap();

    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    let err = wal.start().unwrap_err();
    assert!(matches!(
        err.downcast_ref::<WriteAheadError>(),
        Some(WriteAheadError::FileIdExhausted)
    ));
}

#[tokio::test]
async fn torn_footer_on_non_active_file_is_healed() {
    // File 0's seal footer was torn mid-write during rotation, but file 1
    // already exists. Recovery should re-seal file 0.
    let dir = tempfile::tempdir().unwrap();
    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();
    let id1 = write_str(&wal, "old file record").await;
    drop(wal);

    let path0 = dir.path().join("0000000000.log");
    recover_and_seal::<SimpleFile>(&path0, 1).unwrap();
    // Tear half the footer off
    let len = std::fs::metadata(&path0).unwrap().len();
    let f = std::fs::OpenOptions::new()
        .write(true)
        .open(&path0)
        .unwrap();
    f.set_len(len - FOOTER_SIZE / 2).unwrap();
    // File 1 exists with records
    std::fs::write(dir.path().join("0000000001.log"), b"").unwrap();

    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();

    let id2 = write_str(&wal, "active file record").await;
    assert_eq!(id2.file_id, 1);
    assert_eq!(
        wal.read(id1.file_id, id1.file_offset).unwrap(),
        b"old file record"
    );
    let all = collect_stream(&wal).await;
    assert_eq!(
        all,
        vec![b"old file record".to_vec(), b"active file record".to_vec()]
    );
}

#[tokio::test]
async fn stray_files_are_skipped() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(dir.path().join("notes.txt"), b"hi").unwrap();
    std::fs::write(dir.path().join(".hidden"), b"hi").unwrap();
    std::fs::write(dir.path().join("abc.log"), b"not a wal file").unwrap();

    let mut wal = WriteAhead::<SimpleFile>::with_options(opts(dir.path()));
    wal.start().unwrap();
    let id = write_str(&wal, "works").await;
    assert_eq!(wal.read(id.file_id, id.file_offset).unwrap(), b"works");
}