send cached share on custom job arrival

Author: Shourya742

miner-apps/jd-client/src/lib/channel_manager/upstream_message_handler.rs

@@ -24,7 +24,7 @@ use crate::{
     error::{ChannelSv2Error, JDCError},
     jd_mode::{get_jd_mode, JdMode},
     status::{State, Status},
-    utils::{create_close_channel_msg, UpstreamState},
+    utils::{create_close_channel_msg, validate_share, UpstreamState},
 };
 
 #[hotpath::measure_all]
@@ -557,32 +557,66 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager {
         _tlv_fields: Option<&[Tlv]>,
     ) -> Result<(), Self::Error> {
         info!("Received: {} ✅", msg);
+
+        let mut messages = Vec::new();
+
         self.channel_manager_data.super_safe_lock(|data| {
-            if let Some(last_declare_job) = data.last_declare_job_store.remove(&msg.request_id) {
-                let template_id = last_declare_job.template.template_id;
-                data.last_declare_job_store
-                    .retain(|_, job| job.template.template_id != template_id);
-
-                data.template_id_to_upstream_job_id
-                    .insert(last_declare_job.template.template_id, msg.job_id);
-                debug!(job_id = msg.job_id, "Mapped custom job into template store");
-                if let (Some(upstream_channel), Some(set_custom_job)) = (
-                    data.upstream_channel.as_mut(),
-                    last_declare_job.set_custom_mining_job,
-                ) {
-                    if let Err(e) =
-                        upstream_channel.on_set_custom_mining_job_success(set_custom_job, msg)
-                    {
-                        error!("Custom mining job success validation failed: {e:#?}");
-                    }
-                }
-            } else {
+            let Some(last_declare_job) = data.last_declare_job_store.remove(&msg.request_id) else {
                 warn!(
                     request_id = msg.request_id,
                     "No matching declare job found for custom job success"
                 );
+                return;
+            };
+
+            let template_id = last_declare_job.template.template_id;
+
+            data.last_declare_job_store
+                .retain(|_, job| job.template.template_id != template_id);
+
+            data.template_id_to_upstream_job_id
+                .insert(template_id, msg.job_id);
+
+            let job_id = msg.job_id;
+
+            let cached = data.cached_shares.remove(&template_id);
+
+            let Some(upstream_channel) = data.upstream_channel.as_mut() else {
+                debug!("No upstream channel available");
+                return;
+            };
+
+            let Some(set_custom_job) = last_declare_job.set_custom_mining_job else {
+                debug!("No custom job found");
+                return;
+            };
+
+            if let Err(e) = upstream_channel.on_set_custom_mining_job_success(set_custom_job, msg) {
+                error!("Custom mining job success validation failed: {e:#?}");
+                return;
+            }
+
+            let Some(cached) = cached else {
+                return;
+            };
+
+            debug!(
+                "Flushing {} cached shares for template_id={}",
+                cached.len(),
+                template_id
+            );
+
+            for mut share in cached {
+                share.job_id = job_id;
+
+                validate_share(share, data, &mut messages);
             }
         });
+
+        for msg in messages {
+            _ = msg.forward(&self.channel_manager_channel).await;
+        }
+
         Ok(())
     }
 

miner-apps/jd-client/src/lib/utils.rs

@@ -22,14 +22,24 @@ use std::{
 use stratum_apps::{
     stratum_core::{
         binary_sv2::Str0255,
+        channels_sv2::client,
         common_messages_sv2::{Protocol, SetupConnection},
-        mining_sv2::{CloseChannel, OpenExtendedMiningChannel, OpenStandardMiningChannel},
-        parsers_sv2::Mining,
+        job_declaration_sv2::PushSolution,
+        mining_sv2::{
+            CloseChannel, OpenExtendedMiningChannel, OpenStandardMiningChannel,
+            SubmitSharesExtended,
+        },
+        parsers_sv2::{JobDeclaration, Mining},
     },
     utils::types::{ChannelId, DownstreamId, Hashrate, JobId},
 };
+use tracing::{debug, info};
 
-use crate::{config::ConfigJDCMode, error::JDCError};
+use crate::{
+    channel_manager::{downstream_message_handler::RouteMessageTo, ChannelManagerData},
+    config::ConfigJDCMode,
+    error::JDCError,
+};
 
 /// Represents a message that can trigger shutdown of various system components.
 #[derive(Debug, Clone)]
@@ -303,3 +313,72 @@ impl From<(DownstreamId, ChannelId, JobId)> for DownstreamChannelJobId {
         }
     }
 }
+
+pub fn validate_share(
+    mut upstream_message: SubmitSharesExtended<'static>,
+    channel_manager_data: &mut ChannelManagerData,
+    messages: &mut Vec<RouteMessageTo>,
+) {
+    let Some(upstream_channel) = channel_manager_data.upstream_channel.as_mut() else {
+        return;
+    };
+    let Some(prev_hash) = channel_manager_data.last_new_prev_hash.as_ref() else {
+        return;
+    };
+
+    match upstream_channel.validate_share(upstream_message.clone()) {
+        Ok(client::share_accounting::ShareValidationResult::Valid(share_hash)) => {
+            upstream_message.sequence_number = channel_manager_data
+                .sequence_number_factory
+                .fetch_add(1, Ordering::Relaxed);
+
+            info!(
+                "Flushed cached extended share: valid | ch={}, seq={}, hash={} ✅",
+                upstream_message.channel_id, upstream_message.sequence_number, share_hash
+            );
+
+            messages.push(Mining::SubmitSharesExtended(upstream_message.into_static()).into());
+        }
+
+        Ok(client::share_accounting::ShareValidationResult::BlockFound(share_hash)) => {
+            upstream_message.sequence_number = channel_manager_data
+                .sequence_number_factory
+                .fetch_add(1, Ordering::Relaxed);
+
+            info!("💰 Block Found (cached extended)!!! 💰 {share_hash}");
+
+            let mut channel_extranonce = upstream_channel.get_extranonce_prefix().to_vec();
+            channel_extranonce.extend_from_slice(&upstream_message.extranonce.to_vec());
+
+            let push_solution = PushSolution {
+                extranonce: channel_extranonce.try_into().expect("extranonce"),
+                ntime: upstream_message.ntime,
+                nonce: upstream_message.nonce,
+                version: upstream_message.version,
+                nbits: prev_hash.n_bits,
+                prev_hash: prev_hash.prev_hash.clone(),
+            };
+
+            messages.push(JobDeclaration::PushSolution(push_solution).into());
+            messages.push(Mining::SubmitSharesExtended(upstream_message.into_static()).into());
+        }
+
+        Err(err) => {
+            let code = match err {
+                client::share_accounting::ShareValidationError::Invalid => "invalid-share",
+                client::share_accounting::ShareValidationError::Stale => "stale-share",
+                client::share_accounting::ShareValidationError::InvalidJobId => "invalid-job-id",
+                client::share_accounting::ShareValidationError::DoesNotMeetTarget => {
+                    "difficulty-too-low"
+                }
+                client::share_accounting::ShareValidationError::DuplicateShare => "duplicate-share",
+                _ => unreachable!(),
+            };
+
+            debug!(
+                "❌ Cached extended share rejected: ch={}, seq={}, error={}",
+                upstream_message.channel_id, upstream_message.sequence_number, code
+            );
+        }
+    }
+}