test(harness): pid-only --dir sweep + deadline-poll asserts (test-hygiene)

CHANGELOG.md

@@ -25,6 +25,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `--experimental-per-shard-rewrite` is deprecated (warns, no-op).
 
 ### Fixed
+- **Test-hygiene sweep: pid-only temp dirs + timing-race asserts.** Eight
+  spawn sites across six integration suites named their data dirs by pid
+  only, so a crashed run's leftover dir was silently resurrected once the
+  pid was reused — the server then reloaded stale persistence state and the
+  suite failed on ghosts (the documented stale-reload trap). All eight now
+  use `tempfile` (RAII where a single owner exists; unique `keep()` dirs
+  where the restart flow shares one dir between two server handles).
+  `parked_idle_parity` additionally gets deadline-polled asserts: CLIENT
+  KILL's registry removal is polled instead of read once (the CI
+  assert-too-soon race that fired 3/3 on a starved 2-vCPU runner),
+  connects retry against a listening-but-backlogged server, and read
+  deadlines widened 10s→30s (deadline-bound — green runs are unaffected).
+  The `client_tracking_invalidation` multikey second-key push flake is
+  product-side, not harness-side, and is now tracked as #448.
 - **Central accept loop no longer head-of-line blocks on one wedged shard
   (#438 F3).** Every central-listener delivery (tokio plain/TLS, monoio
   plain/TLS — the monoio sends were *synchronous*, stalling the whole

tests/info_memory_allocator_pagecache.rs

@@ -36,14 +36,17 @@ fn release_binary() -> std::path::PathBuf {
 struct Moon {
     child: Child,
     port: u16,
-    tmp_dir: std::path::PathBuf,
+    /// RAII tempdir (test-hygiene sweep): random unique name, removed on
+    /// drop (after the child is killed in `Drop` above the field drop).
+    /// The old pid-only name resurrected stale dirs after a crashed run
+    /// once the pid was reused, reloading stale persistence state.
+    _tmp_dir: tempfile::TempDir,
 }
 
 impl Drop for Moon {
     fn drop(&mut self) {
         let _ = self.child.kill();
         let _ = self.child.wait();
-        let _ = std::fs::remove_dir_all(&self.tmp_dir);
     }
 }
 
@@ -62,8 +65,7 @@ fn spawn_moon() -> Option<Moon> {
         );
         return None;
     }
-    let tmp_dir = std::env::temp_dir().join(format!("moon-test-info-mem-{}", std::process::id()));
-    let _ = std::fs::create_dir_all(&tmp_dir);
+    let tmp_dir = tempfile::tempdir().expect("tempdir");
     let (child, port) = common::spawn_listening(|port| {
         Command::new(&bin)
             .args([
@@ -76,7 +78,7 @@ fn spawn_moon() -> Option<Moon> {
                 "--appendonly",
                 "no",
                 "--dir",
-                tmp_dir.to_str().unwrap(),
+                tmp_dir.path().to_str().unwrap(),
                 // PageCache only exists when disk-offload is enabled --
                 // required so `pagecache_bytes` has a chance to be non-zero.
                 "--disk-offload",
@@ -94,7 +96,7 @@ fn spawn_moon() -> Option<Moon> {
     let moon = Moon {
         child,
         port,
-        tmp_dir,
+        _tmp_dir: tmp_dir,
     };
 
     let deadline = Instant::now() + Duration::from_secs(5);

tests/memory_doctor_response.rs

@@ -33,14 +33,17 @@ fn release_binary() -> std::path::PathBuf {
 struct Moon {
     child: Child,
     port: u16,
-    tmp_dir: std::path::PathBuf,
+    /// RAII tempdir (test-hygiene sweep): random unique name, removed on
+    /// drop (after the child is killed in `Drop` above the field drop).
+    /// The old pid-only name resurrected stale dirs after a crashed run
+    /// once the pid was reused, reloading stale persistence state.
+    _tmp_dir: tempfile::TempDir,
 }
 
 impl Drop for Moon {
     fn drop(&mut self) {
         let _ = self.child.kill();
         let _ = self.child.wait();
-        let _ = std::fs::remove_dir_all(&self.tmp_dir);
     }
 }
 
@@ -57,8 +60,7 @@ fn spawn_moon() -> Option<Moon> {
         );
         return None;
     }
-    let tmp_dir = std::env::temp_dir().join(format!("moon-test-doctor-{}", std::process::id()));
-    let _ = std::fs::create_dir_all(&tmp_dir);
+    let tmp_dir = tempfile::tempdir().expect("tempdir");
     let (child, port) = common::spawn_listening(|port| {
         Command::new(&bin)
             .args([
@@ -71,7 +73,7 @@ fn spawn_moon() -> Option<Moon> {
                 "--appendonly",
                 "no",
                 "--dir",
-                tmp_dir.to_str().unwrap(),
+                tmp_dir.path().to_str().unwrap(),
                 "--disk-offload",
                 "disable",
             ])
@@ -83,7 +85,7 @@ fn spawn_moon() -> Option<Moon> {
     let moon = Moon {
         child,
         port,
-        tmp_dir,
+        _tmp_dir: tmp_dir,
     };
 
     // Wait up to ~5s for PING to succeed.

tests/memory_prometheus_kinds.rs

@@ -42,14 +42,17 @@ struct Moon {
     child: Child,
     port: u16,
     admin_port: u16,
-    tmp_dir: std::path::PathBuf,
+    /// RAII tempdir (test-hygiene sweep): random unique name, removed on
+    /// drop (after the child is killed in `Drop` above the field drop).
+    /// The old pid-only name resurrected stale dirs after a crashed run
+    /// once the pid was reused, reloading stale persistence state.
+    _tmp_dir: tempfile::TempDir,
 }
 
 impl Drop for Moon {
     fn drop(&mut self) {
         let _ = self.child.kill();
         let _ = self.child.wait();
-        let _ = std::fs::remove_dir_all(&self.tmp_dir);
     }
 }
 
@@ -67,8 +70,7 @@ fn spawn_moon() -> Option<Moon> {
         return None;
     }
     let admin_port = common::reserve_port();
-    let tmp_dir = std::env::temp_dir().join(format!("moon-test-prom-{}", std::process::id()));
-    let _ = std::fs::create_dir_all(&tmp_dir);
+    let tmp_dir = tempfile::tempdir().expect("tempdir");
     let (child, port) = common::spawn_listening(|port| {
         Command::new(&bin)
             .args([
@@ -81,7 +83,7 @@ fn spawn_moon() -> Option<Moon> {
                 "--appendonly",
                 "no",
                 "--dir",
-                tmp_dir.to_str().unwrap(),
+                tmp_dir.path().to_str().unwrap(),
                 "--disk-offload",
                 "disable",
             ])
@@ -94,7 +96,7 @@ fn spawn_moon() -> Option<Moon> {
         child,
         port,
         admin_port,
-        tmp_dir,
+        _tmp_dir: tmp_dir,
     };
 
     // Wait up to ~5s for PING to succeed.

tests/parked_idle_parity.rs

@@ -43,9 +43,29 @@ fn spawn_moon(dir: &std::path::Path, port: u16) -> std::process::Child {
     spawn_moon_with(dir, port, &[])
 }
 
+/// Test-hygiene sweep: connect with a deadline. Under full-suite load a
+/// freshly listening server can still refuse/reset the first attempts
+/// (SYN backlog + debug-binary scheduling), which failed tests at the
+/// connect line rather than on anything they assert.
+fn connect_retry(port: u16) -> TcpStream {
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        match TcpStream::connect(("127.0.0.1", port)) {
+            Ok(s) => return s,
+            Err(e) => {
+                assert!(
+                    std::time::Instant::now() < deadline,
+                    "connect to 127.0.0.1:{port} kept failing: {e}"
+                );
+                std::thread::sleep(Duration::from_millis(50));
+            }
+        }
+    }
+}
+
 fn read_exact_deadline(stream: &mut TcpStream, want: usize) -> Vec<u8> {
     stream
-        .set_read_timeout(Some(Duration::from_secs(10)))
+        .set_read_timeout(Some(Duration::from_secs(30)))
         .expect("set timeout");
     let mut out = Vec::with_capacity(want);
     let mut chunk = [0u8; 4096];
@@ -70,7 +90,7 @@ fn ping(stream: &mut TcpStream) {
 fn command_reply(stream: &mut TcpStream, cmd: &str) -> String {
     stream.write_all(cmd.as_bytes()).expect("write cmd");
     stream
-        .set_read_timeout(Some(Duration::from_secs(10)))
+        .set_read_timeout(Some(Duration::from_secs(30)))
         .expect("set timeout");
     let mut buf = Vec::new();
     let mut chunk = [0u8; 65536];
@@ -104,7 +124,7 @@ fn parked_connection_serves_all_traffic_after_wake() {
     let dir = tempfile::tempdir().expect("tempdir");
     let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p));
 
-    let mut conn = TcpStream::connect(("127.0.0.1", port)).expect("connect");
+    let mut conn = connect_retry(port);
     conn.set_nodelay(true).ok();
     ping(&mut conn);
 
@@ -165,13 +185,13 @@ fn parked_connection_visible_and_killable() {
     let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p));
 
     // Victim: name itself so the control conn can find its id, then park.
-    let mut victim = TcpStream::connect(("127.0.0.1", port)).expect("connect victim");
+    let mut victim = connect_retry(port);
     let r = command_reply(&mut victim, "CLIENT SETNAME parkvictim\r\n");
     assert!(r.starts_with("+OK"), "SETNAME failed: {r}");
     std::thread::sleep(PARK_WAIT);
 
     // Control connection stays active.
-    let mut control = TcpStream::connect(("127.0.0.1", port)).expect("connect control");
+    let mut control = connect_retry(port);
     let list = command_reply(&mut control, "CLIENT LIST\r\n");
     let victim_line = list
         .lines()
@@ -193,7 +213,7 @@ fn parked_connection_visible_and_killable() {
 
     // The victim's socket must observe the close (EOF or reset) promptly.
     victim
-        .set_read_timeout(Some(Duration::from_secs(10)))
+        .set_read_timeout(Some(Duration::from_secs(30)))
         .expect("set timeout");
     let mut buf = [0u8; 16];
     match victim.read(&mut buf) {
@@ -202,12 +222,22 @@ fn parked_connection_visible_and_killable() {
         Ok(n) => panic!("expected close, got {n} bytes"),
     }
 
-    // And it must be gone from CLIENT LIST.
-    let list = command_reply(&mut control, "CLIENT LIST\r\n");
-    assert!(
-        !list.contains("name=parkvictim"),
-        "killed parked connection still listed:\n{list}"
-    );
+    // And it must be gone from CLIENT LIST. The client-side close arrives
+    // instantly (shutdown(2)); the registry entry is released only once the
+    // killed handler task gets scheduled — poll with a deadline instead of
+    // asserting the very first read (the documented CI assert-too-soon race).
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        let list = command_reply(&mut control, "CLIENT LIST\r\n");
+        if !list.contains("name=parkvictim") {
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "killed parked connection still listed after 10s:\n{list}"
+        );
+        std::thread::sleep(Duration::from_millis(100));
+    }
 
     let _ = child.kill();
     let _ = child.wait();
@@ -222,12 +252,12 @@ fn resumed_connection_keeps_registry_identity() {
     let dir = tempfile::tempdir().expect("tempdir");
     let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p));
 
-    let mut victim = TcpStream::connect(("127.0.0.1", port)).expect("connect victim");
+    let mut victim = connect_retry(port);
     victim.set_nodelay(true).ok();
     let r = command_reply(&mut victim, "CLIENT SETNAME wakekeeper\r\n");
     assert!(r.starts_with("+OK"), "SETNAME failed: {r}");
 
-    let mut control = TcpStream::connect(("127.0.0.1", port)).expect("connect control");
+    let mut control = connect_retry(port);
     let list = command_reply(&mut control, "CLIENT LIST\r\n");
     let orig_id: u64 = list
         .lines()
@@ -280,13 +310,13 @@ fn rst_while_parked_tears_down_promptly() {
     let dir = tempfile::tempdir().expect("tempdir");
     let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p));
 
-    let victim = TcpStream::connect(("127.0.0.1", port)).expect("connect victim");
+    let victim = connect_retry(port);
     {
         let mut v = &victim;
         v.write_all(b"CLIENT SETNAME rstvictim\r\n").expect("write");
         let mut buf = [0u8; 8];
         victim
-            .set_read_timeout(Some(Duration::from_secs(10)))
+            .set_read_timeout(Some(Duration::from_secs(30)))
             .expect("timeout");
         let _ = (&victim).read(&mut buf);
     }
@@ -318,7 +348,7 @@ fn rst_while_parked_tears_down_promptly() {
     // The server must fully release the conn: gone from CLIENT LIST and no
     // respawn loop keeping it alive.
     std::thread::sleep(Duration::from_millis(2000));
-    let mut control = TcpStream::connect(("127.0.0.1", port)).expect("connect control");
+    let mut control = connect_retry(port);
     let list = command_reply(&mut control, "CLIENT LIST\r\n");
     assert!(
         !list.contains("name=rstvictim"),
@@ -335,10 +365,10 @@ fn active_sibling_undisturbed_by_parking() {
     let dir = tempfile::tempdir().expect("tempdir");
     let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p));
 
-    let mut idle = TcpStream::connect(("127.0.0.1", port)).expect("connect idle");
+    let mut idle = connect_retry(port);
     ping(&mut idle);
 
-    let mut active = TcpStream::connect(("127.0.0.1", port)).expect("connect active");
+    let mut active = connect_retry(port);
     active.set_nodelay(true).ok();
 
     // ~5s of continuous activity spanning downshift + park of the sibling.
@@ -378,7 +408,7 @@ fn partial_frame_survives_a_park() {
     let dir = tempfile::tempdir().expect("tempdir");
     let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p));
 
-    let mut conn = TcpStream::connect(("127.0.0.1", port)).expect("connect");
+    let mut conn = connect_retry(port);
     conn.set_nodelay(true).ok();
     ping(&mut conn);
 
@@ -429,17 +459,17 @@ fn unauthenticated_conn_never_task_parks() {
         common::spawn_listening(|p| spawn_moon_with(dir.path(), p, &["--requirepass", "park-f6"]));
 
     // Positive control: authenticated, then idle past the park threshold.
-    let mut authed = TcpStream::connect(("127.0.0.1", port)).expect("connect authed");
+    let mut authed = connect_retry(port);
     let r = command_reply(&mut authed, "AUTH park-f6\r\n");
     assert!(r.starts_with("+OK"), "AUTH failed: {r}");
 
     // Victim: connects and never speaks — no AUTH, no bytes.
-    let silent = TcpStream::connect(("127.0.0.1", port)).expect("connect silent");
+    let silent = connect_retry(port);
 
     std::thread::sleep(PARK_WAIT);
 
     // Control conn (freshly active, not parked) reads the gauge.
-    let mut control = TcpStream::connect(("127.0.0.1", port)).expect("connect control");
+    let mut control = connect_retry(port);
     let r = command_reply(&mut control, "AUTH park-f6\r\n");
     assert!(r.starts_with("+OK"), "control AUTH failed: {r}");
     let info = command_reply(&mut control, "INFO clients\r\n");

tests/tls_park_keyupdate.rs

@@ -132,11 +132,12 @@ fn drive_handshake(
 /// the KeyUpdate deadlock the c10k review described becomes real.
 #[test]
 fn keyupdate_reply_is_deferred_until_the_next_outbound_record() {
-    let tmp = std::env::temp_dir().join(format!("moon-tls-keyupdate-{}", std::process::id()));
-    let _ = std::fs::create_dir_all(&tmp);
+    // Test-hygiene sweep: RAII tempdir (unique, removed even on panic/early
+    // return) replaces the pid-only name that could resurrect stale certs.
+    let tmp_guard = tempfile::tempdir().expect("tempdir");
+    let tmp = tmp_guard.path().to_path_buf();
     if !generate_cert(&tmp) {
         eprintln!("skipping: openssl unavailable");
-        let _ = std::fs::remove_dir_all(&tmp);
         return;
     }
 
@@ -224,6 +225,4 @@ fn keyupdate_reply_is_deferred_until_the_next_outbound_record() {
         !s2c.is_empty(),
         "the deferred KeyUpdate must ride out with the next record"
     );
-
-    let _ = std::fs::remove_dir_all(&tmp);
 }