test: count this query's rbtc-tip-flush thread

changelog.d/tip-flush-thread.md

@@ -0,0 +1,6 @@
+Fixed
+
+- **The IBD tip-flush test counts this query's thread.** A process-wide
+  `/proc` scan also sees other queries in the same suite, so coverage
+  failed `two_ibd_flushes_share_one_named_thread` after drop while the
+  uninstrumented job passed.

crates/rbitcoin-query/src/connect.rs

@@ -77,17 +77,54 @@ pub struct TipFlushJob {
 pub struct TipFlushWorker {
     tx: std::sync::Mutex<Option<std::sync::mpsc::Sender<TipFlushJob>>>,
     thread: std::sync::Mutex<Option<std::thread::JoinHandle<()>>>,
+    /// Threads inside this query's barrier loop. The suite reads this
+    /// instead of every `rbtc-tip-flush` comm in the process.
+    #[cfg(debug_assertions)]
+    alive: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+    /// Linux tid of that thread, stored before `alive` increments.
+    #[cfg(all(debug_assertions, target_os = "linux"))]
+    os_tid: std::sync::Arc<std::sync::atomic::AtomicU64>,
 }
 
 impl Default for TipFlushWorker {
     fn default() -> Self {
         Self {
             tx: std::sync::Mutex::new(None),
             thread: std::sync::Mutex::new(None),
+            #[cfg(debug_assertions)]
+            alive: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
+            #[cfg(all(debug_assertions, target_os = "linux"))]
+            os_tid: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)),
         }
     }
 }
 
