Replace custom mutex/dashmap with shared primitves in JDS

integration-tests/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "integration_tests_sv2"
-version = "0.5.0"
+version = "0.6.0"
 authors = ["The Stratum V2 Developers"]
 edition = "2021"
 description = "Sv2 Integration Tests Framework"
@@ -14,10 +14,10 @@ keywords = ["stratum", "mining", "bitcoin", "protocol"]
 exclude = ["resources/high_diff_chain.tar.gz"]
 
 [dependencies]
-stratum-apps = { version = "0.7.0", path = "../stratum-apps", features = ["network", "config", "payout", "bitcoin-core-sv2"] }
-jd_client_sv2 = { version = "0.5.0", path = "../miner-apps/jd-client" }
-pool_sv2 = { version = "0.6.0", path = "../pool-apps/pool" }
-translator_sv2 = { version = "0.5.0", path = "../miner-apps/translator" }
+stratum-apps = { version = "0.8.0", path = "../stratum-apps", features = ["network", "config", "payout", "bitcoin-core-sv2"] }
+jd_client_sv2 = { version = "0.6.0", path = "../miner-apps/jd-client" }
+pool_sv2 = { version = "0.7.0", path = "../pool-apps/pool" }
+translator_sv2 = { version = "0.6.0", path = "../miner-apps/translator" }
 async-channel = { version = "1.5.1", default-features = false }
 corepc-node = { version = "0.7.0", default-features = false, features = ["28_0"] }
 minreq = { version = "2.12.0", default-features = false, features = ["https"] }

miner-apps/jd-client/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "jd_client_sv2"
-version = "0.5.0"
+version = "0.6.0"
 authors = ["The Stratum V2 Developers"]
 edition = "2021"
 description = "Job Declarator Client (JDC) role"
@@ -16,7 +16,7 @@ name = "jd_client_sv2"
 path = "src/lib/mod.rs"
 
 [dependencies]
-stratum-apps = { version = "0.7.0", path = "../../stratum-apps", features = ["jd_client"] }
+stratum-apps = { version = "0.8.0", path = "../../stratum-apps", features = ["jd_client"] }
 async-channel = "1.5.1"
 serde = { version = "1.0.89", default-features = false, features = ["derive", "alloc"] }
 tokio = { version = "1.44.1", features = ["full"] }

miner-apps/translator/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "translator_sv2"
-version = "0.5.0"
+version = "0.6.0"
 authors = ["The Stratum V2 Developers"]
 edition = "2021"
 description = "SV1 to SV2 translation proxy"
@@ -20,7 +20,7 @@ name = "translator_sv2"
 path = "src/main.rs"
 
 [dependencies]
-stratum-apps = { version = "0.7.0", path = "../../stratum-apps", features = ["translator"] }
+stratum-apps = { version = "0.8.0", path = "../../stratum-apps", features = ["translator"] }
 async-channel = "1.5.1"
 serde = { version = "1.0.89", default-features = false, features = ["derive", "alloc"] }
 serde_json = { version = "1.0.64", default-features = false, features = ["alloc"] }

pool-apps/jd-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "jd_server_sv2"
-version = "0.5.0"
+version = "0.6.0"
 authors = ["The Stratum V2 Developers"]
 edition = "2021"
 description = "Sv2 Job Declaration Server"
@@ -16,12 +16,11 @@ name = "jd_server_sv2"
 path = "src/lib/mod.rs"
 
 [dependencies]
-stratum-apps = { version = "0.7.0", path = "../../stratum-apps", features = ["jd_server"] }
+stratum-apps = { version = "0.8.0", path = "../../stratum-apps", features = ["jd_server"] }
 async-channel = "1.5.1"
 serde = { version = "1.0.89", features = ["derive", "alloc"], default-features = false }
 tracing = { version = "0.1" }
 tokio = { version = "1.44.1", features = ["full"] }
-dashmap = "6.1.0"
 async-trait = "0.1"
 hotpath = "0.14.0"
 

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

@@ -12,7 +12,7 @@
 //! The `Owner` type parameter is a zero-sized marker (e.g. [`JobDeclarator`], [`Downstream`])
 //! that controls which constructors (`shutdown`, `disconnect`) are available at the type level.
 
