Catch up with the tree at start even while a 0.6.4 server is still stopping

src/serve.rs

@@ -430,18 +430,32 @@ fn stop_legacy(repo: &Path) -> Result<bool> {
 /// 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<std::thread::JoinHandle<Result<bool>>>);
+struct Legacy {
+    stop: Option<std::thread::JoinHandle<Result<bool>>>,
+    /// 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<bool> + 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,7 +587,7 @@ 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)?; }
+    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());
@@ -583,7 +597,7 @@ pub fn run(repo: &Path, cfg: &config::Config, every: u64, batch: usize, idle: u6
             // 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:#}"); }
             },
@@ -593,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
@@ -725,7 +741,7 @@ fn hello_line_within(stream: &sys::Stream, total: Duration) -> Option<String> {
 
 /// 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<crate::walk::Stamp>, legacy: &mut Legacy) -> Result<bool> {
+fn answer(mut stream: sys::Stream, watcher: &mut crate::Watcher, ctx: &mut ask::Context, build: Option<crate::walk::Stamp>, legacy: &mut Legacy, batch: usize) -> Result<bool> {
     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::<Stop>(&line).is_ok_and(|s| s.stop) {
@@ -747,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

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<String> {
+    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<String>) {
+    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,14 +726,15 @@ fn serve_0_6_4(dir: &std::path::Path) -> std::sync::mpsc::Receiver<String> {
         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.
         std::thread::sleep(Duration::from_millis(300));
         let _ = std::fs::remove_file(&sock);
         let _ = tx.send(line);
     });
-    rx
+    (release, rx)
 }
 
 #[test]
@@ -751,3 +759,31 @@ 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 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();
+    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}");
+}