[POC] Spawn embeded Container chain node on data preservers assignment

Cargo.toml

@@ -58,6 +58,7 @@ pallet-collator-assignment = { path = "pallets/collator-assignment", default-fea
 pallet-collator-assignment-runtime-api = { path = "pallets/collator-assignment/runtime-api", default-features = false }
 pallet-configuration = { path = "pallets/configuration", default-features = false }
 pallet-data-preservers = { path = "pallets/data-preservers", default-features = false }
+pallet-data-preservers-runtime-api = { path = "pallets/data-preservers/runtime-api", default-features = false }
 pallet-inflation-rewards = { path = "pallets/inflation-rewards", default-features = false }
 pallet-initializer = { path = "pallets/initializer", default-features = false }
 pallet-invulnerables = { path = "pallets/invulnerables", default-features = false }

client/orchestrator-chain-rpc-interface/src/lib.rs

@@ -20,8 +20,9 @@ use {
     async_trait::async_trait,
     core::pin::Pin,
     dc_orchestrator_chain_interface::{
-        BlockNumber, ContainerChainGenesisData, OrchestratorChainError, OrchestratorChainInterface,
-        OrchestratorChainResult, PHash, PHeader,
+        BlockNumber, ContainerChainGenesisData, DataPreserverAssignment, DataPreserverProfileId,
+        OrchestratorChainError, OrchestratorChainInterface, OrchestratorChainResult, PHash,
+        PHeader,
     },
     dp_core::ParaId,
     futures::{Stream, StreamExt},
@@ -114,7 +115,7 @@ impl OrchestratorChainRpcClient {
         };
         let res = self
             .request_tracing::<sp_core::Bytes, _>("state_call", params, |err| {
-                tracing::trace!(
+                tracing::debug!(
                     target: LOG_TARGET,
                     %method_name,
                     %hash,
@@ -289,7 +290,7 @@ impl OrchestratorChainInterface for OrchestratorChainRpcClient {
         orchestrator_parent: PHash,
         para_id: ParaId,
     ) -> OrchestratorChainResult<Option<ContainerChainGenesisData>> {
-        self.call_remote_runtime_function("genesis_data", orchestrator_parent, Some(para_id))
+        self.call_remote_runtime_function("RegistrarApi_genesis_data", orchestrator_parent, Some(para_id))
             .await
     }
 
@@ -298,7 +299,7 @@ impl OrchestratorChainInterface for OrchestratorChainRpcClient {
         orchestrator_parent: PHash,
         para_id: ParaId,
     ) -> OrchestratorChainResult<Vec<Vec<u8>>> {
-        self.call_remote_runtime_function("boot_nodes", orchestrator_parent, Some(para_id))
+        self.call_remote_runtime_function("RegistrarApi_boot_nodes", orchestrator_parent, Some(para_id))
             .await
     }
 
@@ -307,7 +308,7 @@ impl OrchestratorChainInterface for OrchestratorChainRpcClient {
         orchestrator_parent: PHash,
         para_id: ParaId,
     ) -> OrchestratorChainResult<Option<BlockNumber>> {
-        self.call_remote_runtime_function("latest_block_number", orchestrator_parent, Some(para_id))
+        self.call_remote_runtime_function("AuthorNotingApi_latest_block_number", orchestrator_parent, Some(para_id))
             .await
     }
 
@@ -318,4 +319,17 @@ impl OrchestratorChainInterface for OrchestratorChainRpcClient {
     async fn finalized_block_hash(&self) -> OrchestratorChainResult<PHash> {
         self.request("chain_getFinalizedHead", rpc_params![]).await
     }
+
+    async fn data_preserver_active_assignment(
+        &self,
+        orchestrator_parent: PHash,
+        profile_id: DataPreserverProfileId,
+    ) -> OrchestratorChainResult<DataPreserverAssignment<ParaId>> {
+        self.call_remote_runtime_function(
+            "DataPreserversApi_get_active_assignment",
+            orchestrator_parent,
+            Some(profile_id),
+        )
+        .await
+    }
 }

client/service-container-chain/Cargo.toml

@@ -27,6 +27,7 @@ dancebox-runtime = { workspace = true, features = [ "std" ] }
 manual-xcm-rpc = { workspace = true }
 node-common = { workspace = true }
 pallet-author-noting-runtime-api = { workspace = true, features = [ "std" ] }
+pallet-data-preservers = { workspace = true, features = [ "std" ] }
 services-payment-rpc = { workspace = true }
 stream-payment-rpc = { workspace = true }
 tc-consensus = { workspace = true }

client/service-container-chain/src/data_preservers.rs

@@ -0,0 +1,90 @@
+// Copyright (C) Moondance Labs Ltd.
+// This file is part of Tanssi.
+
+// Tanssi is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Tanssi is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Tanssi.  If not, see <http://www.gnu.org/licenses/>
+
+use {
+    crate::spawner::{ContainerChainSpawner, TSelectSyncMode},
+    dc_orchestrator_chain_interface::{
+        DataPreserverAssignment, OrchestratorChainInterface, OrchestratorChainResult,
+    },
+    futures::stream::StreamExt,
+    std::{future::Future},
+    tc_consensus::ParaId,
+};
+
+pub type ProfileId = <dancebox_runtime::Runtime as pallet_data_preservers::Config>::ProfileId;
+
+async fn try_fut<T, E>(fut: impl Future<Output = Result<T, E>>) -> Result<T, E> {
+    fut.await
+}
+
+/// Watch assignements by indefinitly listening to finalized block notifications and switching to
+/// the chain the profile is assigned to.
+pub async fn task_watch_assignment<S: TSelectSyncMode>(
+    spawner: ContainerChainSpawner<S>,
+    profile_id: ProfileId,
+) {
+    use dc_orchestrator_chain_interface::DataPreserverAssignment as Assignment;
+
+    if let OrchestratorChainResult::Err(e) = try_fut(async move {
+        let orchestrator_chain_interface = spawner.params.orchestrator_chain_interface.clone();
+
+        let mut current_assignment = DataPreserverAssignment::<ParaId>::NotAssigned;
+
+        let mut stream = orchestrator_chain_interface
+            .finality_notification_stream()
+            .await?;
+
+        while let Some(header) = stream.next().await {
+            let hash = header.hash();
+
+            let new_assignment = orchestrator_chain_interface
+                .data_preserver_active_assignment(hash, profile_id)
+                .await?;
+
+            log::info!("Assignement for block {hash}: {new_assignment:?}");
+
+            match (current_assignment, new_assignment) {
+                // no change
+                (x, y) if x == y => continue,
+                (
+                    Assignment::NotAssigned | Assignment::Inactive(_),
+                    Assignment::Active(para_id),
+                ) => {
+                    spawner.spawn(para_id, false).await;
+                }
+                (Assignment::Active(para_id), Assignment::Inactive(x)) if para_id == x => {
+                    spawner.stop(para_id, true); // keep db
+                }
+                (Assignment::Active(para_id), _) => {
+                    spawner.stop(para_id, false); // don't keep db
+                }
+                // don't do anything yet
+                (
+                    Assignment::NotAssigned | Assignment::Inactive(_),
+                    Assignment::NotAssigned | Assignment::Inactive(_),
+                ) => (),
+            }
+
+            current_assignment = new_assignment;
+        }
+
+        Ok(())
+    })
+    .await
+    {
+        log::error!("Error in data preservers assignement watching task: {e:?}");
+    }
+}

client/service-container-chain/src/lib.rs

@@ -16,6 +16,7 @@
 
 pub mod chain_spec;
 pub mod cli;
+pub mod data_preservers;
 pub mod monitor;
 pub mod rpc;
 pub mod service;