Drop Storage abstraction, revise fs cleanup logic for round reset

phase1-coordinator/src/commands/aggregation.rs

@@ -1,7 +1,7 @@
 use crate::{
     environment::Environment,
     objects::Round,
-    storage::{ContributionLocator, Locator, Object, ObjectReader, Storage},
+    storage::{ContributionLocator, Disk, Locator, Object, ObjectReader, StorageLocator, StorageObject},
     CoordinatorError,
 };
 use phase1::{helpers::CurveKind, Phase1};
@@ -15,7 +15,7 @@ pub(crate) struct Aggregation;
 impl Aggregation {
     /// Runs aggregation for a given environment, storage, and round.
     #[inline]
-    pub(crate) fn run(environment: &Environment, storage: &mut impl Storage, round: &Round) -> anyhow::Result<()> {
+    pub(crate) fn run(environment: &Environment, storage: &mut Disk, round: &Round) -> anyhow::Result<()> {
         let start = Instant::now();
 
         // Fetch the round height.
@@ -95,7 +95,7 @@ impl Aggregation {
     #[inline]
     fn readers<'a>(
         environment: &Environment,
-        storage: &'a impl Storage,
+        storage: &'a Disk,
         round: &Round,
     ) -> anyhow::Result<Vec<ObjectReader<'a>>> {
         let mut readers = vec![];
@@ -149,7 +149,7 @@ mod tests {
         authentication::Dummy,
         commands::{Aggregation, Seed, SigningKey, SEED_LENGTH},
         objects::Task,
-        storage::{Locator, Storage},
+        storage::Locator,
         testing::prelude::*,
         Coordinator,
     };

phase1-coordinator/src/commands/computation.rs

@@ -2,7 +2,7 @@ use crate::{
     authentication::Signature,
     commands::SigningKey,
     environment::Environment,
-    storage::{Locator, Storage},
+    storage::{Disk, Locator, StorageLocator, StorageObject},
     CoordinatorError,
 };
 use phase1::{helpers::CurveKind, Phase1, Phase1Parameters};
@@ -30,7 +30,7 @@ impl Computation {
     ///
     pub(crate) fn run(
         environment: &Environment,
-        storage: &mut impl Storage,
+        storage: &mut Disk,
         signature: Arc<dyn Signature>,
         contributor_signing_key: &SigningKey,
         challenge_locator: &Locator,
@@ -175,7 +175,7 @@ mod tests {
     use crate::{
         authentication::{Dummy, Signature},
         commands::{Computation, Initialization, Seed, SEED_LENGTH},
-        storage::{ContributionLocator, ContributionSignatureLocator, Locator, Object, Storage, StorageObject},
+        storage::{ContributionLocator, ContributionSignatureLocator, Locator, Object, StorageObject},
         testing::prelude::*,
     };
     use setup_utils::calculate_hash;

phase1-coordinator/src/commands/initialization.rs

@@ -1,6 +1,6 @@
 use crate::{
     environment::Environment,
-    storage::{ContributionLocator, Locator, Object, Storage},
+    storage::{ContributionLocator, Disk, Locator, Object, StorageObject},
     CoordinatorError,
 };
 use phase1::{helpers::CurveKind, Phase1, Phase1Parameters};
@@ -23,7 +23,7 @@ impl Initialization {
     #[inline]
     pub(crate) fn run(
         environment: &Environment,
-        storage: &mut impl Storage,
+        storage: &mut Disk,
         round_height: u64,
         chunk_id: u64,
     ) -> anyhow::Result<Vec<u8>> {
@@ -100,7 +100,7 @@ impl Initialization {
     /// Compute both contribution hashes and check for equivalence.
     #[inline]
     fn check_hash(
-        storage: &impl Storage,
+        storage: &Disk,
         contribution_locator: &Locator,
         next_contribution_locator: &Locator,
     ) -> anyhow::Result<Vec<u8>> {

phase1-coordinator/src/commands/mod.rs

@@ -18,7 +18,7 @@ pub(crate) use verification::*;
 use crate::{
     authentication::Signature,
     objects::{ContributionFileSignature, ContributionState},
-    storage::{Locator, Storage},
+    storage::{Disk, Locator, StorageLocator, StorageObject},
     CoordinatorError,
 };
 
@@ -47,7 +47,7 @@ pub type SigningKey = String;
 #[cfg(any(test, feature = "operator"))]
 #[inline]
 pub(crate) fn write_contribution_file_signature(
-    storage: &mut impl Storage,
+    storage: &mut Disk,
     signature: Arc<dyn Signature>,
     signing_key: &SigningKey,
     challenge_locator: &Locator,

phase1-coordinator/src/commands/verification.rs

@@ -2,7 +2,15 @@ use crate::{
     authentication::Signature,
     commands::SigningKey,
     environment::Environment,
-    storage::{ContributionLocator, ContributionSignatureLocator, Locator, Object, Storage},
+    storage::{
+        ContributionLocator,
+        ContributionSignatureLocator,
+        Disk,
+        Locator,
+        Object,
+        StorageLocator,
+        StorageObject,
+    },
     CoordinatorError,
 };
 use phase1::{helpers::CurveKind, Phase1, Phase1Parameters, PublicKey};
@@ -23,7 +31,7 @@ impl Verification {
     #[inline]
     pub(crate) fn run(
         environment: &Environment,
-        storage: &mut impl Storage,
+        storage: &mut Disk,
         signature: Arc<dyn Signature>,
         signing_key: &SigningKey,
         round_height: u64,
@@ -148,7 +156,7 @@ impl Verification {
     #[inline]
     fn verification(
         environment: &Environment,
-        storage: &mut impl Storage,
+        storage: &mut Disk,
         chunk_id: u64,
         challenge_locator: Locator,
         response_locator: Locator,
@@ -341,7 +349,7 @@ mod tests {
     use crate::{
         authentication::Dummy,
         commands::{Computation, Seed, Verification, SEED_LENGTH},
-        storage::{ContributionLocator, ContributionSignatureLocator, Locator, Object, Storage},
+        storage::{ContributionLocator, ContributionSignatureLocator, Locator, Object},
         testing::prelude::*,
         Coordinator,
     };

phase1-coordinator/src/coordinator.rs

@@ -22,8 +22,9 @@ use crate::{
         Locator,
         LocatorPath,
         Object,
-        Storage,
         StorageAction,
+        StorageLocator,
+        StorageObject,
     },
 };
 use setup_utils::calculate_hash;
@@ -333,13 +334,13 @@ impl TimeSource for MockTimeSource {
 /// A core structure for operating the Phase 1 ceremony. This struct
 /// is designed to be [Send] + [Sync]. The state of the ceremony is
 /// stored in a [CoordinatorState] object.
-pub struct Coordinator<S: Storage> {
+pub struct Coordinator {
     /// The parameters and settings of this coordinator.
     environment: Environment,
     /// The signature scheme for contributors & verifiers with this coordinator.
     signature: Arc<dyn Signature>,
     /// The storage of contributions and rounds for this coordinator.
-    storage: S,
+    storage: Disk,
     /// The current round and participant self.
     state: CoordinatorState,
     /// The source of time, allows mocking system time for testing.
@@ -348,7 +349,7 @@ pub struct Coordinator<S: Storage> {
     aggregation_callback: Arc<dyn Fn(Vec<Participant>) -> () + Send + Sync>,
 }
 
-impl Coordinator<Disk> {
+impl Coordinator {
     ///
     /// Creates a new instance of the `Coordinator`, for a given environment.
     ///
@@ -396,10 +397,7 @@ impl Coordinator<Disk> {
     }
 }
 
-impl<S> Coordinator<S>
-where
-    S: Storage + 'static,
-{
+impl Coordinator {
     ///
     /// Runs a set of operations to initialize state and start the coordinator.
     ///
@@ -2200,7 +2198,7 @@ where
     }
 
     #[inline]
-    fn load_current_round_height(storage: &S) -> Result<u64, CoordinatorError> {
+    fn load_current_round_height(storage: &Disk) -> Result<u64, CoordinatorError> {
         // Fetch the current round height from storage.
         match storage.get(&Locator::RoundHeight)? {
             // Case 1 - This is a typical round of the ceremony.
@@ -2211,7 +2209,7 @@ where
     }
 
     #[inline]
-    fn load_current_round(storage: &S) -> Result<Round, CoordinatorError> {
+    fn load_current_round(storage: &Disk) -> Result<Round, CoordinatorError> {
         // Fetch the current round height from storage.
         let current_round_height = Self::load_current_round_height(storage)?;
 
@@ -2220,7 +2218,7 @@ where
     }
 
     #[inline]
-    fn load_round(storage: &S, round_height: u64) -> Result<Round, CoordinatorError> {
+    fn load_round(storage: &Disk, round_height: u64) -> Result<Round, CoordinatorError> {
         // Fetch the current round height from storage.
         let current_round_height = Self::load_current_round_height(storage)?;
 
@@ -2244,7 +2242,7 @@ where
     ///
     #[cfg(test)]
     #[inline]
-    pub(super) fn storage(&self) -> &S {
+    pub(super) fn storage(&self) -> &Disk {
         &self.storage
     }
 
@@ -2254,7 +2252,7 @@ where
     ///
     #[cfg(test)]
     #[inline]
-    pub(super) fn storage_mut(&mut self) -> &mut S {
+    pub(super) fn storage_mut(&mut self) -> &mut Disk {
         &mut self.storage
     }
 
@@ -2300,26 +2298,11 @@ where
         let mut round = Self::load_round(&mut self.storage, current_round_height)?;
 
         tracing::debug!("Resetting round and applying storage changes");
-        if let Some(error) = round
-            .reset(&reset_action.remove_participants)
-            .into_iter()
-            .map(|action| match &action {
-                StorageAction::Remove(a) => match a.clone().try_into_locator(&self.storage) {
-                    Ok(locator) => {
-                        if self.storage.exists(&locator) {
-                            return self.storage.process(action);
-                        } else {
-                            Ok(())
-                        }
-                    }
-                    Err(e) => Err(e),
-                },
-                _ => self.storage.process(action),
-            })
-            .find_map(Result::err)
-        {
-            return Err(error);
-        }
+        self.storage.process(round.reset(&reset_action.remove_participants))?;
+
+        // Clear all files
+        self.storage
+            .process(StorageAction::ClearRoundFiles(current_round_height))?;
 
         if reset_action.rollback {
             if current_round_height == 0 {
@@ -2346,10 +2329,7 @@ where
 use crate::commands::{Computation, Seed, SigningKey, Verification};
 
 #[cfg(any(test, feature = "operator"))]
-impl<S> Coordinator<S>
-where
-    S: Storage + 'static,
-{
+impl Coordinator {
     #[tracing::instrument(
         skip(self, contributor, contributor_signing_key, contributor_seed),
         fields(contributor = %contributor),
@@ -2657,7 +2637,7 @@ mod tests {
     use rand::RngCore;
     use std::{collections::HashMap, sync::Arc};
 
-    fn initialize_to_round_1(coordinator: &mut Coordinator<Disk>, contributors: &[Participant]) -> anyhow::Result<()> {
+    fn initialize_to_round_1(coordinator: &mut Coordinator, contributors: &[Participant]) -> anyhow::Result<()> {
         // Initialize the ceremony and add the contributors and verifiers to the queue.
         {
             // Run initialization.
@@ -2703,7 +2683,7 @@ mod tests {
         Ok(())
     }
 
-    fn initialize_coordinator(coordinator: &mut Coordinator<Disk>) -> anyhow::Result<()> {
+    fn initialize_coordinator(coordinator: &mut Coordinator) -> anyhow::Result<()> {
         // Load the contributors and verifiers.
         let contributors = vec![
             Lazy::force(&TEST_CONTRIBUTOR_ID).clone(),
@@ -2713,7 +2693,7 @@ mod tests {
         initialize_to_round_1(coordinator, &contributors)
     }
 
-    fn initialize_coordinator_single_contributor(coordinator: &mut Coordinator<Disk>) -> anyhow::Result<()> {
+    fn initialize_coordinator_single_contributor(coordinator: &mut Coordinator) -> anyhow::Result<()> {
         // Load the contributors and verifiers.
         let contributors = vec![Lazy::force(&TEST_CONTRIBUTOR_ID).clone()];
 

phase1-coordinator/src/coordinator_state.rs

@@ -4,7 +4,7 @@ use crate::{
         participant::*,
         task::{initialize_tasks, Task},
     },
-    storage::{Locator, Object, Storage},
+    storage::{Disk, Locator, Object},
     CoordinatorError,
     TimeSource,
 };
@@ -3066,7 +3066,7 @@ impl CoordinatorState {
 
     /// Save the coordinator state in storage.
     #[inline]
-    pub(crate) fn save(&self, storage: &mut impl Storage) -> Result<(), CoordinatorError> {
+    pub(crate) fn save(&self, storage: &mut Disk) -> Result<(), CoordinatorError> {
         storage.update(&Locator::CoordinatorState, Object::CoordinatorState(self.clone()))
     }
 }

phase1-coordinator/src/environment.rs

@@ -1,7 +1,4 @@
-use crate::{
-    objects::Participant,
-    storage::{Disk, Storage},
-};
+use crate::{objects::Participant, storage::Disk};
 use phase1::{helpers::CurveKind, ContributionMode, ProvingSystem};
 use setup_utils::{CheckForCorrectness, UseCompression};
 

phase1-coordinator/src/main.rs

@@ -1,7 +1,6 @@
 use phase1_coordinator::{
     authentication::{Dummy, Signature},
     environment::{Development, Environment, Parameters},
-    storage::Disk,
     Coordinator,
 };
 use tracing_subscriber;
@@ -10,7 +9,7 @@ use std::{sync::Arc, time::Duration};
 use tokio::{sync::RwLock, task, time::sleep};
 use tracing::*;
 
-fn coordinator(environment: &Environment, signature: Arc<dyn Signature>) -> anyhow::Result<Coordinator<Disk>> {
+fn coordinator(environment: &Environment, signature: Arc<dyn Signature>) -> anyhow::Result<Coordinator> {
     Ok(Coordinator::new(environment.clone(), signature)?)
 }
 
@@ -29,8 +28,7 @@ pub async fn main() -> anyhow::Result<()> {
     // let environment: Environment = Production::from(Parameters::AleoInner).into();
 
     // Instantiate the coordinator.
-    let coordinator: Arc<RwLock<Coordinator<Disk>>> =
-        Arc::new(RwLock::new(coordinator(&environment, Arc::new(Dummy))?));
+    let coordinator: Arc<RwLock<Coordinator>> = Arc::new(RwLock::new(coordinator(&environment, Arc::new(Dummy))?));
 
     let ceremony_coordinator = coordinator.clone();
     // Initialize the coordinator.