Merge branch 'development' into swarm-list-all-instances

* development: feat(swarm): json-rpc call to delete instance data (#1059) chore(deps): bump docker/build-push-action from 5 to 6 (#1061) refactor(wallet_daemon): use the indexer for substate listing (#1060)
tari-project · Jun 25, 2024 · 1f255b2 · 1f255b2
2 parents aebd5e0 + 3941ac4
commit 1f255b2
Show file tree

Hide file tree

Showing 32 changed files with 284 additions and 122 deletions.
diff --git a/.github/workflows/build_dockers_workflow.yml b/.github/workflows/build_dockers_workflow.yml
@@ -102,7 +102,7 @@ jobs:
 
       - name: Docker image build and push
         id: docker_build
-        uses: docker/build-push-action@v5
+        uses: docker/build-push-action@v6
         with:
           context: .
           file: ./docker_rig/${{ env.IMAGE_NAME }}.Dockerfile

diff --git a/applications/tari_dan_wallet_daemon/src/handlers/substates.rs b/applications/tari_dan_wallet_daemon/src/handlers/substates.rs
@@ -47,19 +47,22 @@ pub async fn handle_list(
     let sdk = context.wallet_sdk().clone();
     sdk.jwt_api().check_auth(token, &[JrpcPermission::SubstatesRead])?;
 
-    // TODO: pagination
-    let substates =
-        sdk.substate_api()
-            .list_substates(req.filter_by_type, req.filter_by_template.as_ref(), None, None)?;
+    let result = sdk
+        .get_network_interface()
+        .list_substates(req.filter_by_template, req.filter_by_type, req.limit, req.offset)
+        .await?;
 
-    let substates = substates
+    let substates = result
+        .substates
         .into_iter()
-        .map(|substate| WalletSubstateRecord {
-            substate_id: substate.address.substate_id,
-            parent_id: substate.parent_address,
-            version: substate.address.version,
-            template_address: substate.template_address,
-            module_name: substate.module_name,
+        // TODO: should also add the "timestamp" and "type" fields from the indexer list items?
+        .map(|s| WalletSubstateRecord {
+            substate_id: s.substate_id,
+            // TODO: should we remove the "parent_id" field from the wallet API? is it really needed somewhere?
+            parent_id: None,
+            version: s.version,
+            template_address: s.template_address,
+            module_name: s.module_name,
         })
         .collect();
 

diff --git a/applications/tari_dan_wallet_daemon/src/indexer_jrpc_impl.rs b/applications/tari_dan_wallet_daemon/src/indexer_jrpc_impl.rs
@@ -5,8 +5,10 @@ use std::sync::{Arc, Mutex};
 
 use axum::async_trait;
 use reqwest::{IntoUrl, Url};
-use tari_dan_common_types::optional::IsNotFoundError;
+use tari_dan_common_types::{optional::IsNotFoundError, substate_type::SubstateType};
 use tari_dan_wallet_sdk::network::{
+    SubstateListItem,
+    SubstateListResult,
     SubstateQueryResult,
     TransactionFinalizedResult,
     TransactionQueryResult,
@@ -20,6 +22,8 @@ use tari_indexer_client::{
         GetSubstateRequest,
         GetTransactionResultRequest,
         IndexerTransactionFinalizedResult,
+        ListSubstateItem,
+        ListSubstatesRequest,
         SubmitTransactionRequest,
     },
 };
@@ -84,6 +88,45 @@ impl WalletNetworkInterface for IndexerJsonRpcNetworkInterface {
         })
     }
 
+    async fn list_substates(
+        &self,
+        filter_by_template: Option<TemplateAddress>,
+        filter_by_type: Option<SubstateType>,
+        limit: Option<u64>,
+        offset: Option<u64>,
+    ) -> Result<SubstateListResult, Self::Error> {
+        let mut client = self.get_client()?;
+        let result = client
+            .list_substates(ListSubstatesRequest {
+                filter_by_template,
+                filter_by_type,
+                limit,
+                offset,
+            })
+            .await?;
+        let substates = result
+            .substates
+            .into_iter()
+            .map(|s| {
+                let ListSubstateItem {
+                    substate_id,
+                    module_name,
+                    version,
+                    template_address,
+                    timestamp,
+                } = s;
+                SubstateListItem {
+                    substate_id,
+                    module_name,
+                    version,
+                    template_address,
+                    timestamp,
+                }
+            })
+            .collect();
+        Ok(SubstateListResult { substates })
+    }
+
     async fn submit_transaction(
         &self,
         transaction: Transaction,

diff --git a/applications/tari_indexer/src/substate_manager.rs b/applications/tari_indexer/src/substate_manager.rs
@@ -25,10 +25,10 @@ use std::{convert::TryInto, sync::Arc};
 use serde::{Deserialize, Serialize};
 use tari_common_types::types::FixedHash;
 use tari_dan_app_utilities::substate_file_cache::SubstateFileCache;
-use tari_dan_common_types::PeerAddress;
+use tari_dan_common_types::{substate_type::SubstateType, PeerAddress};
 use tari_engine_types::substate::{Substate, SubstateId};
 use tari_epoch_manager::base_layer::EpochManagerHandle;
-use tari_indexer_client::types::{ListSubstateItem, SubstateType};
+use tari_indexer_client::types::ListSubstateItem;
 use tari_indexer_lib::{substate_scanner::SubstateScanner, NonFungibleSubstate};
 use tari_template_lib::models::TemplateAddress;
 use tari_transaction::TransactionId;

diff --git a/applications/tari_indexer/src/substate_storage_sqlite/sqlite_substate_store_factory.rs b/applications/tari_indexer/src/substate_storage_sqlite/sqlite_substate_store_factory.rs
@@ -38,11 +38,11 @@ use diesel::{
 };
 use diesel_migrations::{EmbeddedMigrations, MigrationHarness};
 use log::*;
-use tari_dan_common_types::{shard::Shard, Epoch};
+use tari_dan_common_types::{shard::Shard, substate_type::SubstateType, Epoch};
 use tari_dan_storage::{consensus_models::BlockId, StorageError};
 use tari_dan_storage_sqlite::{error::SqliteStorageError, SqliteTransaction};
 use tari_engine_types::substate::SubstateId;
-use tari_indexer_client::types::{ListSubstateItem, SubstateType};
+use tari_indexer_client::types::ListSubstateItem;
 use tari_template_lib::models::TemplateAddress;
 use tari_transaction::TransactionId;
 use thiserror::Error;

diff --git a/applications/tari_swarm_daemon/src/process_definitions/definition.rs b/applications/tari_swarm_daemon/src/process_definitions/definition.rs
@@ -1,6 +1,8 @@
 //   Copyright 2024 The Tari Project
 //   SPDX-License-Identifier: BSD-3-Clause
 
+use std::path::PathBuf;
+
 use async_trait::async_trait;
 use tokio::process::Command;
 
@@ -9,4 +11,7 @@ use super::context::ProcessContext;
 #[async_trait]
 pub trait ProcessDefinition: Send {
     async fn get_command(&self, context: ProcessContext<'_>) -> anyhow::Result<Command>;
+    fn get_relative_data_path(&self) -> Option<PathBuf> {
+        None
+    }
 }
diff --git a/applications/tari_swarm_daemon/src/process_definitions/indexer.rs b/applications/tari_swarm_daemon/src/process_definitions/indexer.rs
@@ -1,6 +1,8 @@
 //   Copyright 2024 The Tari Project
 //   SPDX-License-Identifier: BSD-3-Clause
 
+use std::path::PathBuf;
+
 use anyhow::anyhow;
 use async_trait::async_trait;
 use tokio::process::Command;
@@ -54,4 +56,8 @@ impl ProcessDefinition for Indexer {
 
         Ok(command)
     }
+
+    fn get_relative_data_path(&self) -> Option<PathBuf> {
+        Some("data".into())
+    }
 }
diff --git a/applications/tari_swarm_daemon/src/process_definitions/minotari_node.rs b/applications/tari_swarm_daemon/src/process_definitions/minotari_node.rs
@@ -1,6 +1,8 @@
 //   Copyright 2024 The Tari Project
 //   SPDX-License-Identifier: BSD-3-Clause
 
+use std::path::PathBuf;
+
 use async_trait::async_trait;
 use log::debug;
 use tokio::process::Command;
@@ -63,7 +65,7 @@ impl ProcessDefinition for MinotariNode {
         Ok(command)
     }
 
-    // fn get_relative_data_path(&self) -> Option<PathBuf> {
-    //     Some(Path::new("network/data"))
-    // }
+    fn get_relative_data_path(&self) -> Option<PathBuf> {
+        Some("data".into())
+    }
 }
diff --git a/applications/tari_swarm_daemon/src/process_definitions/minotari_wallet.rs b/applications/tari_swarm_daemon/src/process_definitions/minotari_wallet.rs
@@ -1,6 +1,8 @@
 //   Copyright 2024 The Tari Project
 //   SPDX-License-Identifier: BSD-3-Clause
 
+use std::path::PathBuf;
+
 use anyhow::anyhow;
 use async_trait::async_trait;
 use log::*;
@@ -65,4 +67,8 @@ impl ProcessDefinition for MinotariWallet {
 
         Ok(command)
     }
+
+    fn get_relative_data_path(&self) -> Option<PathBuf> {
+        Some("data".into())
+    }
 }
diff --git a/applications/tari_swarm_daemon/src/process_definitions/validator_node.rs b/applications/tari_swarm_daemon/src/process_definitions/validator_node.rs
@@ -1,6 +1,8 @@
 //   Copyright 2024 The Tari Project
 //   SPDX-License-Identifier: BSD-3-Clause
 
+use std::path::PathBuf;
+
 use anyhow::anyhow;
 use async_trait::async_trait;
 use log::debug;
@@ -63,4 +65,8 @@ impl ProcessDefinition for ValidatorNode {
 
         Ok(command)
     }
+
+    fn get_relative_data_path(&self) -> Option<PathBuf> {
+        Some("data".into())
+    }
 }
