feat(pool): add share bridge and ValidatedShare for bead forwarding [Part 2]

pool-apps/pool/Cargo.toml

@@ -19,6 +19,8 @@ path = "src/lib/mod.rs"
 
 [dependencies]
 stratum-apps = { version = "0.8.0", path = "../../stratum-apps", features = ["pool"] }
+braidpool-common = { path = "../braidpool-common" }
+braidpool-template-provider = { path = "../braidpool-template-provider" }
 async-channel = "1.5.1"
 serde = { version = "1.0.89", features = ["derive", "alloc"], default-features = false }
 tokio = { version = "1.44.1", features = ["full"] }

pool-apps/pool/src/lib/braidpool/mod.rs

@@ -0,0 +1,117 @@
+//! Braidpool-specific types bridging the SV2 pool to the braidpool node.
+//!
+//! This module is the boundary layer between the generic SV2 pool and
+//! Braidpool's bead-based consensus. It defines the types that flow from
+//! the pool (where shares are validated) to the node (where beads are built).
+//!
+//! # Data flow
+//! ```text
+//! SubmitSharesExtended
+//!   → validate_share (ExtendedChannel)
+//!   → ValidatedShare  ──[ShareBridgeSender]──► node::propagate_valid_bead  (PR 4)
+//! ```
+
+use tokio::sync::mpsc;
+
+/// A share that passed pool-level PoW validation, ready for bead construction.
+///
+/// `extranonce1` is `Vec<u8>` rather than `u64` because in audit mode the
+/// pool-assigned prefix is `N + 7` bytes (upstream extranonce1 + 2-byte miner
+/// prefix + 5-byte commitment), which can exceed 8 bytes for common 8-byte
+/// upstream configurations.
+#[derive(Debug, Clone)]
+pub struct ValidatedShare {
+    /// Template the share was mined against. `None` for custom-job shares.
+    pub template_id: Option<u64>,
+    /// SV2 extended channel ID the share arrived on.
+    pub channel_id: u32,
+    /// Pool-assigned extranonce prefix (extranonce1 in Stratum v1 terms).
+    pub extranonce1: Vec<u8>,
+    /// Miner-rolled extranonce suffix (extranonce2).
+    pub extranonce2: Vec<u8>,
+    /// Block version field (may include BIP-320 rolled bits).
+    pub version: u32,
+    /// Block header nTime.
+    pub ntime: u32,
+    /// Block header nonce.
+    pub nonce: u32,
+}
+
+/// Template context attached to a validated share for bead metadata construction.
+///
+/// Carries the information needed to build `CommittedMetadata` in the node.
+/// Populated when a `ValidatedShare` is matched to a template.
+#[derive(Debug)]
+pub struct BeadContext {
+    /// Template that was being mined when the share arrived.
+    pub template_id: u64,
+    /// Wall-clock time the template was received from the template provider.
+    /// Used to measure share propagation latency for bead timing.
+    pub received_at: std::time::Instant,
+    /// Committed transactions from the template (for CommittedMetadata).
+    pub transactions: Vec<Vec<u8>>,
+}
+
+/// Extranonce allocation strategy for the pool↔tproxy Extended Channel.
+///
+/// Determines the split between pool-assigned prefix (extranonce1) and
+/// miner-rolled space (extranonce2) based on the upstream pool configuration.
+///
+/// # Normal mode (4+8)
+/// ```text
+/// upstream negotiates: ext1=4, ext2=8
+/// pool sends to tproxy: prefix=4 bytes (random), extranonce_size=12
+/// ```
+///
+/// # Audit mode
+/// Braidpool embeds commitment bytes inside the extranonce space it sends to
+/// downstream miners, then reconstructs the upstream-agreed format before
+/// submitting shares back upstream.
+///
+/// ```text
+/// upstream negotiates: ext1=N, ext2=8  (common: N=4 or N=8)
+/// pool prefix sent to tproxy: N + 7 bytes  (N upstream + 2 miner prefix + 5 commitment)
+/// miner rolling space: total - (N+7) bytes
+/// on share submission upstream: reconstruct ext1=N, ext2=8
+/// ```
+#[derive(Debug, Clone)]
+pub enum ExtraNonceConfig {
+    /// Standard mode: 4-byte pool prefix, up to 12-byte miner rolling space.
+    Normal,
+    /// Audit mode: prefix embeds upstream extranonce1 + commitment bytes.
+    Audit {
+        /// Size (bytes) of extranonce1 as negotiated with the upstream pool.
+        upstream_ext1_size: usize,
+    },
+}
+
+impl ExtraNonceConfig {
+    /// Pool-assigned prefix size (extranonce1) for this configuration.
+    pub fn extranonce1_size(&self) -> usize {
+        match self {
+            Self::Normal => 4,
+            // N bytes upstream + 2 miner prefix + 5 commitment
+            Self::Audit { upstream_ext1_size } => upstream_ext1_size + 7,
+        }
+    }
+
+    /// Miner-controlled rolling space size (extranonce2) for this configuration,
+    /// given `total_extranonce_bytes` agreed with the upstream pool.
+    pub fn extranonce2_size(&self, total_extranonce_bytes: usize) -> usize {
+        total_extranonce_bytes.saturating_sub(self.extranonce1_size())
+    }
+}
+
+/// Sender half of the share bridge — held by the pool's `ChannelManager`.
+pub type ShareBridgeSender = mpsc::UnboundedSender<ValidatedShare>;
+
+/// Receiver half of the share bridge — consumed by the node IPC task (PR 4).
+pub type ShareBridgeReceiver = mpsc::UnboundedReceiver<ValidatedShare>;
+
+/// Create the share bridge channel connecting pool validation to node bead construction.
+///
+/// The sender is stored on `ChannelManager`; the receiver is kept alive in
+/// `PoolRuntime` until PR 4 wires it to `propagate_valid_bead` over IPC.
+pub fn create_share_bridge() -> (ShareBridgeSender, ShareBridgeReceiver) {
+    mpsc::unbounded_channel()
+}

