From 2d7ad0c3f3771fab6cf6805e9510db84e1a51f5a Mon Sep 17 00:00:00 2001 From: devmaxxx Date: Mon, 19 Oct 2026 13:38:47 +0200 Subject: [PATCH 1/2] fix(serve): catch up at start once a 0.6.4 stop finishes The start-up catch-up was skipped outright when the thread stopping a 0.6.4 server had not finished yet. With a long --every no poll made up for it, so a --stale ask kept answering from the tree as it stood before start. The catch-up is now held until the old server is gone and runs on the first loop turn after that, ahead of any answer. The new test holds a stand-in 0.6.4 server's stop confirmation until the server has answered once, so the race reproduces on every platform rather than only under windows-latest timing. Closes #261 --- src/serve.rs | 20 ++++++++++++++++++-- tests/serve.rs | 48 +++++++++++++++++++++++++++++++++++++++--------- 2 files changed, 57 insertions(+), 11 deletions(-) diff --git a/src/serve.rs b/src/serve.rs index 440c43d..637c9d5 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -573,12 +573,20 @@ pub fn run(repo: &Path, cfg: &config::Config, every: u64, batch: usize, idle: u6 // within seconds: waiting `--every` for the first poll handed that question everything a pull // changed — 1469 vectors, 17 s on beauty-crm. Catching up now spends the same time before // anyone has asked. - if legacy.gone() { adopt_if_moved(&mut watcher, &mut ctx, batch)?; } + // A 0.6.4 server still being stopped holds the catch-up back rather than cancelling it: with + // a long `--every` no poll would come to make up for it, and a `--stale` question would be + // answered from the tree as it stood before this process started. + let mut behind = true; + catch_up_once_gone(&mut behind, &mut legacy, &mut watcher, &mut ctx, batch)?; log(&mut ctx); eprintln!("serve: {} every {every}s, batch {batch}, idle {idle}s, model idle {idle_model}s; Ctrl-C stops", path.display()); let (mut last_poll, mut last_request) = (Instant::now(), Instant::now()); loop { - match rx.recv_timeout(WAKE) { + let received = rx.recv_timeout(WAKE); + // Ahead of the answer, so the first question after the old server leaves is not one + // asked of the tree before the catch-up. + catch_up_once_gone(&mut behind, &mut legacy, &mut watcher, &mut ctx, batch)?; + match received { // Only a question that arrived counts against `--idle`. A bare connect and close is // a liveness probe — another `serve` deciding whether to bind, a client that gave // up — and treating one as a request keeps this process resident, holding 1.3 GB, @@ -657,6 +665,14 @@ fn log(ctx: &mut ask::Context) { for n in ctx.notices() { eprintln!("{}", crate::terminal_safe(&format!("serve: {n}"))); } } +fn catch_up_once_gone(behind: &mut bool, legacy: &mut Legacy, watcher: &mut crate::Watcher, ctx: &mut ask::Context, batch: usize) -> Result<()> { + if *behind && legacy.gone() { + adopt_if_moved(watcher, ctx, batch)?; + *behind = false; + } + Ok(()) +} + /// The poll `ask` does before it answers, and the one case the poll's own answer does not /// name: a store some other process wrote, which the poll read back into the watcher. Either /// way the watcher now holds a graph the context does not, and only an adopt leaves a reader diff --git a/tests/serve.rs b/tests/serve.rs index 5cb2596..19b100d 100644 --- a/tests/serve.rs +++ b/tests/serve.rs @@ -210,9 +210,13 @@ fn serve(dir: &std::path::Path, extra: &[&str]) -> Server { /// The answer once a server is the one giving it. A socket file that exists says a `serve` has /// bound; only a reply says it is listening, so this asks until one comes rather than sleeping. fn ask_until_resident(dir: &std::path::Path, words: &[&str]) -> (String, String) { + ask_until_resident_with(dir, &[], words) +} + +fn ask_until_resident_with(dir: &std::path::Path, extra: &[&str], words: &[&str]) -> (String, String) { let start = Instant::now(); loop { - let (out, err) = ask(dir, &[], words); + let (out, err) = ask(dir, extra, words); if err.contains("serve: answered by the resident process") { return (out, err); } assert!(start.elapsed() < Duration::from_secs(20), "serve never answered over the socket: {err}"); std::thread::sleep(Duration::from_millis(50)); @@ -261,13 +265,7 @@ fn a_server_catches_up_with_the_tree_at_start_rather_than_at_its_first_poll() { wait_for_socket(dir.path()); // `--stale` skips the refresh before the answer, so only the start-up catch-up can have // brought the new document in. - let start = Instant::now(); - let (after, err) = loop { - let (out, err) = ask(dir.path(), &["--stale"], &["возврат", "аванса"]); - if err.contains("serve: answered by the resident process") { break (out, err); } - assert!(start.elapsed() < Duration::from_secs(20), "serve never answered over the socket: {err}"); - std::thread::sleep(Duration::from_millis(50)); - }; + let (after, err) = ask_until_resident_with(dir.path(), &["--stale"], &["возврат", "аванса"]); assert!(after.contains("FR-PAY-2"), "the server refreshed on start: {after}\n{err}"); server.kill().unwrap(); let _ = server.wait(); @@ -709,6 +707,15 @@ fn a_detached_server_answers_after_the_command_returns_and_stop_ends_it() { /// confirmation, then the socket file gone with its exit. The first line it is sent comes back /// on the channel. fn serve_0_6_4(dir: &std::path::Path) -> std::sync::mpsc::Receiver { + let (release, sent) = serve_0_6_4_held(dir); + release.send(()).unwrap(); + sent +} + +/// The same server, confirming the stop only once the test says so — a 0.6.4 server finishing a +/// rerank before it reads the line, made to last as long as the test needs on every platform. +fn serve_0_6_4_held(dir: &std::path::Path) -> (std::sync::mpsc::Sender<()>, std::sync::mpsc::Receiver) { + let (release, held) = std::sync::mpsc::channel::<()>(); let sock = dir.join(".zerograph/serve.sock"); assert!(sock.as_os_str().len() <= 100, "0.6.4 would have bound {} under the temp directory", sock.display()); let listener = transport::bind(&sock); @@ -719,6 +726,7 @@ fn serve_0_6_4(dir: &std::path::Path) -> std::sync::mpsc::Receiver { let _ = writeln!(w, "{{\"ack\":\"0.6.4\"}}"); let mut line = String::new(); let _ = BufReader::new(&s).read_line(&mut line); + let _ = held.recv(); let _ = writeln!(w, "{{\"stopped\":\"0.6.4\"}}"); // The exit takes a moment after the confirmation, as a real one does, so `stop` has to // wait for the file rather than find it already gone. @@ -726,7 +734,7 @@ fn serve_0_6_4(dir: &std::path::Path) -> std::sync::mpsc::Receiver { let _ = std::fs::remove_file(&sock); let _ = tx.send(line); }); - rx + (release, rx) } #[test] @@ -751,3 +759,25 @@ fn serve_stops_a_0_6_4_server_of_its_checkout_before_it_binds() { assert_eq!(sent.recv_timeout(Duration::from_secs(20)).expect("serve never sent the old server the stop line").trim(), r#"{"stop":true}"#); ask_until_resident(dir.path(), &["штраф"]); } + +/// The catch-up at start waits for the old server rather than being skipped for it: with an hour +/// between polls nothing else would bring a document written before start into a `--stale` answer. +#[test] +fn a_server_catches_up_with_the_tree_once_a_slow_0_6_4_server_has_stopped() { + let dir = repo_with_docs(); + std::fs::write(dir.path().join("docs/new.md"), "**FR-PAY-2 · MUST · Возврат аванса**\n\nАванс возвращается при отмене салоном.\n").unwrap(); + let (release, sent) = serve_0_6_4_held(dir.path()); + let _server = serve(dir.path(), &["--every", "3600", "--idle", "60"]); + // A resident answer means the start-up is behind it, with the old server still holding. + let (before, err) = ask_until_resident_with(dir.path(), &["--stale"], &["возврат", "аванса"]); + assert!(!before.contains("FR-PAY-2"), "the server wrote while the old one was still up: {before}\n{err}"); + release.send(()).unwrap(); + assert_eq!(sent.recv_timeout(Duration::from_secs(20)).expect("the old server was never sent the stop line").trim(), r#"{"stop":true}"#); + let start = Instant::now(); + loop { + let (after, err) = ask_until_resident_with(dir.path(), &["--stale"], &["возврат", "аванса"]); + if after.contains("FR-PAY-2") { break; } + assert!(start.elapsed() < Duration::from_secs(20), "the server never caught up once the old one stopped: {after}\n{err}"); + std::thread::sleep(Duration::from_millis(50)); + } +} From 671c3259c6084bf827362cb55ecca2647b225671 Mon Sep 17 00:00:00 2001 From: devmaxxx Date: Mon, 19 Oct 2026 13:47:59 +0200 Subject: [PATCH 2/2] fix(serve): decide the catch-up where the answer reads the old server The loop looked at the 0.6.4 stop before handing a question over, and answer looked again. A stop that finished between the two looks let a --stale question through without being forced stale, and before the catch-up had run, so the reply came from the tree as it was before start. The catch-up now runs inside answer, on the same reading that decides staleness, and a question is forced stale only while the catch-up is still owed. The test no longer retries. It waits for the server to log that the old server stopped, then asserts that the very next --stale answer already has the document written before start. Refs #261 --- src/serve.rs | 56 +++++++++++++++++++++++++++----------------------- tests/serve.rs | 18 ++++++++++------ 2 files changed, 42 insertions(+), 32 deletions(-) diff --git a/src/serve.rs b/src/serve.rs index 637c9d5..cc2349a 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -430,18 +430,32 @@ fn stop_legacy(repo: &Path) -> Result { /// when the rerank is done — up to `RERANK_TIMEOUT` and `STOP_WAIT` on top — and every question /// asked meanwhile would wait that long behind it. Answers go on; it is the watcher's writes, which /// that server would race without the lock, that wait for it to be gone. -struct Legacy(Option>>); +struct Legacy { + stop: Option>>, + /// The start-up catch-up is held back while the old server is up, not cancelled: with a long + /// `--every` no poll would make up for it, and a `--stale` question would be answered from + /// the tree as it stood before this process started. + behind: bool, +} impl Legacy { fn spawn(stop: impl FnOnce() -> Result + Send + 'static) -> Legacy { - Legacy(Some(std::thread::spawn(stop))) + Legacy { stop: Some(std::thread::spawn(stop)), behind: true } + } + + fn catch_up(&mut self, watcher: &mut crate::Watcher, ctx: &mut ask::Context, batch: usize) -> Result<()> { + if self.behind && self.gone() { + adopt_if_moved(watcher, ctx, batch)?; + self.behind = false; + } + Ok(()) } /// Whether the old server is gone, never waiting to find out. A stop that failed is reported /// once and counts as gone: refusing to write would leave the checkout to that server alone. fn gone(&mut self) -> bool { - if self.0.as_ref().is_some_and(|h| !h.is_finished()) { return false; } - if let Some(h) = self.0.take() { + if self.stop.as_ref().is_some_and(|h| !h.is_finished()) { return false; } + if let Some(h) = self.stop.take() { match h.join() { Ok(Ok(_)) => {} Ok(Err(e)) => eprintln!("serve: {e:#}"), @@ -573,25 +587,17 @@ pub fn run(repo: &Path, cfg: &config::Config, every: u64, batch: usize, idle: u6 // within seconds: waiting `--every` for the first poll handed that question everything a pull // changed — 1469 vectors, 17 s on beauty-crm. Catching up now spends the same time before // anyone has asked. - // A 0.6.4 server still being stopped holds the catch-up back rather than cancelling it: with - // a long `--every` no poll would come to make up for it, and a `--stale` question would be - // answered from the tree as it stood before this process started. - let mut behind = true; - catch_up_once_gone(&mut behind, &mut legacy, &mut watcher, &mut ctx, batch)?; + legacy.catch_up(&mut watcher, &mut ctx, batch)?; log(&mut ctx); eprintln!("serve: {} every {every}s, batch {batch}, idle {idle}s, model idle {idle_model}s; Ctrl-C stops", path.display()); let (mut last_poll, mut last_request) = (Instant::now(), Instant::now()); loop { - let received = rx.recv_timeout(WAKE); - // Ahead of the answer, so the first question after the old server leaves is not one - // asked of the tree before the catch-up. - catch_up_once_gone(&mut behind, &mut legacy, &mut watcher, &mut ctx, batch)?; - match received { + match rx.recv_timeout(WAKE) { // Only a question that arrived counts against `--idle`. A bare connect and close is // a liveness probe — another `serve` deciding whether to bind, a client that gave // up — and treating one as a request keeps this process resident, holding 1.3 GB, // for as long as anything at all polls the socket. - Ok(stream) => match answer(stream, &mut watcher, &mut ctx, build, &mut legacy) { + Ok(stream) => match answer(stream, &mut watcher, &mut ctx, build, &mut legacy, batch) { Ok(asked) => if asked { last_request = Instant::now(); }, Err(e) => { last_request = Instant::now(); eprintln!("serve: {e:#}"); } }, @@ -601,6 +607,8 @@ pub fn run(repo: &Path, cfg: &config::Config, every: u64, batch: usize, idle: u6 // Ahead of the poll: a poll that embeds a pull can outlast `stop`'s wait, and a stop // reported as failed while the server is in fact leaving sends the caller to kill it. if terminated() { eprintln!("serve: terminated, exiting"); return Ok(()); } + // For when the old server leaves with nobody asking; a question does its own in `answer`. + legacy.catch_up(&mut watcher, &mut ctx, batch)?; if last_poll.elapsed() >= Duration::from_secs(every) { last_poll = Instant::now(); // The configuration is read once, at start: an id family added, a model or a rerank @@ -665,14 +673,6 @@ fn log(ctx: &mut ask::Context) { for n in ctx.notices() { eprintln!("{}", crate::terminal_safe(&format!("serve: {n}"))); } } -fn catch_up_once_gone(behind: &mut bool, legacy: &mut Legacy, watcher: &mut crate::Watcher, ctx: &mut ask::Context, batch: usize) -> Result<()> { - if *behind && legacy.gone() { - adopt_if_moved(watcher, ctx, batch)?; - *behind = false; - } - Ok(()) -} - /// The poll `ask` does before it answers, and the one case the poll's own answer does not /// name: a store some other process wrote, which the poll read back into the watcher. Either /// way the watcher now holds a graph the context does not, and only an adopt leaves a reader @@ -741,7 +741,7 @@ fn hello_line_within(stream: &sys::Stream, total: Duration) -> Option { /// Whether a question was asked, which is what `--idle` counts — a peer that vanished before /// its hello asked nothing. -fn answer(mut stream: sys::Stream, watcher: &mut crate::Watcher, ctx: &mut ask::Context, build: Option, legacy: &mut Legacy) -> Result { +fn answer(mut stream: sys::Stream, watcher: &mut crate::Watcher, ctx: &mut ask::Context, build: Option, legacy: &mut Legacy, batch: usize) -> Result { let Some(line) = hello_line(&stream) else { return Ok(false) }; // Through the same flag a SIGTERM sets, so the exit is the one every other path takes. if serde_json::from_str::(&line).is_ok_and(|s| s.stop) { @@ -763,8 +763,12 @@ fn answer(mut stream: sys::Stream, watcher: &mut crate::Watcher, ctx: &mut ask:: log(ctx); ctx.begin(); // Until a 0.6.4 server is gone this one writes nothing: the answer is a `--stale` one, which - // skips the dense sync and the detached refresh it may start, as well as the walk. - if !legacy.gone() { hello.req.stale = true; } + // skips the dense sync and the detached refresh it may start, as well as the walk. The + // catch-up is decided here, on the same reading of the old server the answer is: a server + // seen gone only between the loop's look and this one would otherwise answer `--stale` from + // the tree before start. + legacy.catch_up(watcher, ctx, batch)?; + if legacy.behind { hello.req.stale = true; } if hello.req.stale { // `--stale` skips the walk, not the store. A one-shot answers from whatever is on disk // at this instant, so a resident one reads the store back when another process has diff --git a/tests/serve.rs b/tests/serve.rs index 19b100d..932fb63 100644 --- a/tests/serve.rs +++ b/tests/serve.rs @@ -767,17 +767,23 @@ fn a_server_catches_up_with_the_tree_once_a_slow_0_6_4_server_has_stopped() { let dir = repo_with_docs(); std::fs::write(dir.path().join("docs/new.md"), "**FR-PAY-2 · MUST · Возврат аванса**\n\nАванс возвращается при отмене салоном.\n").unwrap(); let (release, sent) = serve_0_6_4_held(dir.path()); - let _server = serve(dir.path(), &["--every", "3600", "--idle", "60"]); + let mut server = serve(dir.path(), &["--every", "3600", "--idle", "60"]); + let (line_tx, lines) = std::sync::mpsc::channel(); + let stderr = server.stderr.take().unwrap(); + std::thread::spawn(move || for line in BufReader::new(stderr).lines().map_while(Result::ok) { let _ = line_tx.send(line); }); // A resident answer means the start-up is behind it, with the old server still holding. let (before, err) = ask_until_resident_with(dir.path(), &["--stale"], &["возврат", "аванса"]); assert!(!before.contains("FR-PAY-2"), "the server wrote while the old one was still up: {before}\n{err}"); release.send(()).unwrap(); assert_eq!(sent.recv_timeout(Duration::from_secs(20)).expect("the old server was never sent the stop line").trim(), r#"{"stop":true}"#); + // The server's own word that the stop is over is the last thing its stopping thread does, so + // the very next question has to be answered from the caught-up tree, with no retry to hide a + // first answer that was not. let start = Instant::now(); - loop { - let (after, err) = ask_until_resident_with(dir.path(), &["--stale"], &["возврат", "аванса"]); - if after.contains("FR-PAY-2") { break; } - assert!(start.elapsed() < Duration::from_secs(20), "the server never caught up once the old one stopped: {after}\n{err}"); - std::thread::sleep(Duration::from_millis(50)); + while !lines.recv_timeout(Duration::from_secs(20)).expect("the server never said the old one stopped").contains("stopped a 0.6.4 server") { + assert!(start.elapsed() < Duration::from_secs(20), "the server never said the old one stopped"); } + let (after, err) = ask(dir.path(), &["--stale"], &["возврат", "аванса"]); + assert!(err.contains("serve: answered by the resident process"), "{err}"); + assert!(after.contains("FR-PAY-2"), "the first answer after the old server stopped missed the catch-up: {after}\n{err}"); }