Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 2 additions & 2 deletions crates/qbit-prism-server/fuzz/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

38 changes: 29 additions & 9 deletions crates/qbit-prism-server/src/coordinator/miner_submit.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1010,23 +1010,43 @@ 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
// this call is still acknowledging, so it lands in the newest
// 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<String>, Option<String>, Option<String>)>(
"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<String>, Option<String>, Option<String>)>(
"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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -535,18 +535,20 @@ 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;
assert!(futures_util::poll!(&mut persist).is_pending());
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",
)
Expand Down Expand Up @@ -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(())
})
})
Expand Down Expand Up @@ -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(())
})
Expand Down
22 changes: 12 additions & 10 deletions docs/prism-pool-acquire-metrics.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion scripts/prism_pool_acquires.json
Original file line number Diff line number Diff line change
Expand Up @@ -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"],
Expand Down