Repository: farcaster-project/farcaster-node
Pull request #773 (2022-11-14 10:59:16)

Auto-restore checkpoint on startup


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index cec9c593..17875121 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,17 +7,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-### Changed
-
-- Refactor: Improve reliability of peerd connections by @TheCharlatan in <https://github.com/farcaster-project/farcaster-node/pull/792>
-- Add richer data to the progress messages by @TheCharlatan in <https://github.com/farcaster-project/farcaster-node/pull/761>
-- Swapd: Simplify checkpoint encoding by @TheCharlatan in <https://github.com/farcaster-project/farcaster-node/pull/786>
-- Swapd: Some patches and improvements by @TheCharlatan in <https://github.com/farcaster-project/farcaster-node/pull/783>
-- Farcaster: Cleanup dangling swap info on restart by @TheCharlatan in <https://github.com/farcaster-project/farcaster-node/pull/779>
-- Database: Change secret key encoding by @TheCharlatan in <https://github.com/farcaster-project/farcaster-node/pull/782>
-- Tests: Use retries until swap id is available in grpc test by @TheCharlatan in <https://github.com/farcaster-project/farcaster-node/pull/797>
-- Cli: Implement endpoint for checking syncer health by @TheCharlatan in <https://github.com/farcaster-project/farcaster-node/pull/755>
-
 ## [0.4.0] - 2022-11-30
 
 ### Changed
diff --git a/Cargo.toml b/Cargo.toml
index 720b11f2..8d7f05f8 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,7 +13,7 @@ include = [
   "CHANGELOG.md",
   "LICENSE",
 ]
-keywords = ["bitcoin", "monero", "atomic-swaps"]
+keywords = ["bitcoin", "monero", "node", "atomic swaps"]
 license = "MIT"
 readme = "README.md"
 repository = "https://github.com/farcaster-project/farcaster-node"
diff --git a/farcasterd.toml b/farcasterd.toml
index aaf5d5a4..5de04c7f 100644
--- a/farcasterd.toml
+++ b/farcasterd.toml
@@ -1,5 +1,10 @@
 # Farcasterd configuration
-# configures farcasterd specific behaviors such as auto-funding swaps
+# configures farcasterd specific behaviors such as auto-funding swaps or
+# auto-restoring checkpoints
+
+[farcasterd]
+# Set this to false if you only want manual restore. Defaut to true
+auto_restore = true
 
 # Defines auto-funding
 [farcasterd.auto_funding]
diff --git a/src/cli/command.rs b/src/cli/command.rs
index c43ac29c..9de84197 100644
--- a/src/cli/command.rs
+++ b/src/cli/command.rs
@@ -37,6 +37,7 @@ use crate::bus::{
     AddressSecretKey,
 };
 use crate::bus::{BusMsg, Failure, FailureCode, HealthReport};
+use crate::cli::opts::CheckpointSelector;
 use crate::client::Client;
 use crate::syncerd::{SweepAddressAddendum, SweepBitcoinAddress, SweepMoneroAddress};
 use crate::{Error, LogStyle, ServiceId};
@@ -158,8 +159,33 @@ impl Exec for Command {
                 runtime.report_response_or_fail()?;
             }
 
-            Command::ListCheckpoints => {
-                runtime.request_info(ServiceId::Database, InfoMsg::RetrieveAllCheckpointInfo)?;
+            Command::ListCheckpoints { select } => {
+                match select {
+                    CheckpointSelector::All => {
+                        runtime.request_info(
+                            ServiceId::Database,
+                            InfoMsg::RetrieveAllCheckpointInfo,
+                        )?;
+                    }
+                    CheckpointSelector::AvailableForRestore => {
+                        runtime.request_info(
+                            ServiceId::Database,
+                            InfoMsg::RetrieveAllCheckpointInfo,
+                        )?;
+                        if let BusMsg::Info(InfoMsg::CheckpointList(list)) =
+                            runtime.report_failure()?
+                        {
+                            runtime.request_info(
+                                ServiceId::Farcasterd,
+                                InfoMsg::CheckpointList(list),
+                            )?;
+                        } else {
+                            return Err(Error::Farcaster(
+                                "Received unexpected response".to_string(),
+                            ));
+                        }
+                    }
+                }
                 runtime.report_response_or_fail()?;
             }
 
diff --git a/src/cli/opts.rs b/src/cli/opts.rs
index e8e62485..98b34514 100644
--- a/src/cli/opts.rs
+++ b/src/cli/opts.rs
@@ -99,7 +99,15 @@ pub enum Command {
 
     /// Lists saved checkpoints of the swaps
     #[clap(aliases = &["lc"])]
-    ListCheckpoints,
+    ListCheckpoints {
+        #[clap(
+            short,
+            long,
+            default_value = "all",
+            possible_values = &["all", "All", "available", "Available", "available-for-restore"],
+        )]
+        select: CheckpointSelector,
+    },
 
     /// Checks the health of the syncers
     #[clap(aliases = &["hc"])]
