feat(engine)!: syncing published templates

applications/tari_dan_app_utilities/Cargo.toml

@@ -29,6 +29,8 @@ tari_bor = { workspace = true, default-features = true }
 tari_indexer_lib = { workspace = true }
 tari_networking = { workspace = true }
 tari_validator_node_rpc = { workspace = true }
+tari_dan_p2p = { workspace = true }
+tari_rpc_framework = { workspace = true }
 
 anyhow = { workspace = true }
 async-trait = { workspace = true }

applications/tari_dan_app_utilities/src/template_manager/implementation/initializer.rs

@@ -20,15 +20,24 @@
 //  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use tari_dan_common_types::NodeAddressable;
+use tari_dan_common_types::{NodeAddressable, PeerAddress, TemplateSyncRequest};
+use tari_epoch_manager::base_layer::EpochManagerHandle;
 use tari_shutdown::ShutdownSignal;
-use tokio::{sync::mpsc, task::JoinHandle};
+use tari_validator_node_rpc::client::TariValidatorNodeRpcClientFactory;
+use tokio::{
+    sync::{broadcast, mpsc},
+    task::JoinHandle,
+};
 
 use super::{downloader::TemplateDownloadWorker, service::TemplateManagerService, TemplateManager};
 use crate::template_manager::interface::TemplateManagerHandle;
 
-pub fn spawn<TAddr: NodeAddressable + 'static>(
+pub fn spawn<TAddr: NodeAddressable + 'static, S: tari_state_store_sqlite::SubstateStore + Send + Sync + 'static>(
     manager: TemplateManager<TAddr>,
+    epoch_manager: EpochManagerHandle<PeerAddress>,
+    client_factory: TariValidatorNodeRpcClientFactory,
+    rx_template_sync: broadcast::Receiver<TemplateSyncRequest>,
+    state_store: S,
     shutdown: ShutdownSignal,
 ) -> (TemplateManagerHandle, JoinHandle<anyhow::Result<()>>) {
     let (tx_request, rx_request) = mpsc::channel(1);
@@ -37,8 +46,17 @@ pub fn spawn<TAddr: NodeAddressable + 'static>(
     let (tx_download_queue, rx_download_queue) = mpsc::channel(1);
     let (tx_completed_downloads, rx_completed_downloads) = mpsc::channel(1);
 
-    let join_handle =
-        TemplateManagerService::spawn(rx_request, manager, tx_download_queue, rx_completed_downloads, shutdown);
+    let join_handle = TemplateManagerService::spawn(
+        rx_request,
+        manager,
+        epoch_manager,
+        tx_download_queue,
+        rx_completed_downloads,
+        rx_template_sync,
+        client_factory,
+        state_store,
+        shutdown,
+    );
     TemplateDownloadWorker::new(rx_download_queue, tx_completed_downloads).spawn();
     (handle, join_handle)
 }

applications/tari_dan_app_utilities/src/template_manager/implementation/manager.rs

@@ -20,13 +20,25 @@
 //  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::{collections::HashMap, convert::TryFrom, fs, sync::Arc};
+use std::{
+    collections::HashMap,
+    convert::TryFrom,
+    fs,
+    sync::Arc,
+    thread,
+    time::{SystemTime, UNIX_EPOCH},
+};
 
-use chrono::Utc;
+use chrono::{Duration, Utc};
 use log::*;
 use tari_common_types::types::{FixedHash, PublicKey};
 use tari_crypto::tari_utilities::ByteArray;
-use tari_dan_common_types::{optional::Optional, services::template_provider::TemplateProvider, NodeAddressable};
+use tari_dan_common_types::{
+    optional::Optional,
+    services::template_provider::TemplateProvider,
+    NodeAddressable,
+    TemplateSyncRequest,
+};
 use tari_dan_engine::{
     flow::FlowFactory,
     function_definitions::FlowFunctionDefinition,
@@ -54,6 +66,12 @@ const LOG_TARGET: &str = "tari::validator_node::template_manager";
 
 const CONCURRENT_ACCESS_LIMIT: isize = 100;
 
+#[derive(Debug, Clone)]
+pub enum TemplateResult {
+    Template(Box<Template>),
+    PendingTemplate,
+}
+
 #[derive(Debug)]
 pub struct TemplateManager<TAddr> {
     global_db: GlobalDb<SqliteGlobalDbAdapter<TAddr>>,
@@ -120,26 +138,31 @@ impl<TAddr: NodeAddressable> TemplateManager<TAddr> {
                 name: name.to_string(),
                 address,
                 binary_sha,
+                author_public_key: Default::default(),
             },
             executable: TemplateExecutable::CompiledWasm(compiled_code),
         }
     }
 
