Fix undefined behavior on simultaneous write in daemon state file

cw-orch-daemon/Cargo.toml

@@ -67,6 +67,10 @@ async-recursion = "1.0.5"
 flate2 = { version = "1.0.26" }
 lazy_static = "1.4.0"
 
+# Lock daemon
+file-lock = { version = "2.1.10" }
+once_cell = { version = "1.19.0" }
+
 [dev-dependencies]
 cw-orch-daemon = { path = "." }
 uid = "0.1.7"
@@ -80,9 +84,12 @@ duct = "0.13"
 mock-contract = { path = "../contracts/mock_contract", features = [
   "interface",
 ] }
-serial_test = "3.0.0"
+serial_test = { version = "3.0.0" }
 
 # Ethereum deps
 ethers-signers = { version = "2.0.7" }
 ethers-core = { version = "2.0.7" }
 tokio-test = "0.4.3"
+
+# File lock test
+nix = { version = "0.28.0", features = ["process"] }

cw-orch-daemon/examples/daemon-capabilities.rs

@@ -16,7 +16,7 @@ pub fn main() -> anyhow::Result<()> {
     // We commit the tx (also resimulates the tx)
     // ANCHOR: send_tx
     let wallet = daemon.wallet();
-    let rt = daemon.rt_handle;
+    let rt = daemon.rt_handle.clone();
     rt.block_on(wallet.bank_send("<address-of-my-sister>", coins(345, "ujunox")))?;
     // ANCHOR_END: send_tx
 

cw-orch-daemon/src/builder.rs

@@ -1,13 +1,14 @@
 use crate::{
-    log::print_if_log_disabled, sender::SenderBuilder, sender::SenderOptions, DaemonAsync,
-    DaemonBuilder,
+    log::print_if_log_disabled,
+    sender::{SenderBuilder, SenderOptions},
+    DaemonAsync, DaemonBuilder, GrpcChannel,
 };
 use std::sync::Arc;
 
 use bitcoin::secp256k1::All;
 
 use super::{error::DaemonError, sender::Sender, state::DaemonState};
-use cw_orch_core::environment::ChainInfoOwned;
+use cw_orch_core::{environment::ChainInfoOwned, log::connectivity_target};
 
 /// The default deployment id if none is provided
 pub const DEFAULT_DEPLOYMENT: &str = "default";
@@ -30,13 +31,17 @@
     pub(crate) chain: Option<ChainInfoOwned>,
     // # Optional
     pub(crate) deployment_id: Option<String>,
+    pub(crate) state_path: Option<String>,
 
     /* Sender related options */
     /// Wallet sender
     /// Will be used in priority when set
     pub(crate) sender: Option<SenderBuilder<All>>,
     /// Specify Daemon Sender Options
     pub(crate) sender_options: SenderOptions,
+
+    /* Rebuilder related options */
+    pub(crate) state: Option<DaemonState>,
 }
 
 impl DaemonAsyncBuilder {
@@ -48,6 +53,8 @@
 
     /// Set the deployment id to use for the daemon interactions
     /// Defaults to `default`
+    ///
+    /// This field is ignored for rebuilt daemon and deployment id of the original daemon used instead
     pub fn deployment_id(&mut self, deployment_id: impl Into<String>) -> &mut Self {
         self.deployment_id = Some(deployment_id.into());
         self
@@ -87,32 +94,70 @@
         self
     }
 
+    /// Specifies path to the daemon state file
+    /// Defaults to env variable.
+    ///
+    /// Variable: STATE_FILE_ENV_NAME.
+    ///
+    /// This field is ignored for rebuilt daemon and path of the original daemon used instead
+    pub fn state_path(&mut self, path: impl ToString) -> &mut Self {
+        self.state_path = Some(path.to_string());
+        self
+    }
+
     /// Build a daemon
     pub async fn build(&self) -> Result<DaemonAsync, DaemonError> {
-        let chain = self
+        let chain_info = self
             .chain
             .clone()
             .ok_or(DaemonError::BuilderMissing("chain information".into()))?;
         let deployment_id = self
             .deployment_id
             .clone()
             .unwrap_or(DEFAULT_DEPLOYMENT.to_string());
-        let state = Arc::new(DaemonState::new(chain, deployment_id, false).await?);
+
+        if chain_info.grpc_urls.is_empty() {
+            return Err(DaemonError::GRPCListIsEmpty);
+        }
+
+        log::debug!(target: &connectivity_target(), "Found {} gRPC endpoints", chain_info.grpc_urls.len());
+
+        // find working grpc channel
+        let grpc_channel =
+            GrpcChannel::connect(&chain_info.grpc_urls, &chain_info.chain_id).await?;
+
+        let state = match &self.state {
+            Some(state) => state.clone(),
+            None => {
+                // If the path is relative, we dis-ambiguate it and take the root at $HOME/$CW_ORCH_STATE_FOLDER
+                let json_file_path = self
+                    .state_path
+                    .clone()
+                    .unwrap_or(DaemonState::state_file_path()?);
+
+                DaemonState::new(json_file_path, chain_info.clone(), deployment_id, false)?
+            }
+        };
         // if mnemonic provided, use it. Else use env variables to retrieve mnemonic
         let sender_options = self.sender_options.clone();
 
         let sender = match self.sender.clone() {
             Some(sender) => match sender {
-                SenderBuilder::Mnemonic(mnemonic) => {
-                    Sender::from_mnemonic_with_options(&state, &mnemonic, sender_options)?
-                }
+                SenderBuilder::Mnemonic(mnemonic) => Sender::from_mnemonic_with_options(
+                    chain_info.clone(),
+                    grpc_channel,
+                    &mnemonic,
+                    sender_options,
+                )?,
                 SenderBuilder::Sender(mut sender) => {
                     sender.set_options(self.sender_options.clone());
+                    sender.grpc_channel = grpc_channel;
                     sender
                 }
             },
-            None => Sender::new_with_options(&state, sender_options)?,
+            None => Sender::new_with_options(chain_info.clone(), grpc_channel, sender_options)?,
         };
+
         let daemon = DaemonAsync {
             state,
             sender: Arc::new(sender),
@@ -129,6 +174,8 @@
             deployment_id: value.deployment_id,
             sender_options: value.sender_options,
             sender: value.sender,
+            state: value.state,
+            state_path: value.state_path,
         }
     }
 }

