Persistence trait

protocols/v2/channels-sv2/src/client/extended.rs

@@ -13,6 +13,7 @@ use crate::{
         share_accounting::{ShareAccounting, ShareValidationError, ShareValidationResult},
     },
     merkle_root::merkle_root_from_path,
+    persistence::PersistenceHandler,
     target::{bytes_to_hex, u256_to_block_hash},
     MAX_EXTRANONCE_PREFIX_LEN,
 };
@@ -59,7 +60,7 @@ pub type ExtendedJob<'a> = (NewExtendedMiningJob<'a>, Vec<u8>);
 /// - Share accounting for the channel (as tracked by the client).
 /// - The channel's current chain tip.
 #[derive(Clone, Debug)]
-pub struct ExtendedChannel<'a> {
+pub struct ExtendedChannel<'a, P> {
     channel_id: u32,
     user_identity: String,
     extranonce_prefix: Vec<u8>,
@@ -74,12 +75,16 @@ pub struct ExtendedChannel<'a> {
     past_jobs: HashMap<u32, ExtendedJob<'a>>,
     // stale jobs are indexed with job_id (u32)
     stale_jobs: HashMap<u32, ExtendedJob<'a>>,
-    share_accounting: ShareAccounting,
+    share_accounting: ShareAccounting<P>,
     chain_tip: Option<ChainTip>,
 }
 
