Replace SingleDiskFarm::on_sector_[plotting|plotted]() with SingleDiskFarm::on_sector_update() that will be extended later

Author: nazar-pc

crates/subspace-farmer-components/src/plotting.rs

@@ -10,7 +10,7 @@ use backoff::future::retry;
 use backoff::{Error as BackoffError, ExponentialBackoff};
 use futures::stream::FuturesUnordered;
 use futures::StreamExt;
-use parity_scale_codec::Encode;
+use parity_scale_codec::{Decode, Encode};
 use rayon::prelude::*;
 use std::error::Error;
 use std::mem;
@@ -97,7 +97,7 @@ impl PieceGetter for ArchivedHistorySegment {
 }
 
 /// Information about sector that was plotted
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Encode, Decode)]
 pub struct PlottedSector {
     /// Sector ID
     pub sector_id: SectorId,

crates/subspace-farmer/src/bin/subspace-farmer/commands/farm.rs

@@ -23,7 +23,7 @@ use subspace_core_primitives::{PublicKey, Record, SectorIndex};
 use subspace_erasure_coding::ErasureCoding;
 use subspace_farmer::piece_cache::PieceCache;
 use subspace_farmer::single_disk_farm::{
-    SingleDiskFarm, SingleDiskFarmError, SingleDiskFarmOptions,
+    SectorPlottingDetails, SectorUpdate, SingleDiskFarm, SingleDiskFarmError, SingleDiskFarmOptions,
 };
 use subspace_farmer::utils::farmer_piece_getter::FarmerPieceGetter;
 use subspace_farmer::utils::piece_validator::SegmentCommitmentPieceValidator;
@@ -633,10 +633,8 @@ where
 
             // Collect newly plotted pieces
             let on_plotted_sector_callback =
-                move |(plotted_sector, maybe_old_plotted_sector): &(
-                    PlottedSector,
-                    Option<PlottedSector>,
-                )| {
+                move |plotted_sector: &PlottedSector,
+                      maybe_old_plotted_sector: &Option<PlottedSector>| {
                     let _span_guard = span.enter();
 
                     {
@@ -645,7 +643,7 @@ where
                             .as_mut()
                             .expect("Initial value was populated above; qed");
 
-                        if let Some(old_plotted_sector) = maybe_old_plotted_sector {
+                        if let Some(old_plotted_sector) = &maybe_old_plotted_sector {
                             readers_and_pieces.delete_sector(disk_farm_index, old_plotted_sector);
                         }
                         readers_and_pieces.add_sector(disk_farm_index, plotted_sector);
@@ -659,7 +657,15 @@ where
             };
 
             single_disk_farm
-                .on_sector_plotted(Arc::new(on_plotted_sector_callback))
+                .on_sector_update(Arc::new(move |(_sector_index, sector_state)| {
+                    if let SectorUpdate::Plotting(SectorPlottingDetails::Finished {
+                        plotted_sector,
+                        old_plotted_sector,
+                    }) = sector_state
+                    {
+                        on_plotted_sector_callback(plotted_sector, old_plotted_sector);
+                    }
+                }))
                 .detach();
 
             single_disk_farm

crates/subspace-farmer/src/single_disk_farm.rs

@@ -539,21 +539,36 @@ type HandlerFn<A> = Arc<dyn Fn(&A) + Send + Sync + 'static>;
 type Handler<A> = Bag<HandlerFn<A>, A>;
 
 /// Details about sector currently being plotted
-pub struct SectorPlottingDetails {
-    /// Sector index
-    pub sector_index: SectorIndex,
-    /// Progress so far in % (not including this sector)
-    pub progress: f32,
-    /// Whether sector is being replotted
-    pub replotting: bool,
-    /// Whether this is the last sector queued so far
-    pub last_queued: bool,
+#[derive(Debug, Clone, Encode, Decode)]
+pub enum SectorPlottingDetails {
+    /// Starting plotting of a sector
+    Starting {
+        /// Progress so far in % (not including this sector)
+        progress: f32,
+        /// Whether sector is being replotted
+        replotting: bool,
+        /// Whether this is the last sector queued so far
+        last_queued: bool,
+    },
+    /// Finished plotting
+    Finished {
+        /// Information about plotted sector
+        plotted_sector: PlottedSector,
+        /// Information about old plotted sector that was replaced
+        old_plotted_sector: Option<PlottedSector>,
+    },
+}
+
+/// Various sector updates
+#[derive(Debug, Clone, Encode, Decode)]
+pub enum SectorUpdate {
+    /// Sector is is being plotted
+    Plotting(SectorPlottingDetails),
 }
 
 #[derive(Default, Debug)]
 struct Handlers {
-    sector_plotting: Handler<SectorPlottingDetails>,
-    sector_plotted: Handler<(PlottedSector, Option<PlottedSector>)>,
+    sector_update: Handler<(SectorIndex, SectorUpdate)>,
     solution: Handler<SolutionResponse>,
     plot_audited: Handler<AuditEvent>,
 }
@@ -1316,17 +1331,9 @@ impl SingleDiskFarm {
         self.piece_reader.clone()
     }
 
-    /// Subscribe to sector plotting notification
-    pub fn on_sector_plotting(&self, callback: HandlerFn<SectorPlottingDetails>) -> HandlerId {
-        self.handlers.sector_plotting.add(callback)
-    }
-
-    /// Subscribe to notification about plotted sectors
-    pub fn on_sector_plotted(
-        &self,
-        callback: HandlerFn<(PlottedSector, Option<PlottedSector>)>,
-    ) -> HandlerId {
-        self.handlers.sector_plotted.add(callback)
+    /// Subscribe to sector updates
+    pub fn on_sector_update(&self, callback: HandlerFn<(SectorIndex, SectorUpdate)>) -> HandlerId {
+        self.handlers.sector_update.add(callback)
     }
 
     /// Subscribe to notification about audited plots

crates/subspace-farmer/src/single_disk_farm/plotting.rs

@@ -1,5 +1,5 @@
 use crate::single_disk_farm::{
-    BackgroundTaskError, Handlers, PlotMetadataHeader, SectorPlottingDetails,
+    BackgroundTaskError, Handlers, PlotMetadataHeader, SectorPlottingDetails, SectorUpdate,
     RESERVED_PLOT_METADATA,
 };
 use crate::thread_pool_manager::PlottingThreadPoolManager;
@@ -184,14 +184,14 @@ where
             info!(%sector_index, "Plotting sector ({progress:.2}% complete)");
         }
 
+        let sector_state = SectorUpdate::Plotting(SectorPlottingDetails::Starting {
+            progress,
+            replotting,
+            last_queued,
+        });
         handlers
-            .sector_plotting
-            .call_simple(&SectorPlottingDetails {
-                sector_index,
-                progress,
-                replotting,
-                last_queued,
-            });
+            .sector_update
+            .call_simple(&(sector_index, sector_state));
 
         // This `loop` is a workaround for edge-case in local setup if expiration is configured to
         // 1. In that scenario we get replotting notification essentially straight from block import
@@ -403,9 +403,13 @@ where
             }
         }
 
+        let sector_state = SectorUpdate::Plotting(SectorPlottingDetails::Finished {
+            plotted_sector,
+            old_plotted_sector: maybe_old_plotted_sector,
+        });
         handlers
-            .sector_plotted
-            .call_simple(&(plotted_sector, maybe_old_plotted_sector));
+            .sector_update
+            .call_simple(&(sector_index, sector_state));
     }
 
     Ok(())