modularize LoopControl on JDS

Author: plebhash

pool-apps/jd-server/src/lib/error.rs

@@ -48,6 +48,13 @@ pub enum Action {
     Shutdown,
 }
 
+/// Loop control signal for message-processing loops.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum LoopControl {
+    Continue,
+    Break,
+}
+
 /// Marker type for errors originating from the [`crate::job_declarator::JobDeclarator`] layer.
 #[derive(Debug)]
 pub struct JobDeclarator;

pool-apps/jd-server/src/lib/job_declarator/downstream/mod.rs

@@ -8,7 +8,7 @@ use super::{
     DownstreamJobDeclarationMessage, JobDeclarationMessage, ALLOCATED_TOKEN_TIMEOUT_SECS,
     JANITOR_INTERVAL_SECS,
 };
-use crate::{error, error::JDSResult, io_task::spawn_io_tasks};
+use crate::{error, error::JDSResult, error::LoopControl, io_task::spawn_io_tasks};
 use async_channel::{unbounded, Receiver, Sender};
 use bitcoin_core_sv2::job_declaration_protocol::CancellationToken;
 use dashmap::DashMap;
@@ -73,6 +73,39 @@ pub struct Downstream {
 
 #[cfg_attr(not(test), hotpath::measure_all)]
 impl Downstream {
+    fn handle_error_action(
+        &self,
+        context: &str,
+        e: &error::JDSError<error::Downstream>,
+    ) -> LoopControl {
+        match e.action {
+            error::Action::Log => {
+                warn!(
+                    downstream_id = self.downstream_id,
+                    error_kind = ?e.kind,
+                    "{context} returned a log-only error"
+                );
+                LoopControl::Continue
+            }
+            error::Action::Disconnect(_) => {
+                warn!(
+                    downstream_id = self.downstream_id,
+                    error_kind = ?e.kind,
+                    "{context} requested disconnect"
+                );
+                LoopControl::Break
+            }
+            error::Action::Shutdown => {
+                warn!(
+                    downstream_id = self.downstream_id,
+                    error_kind = ?e.kind,
+                    "{context} requested shutdown"
+                );
+                LoopControl::Break
+            }
+        }
+    }
+
     /// Creates a new [`Downstream`] and spawns its Noise I/O tasks.
     #[allow(clippy::too_many_arguments)]
     pub fn new(
@@ -171,20 +204,22 @@ impl Downstream {
                     res = self_clone_1.handle_message_from_downstream() => {
                         if let Err(e) = res {
                             error!(?e, "Error handling downstream message for {downstream_id}");
-                            match e.action {
-                                error::Action::Disconnect(_) => break,
-                                error::Action::Shutdown => break,
-                                error::Action::Log => {}
+                            if let LoopControl::Break = self.handle_error_action(
+                                "Downstream::handle_message_from_downstream",
+                                &e,
+                            ) {
+                                break;
                             }
                         }
                     }
                     res = self_clone_2.handle_job_declarator_message() => {
                         if let Err(e) = res {
                             error!(?e, "Error handling job declarator message for {downstream_id}");
-                            match e.action {
-                                error::Action::Disconnect(_) => break,
-                                error::Action::Shutdown => break,
-                                error::Action::Log => {}
+                            if let LoopControl::Break = self.handle_error_action(
+                                "Downstream::handle_job_declarator_message",
+                                &e,
+                            ) {
+                                break;
                             }
                         }
                     }

pool-apps/jd-server/src/lib/job_declarator/mod.rs

@@ -7,7 +7,7 @@
 
 use crate::{
     error,
-    error::{JDSError, JDSErrorKind, JDSResult},
+    error::{JDSError, JDSErrorKind, JDSResult, LoopControl},
     job_declarator::{
         downstream::Downstream,
         job_validation::{JobValidationEngine, SetCustomMiningJobResult},
@@ -37,7 +37,7 @@ use stratum_apps::{
     utils::types::{DownstreamId, JdToken},
 };
 use tokio::net::TcpListener;
-use tracing::{debug, error, info};
+use tracing::{debug, error, info, warn};
 
 // see https://github.com/stratum-mining/sv2-apps/issues/335
 const TEMPORARY_TIMEOUT_MULTIPLIER: u64 = 144;
@@ -147,6 +147,32 @@ impl JobDeclarator {
 
 /// Generic implementation for all [`JobValidationEngine`] types.
 impl JobDeclarator {
+    fn handle_error_action(
+        &self,
+        context: &str,
+        e: &JDSError<error::JobDeclarator>,
+    ) -> LoopControl {
+        match e.action {
+            error::Action::Log => {
+                warn!(error_kind = ?e.kind, "{context} returned a log-only error");
+                LoopControl::Continue
+            }
+            error::Action::Disconnect(downstream_id) => {
+                warn!(
+                    downstream_id,
+                    error_kind = ?e.kind,
+                    "{context} requested downstream disconnect"
+                );
+                self.cleanup_downstream(downstream_id);
+                LoopControl::Continue
+            }
+            error::Action::Shutdown => {
+                warn!(error_kind = ?e.kind, "{context} requested shutdown");
+                LoopControl::Break
+            }
+        }
+    }
+
     /// Binds a TCP listener and spawns the accept loop that creates a `Downstream`
     /// for every new Noise-encrypted connection.
     #[allow(clippy::too_many_arguments)]
@@ -280,12 +306,11 @@ impl JobDeclarator {
                     res = self.handle_jdp_message() => {
                         if let Err(e) = res {
                             error!(?e, "Error handling Job Declaration message");
-                            match e.action {
-                                error::Action::Disconnect(downstream_id) => {
-                                    self.cleanup_downstream(downstream_id);
-                                }
-                                error::Action::Shutdown => break,
-                                error::Action::Log => {}
+                            if let LoopControl::Break = self.handle_error_action(
+                                "JobDeclarator::handle_jdp_message",
+                                &e,
+                            ) {
+                                break;
                             }
                         }
                     }