holys3 0.5.1

Indexed regex search for local files and private S3 buckets
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

#![cfg(unix)]

use anyhow::{Context, Result};
use serde_json::Value;
use std::io::{BufRead, BufReader};
use std::process::{Command, Stdio};
use std::sync::mpsc::{self, Receiver};
use std::time::{Duration, Instant};

fn receive_event(
    receiver: &Receiver<String>,
    event_type: &str,
    minimum_cycle: u64,
    matches: impl Fn(&Value) -> bool,
) -> Result<Value> {
    let deadline = Instant::now() + Duration::from_secs(15);
    loop {
        let remaining = deadline
            .checked_duration_since(Instant::now())
            .with_context(|| format!("timed out waiting for {event_type} event"))?;
        let line = receiver
            .recv_timeout(remaining)
            .with_context(|| format!("receiving {event_type} event"))?;
        let event = serde_json::from_str::<Value>(&line).context("parsing index status JSON")?;
        if event["type"] == event_type
            && event["cycle"]
                .as_u64()
                .is_some_and(|cycle| cycle >= minimum_cycle)
            && matches(&event)
        {
            return Ok(event);
        }
    }
}

#[test]
fn watch_indexes_changes_and_stops_on_sigterm() -> Result<()> {
    let target = tempfile::tempdir()?;
    let index = tempfile::tempdir()?;
    std::fs::write(target.path().join("first.log"), "first\n")?;
    let mut child = Command::new("/bin/sh")
        .arg("-c")
        .arg(
            r#"trap '' INT
exec "$1" index "$2" --out "$3" --watch --interval 1 --json"#,
        )
        .arg("holys3-watch-test")
        .arg(env!("CARGO_BIN_EXE_holys3"))
        .arg(target.path())
        .arg(index.path())
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .spawn()
        .context("starting index watch")?;
    let stdout = child.stdout.take().context("taking index watch stdout")?;
    let (sender, receiver) = mpsc::channel();
    let reader = std::thread::spawn(move || {
        for line in BufReader::new(stdout).lines() {
            let Ok(line) = line else {
                break;
            };
            if sender.send(line).is_err() {
                break;
            }
        }
    });

    let test_result = (|| -> Result<()> {
        let initial = receive_event(&receiver, "indexed", 1, |_| true)?;
        anyhow::ensure!(
            initial["total_docs"] == 1,
            "unexpected initial event: {initial}"
        );
        std::fs::write(target.path().join("second.log"), "WATCHNEEDLE\n")?;
        let changed = receive_event(&receiver, "indexed", 2, |event| {
            event["added"] == 1 && event["total_docs"] == 2
        })?;
        let status = Command::new("kill")
            .args(["-TERM", &child.id().to_string()])
            .status()
            .context("sending SIGTERM to index watch")?;
        anyhow::ensure!(status.success(), "kill -TERM failed with {status}");
        let stopped = receive_event(
            &receiver,
            "stopped",
            changed["cycle"].as_u64().context("missing changed cycle")?,
            |_| true,
        )?;
        anyhow::ensure!(
            stopped["target"].as_str() == target.path().to_str(),
            "unexpected stopped event"
        );
        let status = child.wait().context("waiting for index watch")?;
        anyhow::ensure!(status.success(), "index watch exited with {status}");
        let output = Command::new(env!("CARGO_BIN_EXE_holys3"))
            .arg("WATCHNEEDLE")
            .arg(target.path())
            .arg("--index")
            .arg(index.path())
            .output()
            .context("searching watched index")?;
        anyhow::ensure!(
            output.status.success(),
            "search failed with {}",
            output.status
        );
        Ok(())
    })();

    if child.try_wait()?.is_none() {
        child.kill()?;
        child.wait()?;
    }
    reader
        .join()
        .map_err(|_| anyhow::anyhow!("index stdout reader panicked"))?;
    test_result
}