From 5f4e7e591342b19f764e17b63f9a36ca8d28c227 Mon Sep 17 00:00:00 2001 From: Alexander Miller Date: Mon, 26 Oct 2026 12:59:11 -0400 Subject: [PATCH 1/2] main: update tokio to 1.53.3 for RUSTSEC-2026-0386 The Dependency audit job (`cargo deny --locked check`) fails on RUSTSEC-2026-0386: tokio 1.53.1's safe `AsyncFd` constructors violate I/O safety. Move tokio to its first fixed patch release, 1.53.3, in the root and fuzz lockfiles, as #777 did for the TLS advisories. Nothing in the workspace uses `AsyncFd`. Only tokio's version and checksum change. `cargo update -p tokio --precise 1.53.3` also moved the windows-sys edges of four unrelated crates (errno, quinn-udp, rustix, tempfile) from 0.61.2 to 0.52.0, so the edit keeps those as they were; both lockfiles pass `cargo metadata --locked`. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 4 ++-- crates/qbit-prism-server/fuzz/Cargo.lock | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5a2f431f2..335785cb2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2340,9 +2340,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.53.1" +version = "1.53.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e" dependencies = [ "bytes", "libc", diff --git a/crates/qbit-prism-server/fuzz/Cargo.lock b/crates/qbit-prism-server/fuzz/Cargo.lock index e1e9102fc..5c8e7d75c 100644 --- a/crates/qbit-prism-server/fuzz/Cargo.lock +++ b/crates/qbit-prism-server/fuzz/Cargo.lock @@ -2235,9 +2235,9 @@ checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee" [[package]] name = "tokio" -version = "1.53.1" +version = "1.53.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e" dependencies = [ "bytes", "libc", From 2608df73ea1655172c5a114be8405e78b00babec Mon Sep 17 00:00:00 2001 From: Alexander Miller Date: Mon, 26 Oct 2026 12:59:25 -0400 Subject: [PATCH 2/2] main: keep the block-proof acknowledgement bound from dropping a disposition poll's connection b574_ack_cap's a_captured_proof_is_answered_at_the_cap_and_credited_after_its_landing failed on #780 with `proxy connection task failed: Connection reset by peer (os error 104)`, which the ledger execution proxy reported when the fixture closed. `deferred_share_disposition` wrapped each poll's checkout and SQL in `timeout_at(bound, ...)`. When the bound fell inside the checkout, during the pool's `before_acquire` ping of a connection idle for 100 ms or a new connection's setup, SQLx dropped the raw connection with the reply unread, and the kernel answered with a TCP reset. The test proxy reports that as a failed connection; PostgreSQL logs it as a client reset, and the pool loses a healthy connection. Each poll now runs in its own task, as the duplicate probe and enqueue do (#574), and only the wait for it is bounded. Answers, phases and `block_proof_ack_capped_total` are unchanged. A poll still running at the bound, or when the submission is cancelled, finishes on its own and returns its connection, so `database_pool_acquire_seconds` records its checkout when it ends instead of as a cancelled failure. Two acquire tests and docs/prism-pool-acquire-metrics.md now say so. A bound already past on entry, as for a #657 capture that committed in the share grace, starts no poll: the old code's poll there could only be cut off. Co-Authored-By: Claude Opus 5.5 --- .../src/coordinator/miner_submit.rs | 38 ++++++++++++++----- .../coordinator/miner_submit_acquire_tests.rs | 18 ++++++--- docs/prism-pool-acquire-metrics.md | 22 ++++++----- scripts/prism_pool_acquires.json | 2 +- 4 files changed, 55 insertions(+), 25 deletions(-) diff --git a/crates/qbit-prism-server/src/coordinator/miner_submit.rs b/crates/qbit-prism-server/src/coordinator/miner_submit.rs index 4ab0a3ad1..fdc734469 100644 --- a/crates/qbit-prism-server/src/coordinator/miner_submit.rs +++ b/crates/qbit-prism-server/src/coordinator/miner_submit.rs @@ -1010,6 +1010,12 @@ impl Coordinator { (capped, log_path): (BlockAckPath, &'static str), ) -> SaveOutcome { loop { + // A bound already past, as for a capture whose append committed + // in the share grace (#657), starts no poll: its answer would + // arrive too late to use. + if tokio::time::Instant::now() >= bound { + break; + } // Observe credit and disposition in one MVCC snapshot so // finalization cannot fall between two separate reads. The credit // this polls for is written by the reconciliation of the block @@ -1017,16 +1023,30 @@ impl Coordinator { // leaves and carries the same probe floor as the probe above: // without it, every poll of this loop would descend one per-leaf // `share_id` index per attached partition. - let poll = tokio::time::timeout_at(bound, async { - sqlx::query_as::<_, (bool, Option, Option, Option)>( - "SELECT EXISTS(SELECT 1 FROM qbit_share_ledger WHERE share_id=$1 AND share_seq>=qbit_prism_share_probe_floor()), (SELECT state FROM qbit_block_candidate_outbox WHERE block_hash=$2), (SELECT offer_outcome FROM qbit_block_candidate_outbox WHERE block_hash=$2), (SELECT last_error FROM qbit_block_candidate_outbox WHERE block_hash=$2)", - ) - .bind(share_id) - .bind(block_hash) - .fetch_one(&mut *self.ledger.acquire().await?) + // + // Each poll runs in its own task, as the probe and enqueue do, and + // only the wait for it is bounded. A bound that fell inside the + // poll's checkout, during the pool's probe of an idle connection + // or a new connection's setup, would drop that connection with + // its reply unread: the server sees a reset and the pool loses a + // healthy connection. Past the bound the poll finishes on its own + // and returns its connection. + let poll = tokio::spawn({ + let ledger = self.ledger.clone(); + let (share_id, block_hash) = (share_id.to_owned(), block_hash.to_owned()); + async move { + sqlx::query_as::<_, (bool, Option, Option, Option)>( + "SELECT EXISTS(SELECT 1 FROM qbit_share_ledger WHERE share_id=$1 AND share_seq>=qbit_prism_share_probe_floor()), (SELECT state FROM qbit_block_candidate_outbox WHERE block_hash=$2), (SELECT offer_outcome FROM qbit_block_candidate_outbox WHERE block_hash=$2), (SELECT last_error FROM qbit_block_candidate_outbox WHERE block_hash=$2)", + ) + .bind(share_id) + .bind(block_hash) + .fetch_one(&mut *ledger.acquire().await?) + .await + } + }); + let poll = tokio::time::timeout_at(bound, poll) .await - }) - .await; + .map(|joined| anyhow::Ok(joined??)); match poll { Err(_) => break, Ok(Ok((true, _, _, _))) => return SaveOutcome::Accepted, diff --git a/crates/qbit-prism-server/src/coordinator/miner_submit_acquire_tests.rs b/crates/qbit-prism-server/src/coordinator/miner_submit_acquire_tests.rs index 9626b29c8..e2010eeef 100644 --- a/crates/qbit-prism-server/src/coordinator/miner_submit_acquire_tests.rs +++ b/crates/qbit-prism-server/src/coordinator/miner_submit_acquire_tests.rs @@ -535,7 +535,6 @@ async fn block_only_poll_checkout_cancel_and_deadline_stay_unknown() -> Result<( let mut persist = Box::pin(h.persist(start)); let held = first_poll(h, &mut persist).await?; let before = counts(h.metrics()); - let sum = sample(h.metrics(), "failure", "sum"); tokio::time::pause(); let resume = ResumeClock; tokio::time::advance(Duration::from_millis(50)).await; @@ -543,10 +542,13 @@ async fn block_only_poll_checkout_cancel_and_deadline_stay_unknown() -> Result<( assert_eq!(counts(h.metrics()), before); tokio::time::advance(Duration::from_millis(75)).await; drop(persist); - assert_eq!(counts(h.metrics()), (before.0, before.1 + 1.)); - assert!((sample(h.metrics(), "failure", "sum") - sum - 0.075).abs() < 0.000001); + // The poll runs in a task the submission does not own, so a + // cancelled submission leaves its checkout waiting, not failed, + // and the poll still completes. + assert_eq!(counts(h.metrics()), before); drop(resume); drop(held); + wait_counts(h.metrics(), (before.0 + 1., before.1)).await?; let deferred: Value = sqlx::query_scalar( "SELECT share FROM qbit_prism_deferred_shares WHERE block_hash=$1", ) @@ -584,9 +586,12 @@ async fn block_only_poll_checkout_cancel_and_deadline_stay_unknown() -> Result<( .. } )); - assert_eq!(counts(h.metrics()), (before.0, before.1 + 1.)); + // The bound ends only the wait: the poll's checkout is not + // cancelled with it, and completes once the slot is free. + assert_eq!(counts(h.metrics()), before); drop(resume); drop(held); + wait_counts(h.metrics(), (before.0 + 1., before.1)).await?; Ok(()) }) }) @@ -721,9 +726,12 @@ async fn block_only_poll_missing_and_sql_error_stay_unknown() -> Result<()> { panic!("nonterminal: {answer:?}") }; assert_eq!(phase, expected_phase); - assert_eq!(counts(h.metrics()), (before.0 + 1., before.1 + 1.)); + // The poll still waiting for the slot at the bound is not + // cancelled with the answer; it checks out once released. + assert_eq!(counts(h.metrics()), (before.0 + 1., before.1)); drop(resume); drop(held); + wait_counts(h.metrics(), (before.0 + 2., before.1)).await?; drop(tokio::time::timeout(WAIT, h.ledger().pool.acquire()).await??); Ok(()) }) diff --git a/docs/prism-pool-acquire-metrics.md b/docs/prism-pool-acquire-metrics.md index 57d15be3e..13ec4bbc7 100644 --- a/docs/prism-pool-acquire-metrics.md +++ b/docs/prism-pool-acquire-metrics.md @@ -116,16 +116,18 @@ unchanged. The coordinator database clock and chain-observation state each record one checkout. The latter retains its one-statement revision/epoch/tip snapshot. Each block-only duplicate probe and each credit/disposition poll records its -own checkout, inside the -original acknowledgement deadline. The two block-only probes are not issued-job -reads. Credit and disposition still share one SQL statement and MVCC snapshot. -Connections are released at the statement boundary, before enqueue or poll -sleep. Missing rows, duplicates and terminal dispositions retain successful -checkout observations; SQL errors or cancellation after acquisition do too. -Cancellation while acquiring records a failure. Before enqueue a read failure -is definite; after enqueue a failed or missing read does not prove whether the -durable candidate earned credit. Timing changes neither that distinction nor -the enqueue's existing handling of an unknown outcome. +own checkout. Each runs in a task the acknowledgement does not own (#574), so +neither the acknowledgement deadline nor a cancelled submission cancels its +checkout or SQL: a checkout still waiting then records its own outcome when it +ends, and a poll that answers after the deadline is discarded. The two +block-only probes are not issued-job reads. Credit and disposition still share +one SQL statement and MVCC snapshot. Connections are released at the statement +boundary, before enqueue or poll sleep. Missing rows, duplicates and terminal +dispositions retain successful checkout observations; SQL errors do too. +Before enqueue a read failure is definite; after enqueue a failed or missing +read does not prove whether the durable candidate earned credit. Timing changes +neither that distinction nor the enqueue's existing handling of an unknown +outcome. The server passes its existing metrics handle to the rollup loop. Each valid rollup attempt records one checkout before `BEGIN`; `SET LOCAL`, the single diff --git a/scripts/prism_pool_acquires.json b/scripts/prism_pool_acquires.json index ce610a734..b94a03d6b 100644 --- a/scripts/prism_pool_acquires.json +++ b/scripts/prism_pool_acquires.json @@ -142,7 +142,7 @@ "coordinator.rs::submit_loop_with#1::attempt .run(& self , lease)": [1,"non-database"], "coordinator.rs::with_tracked_heartbeat#1::self . ledger .acquire()": [2,"ledger"], "coordinator/issued_batcher.rs::with_dwell#1::run(ledger , receiver , shutdown . clone ( ) , dwell)": [1,"non-database"], - "coordinator/miner_submit.rs::deferred_share_disposition#1::self . ledger .acquire()": [1,"ledger"], + "coordinator/miner_submit.rs::deferred_share_disposition#1::ledger .acquire()": [1,"ledger"], "coordinator/miner_submit.rs::persist_block_only#1::ledger .acquire()": [1,"ledger"], "coordinator/refresh_window.rs::capture_refresh_window#1::prepare(std :: mem :: take ( & mut snapshot . snapshot . prior_balances ) , & admission ,)": [1,"non-database"], "coordinator/work_ledger.rs::clocked_payout_revision#1::self .acquire()": [1,"ledger"],