tephra 0.4.0

//! End-to-end tests of the read-only follower against a live writer: the committed prefix it
//! exposes, rollovers, subscriptions, and the single-writer lock.

use std::collections::HashSet;
use std::env;
use std::fs::OpenOptions;
use std::os::unix::fs::FileExt;
use std::process::Command;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::{Duration, Instant};

use smallvec::SmallVec;
use tempfile::TempDir;

use tephra::event::{Event, EventType, Tag, Tags};
use tephra::log::set::{LogError, SegmentConfig, SegmentSet};
use tephra::query::{Query, QueryItem};
use tephra::writer::{WriteCoordinator, WriteHandle, WriterConfig};
use tephra::{Follower, FollowerConfig, Position, WaitOutcome};

const SEG_SIZE: usize = 1 << 20;
/// Small enough that a handful of events force rollovers.
const TINY_SEG: usize = 4096;

fn writer_config() -> WriterConfig {
    WriterConfig {
        queue_capacity: 256,
        max_batch_records: 256,
        max_batch_bytes: 2048,
        ..WriterConfig::default()
    }
}

fn start(dir: &TempDir, segment_size: usize) -> (WriteCoordinator, WriteHandle) {
    let set = SegmentSet::open(dir.path(), SegmentConfig::new(segment_size)).unwrap();
    let mut cfg = writer_config();
    cfg.max_batch_bytes = cfg.max_batch_bytes.min(set.segment_capacity());
    WriteCoordinator::start(set, cfg).unwrap()
}

fn follower(dir: &TempDir, segment_size: usize) -> Follower {
    Follower::open(
        dir.path(),
        FollowerConfig::new(SegmentConfig::new(segment_size)),
    )
    .unwrap()
}

fn tags(items: &[&str]) -> Tags {
    Tags::new(
        items
            .iter()
            .map(|s| Tag::new(*s).unwrap())
            .collect::<SmallVec<[Tag; 4]>>(),
    )
    .unwrap()
}

fn event(ty: &str, tag_strs: &[&str], data: &[u8]) -> Event {
    Event::new(&EventType::new(ty).unwrap(), &tags(tag_strs), data).unwrap()
}

fn all() -> Query {
    Query::item(QueryItem::of_types(vec![EventType::new("Ev").unwrap()]))
}

/// Every position the follower can currently see, in order.
fn positions(follower: &Follower, query: &Query) -> Vec<Position> {
    let reader = follower.reader();
    let mut reads = reader.read(query, Position::ZERO, None);
    let mut out = Vec::new();
    while let Some(item) = reads.next() {
        out.push(item.unwrap().position);
    }
    out
}

/// Asserts a sequence of positions is exactly `1..=n`: ascending, no gap, no duplicate.
fn assert_dense_prefix(seen: &[Position]) {
    for (i, position) in seen.iter().enumerate() {
        assert_eq!(
            *position,
            Position::new(i as u64 + 1),
            "position {i} of the prefix is out of order or has a gap: {seen:?}"
        );
    }
}

#[test]
fn follower_sees_a_gap_free_committed_prefix_while_the_writer_appends() {
    const EVENTS: u64 = 2_000;
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, SEG_SIZE);
    handle
        .append(vec![event("Ev", &["k:1"], b"first")], None)
        .unwrap();

    let follower = follower(&dir, SEG_SIZE);
    let done = Arc::new(AtomicBool::new(false));

    let writing = {
        let done = Arc::clone(&done);
        thread::spawn(move || {
            for i in 2..=EVENTS {
                handle
                    .append(
                        vec![event("Ev", &["k:1"], format!("e{i}").as_bytes())],
                        None,
                    )
                    .unwrap();
            }
            done.store(true, Ordering::Release);
            handle
        })
    };

    // Observe repeatedly while the writer runs. Every observation must be a dense prefix,
    // and the tip must never go backwards.
    let query = all();
    let mut highest = Position::ZERO;
    let mut observations = 0;
    while !done.load(Ordering::Acquire) {
        let tip = follower.refresh().unwrap();
        assert!(
            tip >= highest,
            "the tip went backwards: {tip} after {highest}"
        );
        highest = tip;
        let seen = positions(&follower, &query);
        assert_dense_prefix(&seen);
        assert_eq!(
            seen.len() as u64,
            tip.get(),
            "the prefix must reach the tip"
        );
        observations += 1;
    }
    let handle = writing.join().unwrap();

    let tip = follower.refresh().unwrap();
    assert_eq!(tip, Position::new(EVENTS));
    assert_eq!(
        tip,
        handle.head(),
        "the follower must catch up to the writer"
    );
    let seen = positions(&follower, &query);
    assert_dense_prefix(&seen);
    assert_eq!(seen.len() as u64, EVENTS);
    assert!(observations > 0, "the test observed nothing mid-flight");

    coordinator.shutdown();
}