pool-apps/pool/src/lib/channel_manager/mining_message_handler.rs

@@ -19,6 +19,7 @@ use stratum_apps::stratum_core::{
         server::{
             error::{ExtendedChannelError, StandardChannelError},
             extended::ExtendedChannel,
+            jobs::JobOrigin,
             share_accounting::{ShareValidationError, ShareValidationResult},
             standard::StandardChannel,
         },
@@ -36,6 +37,7 @@ use tracing::{error, info, warn};
 use jd_server_sv2::job_declarator::SetCustomMiningJobResponse;
 
 use crate::{
+    braidpool::ValidatedShare,
     channel_manager::{CLIENT_SEARCH_SPACE_BYTES, ChannelManager, RouteMessageTo},
     error::{self, PoolError, PoolErrorKind},
     utils::{PayoutMode, PayoutModeError, create_close_channel_msg},
@@ -943,6 +945,9 @@ impl HandleMiningMessagesFromClientOwnedAsync for ChannelManager {
 
         let channel_id = msg.channel_id;
         let vardiff_key = (downstream_id, channel_id).into();
+        // Clone before the immutable borrow below so the sender can be moved
+        // into the closure without holding a reference to self.
+        let share_bridge = self.share_bridge_sender.clone();
         let messages = self.with_registered_downstream(downstream_id, |downstream| {
                 let messages = if !downstream.extended_channels.contains_key(&channel_id) {
                     let error = SubmitSharesErrorOwned {
@@ -1001,13 +1006,47 @@ impl HandleMiningMessagesFromClientOwnedAsync for ChannelManager {
                                                 downstream_id, channel_id, msg.sequence_number, share_hash, share_work
                                             );
                                         }
+                                        if let Some(sender) = &share_bridge {
+                                            let template_id = extended_channel
+                                                .get_active_job()
+                                                .and_then(|job| match job.get_origin() {
+                                                    JobOrigin::NewTemplate(t) => Some(t.template_id),
+                                                    _ => None,
+                                                });
+                                            let validated = ValidatedShare {
+                                                template_id,
+                                                channel_id,
+                                                extranonce1: extended_channel.get_extranonce_prefix().to_vec(),
+                                                extranonce2: msg.extranonce.as_bytes().to_vec(),
+                                                version: msg.version,
+                                                ntime: msg.ntime,
+                                                nonce: msg.nonce,
+                                            };
+                                            if sender.send(validated).is_err() {
+                                                warn!("share bridge closed — validated share dropped");
+                                            }
+                                        }
                                     }
                                     Ok(ShareValidationResult::BlockFound(
                                         share_hash,
                                         template_id,
                                         coinbase,
                                     )) => {
                                         info!("SubmitSharesExtended: 💰 Block Found!!! 💰{share_hash}");
+                                        if let Some(sender) = &share_bridge {
+                                            let validated = ValidatedShare {
+                                                template_id,
+                                                channel_id,
+                                                extranonce1: extended_channel.get_extranonce_prefix().to_vec(),
+                                                extranonce2: msg.extranonce.as_bytes().to_vec(),
+                                                version: msg.version,
+                                                ntime: msg.ntime,
+                                                nonce: msg.nonce,
+                                            };
+                                            if sender.send(validated).is_err() {
+                                                warn!("share bridge closed — block-found share dropped");
+                                            }
+                                        }
                                         if let Some(template_id) = template_id {
                                             info!("SubmitSharesExtended: Propagating solution to the Template Provider.");
                                             let solution = SubmitSolutionOwned {

pool-apps/pool/src/lib/channel_manager/mod.rs

@@ -118,6 +118,10 @@ pub struct ChannelManager {
     required_extensions: Vec<u16>,
     /// Embedded Job Declaration engine (present when `[jds]` config is set).
     job_declarator: Option<JobDeclarator>,
+    /// Sender half of the Braidpool share bridge. Present when the pool runs in
+    /// Braidpool mode; `None` in standalone SRI mode. Every validated share is
+    /// forwarded here for bead construction in the node (wired by PR 4).
+    share_bridge_sender: Option<crate::braidpool::ShareBridgeSender>,
 }
 
 #[cfg_attr(not(test), hotpath::measure_all)]
@@ -166,6 +170,7 @@ impl ChannelManager {
         downstream_receiver: Receiver<(DownstreamId, MiningOwned, Option<Vec<Tlv>>)>,
         coinbase_outputs: Vec<u8>,
         job_declarator: Option<JobDeclarator>,
+        share_bridge_sender: Option<crate::braidpool::ShareBridgeSender>,
     ) -> PoolResult<Self, error::ChannelManager> {
         // Simulating a scenario where there are multiple mining servers,
         // `server_id` is used as `local_prefix_bytes` so each pool instance
@@ -199,6 +204,7 @@ impl ChannelManager {
             supported_extensions: config.supported_extensions().to_vec(),
             required_extensions: config.required_extensions().to_vec(),
             job_declarator,
+            share_bridge_sender,
         };
 
         Ok(channel_manager)

pool-apps/pool/src/lib/config.rs

@@ -14,11 +14,14 @@ use std::{
 };
 
 pub use jd_server_sv2::config::{JDSConfig, JDSPartialConfig};
-use stratum_apps::{
-    config_helpers::{CoinbaseRewardScript, opt_path_from_toml},
+pub use stratum_apps::{
+    config_helpers::CoinbaseRewardScript,
     key_utils::{Secp256k1PublicKey, Secp256k1SecretKey},
-    stratum_core::bitcoin::{Amount, TxOut},
     tp_type::TemplateProviderType,
+};
+use stratum_apps::{
+    config_helpers::opt_path_from_toml,
+    stratum_core::bitcoin::{Amount, TxOut},
     utils::types::{SharesBatchSize, SharesPerMinute},
 };
 
@@ -93,6 +96,34 @@ impl PoolConfig {
         }
     }
 
+    /// Builds a [`PoolConfig`] for Braidpool's in-process pool mode.
+    ///
+    /// Uses `BraidpoolIpc` as the template provider (no external TDP address needed —
+    /// the template channel is wired at runtime via [`PoolSv2::with_braidpool_template_rx`]).
+    /// Sensible defaults are applied: 1-hour cert validity, 20 shares/min target,
+    /// batch size 100, no JDS, no monitoring.
+    pub fn for_braidpool(
+        listen_address: SocketAddr,
+        authority_public_key: Secp256k1PublicKey,
+        authority_secret_key: Secp256k1SecretKey,
+        coinbase_reward_script: CoinbaseRewardScript,
+    ) -> Self {
+        Self::new(
+            ConnectionConfig::new(listen_address, 3600, "braidpool".to_string()),
+            TemplateProviderType::BraidpoolIpc,
+            AuthorityConfig::new(authority_public_key, authority_secret_key),
+            coinbase_reward_script,
+            20.0,
+            100,
+            0,
+            vec![],
+            vec![],
+            None,
+            None,
+            None,
+        )
+    }
+
     /// Returns the coinbase output.
     pub fn coinbase_reward_script(&self) -> &CoinbaseRewardScript {
         &self.coinbase_reward_script

pool-apps/pool/src/lib/mod.rs

@@ -1,5 +1,7 @@
 use error::PoolErrorKind;
 use pool_runtime::{Init, PoolRuntime};
+
+pub mod braidpool;
 use std::sync::{
     Arc,
     atomic::{AtomicBool, Ordering},
@@ -8,7 +10,9 @@ use stratum_apps::bitcoin_core_sv2::CancellationToken;
 use tokio::sync::Notify;
 use tracing::info;
 
-use crate::config::PoolConfig;
+use braidpool_common::template::BraidpoolTemplate;
+
+use crate::{braidpool::ShareBridgeReceiver, config::PoolConfig};
 
 pub mod channel_manager;
 pub mod config;
@@ -27,6 +31,10 @@ pub struct PoolSv2 {
     cancellation_token: CancellationToken,
     shutdown_notify: Arc<Notify>,
     is_alive: Arc<AtomicBool>,
+    /// In-process template channel for `BraidpoolIpc` mode.
+    /// Wrapped in `Arc<Mutex<Option<...>>>` so `PoolSv2` remains `Clone`.
+    braidpool_template_rx:
+        Arc<std::sync::Mutex<Option<tokio::sync::mpsc::Receiver<BraidpoolTemplate>>>>,
 }
 
 #[cfg_attr(not(test), hotpath::measure_all)]
@@ -37,9 +45,35 @@ impl PoolSv2 {
             cancellation_token: CancellationToken::new(),
             shutdown_notify: Arc::new(Notify::new()),
             is_alive: Arc::new(AtomicBool::new(true)),
+            braidpool_template_rx: Arc::new(std::sync::Mutex::new(None)),
         }
     }
 
+    /// Wires the in-process template channel for `BraidpoolIpc` mode.
+    ///
+    /// Must be called before [`start`] when `template_provider_type = BraidpoolIpc`.
+    /// The receiver is consumed once during bootstrap; calling `start` a second
+    /// time after the channel has been taken will fail with a configuration error.
+    pub fn with_braidpool_template_rx(
+        self,
+        rx: tokio::sync::mpsc::Receiver<BraidpoolTemplate>,
+    ) -> Self {
+        self.braidpool_template_rx
+            .lock()
+            .expect("braidpool_template_rx mutex poisoned")
+            .replace(rx);
+        self
+    }
+
+    pub(crate) fn take_braidpool_template_rx(
+        &self,
+    ) -> Option<tokio::sync::mpsc::Receiver<BraidpoolTemplate>> {
+        self.braidpool_template_rx
+            .lock()
+            .expect("braidpool_template_rx mutex poisoned")
+            .take()
+    }
+
     /// Starts the Pool server and blocks asynchronously on the `PoolRuntime`.
     ///
     /// The startup and execution sequence follows:
@@ -53,7 +87,15 @@ impl PoolSv2 {
     ///
     /// If any error occurs during bootstrapping, `start` receives the partially initialized
     /// runtime, gracefully shuts it down, and then returns the error.
-    pub async fn start(&self) -> Result<(), PoolErrorKind> {
+    ///
+    /// When `share_rx_notify` is `Some`, the [`ShareBridgeReceiver`] is sent over it once
+    /// the channel manager is bootstrapped and the pool is ready to process shares. The
+    /// caller can then forward validated shares to `propagate_valid_bead` without waiting
+    /// for the pool to shut down.
+    pub async fn start(
+        &self,
+        share_rx_notify: Option<tokio::sync::oneshot::Sender<ShareBridgeReceiver>>,
+    ) -> Result<(), PoolErrorKind> {
         let runtime = match PoolRuntime::<Init>::new(self.clone()) {
             Ok(runtime) => runtime,
             Err(err) => {
@@ -62,7 +104,7 @@ impl PoolSv2 {
             }
         };
 
-        let runtime = match runtime.bootstrap().await {
+        let mut runtime = match runtime.bootstrap().await {
             Ok(runtime) => runtime,
             Err(err) => {
                 let (err, runtime) = err.into_parts();
@@ -71,6 +113,14 @@ impl PoolSv2 {
             }
         };
 
+        // Deliver the share bridge receiver to the caller before blocking on shutdown,
+        // so they can start consuming shares without waiting for the pool to exit.
+        if let Some(tx) = share_rx_notify {
+            if let Some(rx) = runtime.take_share_bridge_receiver() {
+                let _ = tx.send(rx);
+            }
+        }
+
         runtime.wait_for_shutdown().await;
         runtime.shutdown().await;