cw-orch-daemon/src/core.rs

@@ -25,7 +25,6 @@
     fmt::Debug,
     io::Write,
     str::{from_utf8, FromStr},
-    sync::Arc,
     time::Duration,
 };
 
@@ -67,7 +66,7 @@
     /// Sender to send transactions to the chain
     pub sender: Wallet,
     /// State of the daemon
-    pub state: Arc<DaemonState>,
+    pub state: DaemonState,
 }
 
 impl DaemonAsync {
@@ -78,12 +77,12 @@
 
     /// Get the channel configured for this DaemonAsync.
     pub fn channel(&self) -> Channel {
-        self.state.grpc_channel.clone()
+        self.sender.grpc_channel.clone()
     }
 }
 
 impl ChainState for DaemonAsync {
-    type Out = Arc<DaemonState>;
+    type Out = DaemonState;
 
     fn state(&self) -> Self::Out {
         self.state.clone()
@@ -100,11 +99,13 @@
     /// Returns a new [`DaemonAsyncBuilder`] with the current configuration.
     /// Does not consume the original [`DaemonAsync`].
     pub fn rebuild(&self) -> DaemonAsyncBuilder {
-        let mut builder = Self::builder();
+        let mut builder = DaemonAsyncBuilder {
+            state: Some(self.state()),
+            ..Default::default()
+        };
         builder
-            .chain(self.state().chain_data.clone())
-            .sender((*self.sender).clone())
-            .deployment_id(&self.state().deployment_id);
+            .chain(self.sender.chain_info.clone())
+            .sender((*self.sender).clone());
         builder
     }
 
@@ -281,7 +282,7 @@
         _uploadable: &T,
     ) -> Result<CosmTxResponse, DaemonError> {
         let sender = &self.sender;
-        let wasm_path = <T as Uploadable>::wasm(&self.state.chain_data);
+        let wasm_path = <T as Uploadable>::wasm(&self.sender.chain_info);
 
         log::debug!(target: &transaction_target(), "Uploading file at {:?}", wasm_path);
 

cw-orch-daemon/src/error.rs

@@ -116,12 +116,14 @@ pub enum DaemonError {
     InsufficientFee(String),
     #[error("Not enough balance, expected {expected}, found {current}")]
     NotEnoughBalance { expected: Coin, current: Coin },
-    #[error("Can't set the daemon state, it's read-only")]
-    StateReadOnly,
+    #[error("Can't set the daemon state, it's read-only {0}")]
+    StateReadOnly(String),
     #[error("You need to pass a runtime to the querier object to do synchronous queries. Use daemon.querier instead")]
     QuerierNeedRuntime,
     #[error(transparent)]
     Instantiate2Error(#[from] Instantiate2AddressError),
+    #[error("State file {0} already locked, use another state file or clone daemon which holds the lock")]
+    StateAlreadyLocked(String),
 }
 
 impl DaemonError {