-use std::{fmt::Debug, marker::PhantomData};
+use std::{fmt::Debug, marker::PhantomData, sync::PoisonError};
 
 use stratum_apps::{
     stratum_core::{
@@ -119,6 +119,13 @@ pub enum JDSErrorKind {
     UnsupportedConnectionFlags,
     OneshotRecv(tokio::sync::oneshot::error::RecvError),
     InvalidConfig(String),
+    PoisonLock,
+}
+
+impl<T> From<PoisonError<T>> for JDSErrorKind {
+    fn from(_: PoisonError<T>) -> Self {
+        JDSErrorKind::PoisonLock
+    }
 }
 
 impl<Owner> From<JDSError<Owner>> for JDSErrorKind {

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

@@ -25,9 +25,7 @@ impl HandleCommonMessagesFromClientAsync for Downstream {
         &self,
         _client_id: Option<usize>,
     ) -> Result<Vec<u16>, Self::Error> {
-        Ok(self
-            .negotiated_extensions
-            .super_safe_lock(|extensions| extensions.clone()))
+        self.negotiated_extensions.get().map_err(JDSError::shutdown)
     }
 
     async fn handle_setup_connection(

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

@@ -14,14 +14,12 @@ use crate::{
     io_task::spawn_io_tasks,
 };
 use async_channel::{unbounded, Receiver, Sender};
-use dashmap::DashMap;
 use std::{
     sync::Arc,
     time::{Duration, Instant},
 };
 use stratum_apps::{
     bitcoin_core_sv2::CancellationToken,
-    custom_mutex::Mutex,
     network_helpers::noise_stream::NoiseTcpStream,
     stratum_core::{
         common_messages_sv2::MESSAGE_TYPE_SETUP_CONNECTION,
@@ -30,6 +28,7 @@ use stratum_apps::{
         job_declaration_sv2::DeclareMiningJob,
         parsers_sv2::{parse_message_frame_with_tlvs, AnyMessage},
     },
+    sync::{SharedLock, SharedMap},
     task_manager::TaskManager,
     utils::{
         protocol_message_type::{protocol_message_type, MessageType},
@@ -58,9 +57,9 @@ pub struct DownstreamIo {
 #[derive(Clone)]
 pub struct Downstream {
     /// Extensions that have been successfully negotiated with this client.
-    pub negotiated_extensions: Arc<Mutex<Vec<u16>>>,
+    pub negotiated_extensions: SharedLock<Vec<u16>>,
     /// Jobs waiting for missing transactions (keyed by `request_id`).
-    pub pending_declare_mining_jobs: Arc<DashMap<RequestId, PendingDeclareMiningJob>>,
+    pub pending_declare_mining_jobs: SharedMap<RequestId, PendingDeclareMiningJob>,
     pub downstream_io: DownstreamIo,
     pub downstream_id: DownstreamId,
     /// Extensions that JDS supports
@@ -137,8 +136,8 @@ impl Downstream {
             downstream_cancellation_token.clone(),
         );
 
-        let negotiated_extensions = Arc::new(Mutex::new(Vec::new()));
-        let pending_declare_mining_jobs = Arc::new(DashMap::new());
+        let negotiated_extensions = SharedLock::new(Vec::new());
+        let pending_declare_mining_jobs = SharedMap::new();
 
         let downstream_io = DownstreamIo {
             to_job_declarator_sender,
@@ -242,7 +241,7 @@ impl Downstream {
     /// older than `ALLOCATED_TOKEN_TIMEOUT_SECS`.
     fn spawn_pending_jobs_janitor(&self, task_manager: Arc<TaskManager>) {
         let cancellation_token = self.downstream_cancellation_token.clone();
-        let pending_declare_mining_jobs = Arc::clone(&self.pending_declare_mining_jobs);
+        let pending_declare_mining_jobs = self.pending_declare_mining_jobs.clone();
         let downstream_id = self.downstream_id;
         let token_timeout = Duration::from_secs(ALLOCATED_TOKEN_TIMEOUT_SECS);
         let janitor_interval = Duration::from_secs(JANITOR_INTERVAL_SECS);
@@ -351,7 +350,8 @@ impl Downstream {
                 debug!("Received mining SV2 frame from downstream.");
                 let negotiated_extensions = self
                     .negotiated_extensions
-                    .super_safe_lock(|extensions| extensions.clone());
+                    .get()
+                    .map_err(error::JDSError::shutdown)?;
                 let (any_message, tlv_fields) = parse_message_frame_with_tlvs(
                     header,
                     sv2_frame.payload(),