diff --git a/changelog.d/deflake-ci.md b/changelog.d/deflake-ci.md new file mode 100644 index 000000000..f292d1f61 --- /dev/null +++ b/changelog.d/deflake-ci.md @@ -0,0 +1,3 @@ +Fixed + +- **A same-slot store probe and a full RPC work queue no longer fail when the schedule is unlucky.** The probe keeps drawing until the mixed key shares the page, and the work-queue check holds the only permit until it has seen HTTP 503. A live follow session counts an unknown BIP324 short id as `*other*` and stays connected. diff --git a/crates/rbitcoin-net/src/peer_tests.rs b/crates/rbitcoin-net/src/peer_tests.rs index bca682643..d40657023 100644 --- a/crates/rbitcoin-net/src/peer_tests.rs +++ b/crates/rbitcoin-net/src/peer_tests.rs @@ -2832,6 +2832,58 @@ async fn write_peer_msg(raw: &mut V2PlainSession, msg: NetworkMessage) { .unwrap(); } +/// Unknown BIP324 short id on a live follow session counts as `*other*` +/// and the peer stays up (Core `test_msgtype`). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unknown_v2_short_id_counts_other_and_stays() { + let mut t = paused_serve("bad-short-id", 0, &["/rbitcoin:test(badtype)/"]).await; + let (raw, sess) = &mut t.clients[0]; + let id = sess.id; + let first_ping = loop { + if let NetworkMessage::Ping(n) = next_peer_msg(raw).await.unwrap() { + break n; + } + }; + write_peer_msg(raw, NetworkMessage::Pong(first_ping)).await; + // short id 99 + compact-size string "d" (Core `msg_unrecognized`). + let unknown = [99u8, 1, b'd']; + raw.write_contents(&unknown).await.unwrap(); + raw.write_contents(&unknown).await.unwrap(); + let want = crate::v2::v2_other_recv_bytes(unknown.len()) * 2; + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let other = loop { + let snap = t.node.peers.snapshot(); + let n = snap + .iter() + .find(|p| p.id == id) + .and_then(|p| p.bytesrecv_per_msg.get("*other*")) + .copied(); + if n == Some(want) { + break want; + } + assert!( + std::time::Instant::now() < deadline, + "unknown short id was not counted as *other*, last={n:?}" + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + }; + assert_eq!(other, want); + write_peer_msg(raw, NetworkMessage::Ping(9)).await; + let pong = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if let NetworkMessage::Pong(n) = next_peer_msg(raw).await.unwrap() { + return n; + } + } + }) + .await + .expect("session stays up after an unknown short id"); + assert_eq!(pong, 9); + assert!(t.node.peers.snapshot().iter().any(|p| p.id == id)); + t.node.shutdown().await; + let _ = std::fs::remove_dir_all(&t.dir); +} + fn getdata_msg(hashes: &[BlockHash]) -> NetworkMessage { NetworkMessage::GetData(hashes.iter().map(|h| Inventory::WitnessBlock(*h)).collect()) } diff --git a/crates/rbitcoin-rpc/src/server.rs b/crates/rbitcoin-rpc/src/server.rs index e782a549f..364402ae8 100644 --- a/crates/rbitcoin-rpc/src/server.rs +++ b/crates/rbitcoin-rpc/src/server.rs @@ -598,6 +598,10 @@ async fn rpc_post(State(state): State, body: Bytes) -> Response { .into_response(); } }; + // Tests arm a hold keyed by this listener's log path so a probe can + // observe the permit while it is still taken. Unarmed calls return. + #[cfg(test)] + tests::hold_work_queue(&state.ctx.logpath).await; let joined = tokio::task::spawn_blocking(move || { let _g = BlockingRegion::enter(); if waited { @@ -853,6 +857,70 @@ fn authorized(auth: &RpcAuth, cookie: Option<&RpcCookie>, headers: &HeaderMap) - mod tests { use super::*; use rbitcoin_primitives::Network; + use std::collections::HashMap; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use tokio::sync::watch; + + struct WorkQueueHold { + entered: Arc, + release_rx: watch::Receiver, + } + + static WORK_QUEUE_HOLDS: Mutex>> = Mutex::new(None); + + fn work_queue_holds() -> std::sync::MutexGuard<'static, Option>> + { + WORK_QUEUE_HOLDS.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// Hold the next RPC on `logpath` after it takes a work-queue permit. + pub(super) fn arm_work_queue_hold(logpath: &str) -> (Arc, watch::Sender) { + let entered = Arc::new(AtomicBool::new(false)); + let (release_tx, release_rx) = watch::channel(false); + work_queue_holds().get_or_insert_with(HashMap::new).insert( + logpath.to_string(), + WorkQueueHold { + entered: Arc::clone(&entered), + release_rx, + }, + ); + (entered, release_tx) + } + + pub(super) fn clear_work_queue_hold(logpath: &str) { + if let Some(holds) = work_queue_holds().as_mut() { + holds.remove(logpath); + } + } + + /// Sends `true` on drop so a failed assert cannot leave the handler parked. + struct ReleaseHold(watch::Sender); + + impl Drop for ReleaseHold { + fn drop(&mut self) { + let _ = self.0.send(true); + } + } + + pub(super) async fn hold_work_queue(logpath: &str) { + let held = { + let holds = work_queue_holds(); + holds + .as_ref() + .and_then(|m| m.get(logpath)) + .map(|h| (Arc::clone(&h.entered), h.release_rx.clone())) + }; + let Some((entered, mut release_rx)) = held else { + return; + }; + entered.store(true, Ordering::SeqCst); + while !*release_rx.borrow() { + if release_rx.changed().await.is_err() { + return; + } + } + } #[test] fn work_queue_zero_and_omitted_are_the_default() { @@ -1704,50 +1772,43 @@ mod tests { .expect("batch json"); assert_eq!(arr.len(), 2, "{body:?}"); - // One POST holds the only permit for every call in the batch. - let mut calls = Vec::new(); - for i in 0..256 { - calls.push(serde_json::json!({ - "jsonrpc": "1.0", - "id": i, - "method": "getblockcount" - })); - } - let hold = serde_json::Value::Array(calls).to_string(); - let done = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + // Hold the only permit across an await the probe can observe. + // A fast getblockcount batch used to finish before the probe ran, + // so every status was 200. + let logpath = dir.path().join("debug.log").display().to_string(); + let (entered, release) = arm_work_queue_hold(&logpath); + let _release_on_drop = ReleaseHold(release.clone()); let holder = { let addr = tcp_addr(&handle); let auth = handle.auth.clone(); - let done = std::sync::Arc::clone(&done); tokio::spawn(async move { - let result = post_raw(addr, &auth, hold.as_bytes()).await; - done.store(true, std::sync::atomic::Ordering::SeqCst); - result + post_raw( + addr, + &auth, + br#"{"jsonrpc":"1.0","id":1,"method":"getblockcount"}"#, + ) + .await }) }; - let mut hits = Vec::new(); - while !done.load(std::sync::atomic::Ordering::SeqCst) { - let (st, _) = post_raw( - tcp_addr(&handle), - &handle.auth, - br#"{"jsonrpc":"1.0","id":1,"method":"getblockcount"}"#, - ) - .await; - hits.push(st); - if hits.len() > 32 { - break; - } + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + while !entered.load(Ordering::SeqCst) { + assert!( + std::time::Instant::now() < deadline, + "handler never took the work-queue permit" + ); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; } - let (hold_st, _) = holder.await.unwrap(); - hits.push(hold_st); - assert!( - hits.contains(&503), - "full permit must HTTP 503, got {hits:?}" - ); - assert!( - hits.contains(&200), - "some occupancy must still succeed, got {hits:?}" - ); + let (st, body) = post_raw( + tcp_addr(&handle), + &handle.auth, + br#"{"jsonrpc":"1.0","id":2,"method":"getblockcount"}"#, + ) + .await; + assert_eq!(st, 503, "full permit must HTTP 503, got {st} {body:?}"); + release.send(true).expect("hold release"); + let (hold_st, hold_body) = holder.await.unwrap(); + assert_eq!(hold_st, 200, "held call still completes: {hold_body:?}"); + clear_work_queue_hold(&logpath); handle.shutdown().await; let _ = std::fs::remove_dir_all(&dir); } diff --git a/crates/rbitcoin-store/src/store.rs b/crates/rbitcoin-store/src/store.rs index 8ed227544..92a993a01 100644 --- a/crates/rbitcoin-store/src/store.rs +++ b/crates/rbitcoin-store/src/store.rs @@ -3468,7 +3468,10 @@ mod tests { let h1_a = crate::address_head::h1_in_page(&mix_a, bits); let mut b = [0xBBu8; 32]; let mut found = false; - for i in 0u64..500_000 { + // Tiny heads take page and h1 from 16 mixed bits (1 in 65536). + // Half a million draws misses often enough to fail CI; the cap is + // many times the mean, and the loop stops at the first hit. + for i in 0u64..8_000_000 { b[24..32].copy_from_slice(&i.to_le_bytes()); let mix_b = s.txs.secret.mix_txid(&b); if crate::address_head::page_base_for_txid(&mix_b, bits) == page_a