diff --git a/applications/tari_swarm_daemon/src/process_definitions/wallet_daemon.rs b/applications/tari_swarm_daemon/src/process_definitions/wallet_daemon.rs
@@ -1,6 +1,8 @@
 //   Copyright 2024 The Tari Project
 //   SPDX-License-Identifier: BSD-3-Clause
 
+use std::path::PathBuf;
+
 use anyhow::anyhow;
 use async_trait::async_trait;
 use tokio::process::Command;
@@ -69,4 +71,8 @@ impl ProcessDefinition for WalletDaemon {
 
         Ok(command)
     }
+
+    fn get_relative_data_path(&self) -> Option<PathBuf> {
+        Some("data".into())
+    }
 }
diff --git a/applications/tari_swarm_daemon/src/process_manager/handle.rs b/applications/tari_swarm_daemon/src/process_manager/handle.rs
@@ -33,6 +33,10 @@ pub enum ProcessManagerRequest {
         instance_id: InstanceId,
         reply: Reply<()>,
     },
+    DeleteInstanceData {
+        instance_id: InstanceId,
+        reply: Reply<()>,
+    },
     MineBlocks {
         blocks: u64,
         reply: Reply<()>,
@@ -200,6 +204,18 @@ impl ProcessManagerHandle {
         rx_reply.await?
     }
 
+    pub async fn delete_instance_data(&self, instance_id: InstanceId) -> anyhow::Result<()> {
+        let (tx_reply, rx_reply) = oneshot::channel();
+        self.tx_request
+            .send(ProcessManagerRequest::DeleteInstanceData {
+                instance_id,
+                reply: tx_reply,
+            })
+            .await?;
+
+        rx_reply.await?
+    }
+
     pub async fn register_validator_node(&self, instance_id: InstanceId) -> anyhow::Result<()> {
         let (tx_reply, rx_reply) = oneshot::channel();
         self.tx_request

diff --git a/applications/tari_swarm_daemon/src/process_manager/instances/manager.rs b/applications/tari_swarm_daemon/src/process_manager/instances/manager.rs
@@ -10,6 +10,7 @@ use std::{
 };
 
 use anyhow::anyhow;
+use log::info;
 use tari_common::configuration::Network;
 use tokio::{
     fs,
@@ -300,6 +301,26 @@ impl InstanceManager {
         Ok(())
     }
 
+    pub async fn delete_instance_data(&mut self, id: InstanceId) -> anyhow::Result<()> {
+        let instance = self
+            .instances_mut()
+            .find(|i| i.id() == id)
+            .ok_or_else(|| anyhow!("Instance not found"))?;
+
+        let definition = get_definition(instance.instance_type());
+
+        if let Some(data_path) = definition.get_relative_data_path() {
+            let path = instance.base_path().join(data_path);
+            info!(
+                "Deleting data directory for instance {}: {}",
+                instance.name(),
+                path.display()
+            );
+            fs::remove_dir_all(path).await?;
+        }
+        Ok(())
+    }
+
     pub fn instances_mut(&mut self) -> impl Iterator<Item = &mut Instance> {
         self.minotari_nodes
             .values_mut()

diff --git a/applications/tari_swarm_daemon/src/process_manager/manager.rs b/applications/tari_swarm_daemon/src/process_manager/manager.rs
@@ -150,6 +150,12 @@ impl ProcessManager {
                     log::warn!("Request cancelled before response could be sent")
                 }
             },
+            DeleteInstanceData { instance_id, reply } => {
+                let result = self.instance_manager.delete_instance_data(instance_id).await;
+                if reply.send(result).is_err() {
+                    log::warn!("Request cancelled before response could be sent")
+                }
+            },
             MineBlocks { blocks, reply } => {
                 let result = self.mine(blocks).await;
                 if reply.send(result).is_err() {

diff --git a/applications/tari_swarm_daemon/src/webserver/rpc/instances.rs b/applications/tari_swarm_daemon/src/webserver/rpc/instances.rs
@@ -104,3 +104,35 @@ pub async fn list(context: &HandlerContext, req: ListInstancesRequest) -> Result
         instances: instances.into_iter().map(Into::into).collect(),
     })
 }
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct DeleteInstanceDataRequest {
+    pub name: String,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct DeleteInstanceDataResponse {
+    pub success: bool,
+}
+
+pub async fn delete_data(
+    context: &HandlerContext,
+    req: DeleteInstanceDataRequest,
+) -> Result<DeleteInstanceDataResponse, anyhow::Error> {
+    let instance = context
+        .process_manager()
+        .get_instance_by_name(req.name)
+        .await?
+        .ok_or_else(|| {
+            JsonRpcError::new(
+                JsonRpcErrorReason::ApplicationError(404),
+                "Instance not found".to_string(),
+                serde_json::Value::Null,
+            )
+        })?;
+
+    context.process_manager().stop_instance(instance.id).await?;
+    context.process_manager().delete_instance_data(instance.id).await?;
+
+    Ok(DeleteInstanceDataResponse { success: true })
+}
diff --git a/applications/tari_swarm_daemon/src/webserver/server.rs b/applications/tari_swarm_daemon/src/webserver/server.rs
@@ -110,6 +110,7 @@ async fn json_rpc_handler(Extension(context): Extension<Arc<HandlerContext>>, va
         "start" => call_handler(context, value, rpc::instances::start).await,
         "stop" => call_handler(context, value, rpc::instances::stop).await,
         "list_instances" => call_handler(context, value, rpc::instances::list).await,
+        "delete_data" => call_handler(context, value, rpc::instances::delete_data).await,
         _ => Ok(value.method_not_found(&value.method)),
     }
 }