From 256b3305eea556233b88ddfd226731a7aa766e7b Mon Sep 17 00:00:00 2001 From: Gary Krause Date: Tue, 6 Oct 2026 16:45:26 -0400 Subject: [PATCH 01/19] sv2: define the job-validation messages and setup flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire layer for the TDP job-validation extension (docs/sv2-job-validation-draft.md §3-5): the four ValidateCustomJob messages (0x77-0x7a) as binary_sv2 structs, and SetupConnection flag bit 0 (REQUIRES_JOB_VALIDATION). Setup now accepts that bit and echoes it in SetupConnection.Success; any other set bit is still rejected with unsupported-feature-flags. The messages are `pub` because the rbitcoin-test journey decodes TDP frames cross-crate (as it does the SRI types today) and they have no in-crate reader yet; the dispatch handlers follow in the next step. Co-Authored-By: Claude Fable 5.1 --- crates/rbitcoin-sv2/src/lib.rs | 1 + crates/rbitcoin-sv2/src/listener_tests.rs | 28 +++-- crates/rbitcoin-sv2/src/messages.rs | 118 ++++++++++++++++++++++ crates/rbitcoin-sv2/src/session.rs | 8 +- 4 files changed, 144 insertions(+), 11 deletions(-) create mode 100644 crates/rbitcoin-sv2/src/messages.rs diff --git a/crates/rbitcoin-sv2/src/lib.rs b/crates/rbitcoin-sv2/src/lib.rs index 6197509c3..8589b9ec1 100644 --- a/crates/rbitcoin-sv2/src/lib.rs +++ b/crates/rbitcoin-sv2/src/lib.rs @@ -3,6 +3,7 @@ //! Noise_NX over TCP is the only transport. Plan and constraints: //! `docs/sv2-template-provider.md`. +pub mod messages; mod session; mod template; pub mod testutil; diff --git a/crates/rbitcoin-sv2/src/listener_tests.rs b/crates/rbitcoin-sv2/src/listener_tests.rs index d524700a1..27537a280 100644 --- a/crates/rbitcoin-sv2/src/listener_tests.rs +++ b/crates/rbitcoin-sv2/src/listener_tests.rs @@ -1,3 +1,4 @@ +use crate::messages::REQUIRES_JOB_VALIDATION; use crate::test_chain::shared_regtest; use crate::testutil::TpClient; use crate::{ @@ -40,6 +41,12 @@ async fn expect_error(c: &mut TpClient, flags: u32, code: &str) { ); } +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); @@ -63,9 +70,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 +89,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; @@ -129,9 +144,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; } @@ -174,8 +187,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; } diff --git a/crates/rbitcoin-sv2/src/messages.rs b/crates/rbitcoin-sv2/src/messages.rs new file mode 100644 index 000000000..cdeaabd0f --- /dev/null +++ b/crates/rbitcoin-sv2/src/messages.rs @@ -0,0 +1,118 @@ +//! Job-validation extension to TDP (`docs/sv2-job-validation-draft.md` §3–5): +//! the `SetupConnection` flag and the four `ValidateCustomJob` messages a +//! Job Declarator Server uses to have this TP check a custom job. + +use binary_sv2::{Deserialize, Seq064K, Serialize, Str0255, B016M, B064K, U256}; + +/// `SetupConnection.flags` bit 0: the client intends to send +/// [`ValidateCustomJob`]. The only TDP flag this TP accepts. +pub const REQUIRES_JOB_VALIDATION: u32 = 1 << 0; + +pub const MESSAGE_TYPE_VALIDATE_CUSTOM_JOB: u8 = 0x77; +pub const MESSAGE_TYPE_VALIDATE_CUSTOM_JOB_MISSING_TRANSACTIONS: u8 = 0x78; +pub const MESSAGE_TYPE_VALIDATE_CUSTOM_JOB_SUCCESS: u8 = 0x79; +pub const MESSAGE_TYPE_VALIDATE_CUSTOM_JOB_ERROR: u8 = 0x7a; + +/// Client → TP: is this custom job a consensus-valid block on the TP tip? +/// `transaction_list` carries the txs a prior +/// [`ValidateCustomJobMissingTransactions`] asked for, in that order. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub struct ValidateCustomJob<'decoder> { + pub request_id: u32, + pub prev_hash: U256<'decoder>, + pub version: u32, + pub coinbase_tx: B064K<'decoder>, + pub wtxid_list: Seq064K<'decoder, U256<'decoder>>, + pub transaction_list: Seq064K<'decoder, B016M<'decoder>>, +} + +/// TP → client: 0-indexed positions in `wtxid_list` the TP cannot resolve. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub struct ValidateCustomJobMissingTransactions<'decoder> { + pub request_id: u32, + pub unknown_tx_position_list: Seq064K<'decoder, u16>, +} + +/// TP → client: the job is valid and retained under `template_id`, which +/// shares the `NewTemplate.template_id` namespace. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub struct ValidateCustomJobSuccess { + pub request_id: u32, + pub template_id: u64, + pub fees: u64, +} + +/// TP → client: the job was not validated. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +pub struct ValidateCustomJobError<'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 (coinbase, tx, details) = ([0xc0u8; 100], [0xeeu8; 300], [1u8, 2, 3]); + let mut bytes = Vec::new(); + round_trip( + ValidateCustomJob { + request_id: 7, + prev_hash: U256::from(&h1), + version: 0x2000_0000, + coinbase_tx: B064K::try_from(&coinbase[..]).unwrap(), + wtxid_list: Seq064K::new(vec![U256::from(&h2), U256::from(&h3)]).unwrap(), + transaction_list: Seq064K::new(vec![B016M::try_from(&tx[..]).unwrap()]).unwrap(), + }, + &mut bytes, + ); + let mut bytes = Vec::new(); + round_trip( + ValidateCustomJobMissingTransactions { + request_id: 7, + unknown_tx_position_list: Seq064K::new(vec![0u16, 5, u16::MAX]).unwrap(), + }, + &mut bytes, + ); + let mut bytes = Vec::new(); + round_trip( + ValidateCustomJobSuccess { + request_id: 7, + template_id: u64::MAX, + fees: 12_345, + }, + &mut bytes, + ); + let mut bytes = Vec::new(); + round_trip( + ValidateCustomJobError { + request_id: 7, + error_code: Str0255::try_from("stale-prevhash").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..b7ab2f2d1 100644 --- a/crates/rbitcoin-sv2/src/session.rs +++ b/crates/rbitcoin-sv2/src/session.rs @@ -1,5 +1,6 @@ //! One TDP session: Noise handshake, `SetupConnection`, then TDP messages. +use crate::messages::REQUIRES_JOB_VALIDATION; use crate::template; use crate::transport::{Frame, NoiseConn, NoiseWriter}; use crate::Sv2TpStats; @@ -527,8 +528,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 { + // sv2-job-validation-draft §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, @@ -557,7 +559,7 @@ async fn on_setup(conn: &mut NoiseConn, mut frame: Frame) -> io::Result { } let reply = SetupConnectionSuccess { used_version: TDP_VERSION, - flags: 0, + flags: setup.flags, }; conn.send(MESSAGE_TYPE_SETUP_CONNECTION_SUCCESS, reply) .await?; From 2ccd4c474136afb7b4ae17806625cdc08a33d493 Mon Sep 17 00:00:00 2001 From: Gary Krause Date: Tue, 6 Oct 2026 16:52:33 -0400 Subject: [PATCH 02/19] sv2: size the client frame cap per message type MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ValidateCustomJob (docs/sv2-job-validation-draft.md §4.1) carries a full wtxid list and the missing transactions, so it cannot fit the 64 KiB cap sized for SubmitSolution. The reader now bounds every client frame at the ValidateCustomJob maximum while reading (40 fixed bytes, a B064K coinbase, 65535 wtxids, and a block's worth of B016M txs: 6_359_306 bytes), then applies the per-type cap once the frame is whole, so every other client message still closes the session past 65557 bytes. codec_sv2 7.0.0 decrypts the header into a private buffer and never exposes it mid-frame, so the type is only known with the whole frame. Seeing it earlier would mean replacing NoiseDecoder's transport path with our own chunked decrypt loop; the per-session RAM bound is the same either way, since a client can always pick 0x77. Named RAM trade: one in-flight client frame of ≤ ~6.4 MB per session, ≤ ~51 MB at MAX_SESSIONS. Co-Authored-By: Claude Fable 5.1 --- crates/rbitcoin-sv2/src/listener_tests.rs | 54 ++++++++++---- crates/rbitcoin-sv2/src/transport.rs | 87 +++++++++++++++++++---- 2 files changed, 115 insertions(+), 26 deletions(-) diff --git a/crates/rbitcoin-sv2/src/listener_tests.rs b/crates/rbitcoin-sv2/src/listener_tests.rs index 27537a280..c1a40eeb2 100644 --- a/crates/rbitcoin-sv2/src/listener_tests.rs +++ b/crates/rbitcoin-sv2/src/listener_tests.rs @@ -1,6 +1,7 @@ -use crate::messages::REQUIRES_JOB_VALIDATION; +use crate::messages::{MESSAGE_TYPE_VALIDATE_CUSTOM_JOB, REQUIRES_JOB_VALIDATION}; use crate::test_chain::shared_regtest; use crate::testutil::TpClient; +use crate::transport::MAX_VALIDATE_CUSTOM_JOB_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, @@ -13,7 +14,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; @@ -34,10 +37,15 @@ 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)) ); } @@ -313,8 +321,9 @@ 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` and a `ValidateCustomJob` well past it 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); @@ -347,14 +356,35 @@ 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; + // No handler yet: a ValidateCustomJob under its own cap is read and ignored. + let big = vec![0u8; 200 << 10]; + c.send_bytes(MESSAGE_TYPE_VALIDATE_CUSTOM_JOB, &big) + .await + .unwrap(); + c.request_transaction_data(1).await.unwrap(); + let f = c + .recv() + .await + .expect("open after a 200 KiB ValidateCustomJob"); + 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 _ = c + .send_bytes( + MESSAGE_TYPE_VALIDATE_CUSTOM_JOB, + &vec![0u8; MAX_VALIDATE_CUSTOM_JOB_PAYLOAD + 1], + ) + .await; + c.request_transaction_data(1).await.ok(); + assert_closed(&mut c, "a ValidateCustomJob over its cap").await; tp.shutdown().await; } diff --git a/crates/rbitcoin-sv2/src/transport.rs b/crates/rbitcoin-sv2/src/transport.rs index 393142722..6b7b62725 100644 --- a/crates/rbitcoin-sv2/src/transport.rs +++ b/crates/rbitcoin-sv2/src/transport.rs @@ -1,11 +1,13 @@ //! Noise_NX transport over one TCP stream: handshake, then encrypted SV2 frames. +use crate::messages::MESSAGE_TYPE_VALIDATE_CUSTOM_JOB; use binary_sv2::{GetSize, Serialize}; use codec_sv2::{ Decoded, Decrypted, Handshake, MessageFrame, NoiseDecoder, NoiseEncoder, TransportDecryptState, TransportEncryptState, ENCRYPTED_SV2_FRAME_HEADER_SIZE, SV2_FRAME_PLAINTEXT_CHUNK_SIZE, }; use noise_sv2::{Initiator, Responder, AEAD_MAC_LEN}; +use rbitcoin_consensus::MAX_BLOCK_WEIGHT; use std::io; use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -18,11 +20,32 @@ pub struct Frame { pub payload: Vec, } -/// Largest client→TP payload. The largest TDP client message is -/// `SubmitSolution`: 20 fixed bytes plus a `B064K` coinbase (2-byte length, +/// Largest client→TP payload for every message but `ValidateCustomJob`: +/// `SubmitSolution`, 20 fixed bytes plus a `B064K` coinbase (2-byte length, /// ≤ 65535 bytes). The 24-bit frame length would otherwise let a client /// make the session buffer ~16 MB per frame. -pub(crate) const MAX_CLIENT_PAYLOAD: usize = 20 + 2 + u16::MAX as usize; +const MAX_CLIENT_PAYLOAD: usize = 20 + 2 + u16::MAX as usize; + +/// Largest `ValidateCustomJob` payload: 40 fixed bytes, a `B064K` coinbase, +/// a full `SEQ0_64K[U256]` wtxid list, and a `SEQ0_64K[B016M]` transaction +/// list whose txs together fit a block (weight is never below serialized +/// size) with a 3-byte length each. +/// +/// RAM trade (CONTRIBUTING 9): a session buffers one in-flight client frame +/// of at most this size (~6.4 MB), so at `MAX_SESSIONS` client frames hold +/// ≤ ~51 MB on top of template retention (lib.rs). +pub(crate) const MAX_VALIDATE_CUSTOM_JOB_PAYLOAD: usize = { + const SEQ: usize = u16::MAX as usize; + 4 + 32 + 4 + (2 + SEQ) + (2 + SEQ * 32) + (2 + SEQ * 3 + MAX_BLOCK_WEIGHT as usize) +}; + +/// Payload cap by client message type. +const fn client_payload_cap(msg_type: u8) -> usize { + match msg_type { + MESSAGE_TYPE_VALIDATE_CUSTOM_JOB => MAX_VALIDATE_CUSTOM_JOB_PAYLOAD, + _ => MAX_CLIENT_PAYLOAD, + } +} /// Encrypted bytes on the wire for a frame carrying `payload` bytes. const fn encrypted_frame_len(payload: usize) -> usize { @@ -31,6 +54,33 @@ const fn encrypted_frame_len(payload: usize) -> usize { + payload.div_ceil(SV2_FRAME_PLAINTEXT_CHUNK_SIZE) * AEAD_MAC_LEN } +fn no_cap(_: u8) -> usize { + usize::MAX +} + +/// Size bounds on the frames one direction carries. +#[derive(Clone, Copy)] +struct FrameCap { + /// Encrypted bytes one frame may take, counted while it is read. + max_frame: usize, + /// Payload cap by message type, applied once the frame is whole. + payload: fn(u8) -> usize, +} + +impl FrameCap { + /// Client→TP: the largest client message while reading, then + /// [`client_payload_cap`] for the type. + const CLIENT: Self = Self { + max_frame: encrypted_frame_len(MAX_VALIDATE_CUSTOM_JOB_PAYLOAD), + payload: client_payload_cap, + }; + /// TP→client: the 24-bit frame length only. + const UNBOUNDED: Self = Self { + max_frame: usize::MAX, + payload: no_cap, + }; +} + pub(crate) struct NoiseConn { reader: NoiseReader, writer: NoiseWriter, @@ -44,8 +94,7 @@ pub(crate) struct NoiseReader { // `next_transport_frame` consumes the state; a failed round leaves `None` // and the connection must close (codec_sv2 nonce rule). rx: Option, - /// Encrypted bytes one frame may take; see [`MAX_CLIENT_PAYLOAD`]. - max_frame: usize, + cap: FrameCap, /// Encrypted bytes read so far for the frame being decoded. frame_read: usize, } @@ -90,7 +139,6 @@ impl NoiseConn { .write_all(encoder.encode_handshake(reply).as_ref()) .await?; let (tx, rx) = transport.split(); - let max_frame = encrypted_frame_len(MAX_CLIENT_PAYLOAD); Ok(Self::new( stream, encoder, @@ -98,7 +146,7 @@ impl NoiseConn { tx, rx, write_timeout, - max_frame, + FrameCap::CLIENT, )) } @@ -129,7 +177,6 @@ impl NoiseConn { .try_into() .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "sv2: handshake size"))?; let (tx, rx) = sent.step_2(reply).map_err(codec_err)?.split(); - // TP→client frames are bounded by the 24-bit frame length only. Ok(Self::new( stream, encoder, @@ -137,7 +184,7 @@ impl NoiseConn { tx, rx, write_timeout, - usize::MAX, + FrameCap::UNBOUNDED, )) } @@ -148,7 +195,7 @@ impl NoiseConn { tx: TransportEncryptState, rx: TransportDecryptState, write_timeout: Duration, - max_frame: usize, + cap: FrameCap, ) -> Self { let (read, write) = stream.into_split(); Self { @@ -156,7 +203,7 @@ impl NoiseConn { stream: read, decoder, rx: Some(rx), - max_frame, + cap, frame_read: 0, }, writer: NoiseWriter { @@ -230,9 +277,21 @@ impl NoiseReader { Decrypted::Frame(mut frame, state) => { self.rx = Some(state); self.frame_read = 0; + let msg_type = frame.header().msg_type(); + let payload = frame.payload(); + // codec_sv2 decrypts the header into a private buffer, so + // the type is only visible with the whole frame: the + // running count bounds every frame at the largest client + // message, and the per-type cap applies here. + if payload.len() > (self.cap.payload)(msg_type) { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "sv2: frame over the size cap", + )); + } return Ok(Frame { - msg_type: frame.header().msg_type(), - payload: frame.payload().to_vec(), + msg_type, + payload: payload.to_vec(), }); } Decrypted::Incomplete(n, state) => { @@ -243,7 +302,7 @@ impl NoiseReader { // the running count closes an oversized frame before the // chunk that would exceed the cap is read. self.frame_read += n; - if self.frame_read > self.max_frame { + if self.frame_read > self.cap.max_frame { return Err(io::Error::new( io::ErrorKind::InvalidData, "sv2: frame over the size cap", From 1290428ed676e30f47464ca99093d961f5fd9a8c Mon Sep 17 00:00:00 2001 From: Gary Krause Date: Tue, 6 Oct 2026 17:21:00 -0400 Subject: [PATCH 03/19] sv2: validate a declared custom job through the template provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit docs/sv2-job-validation-draft.md §4.1–4.3: on a session that negotiated REQUIRES_JOB_VALIDATION, a ValidateCustomJob whose wtxids are all in the mempool is resolved, assembled under a header on the tip (time ≥ MTP + 1, expected nBits, nonce 0, merkle root over the placeholder coinbase and the declared txids) and run through ChainHub::check_block_proposal. Ok(fees) retains the job under the next template id, exactly like a built template, and answers Success{request_id, template_id, fees}; a reject string answers Error{error_code}. Without the flag the message is still ignored. The tip read (height, prev hash, time, bits) moves out of template::build into template::next_header so both paths share it, and Template splits into the NewTemplate coinbase split over a Job: what a session retains under a template id for SubmitSolution and RequestTransactionData, so a validated job carries no dead NewTemplate fields. CPU trade: one full proposal check per request on the blocking pool (no scripts, no PoW). A wtxid not in the mempool still drops the request; MissingTransactions follows in the next step. Co-Authored-By: Claude Fable 5.1 --- crates/rbitcoin-sv2/src/job.rs | 80 ++++++++++ crates/rbitcoin-sv2/src/lib.rs | 1 + crates/rbitcoin-sv2/src/session.rs | 119 +++++++++++---- crates/rbitcoin-sv2/src/template.rs | 91 +++++++---- crates/rbitcoin-sv2/src/template_tests.rs | 175 +++++++++++++++++++++- crates/rbitcoin-sv2/src/testutil.rs | 7 +- 6 files changed, 414 insertions(+), 59 deletions(-) create mode 100644 crates/rbitcoin-sv2/src/job.rs diff --git a/crates/rbitcoin-sv2/src/job.rs b/crates/rbitcoin-sv2/src/job.rs new file mode 100644 index 000000000..b6cf35a24 --- /dev/null +++ b/crates/rbitcoin-sv2/src/job.rs @@ -0,0 +1,80 @@ +//! Custom job validation for a Job Declarator Server +//! (docs/sv2-job-validation.md §4): resolve the declared wtxids, check the +//! job as a block proposal on the tip, and shape it for retention. + +use crate::messages::ValidateCustomJob; +use crate::template::{self, Job}; +use bitcoin::hashes::Hash; +use bitcoin::{block, Block, BlockHash, Transaction, TxMerkleNode, Wtxid}; +use rbitcoin_net::ChainHub; +use std::io; +use std::sync::Arc; + +pub(crate) enum Verdict { + /// Consensus-valid on the tip: the fee total and the job to retain. + Valid { fees: u64, job: Job }, + /// `ValidateCustomJob.Error.error_code`: a Core reject string or one of + /// the draft's own codes. + Rejected(String), +} + +/// Reads the store and the mempool: blocking region only. `None`: the +/// payload is not a `ValidateCustomJob`. +pub(crate) fn validate( + chain: &ChainHub, + mut payload: Vec, +) -> io::Result> { + let Ok(m) = binary_sv2::from_bytes::(&mut payload) else { + return Ok(None); + }; + Ok(Some((m.request_id, check(chain, &m)?))) +} + +fn check(chain: &ChainHub, m: &ValidateCustomJob) -> io::Result { + let Ok(coinbase) = bitcoin::consensus::deserialize::(m.coinbase_tx.as_ref()) + else { + return Ok(Verdict::Rejected("bad-cb-decode".into())); + }; + let mut txs = Vec::with_capacity(m.wtxid_list.len()); + for w in m.wtxid_list.iter() { + let w: [u8; 32] = w.as_ref().try_into().expect("U256 is 32 bytes"); + let Some(tx) = chain + .mempool() + .and_then(|mp| mp.get_tx_by_wtxid(&Wtxid::from_byte_array(w))) + else { + return Err(io::Error::other( + "sv2: ValidateCustomJob wtxid not in the mempool", + )); + }; + txs.push(tx); + } + let next = template::next_header(chain)?; + 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(m.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; no 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), + }) +} diff --git a/crates/rbitcoin-sv2/src/lib.rs b/crates/rbitcoin-sv2/src/lib.rs index 8589b9ec1..f88506cc5 100644 --- a/crates/rbitcoin-sv2/src/lib.rs +++ b/crates/rbitcoin-sv2/src/lib.rs @@ -3,6 +3,7 @@ //! Noise_NX over TCP is the only transport. Plan and constraints: //! `docs/sv2-template-provider.md`. +mod job; pub mod messages; mod session; mod template; diff --git a/crates/rbitcoin-sv2/src/session.rs b/crates/rbitcoin-sv2/src/session.rs index b7ab2f2d1..dc74f9f54 100644 --- a/crates/rbitcoin-sv2/src/session.rs +++ b/crates/rbitcoin-sv2/src/session.rs @@ -1,6 +1,11 @@ //! One TDP session: Noise handshake, `SetupConnection`, then TDP messages. -use crate::messages::REQUIRES_JOB_VALIDATION; +use crate::job::{self, Verdict}; +use crate::messages::{ + ValidateCustomJobError, ValidateCustomJobSuccess, MESSAGE_TYPE_VALIDATE_CUSTOM_JOB, + MESSAGE_TYPE_VALIDATE_CUSTOM_JOB_ERROR, MESSAGE_TYPE_VALIDATE_CUSTOM_JOB_SUCCESS, + REQUIRES_JOB_VALIDATION, +}; use crate::template; use crate::transport::{Frame, NoiseConn, NoiseWriter}; use crate::Sv2TpStats; @@ -75,25 +80,28 @@ 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, } 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, }); @@ -144,9 +152,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 { @@ -171,6 +179,7 @@ pub(crate) async fn serve( stale_grace, fee_push, stats, + job_validation: flags & REQUIRES_JOB_VALIDATION != 0, constraints: None, templates: Templates::default(), built_at: None, @@ -198,6 +207,8 @@ struct Session { stale_grace: 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, @@ -308,6 +319,9 @@ 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_VALIDATE_CUSTOM_JOB if self.job_validation => { + self.on_validate_custom_job(frame).await?; + } t => rbitcoin_log::info!("sv2: ignoring message {t:#x}"), } Ok(true) @@ -348,12 +362,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?; } @@ -400,18 +414,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); @@ -421,16 +436,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(()) } @@ -468,15 +483,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() { @@ -501,9 +516,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); @@ -519,6 +534,52 @@ impl Session { } Ok(()) } + + /// docs/sv2-job-validation.md §4: a valid job is retained under the next + /// template id and answered `Success`; a rejected one answers `Error`. + async fn on_validate_custom_job(&mut self, frame: Frame) -> io::Result<()> { + let c = Arc::clone(&self.chain); + let verdict = tokio::task::spawn_blocking(move || { + let _g = BlockingRegion::enter(); + job::validate(&c, frame.payload) + }) + .await + .map_err(io::Error::other)??; + let Some((request_id, verdict)) = verdict else { + rbitcoin_log::info!("sv2: undecodable ValidateCustomJob"); + return Ok(()); + }; + let wire = |e: binary_sv2::Error| io::Error::other(format!("sv2 ValidateCustomJob: {e:?}")); + match verdict { + 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())); + self.templates.retain(template_id, job, prev_sent); + let reply = ValidateCustomJobSuccess { + request_id, + template_id, + fees, + }; + self.conn + .send(MESSAGE_TYPE_VALIDATE_CUSTOM_JOB_SUCCESS, reply) + .await + } + Verdict::Rejected(code) => { + let reply = ValidateCustomJobError { + 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_VALIDATE_CUSTOM_JOB_ERROR, reply) + .await + } + } + } } fn setup_error(m: &SetupConnection) -> Option<(u32, &'static str)> { @@ -539,15 +600,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) @@ -555,7 +616,7 @@ 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, @@ -563,7 +624,7 @@ async fn on_setup(conn: &mut NoiseConn, mut frame: Frame) -> io::Result { }; 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 @@ -578,7 +639,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 { @@ -593,7 +654,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