@@ -299,7 +307,6 @@ pub enum OfferSelector {
     #[display("All")]
     All,
 }
-
 impl FromStr for OfferSelector {
     type Err = OfferSelectorParseError;
     fn from_str(input: &str) -> Result<OfferSelector, Self::Err> {
@@ -320,6 +327,33 @@ pub enum OfferSelectorParseError {
     Invalid,
 }
 
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Display, From)]
+#[display(Debug)]
+pub enum CheckpointSelector {
+    All,
+    AvailableForRestore,
+}
+
+impl FromStr for CheckpointSelector {
+    type Err = CheckpointSelectorParseError;
+    fn from_str(input: &str) -> Result<CheckpointSelector, Self::Err> {
+        match input {
+            "all" | "All" => Ok(CheckpointSelector::All),
+            "available" | "Available" | "available-for-restore" => {
+                Ok(CheckpointSelector::AvailableForRestore)
+            }
+            _ => Err(CheckpointSelectorParseError::Invalid),
+        }
+    }
+}
+
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Display, Error, From)]
+#[display(doc_comments)]
+pub enum CheckpointSelectorParseError {
+    /// The provided value can't be parsed as an offer selector
+    Invalid,
+}
+
 #[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Display, Error, From)]
 #[display(doc_comments)]
 pub enum AmountOfAssetParseError {
diff --git a/src/config.rs b/src/config.rs
index ddc7dfe8..ce6fbc5e 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -72,6 +72,17 @@ impl Config {
         }
     }
 