#[test]
fn follower_follows_across_many_rollovers() {
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, TINY_SEG);
    handle
        .append(vec![event("Ev", &["k:1"], b"first")], None)
        .unwrap();

    let follower = follower(&dir, TINY_SEG);
    for i in 2..=400u64 {
        handle
            .append(
                vec![event("Ev", &["k:1"], format!("e{i:04}").as_bytes())],
                None,
            )
            .unwrap();
    }
    let set = coordinator.shutdown();
    assert!(
        set.sealed_len() >= 3,
        "expected several rollovers, got {}",
        set.sealed_len()
    );

    let tip = follower.refresh().unwrap();
    assert_eq!(tip, Position::new(400));
    let seen = positions(&follower, &all());
    assert_dense_prefix(&seen);
    assert_eq!(
        seen.len(),
        400,
        "positions stay contiguous across the seams"
    );
}

#[test]
fn follower_query_answers_match_the_writer() {
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, TINY_SEG);
    for i in 1..=120u64 {
        let tag = format!("k:{}", i % 7);
        let ty = if i % 3 == 0 { "Odd" } else { "Ev" };
        handle
            .append(vec![event(ty, &[&tag], format!("e{i}").as_bytes())], None)
            .unwrap();
    }

    let follower = follower(&dir, TINY_SEG);
    follower.refresh().unwrap();
    let reader = follower.reader();

    // Differential: the same queries through the writer's handle and the follower's.
    for tag_id in 0..7u64 {
        for ty in ["Ev", "Odd"] {
            let query = Query::item(QueryItem::new(
                vec![EventType::new(ty).unwrap()],
                tags(&[&format!("k:{tag_id}")]),
            ));
            for after in [Position::ZERO, Position::new(40), Position::new(119)] {
                let collect = |reads: &mut tephra::read::Reads| {
                    let mut out = Vec::new();
                    while let Some(item) = reads.next() {
                        out.push(item.unwrap().position);
                    }
                    out
                };
                let mut writer_reads = handle.read(&query, after, None);
                let expected = collect(&mut writer_reads);
                let mut follower_reads = reader.read(&query, after, None);
                let actual = collect(&mut follower_reads);
                assert_eq!(
                    expected, actual,
                    "follower disagrees for {ty} k:{tag_id} after {after}"
                );
            }
        }
    }

    coordinator.shutdown();
}

#[test]
fn follower_never_observes_an_uncommitted_batch() {
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, SEG_SIZE);
    handle
        .append(vec![event("Ev", &["k:1"], b"committed")], None)
        .unwrap();

    let follower = follower(&dir, SEG_SIZE);
    assert_eq!(follower.refresh().unwrap(), Position::new(1));

    // Records on disk with no commit marker behind them, written straight through seglog.
    let path = dir.path().join(format!("{:020}.log", 1));
    let mut raw = seglog::write::Writer::<0>::open(&path, SEG_SIZE, 64).unwrap();
    let rewind = raw.write_offset();
    raw.append_data(b"not-a-real-event").unwrap();
    raw.flush_writer().unwrap();

    assert_eq!(
        follower.refresh().unwrap(),
        Position::new(1),
        "an uncommitted run must stay invisible"
    );
    assert_dense_prefix(&positions(&follower, &all()));

    raw.rewind_to(rewind).unwrap();
    drop(raw);
    coordinator.shutdown();
}

