tephra 0.4.0

//! The single-writer lock on a data directory, across processes.
//!
//! The in-process half is unit-tested next to the lock itself; what needs a real process
//! boundary lives here: that a second writer process is refused, that a clean shutdown hands
//! the directory back, and that spawning a subprocess does not leak the lock.

use std::env;
use std::process::Command;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;

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

use tephra::Position;
use tephra::event::{Event, EventType, Tag, Tags};
use tephra::log::set::{LogError, SegmentConfig, SegmentSet};
use tephra::writer::{WriteCoordinator, WriteHandle, WriterConfig};

const SEG_SIZE: usize = 1 << 20;
const CHILD_DIR: &str = "TEPHRA_LOCK_DIR";

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

fn event(ty: &str, tag_strs: &[&str], data: &[u8]) -> Event {
    let tags = Tags::new(
        tag_strs
            .iter()
            .map(|s| Tag::new(*s).unwrap())
            .collect::<SmallVec<[Tag; 4]>>(),
    )
    .unwrap();
    Event::new(&EventType::new(ty).unwrap(), &tags, data).unwrap()
}

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

    let (coordinator, handle) = start(&dir);
    assert_eq!(handle.head(), Position::new(1));
    coordinator.shutdown();
}

#[test]
fn spawning_a_subprocess_does_not_leak_the_write_lock() {
    // The lock must not ride the file descriptor. `fork` duplicates descriptors, so between a
    // `Command::spawn`'s fork and its exec a child holds any descriptor-based lock (flock, or
    // an OFD lock) this process took. If the store is then closed and reopened in that window,
    // the reopen is refused by a lock that belongs to nobody, naming this very process as the
    // holder. POSIX record locks are owned by the process and are not inherited, so this must
    // never happen.
    let dir = TempDir::new().unwrap();
    let done = Arc::new(AtomicBool::new(false));
    let flag = Arc::clone(&done);
    let forking = thread::spawn(move || {
        while !flag.load(Ordering::Relaxed) {
            let _ = Command::new("/bin/true").output();
        }
    });

    for i in 0..400 {
        let set = SegmentSet::open(dir.path(), SegmentConfig::new(SEG_SIZE))
            .unwrap_or_else(|err| panic!("iteration {i}: the lock leaked across a fork: {err}"));
        drop(set);
    }

    done.store(true, Ordering::Relaxed);
    forking.join().unwrap();
}

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

    // A whole separate process must see the lock, not just another handle in this one.
    let output = Command::new(env::current_exe().unwrap())
        .args(["--exact", "second_writer_child", "--ignored", "--nocapture"])
        .env(CHILD_DIR, dir.path())
        .output()
        .expect("spawn the second-writer child");
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(
        stdout.contains("LOCKED"),
        "a second writer process should have been refused, got: {stdout}"
    );

    coordinator.shutdown();
}

/// The child half of `a_second_writer_process_is_refused`. Does nothing unless the parent set
/// its environment, so a plain test run never executes it.
#[test]
#[ignore]
fn second_writer_child() {
    let Ok(dir) = env::var(CHILD_DIR) else {
        return;
    };
    match SegmentSet::open(&dir, SegmentConfig::new(SEG_SIZE)) {
        Err(LogError::Locked { .. }) => println!("LOCKED"),
        Err(err) => println!("OTHER-ERROR: {err}"),
        Ok(_) => println!("OPENED"),
    }
}