-impl<'a> ExtendedChannel<'a> {
+impl<'a, P> ExtendedChannel<'a, P>
+where
+    P: PersistenceHandler,
+{
     /// Constructs a new [`ExtendedChannel`].
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         channel_id: u32,
         user_identity: String,
@@ -88,6 +93,7 @@ impl<'a> ExtendedChannel<'a> {
         nominal_hashrate: f32,
         version_rolling: bool,
         rollable_extranonce_size: u16,
+        persistence: P,
     ) -> Self {
         Self {
             channel_id,
@@ -101,7 +107,7 @@ impl<'a> ExtendedChannel<'a> {
             active_job: None,
             past_jobs: HashMap::new(),
             stale_jobs: HashMap::new(),
-            share_accounting: ShareAccounting::new(),
+            share_accounting: ShareAccounting::new(persistence),
             chain_tip: None,
         }
     }
@@ -202,7 +208,7 @@ impl<'a> ExtendedChannel<'a> {
     }
 
     /// Returns a reference to the [`ShareAccounting`] for this channel.
-    pub fn get_share_accounting(&self) -> &ShareAccounting {
+    pub fn get_share_accounting(&self) -> &ShareAccounting<P> {
         &self.share_accounting
     }
 
@@ -533,9 +539,12 @@ impl<'a> ExtendedChannel<'a> {
         // check if a block was found
         if network_target.is_met_by(hash) {
             self.share_accounting.update_share_accounting(
-                self.target.difficulty_float() as u64,
+                self.channel_id,
+                &self.user_identity,
+                self.target.difficulty_float(),
                 share.sequence_number,
                 hash.to_raw_hash(),
+                true,
             );
             return Ok(ShareValidationResult::BlockFound);
         }
@@ -547,13 +556,17 @@ impl<'a> ExtendedChannel<'a> {
             }
 
             self.share_accounting.update_share_accounting(
-                self.target.difficulty_float() as u64,
+                self.channel_id,
+                &self.user_identity,
+                self.target.difficulty_float(),
                 share.sequence_number,
                 hash.to_raw_hash(),
+                false,
             );
 
             // update the best diff
-            self.share_accounting.update_best_diff(hash_as_diff);
+            self.share_accounting
+                .update_best_diff(self.channel_id, hash_as_diff);
 
             return Ok(ShareValidationResult::Valid);
         }
@@ -564,9 +577,12 @@ impl<'a> ExtendedChannel<'a> {
 
 #[cfg(test)]
 mod tests {
-    use crate::client::{
-        extended::ExtendedChannel,
-        share_accounting::{ShareValidationError, ShareValidationResult},
+    use crate::{
+        client::{
+            extended::ExtendedChannel,
+            share_accounting::{ShareValidationError, ShareValidationResult},
+        },
+        persistence::{Persistence, PersistenceHandler, ShareAccountingEvent},
     };
     use binary_sv2::Sv2Option;
     use bitcoin::Target;
@@ -575,6 +591,16 @@ mod tests {
     };
     use std::convert::TryInto;
 
+    /// Unit-like type for persistence in tests
+    #[derive(Debug, Clone)]
+    struct TestPersistence;
+
+    impl PersistenceHandler for TestPersistence {
+        fn persist_event(&self, _event: ShareAccountingEvent) {
+            // No-op for tests
+        }
+    }
+
     #[test]
     fn test_future_job_activation_flow() {
         let channel_id = 1;
@@ -589,14 +615,15 @@ mod tests {
         let version_rolling = true;
         let rollable_extranonce_size = 4u16;
 
-        let mut channel = ExtendedChannel::new(
+        let mut channel = ExtendedChannel::<Persistence<TestPersistence>>::new(
             channel_id,
             user_identity,
             extranonce_prefix.clone(),
             target,
             nominal_hashrate,
             version_rolling,
             rollable_extranonce_size,
+            Persistence::default(),
         );
 
         let future_job = NewExtendedMiningJob {
@@ -671,14 +698,15 @@ mod tests {
         let version_rolling = true;
         let rollable_extranonce_size = 4u16;
 
-        let mut channel = ExtendedChannel::new(
+        let mut channel = ExtendedChannel::<Persistence<TestPersistence>>::new(
             channel_id,
             user_identity,
             extranonce_prefix.clone(),
             target,
             nominal_hashrate,
             version_rolling,
             rollable_extranonce_size,
+            Persistence::default(),
         );
 
         let ntime: u32 = 1746839905;
@@ -744,14 +772,15 @@ mod tests {
         let version_rolling = true;
         let rollable_extranonce_size = 8u16;
 
-        let mut channel = ExtendedChannel::new(
+        let mut channel = ExtendedChannel::<Persistence<TestPersistence>>::new(
             channel_id,
             user_identity,
             extranonce_prefix.clone(),
             target,
             nominal_hashrate,
             version_rolling,
             rollable_extranonce_size,
+            Persistence::default(),
         );
 
         let future_job = NewExtendedMiningJob {
@@ -836,14 +865,15 @@ mod tests {
         let version_rolling = true;
         let rollable_extranonce_size = 8u16;
 
-        let mut channel = ExtendedChannel::new(
+        let mut channel = ExtendedChannel::<Persistence<TestPersistence>>::new(
             channel_id,
             user_identity,
             extranonce_prefix.clone(),
             target,
             nominal_hashrate,
             version_rolling,
             rollable_extranonce_size,
+            Persistence::default(),
         );
 
         let future_job = NewExtendedMiningJob {
@@ -931,14 +961,15 @@ mod tests {
         let version_rolling = true;
         let rollable_extranonce_size = 8u16;
 
-        let mut channel = ExtendedChannel::new(
+        let mut channel = ExtendedChannel::<Persistence<TestPersistence>>::new(
             channel_id,
             user_identity,
             extranonce_prefix.clone(),
             target,
             nominal_hashrate,
             version_rolling,
             rollable_extranonce_size,
+            Persistence::default(),
         );
 
         let future_job = NewExtendedMiningJob {

protocols/v2/channels-sv2/src/client/share_accounting.rs

@@ -5,6 +5,7 @@
 //! are intended for use in Mining Clients.
 
 use super::HashSet;
+use crate::persistence::{PersistenceHandler, ShareAccountingEvent};
 use bitcoin::hashes::sha256d::Hash;
 
 /// The outcome of share validation, as seen by a Mining Client.
@@ -46,33 +47,34 @@ pub enum ShareValidationError {
 /// Keeps statistics and state for shares submitted through the channel:
 /// - last received share's sequence number
 /// - total accepted shares
-/// - cumulative work from accepted shares
+/// - cumulative work from accepted shares (sum of share difficulties as floating point)
 /// - hashes of seen shares (for duplicate detection)
 /// - highest difficulty seen in accepted shares
 #[derive(Clone, Debug)]
-pub struct ShareAccounting {
+pub struct ShareAccounting<P> {
     last_share_sequence_number: u32,
     shares_accepted: u32,
-    share_work_sum: u64,
+    share_work_sum: f64,
     seen_shares: HashSet<Hash>,
     best_diff: f64,
+    persistence: P,
 }
 
-impl Default for ShareAccounting {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl ShareAccounting {
+impl<P> ShareAccounting<P>
+where
+    P: PersistenceHandler,
+{
     /// Creates a new [`ShareAccounting`] instance, initializing all statistics to zero.
-    pub fn new() -> Self {
+    ///
+    /// `persistence` handles background persistence of share accounting events.
+    pub fn new(persistence: P) -> Self {
         Self {
             last_share_sequence_number: 0,
             shares_accepted: 0,
-            share_work_sum: 0,
+            share_work_sum: 0.0,
             seen_shares: HashSet::new(),
             best_diff: 0.0,
+            persistence,
         }
     }
 
@@ -83,14 +85,31 @@ impl ShareAccounting {
     /// - Records share hash to detect duplicates.
     pub fn update_share_accounting(
         &mut self,
-        share_work: u64,
+        channel_id: u32,
+        user_identity: &str,
+        share_work: f64,
         share_sequence_number: u32,
         share_hash: Hash,
+        block_found: bool,
     ) {
         self.last_share_sequence_number = share_sequence_number;
         self.shares_accepted += 1;
         self.share_work_sum += share_work;
         self.seen_shares.insert(share_hash);
+
+        // Persist the share accepted event
+        let event = ShareAccountingEvent::ShareAccepted {
+            channel_id,
+            user_identity: user_identity.to_string(),
+            share_work,
+            share_sequence_number,
+            share_hash,
+            total_shares_accepted: self.shares_accepted,
+            total_share_work_sum: self.share_work_sum,
+            timestamp: std::time::SystemTime::now(),
+            block_found,
+        };
+        self.persistence.persist_event(event);
     }
 
     /// Clears the set of seen share hashes.
@@ -99,6 +118,9 @@ impl ShareAccounting {
     /// to prevent unbounded memory growth.
     pub fn flush_seen_shares(&mut self) {
         self.seen_shares.clear();
+
+        // Ensure any pending persistence events are flushed
+        self.persistence.flush();
     }
 
     /// Returns the sequence number of the last share received.
@@ -112,7 +134,7 @@ impl ShareAccounting {
     }
 
     /// Returns the cumulative work of all accepted shares.
-    pub fn get_share_work_sum(&self) -> u64 {
+    pub fn get_share_work_sum(&self) -> f64 {
         self.share_work_sum
     }
 
@@ -127,9 +149,19 @@ impl ShareAccounting {
     }
 
     /// Updates the best difficulty if the new difficulty is higher than the current best.
-    pub fn update_best_diff(&mut self, diff: f64) {
+    pub fn update_best_diff(&mut self, channel_id: u32, diff: f64) {
+        let previous_best_diff = self.best_diff;
         if diff > self.best_diff {
             self.best_diff = diff;
+
+            // Persist the best difficulty updated event
+            let event = ShareAccountingEvent::BestDifficultyUpdated {
+                channel_id,
+                new_best_diff: diff,
+                previous_best_diff,
+                timestamp: std::time::SystemTime::now(),
+            };
+            self.persistence.persist_event(event);
         }
     }
 }