#[test]
fn follower_subscription_tails_the_writer() {
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, SEG_SIZE);
    handle
        .append(vec![event("Ev", &["k:1"], b"first")], None)
        .unwrap();

    let follower = Arc::new(follower(&dir, SEG_SIZE));
    let poller = follower.poll_every(Duration::from_millis(2));

    let reader = follower.reader();
    let subscriber = thread::spawn(move || {
        let mut subscription = reader.subscribe(all(), Position::ZERO);
        let mut seen = Vec::new();
        let deadline = Instant::now() + Duration::from_secs(20);
        while seen.len() < 50 && Instant::now() < deadline {
            match subscription.next_batch() {
                Some(batch) => seen.extend(batch.unwrap().into_iter().map(|(p, _)| p)),
                None => break,
            }
        }
        seen
    });

    for i in 2..=50u64 {
        handle
            .append(
                vec![event("Ev", &["k:1"], format!("e{i}").as_bytes())],
                None,
            )
            .unwrap();
        thread::sleep(Duration::from_millis(1));
    }

    let seen = subscriber.join().unwrap();
    assert_eq!(
        seen.len(),
        50,
        "the subscription should have tailed every event"
    );
    assert_dense_prefix(&seen);
    assert_eq!(
        seen.iter().collect::<HashSet<_>>().len(),
        50,
        "no duplicates across the catch-up/live seam"
    );

    // Dropping the poller must close the follower, so a parked subscriber ends.
    let reader = follower.reader();
    let parked = reader.subscribe(all(), Position::new(50));
    drop(poller);
    assert_eq!(
        parked.wait_timeout(Duration::from_secs(5)),
        WaitOutcome::Closed
    );

    coordinator.shutdown();
}

#[test]
fn follower_open_before_any_writer_errors() {
    let dir = TempDir::new().unwrap();
    let message = match Follower::open(
        dir.path(),
        FollowerConfig::new(SegmentConfig::new(SEG_SIZE)),
    ) {
        Ok(_) => panic!("an uninitialized store must not open"),
        Err(err) => err.to_string(),
    };
    assert!(
        message.contains("read-only open"),
        "expected an uninitialized-store error, got {message}"
    );
    assert!(
        !dir.path().join("index").exists(),
        "a failed follower open must not create anything"
    );
}

#[test]
fn a_follower_opens_while_a_writer_holds_the_lock() {
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, SEG_SIZE);
    handle
        .append(vec![event("Ev", &["k:1"], b"first")], None)
        .unwrap();

    let follower = follower(&dir, SEG_SIZE);
    assert_eq!(follower.refresh().unwrap(), Position::new(1));

    // A second writer, however, is refused.
    let err = SegmentSet::open(dir.path(), SegmentConfig::new(SEG_SIZE)).unwrap_err();
    assert!(matches!(err, LogError::Locked { .. }), "got {err:?}");

    coordinator.shutdown();
}

// --- Cross-process ---
//
// The only test that proves the page-cache visibility a follower actually depends on, that the
// writer's exclusive lock does not block a follower, and that `O_RDONLY` descriptors suffice.
// The child is this same test binary, re-executed into the ignored test below.

const CHILD_DIR: &str = "TEPHRA_FOLLOW_DIR";
const CHILD_EVENTS: &str = "TEPHRA_FOLLOW_EVENTS";

#[test]
fn follower_and_writer_in_separate_processes() {
    const EVENTS: u64 = 500;
    let dir = TempDir::new().unwrap();

    let mut child = Command::new(env::current_exe().unwrap())
        .args([
            "--exact",
            "writer_child_process",
            "--ignored",
            "--nocapture",
        ])
        .env(CHILD_DIR, dir.path())
        .env(CHILD_EVENTS, EVENTS.to_string())
        .spawn()
        .expect("spawn the writer child");

    // The child creates the store, so wait for it to exist before attaching.
    let deadline = Instant::now() + Duration::from_secs(30);
    let follower = loop {
        match Follower::open(
            dir.path(),
            FollowerConfig::new(SegmentConfig::new(TINY_SEG)),
        ) {
            Ok(follower) => break follower,
            Err(err) => {
                assert!(
                    Instant::now() < deadline,
                    "the child never initialized the store: {err}"
                );
                thread::sleep(Duration::from_millis(5));
            }
        }
    };

    let query = all();
    let mut highest = Position::ZERO;
    while highest < Position::new(EVENTS) {
        assert!(
            Instant::now() < deadline,
            "the follower stalled at {highest}"
        );
        let tip = follower.refresh().unwrap();
        assert!(
            tip >= highest,
            "the tip went backwards: {tip} after {highest}"
        );
        highest = tip;
        // Every intermediate observation is a dense prefix of the writer's log.
        assert_dense_prefix(&positions(&follower, &query));
    }

    let status = child.wait().expect("await the writer child");
    assert!(status.success(), "the writer child failed: {status}");

    let tip = follower.refresh().unwrap();
    assert_eq!(tip, Position::new(EVENTS));
    let seen = positions(&follower, &query);
    assert_dense_prefix(&seen);
    assert_eq!(seen.len() as u64, EVENTS);
}