+    /// Returns if auto restore is enabled. Default to true
+    pub fn auto_restore_enable(&self) -> bool {
+        match &self.farcasterd {
+            Some(FarcasterdConfig {
+                auto_restore: Some(enable),
+                ..
+            }) => *enable,
+            _ => true,
+        }
+    }
+
     /// Returns the auto-funding configuration for a given network if enable, if None no
     /// configuration is found
     pub fn get_auto_funding_config(&self, network: Network) -> Option<AutoFundingServers> {
@@ -106,18 +117,29 @@ impl Config {
 impl Default for Config {
     fn default() -> Self {
         Config {
-            farcasterd: None,
+            farcasterd: Some(FarcasterdConfig::default()),
             grpc: None,
             syncers: Some(SyncersConfig::default()),
         }
     }
 }
 
-#[derive(Deserialize, Serialize, Debug, Clone, Default)]
+#[derive(Deserialize, Serialize, Debug, Clone)]
 #[serde(crate = "serde_crate")]
 pub struct FarcasterdConfig {
     /// Sets the auto-funding parameters, default to no auto-fund
     pub auto_funding: Option<AutoFundingConfig>,
+    /// Whether checkpoints should be auto restored at start-up, or not
+    pub auto_restore: Option<bool>,
+}
+
+impl Default for FarcasterdConfig {
+    fn default() -> Self {
+        FarcasterdConfig {
+            auto_funding: None,
+            auto_restore: Some(true),
+        }
+    }
 }
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
diff --git a/src/farcasterd/runtime.rs b/src/farcasterd/runtime.rs
index 98d99381..44d2d081 100644
--- a/src/farcasterd/runtime.rs
+++ b/src/farcasterd/runtime.rs
@@ -91,6 +91,7 @@ pub fn run(
         node_public_key: None,
         listens: none!(),
         started: SystemTime::now(),
+        auto_restored: false,
         spawning_services: none!(),
         registered_services: none!(),
         public_offers: none!(),
@@ -112,6 +113,7 @@ pub struct Runtime {
     identity: ServiceId,                         // Set on Runtime instantiation
     wallet_token: Token,                         // Set on Runtime instantiation
     started: SystemTime,                         // Set on Runtime instantiation
+    auto_restored: bool,                         // Set on Runtime instantiation
     node_secret_key: Option<SecretKey>, // Set by Keys request shortly after Hello from walletd
     node_public_key: Option<PublicKey>, // Set by Keys request shortly after Hello from walletd
     pub listens: HashSet<InetSocketAddr>, // Set by MakeOffer, contains unique socket addresses of the binding peerd listeners.
@@ -248,6 +250,7 @@ impl Runtime {
                             ServiceId::Database,
                             BusMsg::Ctl(CtlMsg::CleanDanglingOffers),
                         )?;
+                        self.handle_auto_restore(endpoints)?;
                     }
                     ServiceId::Wallet => {
                         self.registered_services.insert(source.clone());
@@ -337,6 +340,7 @@ impl Runtime {
                 debug!("received peerd keys {}", sk.display_secret());
                 self.node_secret_key = Some(sk);
                 self.node_public_key = Some(pk);
+                self.handle_auto_restore(endpoints)?;
             }
 
             CtlMsg::PeerdTerminated if matches!(source, ServiceId::Peer(..)) => {
@@ -555,6 +559,33 @@ impl Runtime {
                 }
             }
 
+            // From client: Request a list of checkpoints available for restore.
+            // From internal: Trigger restore on a list of checkpoints.
+            //
+            // If the request commes from a client, return the diff between the list and running
+            // swaps, otherwise handle a restore command for each checkpoint.
+            InfoMsg::CheckpointList(mut list) => {
+                if matches!(source, ServiceId::Client(_) | ServiceId::GrpcdClient(_)) {
+                    self.send_client_info(
+                        endpoints,
+                        source,
+                        InfoMsg::CheckpointList(
+                            list.drain(..)
+                                .filter(|c| !self.running_swaps_contain(&c.swap_id))
+                                .collect(),
+                        ),
+                    )?;
+                } else {
+                    for checkpoint in list.drain(..) {
+                        self.handle_ctl(
+                            endpoints,
+                            source.clone(),
+                            CtlMsg::RestoreCheckpoint(checkpoint),
+                        )?;
+                    }
+                }
+            }
+
             // Add the request's source to the subscription list for later progress notifications
             // and send all notifications already in the queue
             InfoMsg::SubscribeProgress(swap_id) => {
@@ -666,6 +697,29 @@ impl Runtime {
         self.process_request_with_state_machines(BusMsg::Sync(request), source, endpoints)
     }
 
+    fn handle_auto_restore(&mut self, endpoints: &mut Endpoints) -> Result<(), Error> {
+        if self.config.auto_restore_enable()
+            && self.services_ready().is_ok()
+            && self.peer_keys_ready().is_ok()
+            && !self.auto_restored
+        {
+            info!(
+                "{} will {} checkpoints",
+                "farcasterd".label(),
+                "auto restore".label()
+            );
+            // Retrieve all checkpoint info from farcasterd triggers restore of all checkpoints
+            endpoints.send_to(
+                ServiceBus::Info,
+                self.identity(),
+                ServiceId::Database,
+                BusMsg::Info(InfoMsg::RetrieveAllCheckpointInfo),
+            )?;
+            self.auto_restored = true;
+        }
+        Ok(())
+    }
+
     pub fn services_ready(&self) -> Result<(), Error> {
         if !self.registered_services.contains(&ServiceId::Wallet) {
             Err(Error::Farcaster(
diff --git a/src/grpcd/proto/farcaster.proto b/src/grpcd/proto/farcaster.proto
index 0bc60bfe..6c50f41e 100644
--- a/src/grpcd/proto/farcaster.proto
+++ b/src/grpcd/proto/farcaster.proto
@@ -115,6 +115,7 @@ message ListOffersResponse {
 
 message CheckpointsRequest {
     uint32 id = 1;
+    CheckpointSelector selector = 2;
 }
 
 message CheckpointsResponse {
@@ -122,6 +123,11 @@ message CheckpointsResponse {
     repeated CheckpointEntry checkpoint_entries = 2;
 }
 
+enum CheckpointSelector {
+    ALL_CHECKPOINTS = 0;
+    AVAILABLE_FOR_RESTORE = 1;
+}
+
 message CheckpointEntry {
     string swap_id = 1;
     string public_offer = 2;
diff --git a/src/grpcd/runtime.rs b/src/grpcd/runtime.rs
index 1a1efd55..4afaa182 100644
--- a/src/grpcd/runtime.rs
+++ b/src/grpcd/runtime.rs
@@ -587,16 +587,52 @@ impl Farcaster for FarcasterService {
         request: GrpcRequest<CheckpointsRequest>,
     ) -> Result<GrpcResponse<CheckpointsResponse>, Status> {
         debug!("Received a grpc checkpoints request: {:?}", request);
-        let oneshot_rx = self
-            .process_request(BusMsg::Bridge(BridgeMsg::Info {
-                request: InfoMsg::RetrieveAllCheckpointInfo,
-                service_id: ServiceId::Database,
-            }))
-            .await?;
+        let CheckpointsRequest { id, selector } = request.into_inner();
+
+        let selector = farcaster::CheckpointSelector::from_i32(selector)
+            .ok_or(Status::invalid_argument("selector"))?
+            .into();
+
+        let oneshot_rx = match selector {
+            farcaster::CheckpointSelector::AllCheckpoints => {
+                self.process_request(BusMsg::Bridge(BridgeMsg::Info {
+                    request: InfoMsg::RetrieveAllCheckpointInfo,
+                    service_id: ServiceId::Database,
+                }))
+                .await?
+            }
+            farcaster::CheckpointSelector::AvailableForRestore => {
+                let oneshot_rx = self
+                    .process_request(BusMsg::Bridge(BridgeMsg::Info {
+                        request: InfoMsg::RetrieveAllCheckpointInfo,
+                        service_id: ServiceId::Database,
+                    }))
+                    .await?;
+                match oneshot_rx.await {
+                    Ok(BusMsg::Info(InfoMsg::CheckpointList(checkpoint_entries))) => {
+                        self.process_request(BusMsg::Bridge(BridgeMsg::Info {
+                            request: InfoMsg::CheckpointList(checkpoint_entries),
+                            service_id: ServiceId::Farcasterd,
+                        }))
+                        .await?
+                    }
+                    Err(error) => {
+                        return Err(Status::internal(format!("{}", error)));
+                    }
+                    Ok(BusMsg::Ctl(CtlMsg::Failure(Failure { info, .. }))) => {
+                        return Err(Status::internal(info));
+                    }
+                    _ => {
+                        return Err(Status::invalid_argument("received invalid response"));
+                    }
+                }
+            }
+        };
+
         match oneshot_rx.await {
             Ok(BusMsg::Info(InfoMsg::CheckpointList(checkpoint_entries))) => {
                 let reply = farcaster::CheckpointsResponse {
-                    id: request.into_inner().id,
+                    id,
                     checkpoint_entries: checkpoint_entries
                         .iter()
                         .map(|entry| farcaster::CheckpointEntry {


Test patch:
diff --git a/tests/cfg/fc1.ci.toml b/tests/cfg/fc1.ci.toml
index 66159767..fd21cb49 100644
--- a/tests/cfg/fc1.ci.toml
+++ b/tests/cfg/fc1.ci.toml
@@ -1,3 +1,6 @@
+[farcasterd]
+auto_restore = false
+
 [grpc]
 enable = true
 bind_port = 23432
diff --git a/tests/cfg/fc1.toml b/tests/cfg/fc1.toml
index e99b1d60..2fd03ce7 100644
--- a/tests/cfg/fc1.toml
+++ b/tests/cfg/fc1.toml
@@ -1,3 +1,6 @@
+[farcasterd]
+auto_restore = false
+
 [grpc]
 enable = true
 bind_port = 23432
diff --git a/tests/cfg/fc2.ci.toml b/tests/cfg/fc2.ci.toml
index c4766b4e..9ba9e7d2 100644
--- a/tests/cfg/fc2.ci.toml
+++ b/tests/cfg/fc2.ci.toml
@@ -1,3 +1,6 @@
+[farcasterd]
+auto_restore = false
+
 [grpc]
 enable = true
 bind_port = 23433
diff --git a/tests/cfg/fc2.toml b/tests/cfg/fc2.toml
index 709e5d31..b0920b9d 100644
--- a/tests/cfg/fc2.toml
+++ b/tests/cfg/fc2.toml
@@ -1,3 +1,6 @@
+[farcasterd]
+auto_restore = false
+
 [grpc]
 enable = true
 bind_port = 23433
diff --git a/tests/grpc.rs b/tests/grpc.rs
index 76920a91..aab7f0d8 100644
--- a/tests/grpc.rs
+++ b/tests/grpc.rs
@@ -2,10 +2,10 @@
 extern crate log;
 
 use crate::farcaster::{
-    farcaster_client::FarcasterClient, AbortSwapRequest, CheckpointsRequest, InfoResponse,
-    ListOffersRequest, MakeRequest, NeedsFundingRequest, OfferInfoRequest, OfferSelector,
-    PeersRequest, ProgressRequest, RestoreCheckpointRequest, RevokeOfferRequest, SwapInfoRequest,
-    SweepAddressRequest, TakeRequest,
+    farcaster_client::FarcasterClient, AbortSwapRequest, CheckpointSelector, CheckpointsRequest,
+    InfoResponse, ListOffersRequest, MakeRequest, NeedsFundingRequest, OfferInfoRequest,
+    OfferSelector, PeersRequest, ProgressRequest, RestoreCheckpointRequest, RevokeOfferRequest,
+    SwapInfoRequest, SweepAddressRequest, TakeRequest,
 };
 use bitcoincore_rpc::RpcApi;
 use farcaster::{InfoRequest, MakeResponse, NeedsFundingResponse};
@@ -56,7 +56,10 @@ async fn grpc_server_functional_test() {
     assert_eq!(response.unwrap().into_inner().id, 1);
 
     // Test list checkpoints
-    let request = tonic::Request::new(CheckpointsRequest { id: 2 });
+    let request = tonic::Request::new(CheckpointsRequest {
+        id: 2,
+        selector: CheckpointSelector::AllCheckpoints.into(),
+    });
     let response = farcaster_client_1.checkpoints(request).await;
     assert_eq!(response.unwrap().into_inner().id, 2);