Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 64 additions & 6 deletions crates/nms-watch/src/watcher.rs
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,22 @@ use crate::snapshot::SaveSnapshot;
/// After this many consecutive parse failures, log a warning.
const MAX_CONSECUTIVE_FAILURES: usize = 5;

#[derive(Default)]
struct ParseFailureStreak {
count: usize,
}

impl ParseFailureStreak {
fn record_failure(&mut self) -> bool {
self.count = self.count.saturating_add(1);
self.count == MAX_CONSECUTIVE_FAILURES
}

fn record_success(&mut self) {
self.count = 0;
}
}

/// Milliseconds to wait between file size checks for stability.
const FILE_STABILITY_CHECK_MS: u64 = 100;

Expand Down Expand Up @@ -498,7 +514,7 @@ fn start_watching_inner(
.name("nms-save-watch".into())
.spawn(move || {
let mut source = initial_source;
let mut consecutive_failures: usize = 0;
let mut failure_streak = ParseFailureStreak::default();

while !shutdown_worker.load(Ordering::Acquire) {
let events = match notify_rx.recv_timeout(Duration::from_millis(100)) {
Expand Down Expand Up @@ -528,7 +544,7 @@ fn start_watching_inner(
// Re-parse save file
match SourceSnapshot::from_file(&save_path_bg) {
Ok(new_source) => {
consecutive_failures = 0;
failure_streak.record_success();
let delta = compute_delta(&source.snapshot, &new_source.snapshot);
let changed = !delta.is_empty()
|| source.player_state != new_source.player_state
Expand Down Expand Up @@ -573,8 +589,7 @@ fn start_watching_inner(
source = new_source;
}
Err(e) => {
consecutive_failures += 1;
if consecutive_failures == MAX_CONSECUTIVE_FAILURES {
if failure_streak.record_failure() {
eprintln!(
"Warning: {MAX_CONSECUTIVE_FAILURES} consecutive parse failures. \
Save file may be corrupt or format changed: {e}"
Expand Down Expand Up @@ -1764,8 +1779,51 @@ mod tests {
}

#[test]
fn test_max_consecutive_failures_constant() {
const { assert!(MAX_CONSECUTIVE_FAILURES > 0) };
fn test_parse_failure_streak_warns_once_at_five_and_suppresses_until_success() {
assert_eq!(MAX_CONSECUTIVE_FAILURES, 5);

let mut streak = ParseFailureStreak::default();
for failure in 1..=8 {
assert_eq!(
streak.record_failure(),
failure == 5,
"unexpected warning decision at failure {failure}"
);
}
}

#[test]
fn test_parse_failure_streak_success_resets_before_and_after_warning() {
let mut streak = ParseFailureStreak::default();
for _ in 0..3 {
assert!(!streak.record_failure());
}

streak.record_success();
for _ in 0..4 {
assert!(!streak.record_failure());
}
assert!(streak.record_failure());

streak.record_success();
for _ in 0..4 {
assert!(!streak.record_failure());
}
assert!(streak.record_failure());
}

#[test]
fn test_parse_failure_streak_zero_reset_and_instances_are_independent() {
let mut first = ParseFailureStreak::default();
let mut second = ParseFailureStreak::default();
first.record_success();

for _ in 0..4 {
assert!(!first.record_failure());
assert!(!second.record_failure());
}
assert!(first.record_failure());
assert!(second.record_failure());
}

#[test]
Expand Down
85 changes: 4 additions & 81 deletions crates/nms-watch/tests/validated_save.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,43 +5,10 @@ use nms_core::system::{Planet, System, SystemId};
use nms_save::address::AddressDecodeError;
use nms_watch::SaveSnapshot;
use std::io::Write;
use std::process::{Child, Command, ExitStatus, Stdio};
use std::time::{Duration, Instant};
use std::time::Duration;

const MINIMAL: &[u8] = include_bytes!("../../../data/test/save56_minimal.json");

struct ChildGuard(Option<Child>);

impl ChildGuard {
fn try_wait(&mut self) -> std::io::Result<Option<ExitStatus>> {
let status = self.0.as_mut().expect("owned child").try_wait()?;
if status.is_some() {
self.0.take();
}
Ok(status)
}
}

impl Drop for ChildGuard {
fn drop(&mut self) {
let Some(mut child) = self.0.take() else {
return;
};
if let Err(error) = child.kill() {
let _ = writeln!(
std::io::stderr().lock(),
"watcher child cleanup kill failed: {error}"
);
}
if let Err(error) = child.wait() {
let _ = writeln!(
std::io::stderr().lock(),
"watcher child cleanup wait failed: {error}"
);
}
}
}

fn replace_once(source: &[u8], from: &str, to: &str) -> Vec<u8> {
let source = std::str::from_utf8(source).unwrap();
assert_eq!(
Expand Down Expand Up @@ -296,53 +263,7 @@ fn test_snapshot_file_path_rejects_late_bad_address_before_publishing_any_snapsh
}

#[test]
fn test_five_semantic_address_failures_keep_last_good_state_and_emit_one_warning() {
const CHILD: &str = "NMS_WATCH_VALIDATED_SAVE_CHILD";
if std::env::var_os(CHILD).is_some() {
run_five_failure_watcher_child();
return;
}

let dir = tempfile::tempdir().unwrap();
let stdout_path = dir.path().join("child.stdout");
let stderr_path = dir.path().join("child.stderr");
let stdout = std::fs::File::create(&stdout_path).unwrap();
let stderr = std::fs::File::create(&stderr_path).unwrap();
let child = Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"test_five_semantic_address_failures_keep_last_good_state_and_emit_one_warning",
"--nocapture",
])
.env(CHILD, "1")
.stdout(Stdio::from(stdout))
.stderr(Stdio::from(stderr))
.spawn()
.unwrap();
let mut child = ChildGuard(Some(child));

let deadline = Instant::now() + Duration::from_secs(10);
let status = loop {
if let Some(status) = child.try_wait().unwrap() {
break status;
}
if Instant::now() >= deadline {
panic!("watcher child exceeded its bounded 10-second runtime");
}
std::thread::sleep(Duration::from_millis(10));
};
assert!(status.success(), "watcher child failed: {status}");
let stderr = std::fs::read_to_string(stderr_path).unwrap();
assert_eq!(
stderr
.matches("Warning: 5 consecutive parse failures.")
.count(),
1,
"warning output: {stderr}"
);
}

fn run_five_failure_watcher_child() {
fn test_semantic_address_failures_preserve_last_good_state_until_recovery() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("save.json");
std::fs::write(&path, MINIMAL).unwrap();
Expand Down Expand Up @@ -381,6 +302,8 @@ fn run_five_failure_watcher_child() {
std::fs::rename(replacement, &path).unwrap();
assert_eq!(std::fs::read(&path).unwrap(), bytes);
let observed_invalid_mtime = std::fs::metadata(&path).unwrap().modified().unwrap();
// This bounded window proves no update was published; it does not prove
// that the watcher parsed this particular replacement.
assert!(matches!(
watcher.receiver.recv_timeout(Duration::from_millis(400)),
Err(std::sync::mpsc::RecvTimeoutError::Timeout)
Expand Down
Loading