diff --git a/COMPAT.md b/COMPAT.md index a72415040..161f24730 100644 --- a/COMPAT.md +++ b/COMPAT.md @@ -72,7 +72,7 @@ Full `/tx/:txid` JSON still has `vin[]`. Electrum has no outspend-vin surface. | Package submit | RPC `submitpackage` / Esplora `POST /txs/package` (no P2P package command) | BIP331 wire | | Pruning / GUI | Unpruned: `seqsigwit.body`. `--prune-seqsigwit`: watermark + `NETWORK_LIMITED`, kept witness is 288 height files under `store/seqsigwit.window/` plus a RAM cap (`--prune-seqsigwit-ram-threshold-bytes`, `0` = files only). Not a rolling stem. Not Core `-prune` of headers/txout | Supported | | Mining template RPC | `getblocktemplate` / `getmininginfo` / `prioritisetransaction` (selector; no stratum) | GBT + stratum / pool stack | -| SV2 Template Distribution | In-process TDP v2 server over Noise NX (`--sv2-tp-listen`, default off): `NewTemplate` / `SetNewPrevHash` per tip, `NewTemplate` on a same-tip fee gain (`--sv2-tp-fee-delta`, `--sv2-tp-template-interval`, sv2-tp defaults), `RequestTransactionData`, `SubmitSolution` → block accept. No templates during IBD; every template on the tip kept, up to 64 per session. No plaintext, no pool / JD roles | No TDP in Core; sv2-apps `bitcoin-core-sv2` bridges Core's IPC mining interface to TDP | +| SV2 Template Distribution | In-process TDP v2 server over Noise NX (`--sv2-tp-listen`, default off): `NewTemplate` / `SetNewPrevHash` per tip, `NewTemplate` on a same-tip fee gain (`--sv2-tp-fee-delta`, `--sv2-tp-template-interval`, sv2-tp defaults), `RequestTransactionData`, `SubmitSolution` → block accept; `ProposeTemplate` for a Job Declarator Server's node backend (proposed TDP messages, sv2-spec discussion #239, `SetupConnection` flag bit 0). No templates during IBD; every template on the tip kept, up to 64 per session. No plaintext, no pool / JD roles | No TDP in Core; sv2-apps `bitcoin-core-sv2` bridges Core's IPC mining interface to TDP | | Wallets | Electrum clients (requires `--sh-index`) | Descriptor + legacy | | Scripthash index | Optional (`--sh-index`, default **off**); bulk at tip when on | External ElectrumX / Fulcrum; Core `-txindex` is different (txid→block) | | JSON-RPC | Documented **subset** ([`docs/rpc.md`](./docs/rpc.md)); Bearer token or opt-in Core cookie (`--rpc-cookie-file`), no `--rpcuser` / `--rpcpassword`; `rbitcoin-cli`. `--rpc-work-queue` defaults to **16** (HTTP occupancy, 503 when full; **0** is that default); a JSON-RPC array is one POST | Full Core RPC; `-rpcworkqueue` is in-flight HTTP jobs (503) | diff --git a/changelog.d/sv2-job-validation.md b/changelog.d/sv2-job-validation.md new file mode 100644 index 000000000..9f9725d9d --- /dev/null +++ b/changelog.d/sv2-job-validation.md @@ -0,0 +1,20 @@ +Added + +- **SV2 TP: custom job validation for Job Declarator Servers.** A TDP + client that sets `SetupConnection` flag bit 0 (`REQUIRES_JOB_VALIDATION`) + can send `ProposeTemplate` (proposed TDP messages 0x77–0x7b, sv2-spec + discussion #239) with the `DeclareMiningJob` fields relayed unchanged. + The node rebuilds the placeholder coinbase from the declared prefix and + suffix, resolves the declared wtxids against its mempool, asks for the + ones it lacks with `ProvideMissingTransactions` and takes them back in + `ProvideMissingTransactions.Success` (the Job Declaration Protocol's own + shapes, so a JDS relays both unchanged; the proposal is held up to 30 s, + at most eight per connection), checks the job as a block on its tip, and + answers that tip, + the fee total of the declared transactions, and a template id that + `SubmitSolution` accepts like one of the node's own templates; the job is + retained like one too (same ring, same stale grace). Duplicated, + undeclared, or malformed input is refused before anything is decoded. + Validation runs off the session loop, up to four at once per + connection, so a `SubmitSolution` or a tip push never waits behind + another client's proposal. diff --git a/crates/rbitcoin-node/src/run.rs b/crates/rbitcoin-node/src/run.rs index 941dce14a..23be4b0de 100644 --- a/crates/rbitcoin-node/src/run.rs +++ b/crates/rbitcoin-node/src/run.rs @@ -1956,6 +1956,7 @@ async fn start_sv2_tp(config: &NodeConfig, hub: &Arc) -> Option), + /// Consensus-valid on the tip: the fee total and the job to retain. + Valid { fees: u64, job: Job }, + /// `ProposeTemplate.Error.error_code`: a Core reject string or one of + /// the draft's own codes. + Rejected(String), +} + +/// A `ProposeTemplate` as the session keeps it: owned, so it outlives its +/// frame while the TP waits for the transactions it asked for. `excess_data` +/// is not kept (opaque to the TP). +pub(crate) struct Proposal { + pub request_id: u32, + version: u32, + coinbase_prefix: Vec, + coinbase_suffix: Vec, + wtxids: Vec<[u8; 32]>, +} + +/// Supplied transactions keyed by the wtxid of their bytes as sent. +pub(crate) type Supplied = HashMap<[u8; 32], Vec>; + +impl Proposal { + /// `None`: the payload is not a `ProposeTemplate`. + pub(crate) fn decode(payload: &mut [u8]) -> Option { + let m: ProposeTemplate = binary_sv2::from_bytes(payload).ok()?; + Some(Self { + request_id: m.request_id, + version: m.version, + coinbase_prefix: m.coinbase_tx_prefix.as_ref().to_vec(), + coinbase_suffix: m.coinbase_tx_suffix.as_ref().to_vec(), + wtxids: m + .wtxid_list + .iter() + .map(|w| w.as_ref().try_into().expect("U256 is 32 bytes")) + .collect(), + }) + } + + /// §4.1: `wtxid_list` has no duplicates. No chain read, so the session + /// runs it on arrival; [`validate`] runs it again and stays complete on + /// its own. + pub(crate) fn precheck(&self) -> Result<(), &'static str> { + let mut declared = HashSet::with_capacity(self.wtxids.len()); + if self.wtxids.iter().all(|w| declared.insert(w)) { + Ok(()) + } else { + Err("duplicate-wtxid") + } + } + + /// §4.3: every supplied transaction hashes to a wtxid at one of the + /// positions the TP asked for, and every asked position is covered, + /// before any of them is decoded or copied, so a transaction nobody + /// asked for costs one hash and a short provide costs none. `Err` is + /// the `Error.error_code`. + pub(crate) fn accept_supplied( + &self, + missing: &[u16], + transaction_list: &Seq064K<'_, B016M<'_>>, + ) -> Result { + let asked: HashSet<&[u8; 32]> = missing + .iter() + .filter_map(|&pos| self.wtxids.get(usize::from(pos))) + .collect(); + let mut supplied = Supplied::with_capacity(transaction_list.len()); + for raw in transaction_list.iter() { + let w = sha256d::Hash::hash(raw.as_ref()).to_byte_array(); + if !asked.contains(&w) { + return Err("bad-missing-tx"); + } + supplied.insert(w, raw.as_ref().to_vec()); + } + if supplied.len() != asked.len() { + return Err("bad-missing-tx"); + } + Ok(supplied) + } +} + +/// Reads the store and the mempool: blocking region only. +pub(crate) fn validate(chain: &ChainHub, p: &Proposal, supplied: &Supplied) -> io::Result { + let rejected = |code: &str| Ok(Verdict::Rejected(code.into())); + // Same gate as the templates: a stale tip validates nothing. + if chain.in_ibd() { + return rejected("job-validation-unavailable"); + } + let next = template::next_header(chain)?; + if let Err(code) = p.precheck() { + return rejected(code); + } + let Some(extranonce) = extranonce_len(&p.coinbase_prefix) else { + return rejected("bad-cb-decode"); + }; + let zeros = vec![0u8; extranonce]; + let raw = [&p.coinbase_prefix[..], &zeros[..], &p.coinbase_suffix[..]].concat(); + let Ok(coinbase) = bitcoin::consensus::deserialize::(&raw) else { + return rejected("bad-cb-decode"); + }; + let mut txs = Vec::with_capacity(p.wtxids.len()); + let mut missing = Vec::new(); + let mut provided = Vec::new(); + for (pos, w) in p.wtxids.iter().enumerate() { + let pos = u16::try_from(pos).expect("Seq064K holds at most 65535"); + let tx = match supplied.get(w) { + Some(raw) => match bitcoin::consensus::deserialize::(raw) { + Ok(tx) => { + provided.push(pos); + Some(tx) + } + Err(_) => return rejected("bad-missing-tx"), + }, + None => chain + .mempool() + .and_then(|mp| mp.get_tx_by_wtxid(&Wtxid::from_byte_array(*w))), + }; + match tx { + Some(tx) => txs.push(tx), + None => missing.push(pos), + } + } + if !missing.is_empty() { + missing.extend(provided); + missing.sort_unstable(); + return Ok(Verdict::Missing(missing)); + } + let mut leaves = Vec::with_capacity(1 + txs.len()); + leaves.push(coinbase.compute_txid().to_byte_array()); + leaves.extend(txs.iter().map(|tx| tx.compute_txid().to_byte_array())); + let header = block::Header { + version: block::Version::from_consensus(p.version as i32), + prev_blockhash: BlockHash::from_byte_array(next.prev_hash), + merkle_root: TxMerkleNode::from_byte_array(rbitcoin_store::merkle_root_from_txids(&leaves)), + time: next.time, + bits: next.bits, + nonce: 0, + }; + let mut txdata = Vec::with_capacity(1 + txs.len()); + txdata.push(coinbase); + txdata.extend(txs); + let mut block = Block { header, txdata }; + // CPU trade (CONTRIBUTING 9): one full proposal check per request + // (every spend against the chain, structure, weight, sigops, coinbase + // value, scripts; no PoW), on the blocking pool. A JDS sends one per + // declaration; a flood costs blocking threads, not the reactor. + let fees = match chain.check_block_proposal(&block) { + Ok(fees) => fees, + Err(code) => return Ok(Verdict::Rejected(code)), + }; + let txs = block + .txdata + .split_off(1) + .into_iter() + .map(Arc::new) + .collect(); + Ok(Verdict::Valid { + fees, + job: next.job(rbitcoin_store::merkle_branch(&leaves, 0), txs), + }) +} + +/// §4.1: `coinbase_tx_prefix` ends inside the scriptSig and +/// `coinbase_tx_suffix` starts at nSequence, so the extranonce is the +/// scriptSig length the prefix declares minus the scriptSig bytes it +/// carries. `None`: not one input, a length outside the coinbase bounds +/// (2..=100), or fewer bytes declared than present. +fn extranonce_len(prefix: &[u8]) -> Option { + let mut at = 4; + // BIP144: a zero where the input count would be, then flag 1. + if prefix.get(4..6) == Some(&[0u8, 1][..]) { + at += 2; + } + let (inputs, n) = compact_size(prefix.get(at..)?)?; + if inputs != 1 { + return None; + } + at += n + 36; + let (len, n) = compact_size(prefix.get(at..)?)?; + at += n; + if !(2..=100).contains(&len) { + return None; + } + (len as usize).checked_sub(prefix.len() - at) +} + +fn compact_size(bytes: &[u8]) -> Option<(u64, usize)> { + let (v, n) = deserialize_partial::(bytes).ok()?; + Some((v.0, n)) +} + +#[cfg(test)] +mod tests { + use super::extranonce_len; + + fn prefix(segwit: bool, inputs: u8, len: u8, present: usize) -> Vec { + let mut p = vec![2, 0, 0, 0]; + if segwit { + p.extend([0, 1]); + } + p.push(inputs); + p.extend([0; 32]); + p.extend([0xff; 4]); + p.push(len); + p.resize(p.len() + present, 0x51); + p + } + + #[test] + fn extranonce_is_the_declared_script_sig_length_past_the_prefix() { + assert_eq!(extranonce_len(&prefix(false, 1, 11, 3)), Some(8)); + assert_eq!(extranonce_len(&prefix(true, 1, 11, 3)), Some(8)); + assert_eq!(extranonce_len(&prefix(true, 1, 3, 3)), Some(0)); + assert_eq!(extranonce_len(&prefix(true, 1, 100, 0)), Some(100)); + assert_eq!(extranonce_len(&prefix(true, 2, 11, 3)), None, "two inputs"); + assert_eq!( + extranonce_len(&prefix(true, 1, 2, 3)), + None, + "shorter than present" + ); + assert_eq!( + extranonce_len(&prefix(true, 1, 101, 0)), + None, + "over the coinbase max" + ); + assert_eq!( + extranonce_len(&prefix(true, 1, 1, 0)), + None, + "under the coinbase min" + ); + assert_eq!( + extranonce_len(&prefix(true, 1, 11, 3)[..40]), + None, + "truncated" + ); + assert_eq!(extranonce_len(&[]), None); + } +} diff --git a/crates/rbitcoin-sv2/src/lib.rs b/crates/rbitcoin-sv2/src/lib.rs index 6197509c3..84c1f810c 100644 --- a/crates/rbitcoin-sv2/src/lib.rs +++ b/crates/rbitcoin-sv2/src/lib.rs @@ -3,6 +3,8 @@ //! Noise_NX over TCP is the only transport. Plan and constraints: //! `docs/sv2-template-provider.md`. +mod job; +mod messages; mod session; mod template; pub mod testutil; @@ -35,6 +37,10 @@ pub const SETUP_TIMEOUT: Duration = Duration::from_secs(10); /// Default [`Sv2TpConfig::write_timeout`]. pub const WRITE_TIMEOUT: Duration = Duration::from_secs(30); +/// Default [`Sv2TpConfig::provide_timeout`]: a JDS relays the request to +/// its JDC and the answer back, so the wait is two hops over the internet. +pub const PROVIDE_TIMEOUT: Duration = Duration::from_secs(30); + /// Default [`Sv2TpConfig::fee_delta`] (stratum-mining `sv2-tp` /// `-sv2feedelta`). pub const FEE_DELTA: u64 = 1000; @@ -74,6 +80,11 @@ pub struct Sv2TpConfig { /// A socket write that makes no progress this long closes the session, /// so a client that stops reading cannot hold a slot. pub write_timeout: Duration, + /// How long a `ProposeTemplate` the node answered + /// `ProvideMissingTransactions` waits for the + /// `ProvideMissingTransactions.Success`; a later one is + /// `unknown-request-id`. + pub provide_timeout: Duration, /// With the tip unchanged, a rebuild is pushed only when its fees are at /// least this many sats above the session's last template. pub fee_delta: u64, @@ -167,6 +178,7 @@ pub async fn run_sv2_tp(config: Sv2TpConfig) -> io::Result { let stale_grace = config.stale_grace; let setup_timeout = config.setup_timeout; let write_timeout = config.write_timeout; + let provide_timeout = config.provide_timeout; let fee_push = session::FeePush { delta: config.fee_delta, interval: config.template_interval, @@ -224,6 +236,7 @@ pub async fn run_sv2_tp(config: Sv2TpConfig) -> io::Result { stale_grace, setup_timeout, write_timeout, + provide_timeout, fee_push, stats, ) diff --git a/crates/rbitcoin-sv2/src/listener_tests.rs b/crates/rbitcoin-sv2/src/listener_tests.rs index d524700a1..b5c12de97 100644 --- a/crates/rbitcoin-sv2/src/listener_tests.rs +++ b/crates/rbitcoin-sv2/src/listener_tests.rs @@ -1,8 +1,13 @@ +use crate::messages::{ + MESSAGE_TYPE_PROPOSE_TEMPLATE, MESSAGE_TYPE_PROVIDE_MISSING_TRANSACTIONS_SUCCESS, + REQUIRES_JOB_VALIDATION, +}; use crate::test_chain::shared_regtest; use crate::testutil::TpClient; +use crate::transport::MAX_PROPOSE_TEMPLATE_PAYLOAD; use crate::{ run_sv2_tp, Sv2TpConfig, FEE_DELTA, MAX_SESSIONS, MAX_STALE_GRACE, MAX_TEMPLATE_INTERVAL, - MIN_TEMPLATE_INTERVAL, SETUP_TIMEOUT, TEMPLATE_INTERVAL, WRITE_TIMEOUT, + MIN_TEMPLATE_INTERVAL, PROVIDE_TIMEOUT, SETUP_TIMEOUT, TEMPLATE_INTERVAL, WRITE_TIMEOUT, }; use common_messages_sv2::{ SetupConnectionError, SetupConnectionSuccess, MESSAGE_TYPE_SETUP_CONNECTION_ERROR, @@ -12,7 +17,9 @@ use std::io; use std::net::SocketAddr; use std::sync::Arc; use std::time::{Duration, Instant}; -use template_distribution_sv2::MESSAGE_TYPE_REQUEST_TRANSACTION_DATA_ERROR; +use template_distribution_sv2::{ + MESSAGE_TYPE_REQUEST_TRANSACTION_DATA_ERROR, MESSAGE_TYPE_SUBMIT_SOLUTION, +}; const TDP: u8 = 2; @@ -33,13 +40,24 @@ async fn expect_error(c: &mut TpClient, flags: u32, code: &str) { let e: SetupConnectionError = binary_sv2::from_bytes(&mut f.payload).expect("decode"); assert_eq!(e.flags, flags); assert_eq!(e.error_code.as_utf8_or_hex(), code); + assert_closed(c, "SetupConnection.Error").await; +} + +async fn assert_closed(c: &mut TpClient, what: &str) { let closed = tokio::time::timeout(Duration::from_secs(5), c.recv()).await; assert!( matches!(closed, Ok(Err(_))), - "connection must close after SetupConnection.Error" + "{what} must close the session, got {:?}", + closed.map(|r| r.map(|f| f.msg_type)) ); } +async fn expect_success(c: &mut TpClient) -> SetupConnectionSuccess { + let mut f = c.recv().await.expect("setup reply"); + assert_eq!(f.msg_type, MESSAGE_TYPE_SETUP_CONNECTION_SUCCESS); + binary_sv2::from_bytes(&mut f.payload).expect("decode") +} + #[tokio::test] async fn setup_connection_success_errors_and_session_cap() { let tc = shared_regtest(0); @@ -52,6 +70,7 @@ async fn setup_connection_success_errors_and_session_cap() { stale_grace: Duration::from_secs(10), setup_timeout: SETUP_TIMEOUT, write_timeout: WRITE_TIMEOUT, + provide_timeout: PROVIDE_TIMEOUT, fee_delta: FEE_DELTA, template_interval: TEMPLATE_INTERVAL, }) @@ -63,9 +82,7 @@ async fn setup_connection_success_errors_and_session_cap() { for _ in 0..MAX_SESSIONS { let mut c = TpClient::connect(addr, pk).await.expect("handshake"); c.setup_connection(TDP, 2, 2, 0).await.unwrap(); - let mut f = c.recv().await.expect("setup reply"); - assert_eq!(f.msg_type, MESSAGE_TYPE_SETUP_CONNECTION_SUCCESS); - let ok: SetupConnectionSuccess = binary_sv2::from_bytes(&mut f.payload).expect("decode"); + let ok = expect_success(&mut c).await; assert_eq!((ok.used_version, ok.flags), (2, 0)); live.push(c); } @@ -84,6 +101,16 @@ async fn setup_connection_success_errors_and_session_cap() { } drop(live); + let mut c = connect_when_free(addr, pk).await; + c.setup_connection(TDP, 2, 2, REQUIRES_JOB_VALIDATION) + .await + .unwrap(); + assert_eq!( + expect_success(&mut c).await.flags, + REQUIRES_JOB_VALIDATION, + "REQUIRES_JOB_VALIDATION is accepted and echoed" + ); + let mut c = connect_when_free(addr, pk).await; c.setup_connection(TDP, 2, 2, 0b101).await.unwrap(); expect_error(&mut c, 0b101, "unsupported-feature-flags").await; @@ -115,6 +142,7 @@ async fn authority_key_prints_in_key_utils_base58check() { stale_grace: Duration::from_secs(10), setup_timeout: SETUP_TIMEOUT, write_timeout: WRITE_TIMEOUT, + provide_timeout: PROVIDE_TIMEOUT, fee_delta: FEE_DELTA, template_interval: TEMPLATE_INTERVAL, }) @@ -129,9 +157,7 @@ async fn authority_key_prints_in_key_utils_base58check() { .await .expect("handshake against the printed key"); c.setup_connection(TDP, 2, 2, 0).await.unwrap(); - let mut f = c.recv().await.expect("setup reply"); - assert_eq!(f.msg_type, MESSAGE_TYPE_SETUP_CONNECTION_SUCCESS); - let _: SetupConnectionSuccess = binary_sv2::from_bytes(&mut f.payload).expect("decode"); + expect_success(&mut c).await; tp.shutdown().await; } @@ -152,6 +178,7 @@ async fn silent_sockets_are_dropped_at_the_setup_deadline() { stale_grace: Duration::from_secs(10), setup_timeout, write_timeout: WRITE_TIMEOUT, + provide_timeout: PROVIDE_TIMEOUT, fee_delta: FEE_DELTA, template_interval: TEMPLATE_INTERVAL, }) @@ -174,8 +201,7 @@ async fn silent_sockets_are_dropped_at_the_setup_deadline() { let mut c = connect_when_free(addr, pk).await; c.setup_connection(TDP, 2, 2, 0).await.unwrap(); - let f = c.recv().await.expect("setup reply"); - assert_eq!(f.msg_type, MESSAGE_TYPE_SETUP_CONNECTION_SUCCESS); + expect_success(&mut c).await; tp.shutdown().await; } @@ -195,6 +221,7 @@ async fn session_without_constraints_is_dropped_at_the_setup_deadline() { stale_grace: Duration::from_secs(10), setup_timeout, write_timeout: WRITE_TIMEOUT, + provide_timeout: PROVIDE_TIMEOUT, fee_delta: FEE_DELTA, template_interval: TEMPLATE_INTERVAL, }) @@ -252,6 +279,7 @@ async fn client_that_stops_reading_is_dropped_at_the_write_deadline() { stale_grace: Duration::from_secs(10), setup_timeout: SETUP_TIMEOUT, write_timeout, + provide_timeout: PROVIDE_TIMEOUT, fee_delta: FEE_DELTA, template_interval: TEMPLATE_INTERVAL, }) @@ -301,8 +329,10 @@ async fn client_that_stops_reading_is_dropped_at_the_write_deadline() { tp.shutdown().await; } -/// The largest legitimate client frame (a `SubmitSolution` with a full -/// `B064K` coinbase) keeps the session; a larger frame closes it. +/// Each client message type has its own payload cap: the largest legitimate +/// `SubmitSolution`, a `ProposeTemplate` well past it, and a +/// `ProvideMissingTransactions.Success` past the `ProposeTemplate` cap keep +/// the session; a frame over the cap for its type closes it. #[tokio::test] async fn oversized_client_frame_closes_the_session() { let tc = shared_regtest(0); @@ -315,6 +345,7 @@ async fn oversized_client_frame_closes_the_session() { stale_grace: Duration::from_secs(10), setup_timeout: SETUP_TIMEOUT, write_timeout: WRITE_TIMEOUT, + provide_timeout: PROVIDE_TIMEOUT, fee_delta: FEE_DELTA, template_interval: TEMPLATE_INTERVAL, }) @@ -335,14 +366,41 @@ async fn oversized_client_frame_closes_the_session() { .expect("open after a max-size SubmitSolution"); assert_eq!(f.msg_type, MESSAGE_TYPE_REQUEST_TRANSACTION_DATA_ERROR); - let _ = c.send_bytes(0xff, &vec![0u8; 1 << 20]).await; + // Without the flag a ProposeTemplate under its own cap is read and + // ignored. + let big = vec![0u8; 200 << 10]; + c.send_bytes(MESSAGE_TYPE_PROPOSE_TEMPLATE, &big) + .await + .unwrap(); + c.request_transaction_data(1).await.unwrap(); + let f = c + .recv() + .await + .expect("open after a 200 KiB ProposeTemplate"); + assert_eq!(f.msg_type, MESSAGE_TYPE_REQUEST_TRANSACTION_DATA_ERROR); + + let _ = c.send_bytes(MESSAGE_TYPE_SUBMIT_SOLUTION, &big).await; c.request_transaction_data(1).await.ok(); - let closed = tokio::time::timeout(Duration::from_secs(5), c.recv()).await; - assert!( - matches!(closed, Ok(Err(_))), - "an oversized frame must close the session, got {:?}", - closed.map(|r| r.map(|f| f.msg_type)) - ); + assert_closed(&mut c, "a 200 KiB SubmitSolution").await; + + let mut c = TpClient::connect(tp.local_addr, tp.authority_pubkey) + .await + .expect("handshake"); + c.setup_connection(TDP, 2, 2, 0).await.unwrap(); + c.recv().await.expect("setup reply"); + let over = vec![0u8; MAX_PROPOSE_TEMPLATE_PAYLOAD + 1]; + c.send_bytes(MESSAGE_TYPE_PROVIDE_MISSING_TRANSACTIONS_SUCCESS, &over) + .await + .unwrap(); + c.request_transaction_data(1).await.unwrap(); + let f = c + .recv() + .await + .expect("open after a ProvideMissingTransactions.Success past the ProposeTemplate cap"); + assert_eq!(f.msg_type, MESSAGE_TYPE_REQUEST_TRANSACTION_DATA_ERROR); + let _ = c.send_bytes(MESSAGE_TYPE_PROPOSE_TEMPLATE, &over).await; + c.request_transaction_data(1).await.ok(); + assert_closed(&mut c, "a ProposeTemplate over its cap").await; tp.shutdown().await; } @@ -375,6 +433,7 @@ async fn out_of_range_timing_refuses_to_start() { stale_grace, setup_timeout: SETUP_TIMEOUT, write_timeout: WRITE_TIMEOUT, + provide_timeout: PROVIDE_TIMEOUT, fee_delta: FEE_DELTA, template_interval, }) diff --git a/crates/rbitcoin-sv2/src/messages.rs b/crates/rbitcoin-sv2/src/messages.rs new file mode 100644 index 000000000..59a006ff5 --- /dev/null +++ b/crates/rbitcoin-sv2/src/messages.rs @@ -0,0 +1,145 @@ +//! Job-validation extension to TDP (`docs/sv2-job-validation.md` §3–5): +//! the `SetupConnection` flag, the `ProposeTemplate` request and its two +//! replies, and the `ProvideMissingTransactions` pair a Job Declarator +//! Server relays between this TP and its JDC. + +use binary_sv2::{Deserialize, Seq064K, Serialize, Str0255, B016M, B064K, U256}; + +/// `SetupConnection.flags` bit 0: the client intends to send +/// [`ProposeTemplate`]. The only TDP flag this TP accepts. +pub(crate) const REQUIRES_JOB_VALIDATION: u32 = 1 << 0; + +pub(crate) const MESSAGE_TYPE_PROPOSE_TEMPLATE: u8 = 0x77; +pub(crate) const MESSAGE_TYPE_PROVIDE_MISSING_TRANSACTIONS: u8 = 0x78; +pub(crate) const MESSAGE_TYPE_PROPOSE_TEMPLATE_SUCCESS: u8 = 0x79; +pub(crate) const MESSAGE_TYPE_PROPOSE_TEMPLATE_ERROR: u8 = 0x7a; +pub(crate) const MESSAGE_TYPE_PROVIDE_MISSING_TRANSACTIONS_SUCCESS: u8 = 0x7b; + +/// Client → TP: is this custom job a consensus-valid block on the TP tip? +/// The `DeclareMiningJob` subset a node needs, relayed unchanged: the +/// coinbase split around the extranonce and the declared wtxids. Carries no +/// transactions; the TP asks for the ones it lacks with +/// [`ProvideMissingTransactions`]. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub(crate) struct ProposeTemplate<'decoder> { + pub request_id: u32, + pub version: u32, + pub coinbase_tx_prefix: B064K<'decoder>, + pub coinbase_tx_suffix: B064K<'decoder>, + pub wtxid_list: Seq064K<'decoder, U256<'decoder>>, + /// Opaque to the TP (TDP 7.6 semantics belong to the Pool). + pub excess_data: B064K<'decoder>, +} + +/// TP → client: 0-indexed positions in `wtxid_list` the TP cannot resolve. +/// The layout of JDP 6.4.7, so a JDS copies the payload to its JDC. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub(crate) struct ProvideMissingTransactions<'decoder> { + pub request_id: u32, + pub unknown_tx_position_list: Seq064K<'decoder, u16>, +} + +/// Client → TP: the transactions a [`ProvideMissingTransactions`] asked +/// for, in that order. The layout of JDP 6.4.8, copied from the JDC. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub(crate) struct ProvideMissingTransactionsSuccess<'decoder> { + pub request_id: u32, + pub transaction_list: Seq064K<'decoder, B016M<'decoder>>, +} + +/// TP → client: the job is valid on the tip `prev_hash` and retained under +/// `template_id`, which shares the `NewTemplate.template_id` namespace. +/// `fees` is the fee total of the declared transactions as the node's +/// validation computed it. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub(crate) struct ProposeTemplateSuccess<'decoder> { + pub request_id: u32, + pub template_id: u64, + pub prev_hash: U256<'decoder>, + pub fees: u64, +} + +/// TP → client: the job was not validated. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub(crate) struct ProposeTemplateError<'decoder> { + pub request_id: u32, + pub error_code: Str0255<'decoder>, + pub error_details: B064K<'decoder>, +} + +#[cfg(test)] +mod tests { + use super::*; + use binary_sv2::{Seq064K, Str0255, B016M, B064K, U256}; + use std::fmt::Debug; + + /// Encode with the codec `NoiseWriter::send` frames and decode with the + /// one `Session::on_frame` uses. + fn round_trip<'a, T>(msg: T, bytes: &'a mut Vec) + where + T: Clone + + PartialEq + + Debug + + binary_sv2::Serialize + + binary_sv2::GetSize + + binary_sv2::Deserialize<'a>, + { + *bytes = binary_sv2::to_bytes(msg.clone()).expect("encode"); + let back: T = binary_sv2::from_bytes(bytes).expect("decode"); + assert_eq!(back, msg); + } + + #[test] + fn job_validation_messages_round_trip() { + let (h1, h2, h3) = ([1u8; 32], [2u8; 32], [3u8; 32]); + let (prefix, suffix, tx, details) = + ([0xc0u8; 50], [0xc1u8; 100], [0xeeu8; 300], [1u8, 2, 3]); + let mut bytes = Vec::new(); + round_trip( + ProposeTemplate { + request_id: 7, + version: 0x2000_0000, + coinbase_tx_prefix: B064K::try_from(&prefix[..]).unwrap(), + coinbase_tx_suffix: B064K::try_from(&suffix[..]).unwrap(), + wtxid_list: Seq064K::new(vec![U256::from(&h2), U256::from(&h3)]).unwrap(), + excess_data: B064K::try_from(&details[..]).unwrap(), + }, + &mut bytes, + ); + let mut bytes = Vec::new(); + round_trip( + ProvideMissingTransactions { + request_id: 7, + unknown_tx_position_list: Seq064K::new(vec![0u16, 5, u16::MAX]).unwrap(), + }, + &mut bytes, + ); + let mut bytes = Vec::new(); + round_trip( + ProvideMissingTransactionsSuccess { + request_id: 7, + transaction_list: Seq064K::new(vec![B016M::try_from(&tx[..]).unwrap()]).unwrap(), + }, + &mut bytes, + ); + let mut bytes = Vec::new(); + round_trip( + ProposeTemplateSuccess { + request_id: 7, + template_id: u64::MAX, + prev_hash: U256::from(&h1), + fees: 12_345, + }, + &mut bytes, + ); + let mut bytes = Vec::new(); + round_trip( + ProposeTemplateError { + request_id: 7, + error_code: Str0255::try_from("bad-cb-decode").unwrap(), + error_details: B064K::try_from(&details[..]).unwrap(), + }, + &mut bytes, + ); + } +} diff --git a/crates/rbitcoin-sv2/src/session.rs b/crates/rbitcoin-sv2/src/session.rs index c9c00c236..d5085f1bd 100644 --- a/crates/rbitcoin-sv2/src/session.rs +++ b/crates/rbitcoin-sv2/src/session.rs @@ -1,9 +1,17 @@ //! One TDP session: Noise handshake, `SetupConnection`, then TDP messages. +use crate::job::{self, Proposal, Supplied, Verdict}; +use crate::messages::{ + ProposeTemplateError, ProposeTemplateSuccess, ProvideMissingTransactions, + ProvideMissingTransactionsSuccess, MESSAGE_TYPE_PROPOSE_TEMPLATE, + MESSAGE_TYPE_PROPOSE_TEMPLATE_ERROR, MESSAGE_TYPE_PROPOSE_TEMPLATE_SUCCESS, + MESSAGE_TYPE_PROVIDE_MISSING_TRANSACTIONS, MESSAGE_TYPE_PROVIDE_MISSING_TRANSACTIONS_SUCCESS, + REQUIRES_JOB_VALIDATION, +}; use crate::template; use crate::transport::{Frame, NoiseConn, NoiseWriter}; use crate::Sv2TpStats; -use binary_sv2::{Seq064K, Str0255, B016M, B064K}; +use binary_sv2::{Seq064K, Str0255, B016M, B064K, U256}; use bitcoin::hashes::Hash; use bitcoin::{block, Block, BlockHash, CompactTarget, Transaction, TxMerkleNode, Witness}; use common_messages_sv2::{ @@ -56,6 +64,25 @@ const CONSTRAINTS_COOLDOWN: Duration = Duration::from_secs(1); /// session is closed so the slot goes back to a real client. const MAX_SUPERSEDED_CONSTRAINTS: u32 = 8; +/// CPU and RAM trade (CONTRIBUTING 9): this many `ProposeTemplate` +/// validations run at once per session, each one full proposal check on a +/// blocking thread holding up to a block of decoded transactions. Later +/// proposals wait in arrival order, each holding its wtxids and the +/// transactions its JDS supplied (at most +/// `MAX_PROVIDE_MISSING_TRANSACTIONS_PAYLOAD`), and are not refused: a JDS +/// declares one job per JDC at every tip change, so a burst is the normal +/// shape. The queue has no cap of its own (docs/sv2-template-provider.md, +/// Plan D risks). +const MAX_INFLIGHT_VALIDATIONS: usize = 4; + +/// RAM trade (CONTRIBUTING 9): a proposal the TP answered +/// `ProvideMissingTransactions` waits here for the `.Success`, holding its +/// wtxids (≤ 2 MiB) and coinbase split (≤ 128 KiB) but no transactions: +/// ≤ 8 × ~2.2 MiB per session. Past the bound the oldest is dropped, and a +/// provide for it is `unknown-request-id`, the same recovery a JDS runs +/// when the hold timed out. +pub(crate) const MAX_PENDING_PROPOSALS: usize = 8; + /// Fee-delta push settings (docs/sv2-template-provider.md C1). #[derive(Clone, Copy)] pub(crate) struct FeePush { @@ -74,25 +101,42 @@ struct Templates { sent_bits: Option<(u32, [u8; 32])>, /// A template was built since the last tip event. built_since_tip: bool, + /// `coinbase_tx_value_remaining` of the last `NewTemplate` sent; the fee + /// delta is against it, not against a validated job retained after it. + last_value_remaining: Option, retained: VecDeque, } struct Retained { id: u64, - t: template::Template, + job: template::Job, prev_sent: (u32, Instant), /// Set once the tip moves on. retire_at: Option, } +/// A proposal waiting for its `ProvideMissingTransactions.Success`. +struct Pending { + proposal: Arc, + /// The positions the `ProvideMissingTransactions` named. + missing: Vec, + expires_at: Instant, +} + +/// A proposal and what its JDS supplied, waiting for a validation slot. +struct Work { + proposal: Arc, + supplied: Supplied, +} + impl Templates { - fn retain(&mut self, id: u64, t: template::Template, prev_sent: (u32, Instant)) { + fn retain(&mut self, id: u64, job: template::Job, prev_sent: (u32, Instant)) { if self.retained.len() == MAX_RETAINED { self.retained.pop_front(); } self.retained.push_back(Retained { id, - t, + job, prev_sent, retire_at: None, }); @@ -136,6 +180,7 @@ pub(crate) async fn serve( stale_grace: Duration, setup_timeout: Duration, write_timeout: Duration, + provide_timeout: Duration, fee_push: FeePush, stats: Arc, ) -> io::Result<()> { @@ -143,9 +188,9 @@ pub(crate) async fn serve( let setup = async { let mut conn = NoiseConn::accept(stream, responder, write_timeout).await?; let frame = conn.recv().await?; - Ok::<_, io::Error>(on_setup(&mut conn, frame).await?.then_some(conn)) + Ok::<_, io::Error>(on_setup(&mut conn, frame).await?.map(|flags| (conn, flags))) }; - let Some(conn) = tokio::time::timeout_at(deadline, setup) + let Some((conn, flags)) = tokio::time::timeout_at(deadline, setup) .await .map_err(|_| missed_setup_deadline())?? else { @@ -168,8 +213,10 @@ pub(crate) async fn serve( conn: writer, chain, stale_grace, + provide_timeout, fee_push, stats, + job_validation: flags & REQUIRES_JOB_VALIDATION != 0, constraints: None, templates: Templates::default(), built_at: None, @@ -179,6 +226,10 @@ pub(crate) async fn serve( held_logged: false, fee_check_at: None, seen_updates: 0, + validations: JoinSet::new(), + queued: VecDeque::new(), + pending: VecDeque::new(), + open: Vec::new(), }; s.run(&mut frames, deadline).await } @@ -195,8 +246,11 @@ struct Session { conn: NoiseWriter, chain: Arc, stale_grace: Duration, + provide_timeout: Duration, fee_push: FeePush, stats: Arc, + /// `SetupConnection` negotiated `REQUIRES_JOB_VALIDATION`. + job_validation: bool, /// Last `CoinbaseOutputConstraints`: `(max_additional_size, sigops)`. constraints: Option<(u32, u16)>, templates: Templates, @@ -214,6 +268,17 @@ struct Session { fee_check_at: Option, /// `MempoolHub::template_updates` read before the last build. seen_updates: u64, + /// `ProposeTemplate` validations in flight, at most + /// `MAX_INFLIGHT_VALIDATIONS`. Dropping the set aborts them with the + /// session. + validations: JoinSet, Verdict)>>, + /// Prechecked proposals waiting for a validation slot, oldest first. + queued: VecDeque, + /// Proposals waiting for their missing transactions, oldest first, at + /// most `MAX_PENDING_PROPOSALS`. + pending: VecDeque, + /// Request ids accepted and not yet answered: queued or in flight. + open: Vec, } impl Session { @@ -227,7 +292,12 @@ impl Session { ) -> io::Result<()> { let mut tips = self.chain.subscribe_tips(); loop { - let retire_at = self.templates.next_retire(); + let retire_at = self + .templates + .next_retire() + .into_iter() + .chain(self.pending.front().map(|p| p.expires_at)) + .min(); tokio::select! { f = frames.recv() => { let Some(f) = f else { return Ok(()) }; @@ -240,10 +310,22 @@ impl Session { Ok(_) | Err(RecvError::Lagged(_)) => self.publish().await?, Err(RecvError::Closed) => return Ok(()), }, + Some(done) = self.validations.join_next(), if !self.validations.is_empty() => { + let (proposal, verdict) = done.map_err(io::Error::other)??; + let request_id = proposal.request_id; + self.open.retain(|&id| id != request_id); + if let Verdict::Missing(positions) = &verdict { + self.hold(proposal, positions.clone()); + } + self.reply_propose(request_id, verdict).await?; + self.start_validations(); + } _ = tokio::time::sleep_until(retire_at.unwrap_or_else(Instant::now)), if retire_at.is_some() => { - self.templates.retire(Instant::now()); + let now = Instant::now(); + self.templates.retire(now); + self.pending.retain(|p| p.expires_at > now); } _ = tokio::time::sleep_until(self.rebuild_at.unwrap_or_else(Instant::now)), if self.rebuild_at.is_some() => @@ -307,6 +389,67 @@ impl Session { on_request_transaction_data(&mut self.conn, frame, &self.templates).await?; } MESSAGE_TYPE_SUBMIT_SOLUTION => self.on_submit_solution(frame).await?, + MESSAGE_TYPE_PROPOSE_TEMPLATE if self.job_validation => { + let Some(p) = Proposal::decode(&mut frame.payload) else { + rbitcoin_log::info!("sv2: undecodable ProposeTemplate"); + return Ok(true); + }; + let held = self.open.contains(&p.request_id) + || self + .pending + .iter() + .any(|w| w.proposal.request_id == p.request_id); + let code = if held { + Err("duplicate-request-id") + } else { + p.precheck() + }; + if let Err(code) = code { + self.reply_propose(p.request_id, Verdict::Rejected(code.into())) + .await?; + return Ok(true); + } + self.open.push(p.request_id); + self.queued.push_back(Work { + proposal: Arc::new(p), + supplied: Supplied::new(), + }); + self.start_validations(); + } + MESSAGE_TYPE_PROVIDE_MISSING_TRANSACTIONS_SUCCESS if self.job_validation => { + let Ok(m) = + binary_sv2::from_bytes::(&mut frame.payload) + else { + rbitcoin_log::info!("sv2: undecodable ProvideMissingTransactions.Success"); + return Ok(true); + }; + let at = self + .pending + .iter() + .position(|w| w.proposal.request_id == m.request_id); + let Some(Pending { + proposal, missing, .. + }) = at.and_then(|at| self.pending.remove(at)) + else { + self.reply_propose( + m.request_id, + Verdict::Rejected("unknown-request-id".into()), + ) + .await?; + return Ok(true); + }; + match proposal.accept_supplied(&missing, &m.transaction_list) { + Ok(supplied) => { + self.open.push(m.request_id); + self.queued.push_back(Work { proposal, supplied }); + self.start_validations(); + } + Err(code) => { + self.reply_propose(m.request_id, Verdict::Rejected(code.into())) + .await?; + } + } + } t => rbitcoin_log::info!("sv2: ignoring message {t:#x}"), } Ok(true) @@ -347,12 +490,12 @@ impl Session { return Ok(()); } }; - let Some(last) = self.templates.retained.back() else { + let Some(last) = self.templates.last_value_remaining else { return Ok(()); }; // Same prev hash, same height: value_remaining differs by fees only. - if self.templates.current_prev == Some(t.prev_hash) - && t.value_remaining >= last.t.value_remaining.saturating_add(self.fee_push.delta) + if self.templates.current_prev == Some(t.job.prev_hash) + && t.value_remaining >= last.saturating_add(self.fee_push.delta) { self.send(t).await?; } @@ -399,18 +542,19 @@ impl Session { self.templates.last_id += 1; let template_id = self.templates.last_id; // sv2-spec 07 §7.3: a template on a new prev hash is future, then activated. - let new_prev = self.templates.current_prev != Some(t.prev_hash); + let new_prev = self.templates.current_prev != Some(t.job.prev_hash); // §7.4: nBits comes once per prev hash. On min-difficulty networks the // build's bits follow the clock; a solution on this template is hashed // with the bits the client was sent. if let Some((n_bits, target)) = self.templates.sent_bits.filter(|_| !new_prev) { - t.n_bits = n_bits; - t.target = target; + t.job.n_bits = n_bits; + t.job.target = target; } let msg = t .to_message(template_id, new_prev) .map_err(|e| io::Error::other(format!("sv2 NewTemplate: {e:?}")))?; self.conn.send(MESSAGE_TYPE_NEW_TEMPLATE, msg).await?; + self.templates.last_value_remaining = Some(t.value_remaining); let sent = Instant::now(); self.built_at = Some(sent); self.fee_check_at = Some(sent + self.fee_push.interval); @@ -420,16 +564,16 @@ impl Session { .send(MESSAGE_TYPE_SET_NEW_PREV_HASH, t.to_prev_hash(template_id)) .await?; let now = Instant::now(); - self.templates.current_prev = Some(t.prev_hash); - self.templates.prev_sent = Some((t.header_timestamp, now)); - self.templates.sent_bits = Some((t.n_bits, t.target)); + self.templates.current_prev = Some(t.job.prev_hash); + self.templates.prev_sent = Some((t.job.header_timestamp, now)); + self.templates.sent_bits = Some((t.job.n_bits, t.job.target)); self.templates.start_grace(now + self.stale_grace); } let prev_sent = self .templates .prev_sent .ok_or_else(|| io::Error::other("sv2: template before SetNewPrevHash"))?; - self.templates.retain(template_id, t, prev_sent); + self.templates.retain(template_id, t.job, prev_sent); Ok(()) } @@ -467,15 +611,15 @@ impl Session { // and hash of every template tx plus ChainHub's connect path. let root = rbitcoin_store::merkle_root_from_branch( coinbase.compute_txid().to_byte_array(), - &r.t.merkle_path, + &r.job.merkle_path, 0, ); let header = block::Header { version: block::Version::from_consensus(m.version as i32), - prev_blockhash: BlockHash::from_byte_array(r.t.prev_hash), + prev_blockhash: BlockHash::from_byte_array(r.job.prev_hash), merkle_root: TxMerkleNode::from_byte_array(root), time: m.header_timestamp, - bits: CompactTarget::from_consensus(r.t.n_bits), + bits: CompactTarget::from_consensus(r.job.n_bits), nonce: m.header_nonce, }; if header.validate_pow(header.target()).is_err() { @@ -500,9 +644,9 @@ impl Session { m.header_timestamp ); } - let mut txdata = Vec::with_capacity(1 + r.t.txs.len()); + let mut txdata = Vec::with_capacity(1 + r.job.txs.len()); txdata.push(coinbase); - txdata.extend(r.t.txs.iter().map(|tx| Transaction::clone(tx))); + txdata.extend(r.job.txs.iter().map(|tx| Transaction::clone(tx))); let block = Block { header, txdata }; let hash = block.block_hash(); let c = Arc::clone(&self.chain); @@ -518,6 +662,83 @@ impl Session { } Ok(()) } + + /// Fill the validation slots from the queue, oldest first. Other frames + /// and the loop's timers run while a validation is in flight; the reply + /// comes from the `join_next` arm, paired by `request_id`. + fn start_validations(&mut self) { + while self.validations.len() < MAX_INFLIGHT_VALIDATIONS { + let Some(Work { proposal, supplied }) = self.queued.pop_front() else { + return; + }; + let c = Arc::clone(&self.chain); + self.validations.spawn_blocking(move || { + let _g = BlockingRegion::enter(); + let verdict = job::validate(&c, &proposal, &supplied)?; + Ok((proposal, verdict)) + }); + } + } + + /// docs/sv2-job-validation.md §4.2: keep the proposal for its + /// `ProvideMissingTransactions.Success`, `provide_timeout` at most. + fn hold(&mut self, proposal: Arc, missing: Vec) { + if self.pending.len() == MAX_PENDING_PROPOSALS { + self.pending.pop_front(); + } + self.pending.push_back(Pending { + proposal, + missing, + expires_at: Instant::now() + self.provide_timeout, + }); + } + + /// docs/sv2-job-validation.md §4: a valid job is retained under the next + /// template id and answered `Success`; a rejected one answers `Error`; one + /// the mempool cannot complete is asked for its missing transactions. + async fn reply_propose(&mut self, request_id: u32, verdict: Verdict) -> io::Result<()> { + let wire = |e: binary_sv2::Error| io::Error::other(format!("sv2 ProposeTemplate: {e:?}")); + match verdict { + Verdict::Missing(positions) => { + let reply = ProvideMissingTransactions { + request_id, + unknown_tx_position_list: Seq064K::new(positions).map_err(wire)?, + }; + self.conn + .send(MESSAGE_TYPE_PROVIDE_MISSING_TRANSACTIONS, reply) + .await + } + Verdict::Valid { fees, job } => { + self.templates.last_id += 1; + let template_id = self.templates.last_id; + let prev_sent = self + .templates + .prev_sent + .unwrap_or((job.header_timestamp, Instant::now())); + let prev_hash = job.prev_hash; + self.templates.retain(template_id, job, prev_sent); + let reply = ProposeTemplateSuccess { + request_id, + template_id, + prev_hash: U256::from(&prev_hash), + fees, + }; + self.conn + .send(MESSAGE_TYPE_PROPOSE_TEMPLATE_SUCCESS, reply) + .await + } + Verdict::Rejected(code) => { + let reply = ProposeTemplateError { + request_id, + error_code: Str0255::try_from(code.as_str()).map_err(wire)?, + error_details: B064K::try_from(&[][..]).map_err(wire)?, + }; + self.conn + .send(MESSAGE_TYPE_PROPOSE_TEMPLATE_ERROR, reply) + .await + } + } + } } fn setup_error(m: &SetupConnection) -> Option<(u32, &'static str)> { @@ -527,8 +748,9 @@ fn setup_error(m: &SetupConnection) -> Option<(u32, &'static str)> { if !(m.min_version..=m.max_version).contains(&TDP_VERSION) { return Some((0, ERROR_CODE_SETUP_CONNECTION_PROTOCOL_VERSION_MISMATCH)); } - // TDP defines no SetupConnection flags: every set bit is unsupported. - if m.flags != 0 { + // docs/sv2-job-validation.md §3 defines bit 0; every other set bit is + // unsupported. + if m.flags & !REQUIRES_JOB_VALIDATION != 0 { return Some(( m.flags, ERROR_CODE_SETUP_CONNECTION_UNSUPPORTED_FEATURE_FLAGS, @@ -537,15 +759,15 @@ fn setup_error(m: &SetupConnection) -> Option<(u32, &'static str)> { None } -/// `false`: close the session. -async fn on_setup(conn: &mut NoiseConn, mut frame: Frame) -> io::Result { +/// The negotiated flags; `None`: close the session. +async fn on_setup(conn: &mut NoiseConn, mut frame: Frame) -> io::Result> { if frame.msg_type != MESSAGE_TYPE_SETUP_CONNECTION { rbitcoin_log::info!("sv2: message {:#x} before SetupConnection", frame.msg_type); - return Ok(false); + return Ok(None); } let Ok(setup) = binary_sv2::from_bytes::(&mut frame.payload) else { rbitcoin_log::info!("sv2: undecodable SetupConnection"); - return Ok(false); + return Ok(None); }; if let Some((flags, code)) = setup_error(&setup) { let error_code = Str0255::try_from(code) @@ -553,15 +775,15 @@ async fn on_setup(conn: &mut NoiseConn, mut frame: Frame) -> io::Result { let reply = SetupConnectionError { flags, error_code }; conn.send(MESSAGE_TYPE_SETUP_CONNECTION_ERROR, reply) .await?; - return Ok(false); + return Ok(None); } let reply = SetupConnectionSuccess { used_version: TDP_VERSION, - flags: 0, + flags: setup.flags, }; conn.send(MESSAGE_TYPE_SETUP_CONNECTION_SUCCESS, reply) .await?; - Ok(true) + Ok(Some(setup.flags)) } /// An id this session was sent but no longer retains is stale; any other @@ -576,7 +798,7 @@ async fn on_request_transaction_data( rbitcoin_log::info!("sv2: undecodable RequestTransactionData"); return Ok(()); }; - let Some(Retained { t, .. }) = templates.get(template_id) else { + let Some(Retained { job, .. }) = templates.get(template_id) else { let code = if template_id <= templates.last_id { ERROR_CODE_REQUEST_TRANSACTION_DATA_STALE_TEMPLATE_ID } else { @@ -591,7 +813,7 @@ async fn on_request_transaction_data( .send(MESSAGE_TYPE_REQUEST_TRANSACTION_DATA_ERROR, reply) .await; }; - let raw: Vec> = t + let raw: Vec> = job .txs .iter() .map(|tx| bitcoin::consensus::encode::serialize(tx.as_ref())) diff --git a/crates/rbitcoin-sv2/src/template.rs b/crates/rbitcoin-sv2/src/template.rs index 80f5fa8bd..dec31be31 100644 --- a/crates/rbitcoin-sv2/src/template.rs +++ b/crates/rbitcoin-sv2/src/template.rs @@ -4,7 +4,7 @@ use binary_sv2::{Seq0255, B0255, B064K, U256}; use bitcoin::consensus::encode::serialize; use bitcoin::hashes::Hash; -use bitcoin::{Amount, ScriptBuf, Target, Transaction, TxOut}; +use bitcoin::{Amount, CompactTarget, ScriptBuf, Target, Transaction, TxOut}; use rbitcoin_consensus::{ bip34_height_script, block_subsidy, expected_next_bits, median_time_past, witness_commitment_script, MAX_BLOCK_WEIGHT, @@ -24,12 +24,20 @@ pub(crate) const WITNESS_RESERVED_VALUE: [u8; 32] = [0u8; 32]; const COINBASE_BASE_WU: u64 = 1168; const MIN_COINBASE_RESERVE_WU: u64 = 2000; +/// A `NewTemplate`: the coinbase split the client completes, over a [`Job`]. pub(crate) struct Template { pub version: u32, pub coinbase_prefix: Vec, pub value_remaining: u64, /// The witness commitment output, serialized with no count prefix. pub coinbase_outputs: Vec, + pub job: Job, +} + +/// What a session retains under a template id, for a built template or a +/// validated custom job: the header fields a `SubmitSolution` is assembled +/// with and the txs `RequestTransactionData` returns. +pub(crate) struct Job { pub merkle_path: Vec<[u8; 32]>, pub prev_hash: [u8; 32], pub header_timestamp: u32, @@ -60,29 +68,45 @@ impl Template { coinbase_tx_outputs_count: 1, coinbase_tx_outputs: B064K::try_from(&self.coinbase_outputs[..])?, coinbase_tx_locktime: 0, - merkle_path: Seq0255::new(self.merkle_path.iter().map(U256::from).collect())?, + merkle_path: Seq0255::new(self.job.merkle_path.iter().map(U256::from).collect())?, }) } pub fn to_prev_hash(&self, template_id: u64) -> SetNewPrevHash<'_> { SetNewPrevHash { template_id, - prev_hash: U256::from(&self.prev_hash), - header_timestamp: self.header_timestamp, - n_bits: self.n_bits, - target: U256::from(&self.target), + prev_hash: U256::from(&self.job.prev_hash), + header_timestamp: self.job.header_timestamp, + n_bits: self.job.n_bits, + target: U256::from(&self.job.target), } } } -/// Template on the current tip for one client's coinbase constraints. The -/// client's sigops replace the default reserve (Core `BlockAssembler`). -/// Takes the mempool lock and reads the store: blocking region only. -pub(crate) fn build( - chain: &ChainHub, - max_additional_size: u32, - max_additional_sigops: u16, -) -> io::Result