/// The child half of `follower_and_writer_in_separate_processes`. Does nothing unless the
/// parent set its environment, so a plain `cargo test` run never executes it.
#[test]
#[ignore]
fn writer_child_process() {
    let Ok(dir) = env::var(CHILD_DIR) else {
        return;
    };
    let events: u64 = env::var(CHILD_EVENTS).unwrap().parse().unwrap();

    let set = SegmentSet::open(&dir, SegmentConfig::new(TINY_SEG)).unwrap();
    let mut cfg = writer_config();
    cfg.max_batch_bytes = cfg.max_batch_bytes.min(set.segment_capacity());
    let (coordinator, handle) = WriteCoordinator::start(set, cfg).unwrap();

    for i in 1..=events {
        handle
            .append(
                vec![event("Ev", &["k:1"], format!("e{i:04}").as_bytes())],
                None,
            )
            .unwrap();
        // Give the follower a chance to observe intermediate states.
        if i % 50 == 0 {
            thread::sleep(Duration::from_millis(2));
        }
    }
    coordinator.shutdown();
}

#[test]
fn dropping_a_poller_does_not_wait_out_the_interval() {
    // The loop must wait on a cancellable primitive, not `thread::sleep`. With a sleep, drop
    // blocks for up to a full interval, which also stalls unwinding and process shutdown.
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, SEG_SIZE);
    handle
        .append(vec![event("Ev", &["k:1"], b"first")], None)
        .unwrap();

    let follower = Arc::new(follower(&dir, SEG_SIZE));
    let poller = follower.poll_every(Duration::from_secs(30));
    thread::sleep(Duration::from_millis(20));

    let start = Instant::now();
    drop(poller);
    let elapsed = start.elapsed();
    assert!(
        elapsed < Duration::from_secs(2),
        "dropping the poller waited {elapsed:?}, so it slept through the interval"
    );

    coordinator.shutdown();
}

#[test]
fn poller_health_surfaces_a_persistent_failure() {
    // A follower that cannot advance must be distinguishable from an idle store. Corrupting a
    // header in the middle of the chain makes every refresh fail permanently.
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, TINY_SEG);
    for i in 1..=400u64 {
        handle
            .append(
                vec![event("Ev", &["k:1"], format!("e{i:04}").as_bytes())],
                None,
            )
            .unwrap();
    }
    let set = coordinator.shutdown();
    assert!(
        set.sealed_len() >= 2,
        "need a non-trailing segment to corrupt"
    );
    let second_base = set
        .sealed_segments()
        .nth(1)
        .expect("a second sealed segment")
        .0;
    let victim = dir.path().join(format!("{:020}.log", second_base.get()));
    drop(set);

    let follower = Arc::new(follower(&dir, TINY_SEG));
    let poller = follower.poll_every(Duration::from_millis(5));
    assert_eq!(poller.health().consecutive_failures, 0);

    // Corrupt a non-trailing header's CRC: not a mid-rollover race, so it is a hard error.
    let file = OpenOptions::new().write(true).open(&victim).unwrap();
    file.write_all_at(&[0xAB; 4], 60).unwrap();
    file.sync_all().unwrap();

    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        let health = poller.health();
        if health.consecutive_failures > 0 {
            assert!(
                health.last_error.is_some(),
                "a failing poller must report why"
            );
            break;
        }
        assert!(
            Instant::now() < deadline,
            "the poller never reported the corruption it cannot get past"
        );
        thread::sleep(Duration::from_millis(10));
    }
}

#[test]
fn refresh_leaves_the_published_head_at_the_tip() {
    // The watermark must never be left behind the log set's tip: a refresh that advanced the
    // set but did not publish would freeze the follower with no error, because every later
    // refresh would see nothing new to do.
    let dir = TempDir::new().unwrap();
    let (coordinator, handle) = start(&dir, TINY_SEG);
    handle
        .append(vec![event("Ev", &["k:1"], b"first")], None)
        .unwrap();

    let follower = follower(&dir, TINY_SEG);
    for i in 2..=400u64 {
        handle
            .append(
                vec![event("Ev", &["k:1"], format!("e{i:04}").as_bytes())],
                None,
            )
            .unwrap();
        let tip = follower.refresh().unwrap();
        assert_eq!(
            follower.head(),
            tip,
            "the published head must equal the tip refresh reported"
        );
        // And an immediate second refresh must agree, not creep forward.
        assert_eq!(follower.refresh().unwrap(), tip);
        assert_eq!(follower.head(), tip);
    }

    let set = coordinator.shutdown();
    assert!(set.sealed_len() >= 2, "expected rollovers along the way");
}