-    pub fn template_exists(&self, address: &TemplateAddress) -> Result<bool, TemplateManagerError> {
+    pub fn template_exists(
+        &self,
+        address: &TemplateAddress,
+        status: TemplateStatus,
+    ) -> Result<bool, TemplateManagerError> {
         if self.builtin_templates.contains_key(address) {
             return Ok(true);
         }
         let mut tx = self.global_db.create_transaction()?;
         self.global_db
             .templates(&mut tx)
-            .template_exists(address)
+            .template_exists(address, status)
             .map_err(|_| TemplateManagerError::TemplateNotFound { address: *address })
     }
 
-    pub fn fetch_template(&self, address: &TemplateAddress) -> Result<Template, TemplateManagerError> {
+    pub fn fetch_template(&self, address: &TemplateAddress) -> Result<TemplateResult, TemplateManagerError> {
         // first of all, check if the address is for a bulitin template
         if let Some(template) = self.builtin_templates.get(address) {
-            return Ok(template.to_owned());
+            return Ok(TemplateResult::Template(Box::new(template.to_owned())));
         }
 
         let mut tx = self.global_db.create_transaction()?;
@@ -149,6 +172,11 @@ impl<TAddr: NodeAddressable> TemplateManager<TAddr> {
             .get_template(address)?
             .ok_or(TemplateManagerError::TemplateNotFound { address: *address })?;
 
+        // notify the caller that the template is under sync, so not yet ready
+        if matches!(template.status, TemplateStatus::Pending) {
+            return Ok(TemplateResult::PendingTemplate);
+        }
+
         if !matches!(template.status, TemplateStatus::Active | TemplateStatus::Deprecated) {
             return Err(TemplateManagerError::TemplateUnavailable);
         }
@@ -167,9 +195,9 @@ impl<TAddr: NodeAddressable> TemplateManager<TAddr> {
                 _ => return Err(TemplateManagerError::TemplateUnavailable),
             }
 
-            Ok(result)
+            Ok(TemplateResult::Template(Box::new(result)))
         } else {
-            Ok(template.into())
+            Ok(TemplateResult::Template(Box::new(template.into())))
         }
     }
 
@@ -185,6 +213,27 @@ impl<TAddr: NodeAddressable> TemplateManager<TAddr> {
         Ok(templates)
     }
 
+    pub(super) fn add_pending_template(
+        &self,
+        template_address: tari_engine_types::TemplateAddress,
+    ) -> Result<(), TemplateManagerError> {
+        let template = DbTemplate::empty_pending(template_address);
+
+        let mut tx = self.global_db.create_transaction()?;
+        let mut templates_db = self.global_db.templates(&mut tx);
+        match templates_db.get_template(&template.template_address)? {
+            Some(_) => templates_db.update_template(
+                &template.template_address,
+                DbTemplateUpdate::status(TemplateStatus::Pending),
+            )?,
+            None => templates_db.insert_template(template)?,
+        }
+
+        tx.commit()?;
+
+        Ok(())
+    }
+
     pub(super) fn add_template(
         &self,
         author_public_key: PublicKey,
@@ -301,10 +350,34 @@ impl<TAddr: NodeAddressable + Send + Sync + 'static> TemplateProvider for Templa
             return Ok(Some(template));
         }
 
-        let Some(template) = self.fetch_template(address).optional()? else {
+        let Some(template_result) = self.fetch_template(address).optional()? else {
             return Ok(None);
         };
+
         debug!(target: LOG_TARGET, "CACHE MISS: Template {}", address);
+
+        // getting template
+        let template = match template_result {
+            TemplateResult::Template(template) => template,
+            TemplateResult::PendingTemplate => {
+                let start = SystemTime::now();
+                let mut template = None;
+                while template.is_none() {
+                    let elapsed = start.duration_since(UNIX_EPOCH).expect("Time went backwards");
+                    if elapsed.gt(&self.config.pending_templates_wait_timeout()) {
+                        break;
+                    }
+                    if let Some(TemplateResult::Template(fetched_template)) = self.fetch_template(address).optional()? {
+                        template = Some(fetched_template);
+                    }
+                    // sleeping here to not overload the local database while waiting for the template to be ready
+                    thread::sleep(std::time::Duration::from_millis(100));
+                }
+                debug!(target: LOG_TARGET, "Failed to fetch template {} within {:?}", address, self.config.pending_templates_wait_timeout());
+                template.ok_or(Self::Error::TemplateUnavailable)?
+            },
+        };
+
         let loaded = match template.executable {
             TemplateExecutable::CompiledWasm(wasm) => {
                 let module = WasmModule::from_code(wasm);

applications/tari_dan_app_utilities/src/template_manager/implementation/mod.rs

@@ -26,7 +26,8 @@ mod initializer;
 pub use initializer::spawn;
 
 mod manager;
-pub use manager::TemplateManager;
+pub use manager::{TemplateManager, TemplateResult};
+
 mod service;
 
 mod cmap_semaphore;