+/// Decrements the worker count when the barrier thread leaves its loop.
+#[cfg(debug_assertions)]
+struct TipFlushAlive(std::sync::Arc<std::sync::atomic::AtomicUsize>);
+
+#[cfg(debug_assertions)]
+impl TipFlushAlive {
+    fn enter(alive: std::sync::Arc<std::sync::atomic::AtomicUsize>) -> Self {
+        alive.fetch_add(1, Ordering::Release);
+        Self(alive)
+    }
+}
+
+#[cfg(debug_assertions)]
+impl Drop for TipFlushAlive {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, Ordering::Release);
+    }
+}
+
+#[cfg(all(debug_assertions, target_os = "linux"))]
+fn linux_thread_id() -> Option<u64> {
+    // The link target is `{pid}/task/{tid}`, not a bare tid.
+    let name = std::fs::read_link("/proc/thread-self").ok()?;
+    name.file_name()?.to_str()?.parse().ok()
+}
+
 impl Query {
     /// Tip hash from the last confirm or disconnect, if it still matches `height`.
     pub fn published_tip(&self) -> Option<PublishedTip> {
@@ -275,9 +312,19 @@ impl Query {
         let store = Arc::clone(&self.store);
         #[cfg(debug_assertions)]
         let hold = Arc::clone(&self.tip_flush_hold);
+        #[cfg(debug_assertions)]
+        let alive = Arc::clone(&self.tip_flush_worker.alive);
+        #[cfg(all(debug_assertions, target_os = "linux"))]
+        let os_tid = Arc::clone(&self.tip_flush_worker.os_tid);
         let handle = std::thread::Builder::new()
             .name("rbtc-tip-flush".into())
             .spawn(move || {
+                #[cfg(all(debug_assertions, target_os = "linux"))]
+                if let Some(tid) = linux_thread_id() {
+                    os_tid.store(tid, Ordering::Release);
+                }
+                #[cfg(debug_assertions)]
+                let _alive = TipFlushAlive::enter(alive);
                 while let Ok(job) = rx.recv() {
                     #[cfg(debug_assertions)]
                     while hold.load(Ordering::Acquire) {
@@ -322,6 +369,20 @@ impl Query {
         false
     }
 
+    /// Threads inside this query's `rbtc-tip-flush` loop.
+    #[cfg(debug_assertions)]
+    #[doc(hidden)]
+    pub fn tip_flush_alive(&self) -> Arc<std::sync::atomic::AtomicUsize> {
+        Arc::clone(&self.tip_flush_worker.alive)
+    }
+
+    /// Linux tid of this query's barrier thread, or 0 before it publishes one.
+    #[cfg(all(debug_assertions, target_os = "linux"))]
+    #[doc(hidden)]
+    pub fn tip_flush_os_tid(&self) -> u64 {
+        self.tip_flush_worker.os_tid.load(Ordering::Acquire)
+    }
+
     /// IBD tip flush threads wait before writing until the returned flag is cleared.
     #[cfg(debug_assertions)]
     #[doc(hidden)]

crates/rbitcoin-query/tests/tip_flush.rs

@@ -1,45 +1,87 @@
 //! IBD Class C barrier thread.
 //!
-//! The OS name is read from `/proc`, so this file is Linux-only.
+//! The worker publishes its own tid. A process-wide comm scan also counts
+//! other queries running in this suite.
 
 #![cfg(target_os = "linux")]
 
-use std::sync::atomic::Ordering;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+/// Clears the flush hold on drop so a failed assert cannot leave the worker
+/// parked while `Query`'s drop joins it.
+struct ReleaseHold(Arc<AtomicBool>);
+
+impl Drop for ReleaseHold {
+    fn drop(&mut self) {
+        self.0.store(false, Ordering::SeqCst);
+    }
+}
+
 /// Two barriers share one `rbtc-tip-flush` thread. Tip mode does not start it.
 #[test]
 fn two_ibd_flushes_share_one_named_thread() {
     let (dir, q) = rbitcoin_query::testutil::tiny_query_labeled("tip-flush-one");
     assert!(q.index_mode().unwrap().is_tip());
+    let alive = q.tip_flush_alive();
     q.flush_tip(None).unwrap();
-    assert_eq!(comm_now("rbtc-tip-flush"), 0, "tip mode flushes inline");
+    assert_eq!(alive.load(Ordering::Acquire), 0, "tip mode flushes inline");
 
     q.enter_direct_index_mode_sh(false).unwrap();
     let hold = q.hold_tip_flushes();
+    let _release = ReleaseHold(Arc::clone(&hold));
     q.spawn_tip_flush().unwrap();
+    let tid = wait_tid(&q, Duration::from_millis(200));
+    assert_eq!(
+        task_comm(tid).as_deref(),
+        Some("rbtc-tip-flush"),
+        "IBD barrier thread name"
+    );
     q.spawn_tip_flush().unwrap();
-    let n = comm_max("rbtc-tip-flush", Duration::from_millis(200));
+    let n = alive_max(&alive, Duration::from_millis(200));
+    assert_eq!(n, 1, "two IBD barriers use one rbtc-tip-flush thread");
+    assert_eq!(
+        q.tip_flush_os_tid(),
+        tid,
+        "second barrier keeps the same thread"
+    );
     hold.store(false, Ordering::SeqCst);
     q.join_tip_flush().unwrap();
+    drop(_release);
     drop(q);
     drop(dir);
-    assert_eq!(n, 1, "two IBD barriers use one rbtc-tip-flush thread");
-    assert_eq!(comm_now("rbtc-tip-flush"), 0, "drop joins rbtc-tip-flush");
+    assert_eq!(
+        alive.load(Ordering::Acquire),
+        0,
+        "drop joins rbtc-tip-flush"
+    );
+    assert!(
+        task_gone(tid, Duration::from_secs(2)),
+        "rbtc-tip-flush tid {tid} still in /proc"
+    );
 }
 
-fn comm_now(want: &str) -> usize {
-    linux_comms()
-        .iter()
-        .filter(|name| name.as_str() == want)
-        .count()
+fn wait_tid(q: &rbitcoin_query::Query, window: Duration) -> u64 {
+    let deadline = Instant::now() + window;
+    loop {
+        let tid = q.tip_flush_os_tid();
+        if tid != 0 {
+            return tid;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "rbtc-tip-flush did not publish a tid"
+        );
+        std::thread::sleep(Duration::from_millis(1));
+    }
 }
 
-fn comm_max(want: &str, window: Duration) -> usize {
+fn alive_max(alive: &Arc<AtomicUsize>, window: Duration) -> usize {
     let deadline = Instant::now() + window;
     let mut n = 0;
     loop {
-        n = n.max(comm_now(want));
+        n = n.max(alive.load(Ordering::Acquire));
         if Instant::now() >= deadline {
             break;
         }
@@ -48,15 +90,20 @@ fn comm_max(want: &str, window: Duration) -> usize {
     n
 }
 
-fn linux_comms() -> Vec<String> {
-    let mut names = Vec::new();
-    let Ok(tasks) = std::fs::read_dir("/proc/self/task") else {
-        return names;
-    };
-    for task in tasks.flatten() {
-        if let Ok(raw) = std::fs::read_to_string(task.path().join("comm")) {
-            names.push(raw.trim_end_matches('\n').to_string());
+fn task_comm(tid: u64) -> Option<String> {
+    let raw = std::fs::read_to_string(format!("/proc/self/task/{tid}/comm")).ok()?;
+    Some(raw.trim_end_matches('\n').to_string())
+}
+
+fn task_gone(tid: u64, window: Duration) -> bool {
+    let deadline = Instant::now() + window;
+    loop {
+        if std::fs::metadata(format!("/proc/self/task/{tid}")).is_err() {
+            return true;
+        }
+        if Instant::now() >= deadline {
+            return false;
         }
+        std::thread::sleep(Duration::from_millis(1));
     }
-    names
 }