Feat/platform orchestrator stage 2 (#245)

* add platform-orchestrator canister

* user_index canister changes

* individual_user canister changes

* added platform orchestrator

* chore: refactor and fixes

* fix: new user signups

* user_index stage 1 upgrade

* update upgrade local canister script

* remove sns

* fix nix file

* Revert "user_index stage 1 upgrade"

This reverts commit 139d685.

* chore: refactor

* update github workflows

* feat: add last subnet upgrade status in platform orchestrator canister

* update cycle constants

* added get_cycle_balance in post_cache

* added get_next_available_subnet in platform_orchestrator

* chore: fix platform orchestrator

* fix: upgrade individual canisters for not yet assigned canisters

* added existing subnet_orchestrator and post_cache to plaform_orchestrator

.github/workflows/all_canisters_test_suite_on_any_push.yml

@@ -68,5 +68,3 @@ jobs:
           nix-shell --run "dfx build post_cache"
       - name: Run canister test suite
         run: nix-shell --run "cargo test"
-      - name: Run canister test suite - feature mockdata
-        run: nix-shell --run "cargo test --features mockdata --package post_cache"

.github/workflows/create_release_on_tag_push.yml

@@ -131,27 +131,5 @@ jobs:
             --canister-upgrade-arg "(record {version=\"${{ github.ref_name }}\"})" \
             $NEURON_ID > "proposals/${CANISTER_NAME}/upgrade.json"
           ./quill send proposals/${CANISTER_NAME}/upgrade.json --yes
-      - name: Submit upgrade proposal for configuration canister
-        run: |
-          CANISTER_NAME=configuration
-          export CANISTER_ID=$(nix-shell --run "dfx canister id ${CANISTER_NAME} --network=ic")
-          mkdir -p "proposals/${CANISTER_NAME}"
-          touch "proposals/${CANISTER_NAME}/upgrade.json"
-          ./quill sns \
-            --canister-ids-file ./sns/sns_canister_ids.json \
-            --pem-file actions_identity.pem \
-            make-upgrade-canister-proposal \
-            --title "Upgrade ${CANISTER_NAME} Canisters" \
-            --summary "
-            # Upgrade ${CANISTER_NAME}
-
-            ${{ steps.changelog.outputs.changes }}
-            " \
-            --url 'https://hotornot.wtf' \
-            --target-canister-id $CANISTER_ID \
-            --wasm-path .dfx/ic/canisters/${CANISTER_NAME}/${CANISTER_NAME}.wasm.gz \
-            --canister-upgrade-arg '(record {})' \
-            $NEURON_ID > "proposals/${CANISTER_NAME}/upgrade.json"
-          ./quill send proposals/${CANISTER_NAME}/upgrade.json --yes
       - name: Remove messages
         run: rm -r proposals

src/canister/platform_orchestrator/can.did

@@ -1,3 +1,8 @@
+type CanisterUpgradeStatus = record {
+  failures : vec record { principal; text };
+  count : nat64;
+  upgrade_arg : UpgradeCanisterArg;
+};
 type PlatformOrchestratorInitArgs = record { version : text };
 type Result = variant { Ok : principal; Err : text };
 type Result_1 = variant { Ok : text; Err : text };
@@ -12,6 +17,8 @@ type WasmType = variant {
   SubnetOrchestratorWasm;
 };
 service : (PlatformOrchestratorInitArgs) -> {
+  get_next_available_subnet : () -> (principal) query;
+  get_subnet_last_upgrade_status : () -> (CanisterUpgradeStatus) query;
   get_version : () -> (text) query;
   provision_subnet_orchestrator_canister : (principal) -> (Result);
   subnet_orchestrator_maxed_out : () -> ();

src/canister/platform_orchestrator/src/api/canister_lifecycle/init.rs

@@ -1,6 +1,5 @@
-use std::borrow::Cow;
-use ic_stable_structures::{storable::Blob, StableVec, Storable};
-use shared_utils::{canister_specific::platform_orchestrator::types::args::PlatformOrchestratorInitArgs, common::types::wasm::WasmType};
+use candid::Principal;
+use shared_utils::canister_specific::platform_orchestrator::types::args::PlatformOrchestratorInitArgs;
 use crate::CANISTER_DATA;
 
 
@@ -9,7 +8,10 @@ use crate::CANISTER_DATA;
 #[ic_cdk::init]
 #[candid::candid_method(init)]
 fn init(init_args: PlatformOrchestratorInitArgs) {
-    CANISTER_DATA.with(|canister_data_ref_cell| {
-        canister_data_ref_cell.borrow_mut().version_detail.version = init_args.version;
+    CANISTER_DATA.with_borrow_mut(|canister_data| {
+        canister_data.version_detail.version = init_args.version;
+        canister_data.all_subnet_orchestrator_canisters_list.insert(Principal::from_text("rimrc-piaaa-aaaao-aaljq-cai").unwrap());
+        canister_data.subet_orchestrator_with_capacity_left.insert(Principal::from_text("rimrc-piaaa-aaaao-aaljq-cai").unwrap());
+        canister_data.all_post_cache_orchestrator_list.insert(Principal::from_text("y6yjf-jyaaa-aaaal-qbd6q-cai").unwrap());
     })
 }

src/canister/platform_orchestrator/src/api/canister_management/get_last_subnet_upgrade_status.rs

@@ -0,0 +1,13 @@
+use candid::candid_method;
+
+use crate::{data_model::CanisterUpgradeStatus, CANISTER_DATA};
+
+
+
+#[candid_method(query)]
+#[ic_cdk::query]
+pub fn get_subnet_last_upgrade_status() -> CanisterUpgradeStatus {
+    CANISTER_DATA.with_borrow(|canister_data| {
+        canister_data.last_subnet_canister_upgrade_status.clone()
+    })
+}

src/canister/platform_orchestrator/src/api/canister_management/get_next_available_subnet.rs

@@ -0,0 +1,13 @@
+use candid::Principal;
+
+use crate::CANISTER_DATA;
+
+
+
+#[ic_cdk::query]
+#[candid::candid_method(query)]
+fn get_next_available_subnet() -> Principal {
+    CANISTER_DATA.with_borrow(|canister_data| {
+        *canister_data.subet_orchestrator_with_capacity_left.iter().next().unwrap()
+    })
+}

src/canister/platform_orchestrator/src/api/canister_management/mod.rs

@@ -7,6 +7,8 @@ mod provision_subnet_orchestrator;
 mod upgrade_canister;
 mod upload_wasms;
 mod subnet_orchestrator_maxed_out;
+mod get_last_subnet_upgrade_status;
+mod get_next_available_subnet;
 
 #[ic_cdk::query]
 #[candid_method(query)]

src/canister/platform_orchestrator/src/api/canister_management/provision_subnet_orchestrator.rs

@@ -1,9 +1,8 @@
-use std::{str::FromStr, vec};
-
+use std::{borrow::BorrowMut, str::FromStr, vec};
 use candid::{Principal, CandidType};
-use ic_cdk::{api::{self, call, is_controller, management_canister::{main::{self,  CanisterInstallMode, InstallCodeArgument}, provisional::CanisterSettings}}, caller, id};
+use ic_cdk::{api::{self, call, is_controller, management_canister::{main::{self,  CanisterInstallMode, InstallCodeArgument}, provisional::CanisterSettings}}, call, caller, id};
 use serde::{Deserialize, Serialize};
-use shared_utils::{canister_specific::{individual_user_template, post_cache::types::arg::PostCacheInitArgs, user_index::types::args::UserIndexInitArgs}, common::types::{known_principal::{KnownPrincipalMap, KnownPrincipalType}, wasm::WasmType}, constant::{INDIVIDUAL_USER_CANISTER_RECHARGE_AMOUNT, NNS_CYCLE_MINTING_CANISTER, SUBNET_ORCHESTRATOR_CANISTER_INITIAL_CYCLES}};
+use shared_utils::{canister_specific::{individual_user_template, post_cache::types::arg::PostCacheInitArgs, user_index::types::args::UserIndexInitArgs}, common::types::{known_principal::{KnownPrincipalMap, KnownPrincipalType}, wasm::WasmType}, constant::{INDIVIDUAL_USER_CANISTER_RECHARGE_AMOUNT, NNS_CYCLE_MINTING_CANISTER, POST_CACHE_CANISTER_CYCLES_RECHARGE_AMOUMT, SUBNET_ORCHESTRATOR_CANISTER_INITIAL_CYCLES}};
 
 use crate::CANISTER_DATA;
 
@@ -87,7 +86,7 @@ pub async fn provision_subnet_orchestrator_canister(subnet: Principal) -> Result
         Principal::from_str(NNS_CYCLE_MINTING_CANISTER).unwrap(), 
         "create_canister",
        (create_canister_arg,),
-        INDIVIDUAL_USER_CANISTER_RECHARGE_AMOUNT as u64
+        POST_CACHE_CANISTER_CYCLES_RECHARGE_AMOUMT as u64
     )
     .await
     .unwrap();
@@ -102,6 +101,11 @@ pub async fn provision_subnet_orchestrator_canister(subnet: Principal) -> Result
     known_principal_map.insert(KnownPrincipalType::CanisterIdUserIndex, subnet_orchestrator_canister_id);
     known_principal_map.insert(KnownPrincipalType::CanisterIdPostCache, post_cache_canister_id);
 
+    CANISTER_DATA.with_borrow_mut(|canister_data| {
+        canister_data.all_post_cache_orchestrator_list.insert(post_cache_canister_id);
+        canister_data.all_subnet_orchestrator_canisters_list.insert(subnet_orchestrator_canister_id);
+        canister_data.subet_orchestrator_with_capacity_left.insert(subnet_orchestrator_canister_id);
+    });
 
     let user_index_init_arg = UserIndexInitArgs {
         known_principal_ids: Some(known_principal_map.clone()),
@@ -135,7 +139,13 @@ pub async fn provision_subnet_orchestrator_canister(subnet: Principal) -> Result
 
     main::install_code(post_cache_install_code_arg).await.unwrap();
 
+    //provision pool of canisters
+    let individual_user_template_wasm = CANISTER_DATA.with_borrow(|canister_data| canister_data.wasms.get(&WasmType::IndividualUserWasm).clone()).unwrap();
+
+    let (create_pool_of_individual_canister_res, ): (Result<String, String>, ) = call(subnet_orchestrator_canister_id, "create_pool_of_individual_user_available_canisters", (individual_user_template_wasm.version, individual_user_template_wasm.wasm_blob)).await.unwrap();
 
+    create_pool_of_individual_canister_res.unwrap();
+
     Ok(subnet_orchestrator_canister_id)
 
 }

src/canister/platform_orchestrator/src/api/canister_management/upgrade_canister.rs

@@ -71,14 +71,15 @@ async fn upgrade_individual_canisters(upgrade_arg: UpgradeCanisterArg) {
                 continue;
             }
         }
-        let res = ic_cdk::notify(
+        let res: Result<(String, ), String> = ic_cdk::call(
                 *subnet_orchestrator, 
                 "start_upgrades_for_individual_canisters", 
                 (
-                    upgrade_arg.wasm_blob.clone(), 
-                    upgrade_arg.version.clone()
+                    upgrade_arg.version.clone(),
+                    upgrade_arg.wasm_blob.clone() 
                 )
             )
+            .await
             .map_err(|e| format!("Failed to start upgrades on {}", subnet_orchestrator));
 
         match res {
@@ -91,6 +92,9 @@ async fn upgrade_individual_canisters(upgrade_arg: UpgradeCanisterArg) {
                 })
             }
         }
+        CANISTER_DATA.with_borrow_mut(|canister_data| {
+            canister_data.last_subnet_canister_upgrade_status.count += 1;
+        })
     }
 }
 

src/canister/platform_orchestrator/src/lib.rs

@@ -5,7 +5,7 @@ use candid::export_service;
 use data_model::CanisterData;
 use candid::Principal;
 use shared_utils::{canister_specific::platform_orchestrator::types::args::PlatformOrchestratorInitArgs, common::types::wasm::WasmType};
-use crate::data_model::UpgradeCanisterArg;
+use crate::data_model::{UpgradeCanisterArg, CanisterUpgradeStatus};
 
 mod data_model;
 #[cfg(test)]

src/canister/post_cache/can.did

@@ -53,6 +53,7 @@ type TopPostsFetchError = variant {
   ExceededMaxNumberOfItemsAllowedInOneRequest;
 };
 service : (PostCacheInitArgs) -> {
+  get_cycle_balance : () -> (nat) query;
   get_top_posts_aggregated_from_canisters_on_this_network_for_home_feed : (
       nat64,
       nat64,

src/canister/post_cache/src/api/canister_management/get_cycle_balance.rs

@@ -0,0 +1,9 @@
+use ic_cdk::api;
+
+
+
+#[ic_cdk::query]
+#[candid::candid_method(query)]
+fn get_cycle_balance() -> u128 {
+    api::canister_balance128()
+}

src/canister/post_cache/src/api/canister_management/mod.rs

@@ -0,0 +1 @@
+pub mod get_cycle_balance;

src/canister/post_cache/src/api/mod.rs

@@ -3,3 +3,4 @@ pub mod feed;
 pub mod home_feed;
 pub mod hot_or_not_feed;
 pub mod well_known_principal;
+pub mod canister_management;

src/canister/post_cache/src/lib.rs

@@ -12,8 +12,6 @@ use shared_utils::{
     types::canister_specific::post_cache::error_types::TopPostsFetchError,
 };
 
-//TODO: add method get_cycle_balance in post_cache canister.
-
 mod api;
 mod data_model;
 #[cfg(test)]

src/canister/user_index/src/api/canister_lifecycle/post_upgrade.rs

@@ -1,17 +1,13 @@
-use std::time::Duration;
 use ciborium::de;
 
 use ic_stable_structures::Memory;
-use shared_utils::{canister_specific::user_index::types::args::UserIndexInitArgs, common::utils::{stable_memory_serializer_deserializer, system_time}};
+use shared_utils::{common::utils::system_time, canister_specific::user_index::types::args::UserIndexInitArgs};
 
 use crate::{
-    api:: well_known_principal::update_locally_stored_well_known_principals,
-    data_model::{canister_upgrade::UpgradeStatus, memory, CanisterData},
+    data_model::{canister_upgrade::UpgradeStatus, memory},
     CANISTER_DATA,
 };
 
-use super::pre_upgrade::BUFFER_SIZE_BYTES;
-
 #[ic_cdk::post_upgrade]
 fn post_upgrade() {
     restore_data_from_stable_memory();
@@ -34,17 +30,15 @@ fn update_version_from_args() {
 }
 
 fn restore_data_from_stable_memory() {
-    match stable_memory_serializer_deserializer::deserialize_from_stable_memory::<CanisterData>(
-        BUFFER_SIZE_BYTES,
-    ) {
-        Ok(canister_data) => {
-            CANISTER_DATA.with(|canister_data_ref_cell| {
-                *canister_data_ref_cell.borrow_mut() = canister_data;
-            });
-        }
-        Err(e) => {
-            ic_cdk::print(format!("Error: {:?}", e));
-            panic!("Failed to restore canister data from stable memory");
-        }
-    }
+   let heap_data = memory::get_upgrades_memory();
+   let mut heap_data_len_bytes = [0; 4];
+   heap_data.read(0, &mut heap_data_len_bytes);
+   let heap_data_len = u32::from_le_bytes(heap_data_len_bytes) as usize;
+
+   let mut canister_data_bytes = vec![0; heap_data_len];
+   heap_data.read(4, &mut canister_data_bytes);
+   let canister_data = de::from_reader(&*canister_data_bytes).expect("Failed to deserialize heap data");
+   CANISTER_DATA.with_borrow_mut(|cd| {
+        *cd = canister_data;
+   })
 }

src/canister/user_index/src/api/canister_management/mod.rs

@@ -36,10 +36,10 @@ pub async fn set_permission_to_upgrade_individual_canisters(flag: bool) -> Strin
 
 #[candid::candid_method(update)]
 #[ic_cdk::update]
-pub async fn start_upgrades_for_individual_canisters(version: String, individual_user_wasm: Vec<u8>) -> String {
+async fn start_upgrades_for_individual_canisters(version: String, individual_user_wasm: Vec<u8>) -> String {
 
     if !is_controller(&caller()) {
-        return "Unauthorized caller".to_string();
+        panic!("Unauthorized caller");
     }
 
     CANISTER_DATA.with_borrow_mut(|canister_data| {
@@ -101,7 +101,7 @@ pub async fn reset_user_individual_canisters(canisters: Vec<Principal>) -> Resul
     });
 
     let canister_reinstall_futures = canisters.iter().map(|canister| async move {
-        canister_management::recharge_canister_if_below_threshold(&canister).await?;
+        canister_management::recharge_canister_if_below_threshold(&canister).await.map_err(|e| (*canister, e))?;
         canister_management::upgrade_individual_user_canister(canister.clone(), CanisterInstallMode::Reinstall, IndividualUserTemplateInitArgs {
             known_principal_ids: Some(CANISTER_DATA.with(|canister_data_ref| {canister_data_ref.borrow().configuration.known_principal_ids.clone()})),
             profile_owner: None,

src/canister/user_index/src/api/upgrade_individual_user_template/update_user_index_upgrade_user_canisters_with_latest_wasm.rs

@@ -20,12 +20,21 @@ pub async fn upgrade_user_canisters_with_latest_wasm(_version: String, individua
     let mut upgrade_count = 0;
     let mut failed_canister_ids = Vec::new();
 
-    let user_principal_id_to_canister_id_map = CANISTER_DATA.with(|canister_data_ref_cell| {
+    let mut user_principal_id_to_canister_id_vec: Vec<(Principal, Principal)> = CANISTER_DATA.with(|canister_data_ref_cell| {
         canister_data_ref_cell
             .borrow()
             .user_principal_id_to_canister_id_map
             .clone()
+            .into_iter()
+            .collect()
     });
+
+    CANISTER_DATA.with_borrow(|canister_data| {
+        canister_data.available_canisters.iter().for_each(|canister_id| {
+            user_principal_id_to_canister_id_vec.push((Principal::anonymous(), *canister_id));
+        })
+    });
+
 
     let saved_upgrade_status = CANISTER_DATA.with(|canister_data_ref_cell| {
         canister_data_ref_cell
@@ -37,22 +46,21 @@ pub async fn upgrade_user_canisters_with_latest_wasm(_version: String, individua
     let configuration = CANISTER_DATA
         .with(|canister_data_ref_cell| canister_data_ref_cell.borrow().configuration.clone());
 
-   let upgrade_individual_canister_futures = user_principal_id_to_canister_id_map.iter()
+   let upgrade_individual_canister_futures = user_principal_id_to_canister_id_vec.iter()
     .map(|(user_principal_id, user_canister_id)| {
         recharge_and_upgrade(*user_canister_id, *user_principal_id, saved_upgrade_status.version_number, configuration.clone(), saved_upgrade_status.version.clone(), individual_user_wasm.clone())
     });
 
-    let result_callback = |upgrade_result: Result<Principal, (Principal, String)>| {
+    let result_callback = |upgrade_result: Result<(Principal, Principal), ((Principal, Principal), String)>| {
 
         if upgrade_result.is_err() {
-            let (done_user_principal_id, err) = upgrade_result.err().unwrap();
-            let done_user_canister_id = user_principal_id_to_canister_id_map.get(&done_user_principal_id).unwrap();
+            let ((done_user_principal_id, done_user_canister_id), err) = upgrade_result.err().unwrap();
             ic_cdk::print(format!(
                 "Failed to upgrade canister: {:?} with error: {:?}",
                 done_user_canister_id.to_text(),
                 err
             ));
-            failed_canister_ids.push((done_user_principal_id, *done_user_canister_id, err));
+            failed_canister_ids.push((done_user_principal_id, done_user_canister_id, err));
         }
 
         upgrade_count += 1;
@@ -85,12 +93,12 @@ pub async fn upgrade_user_canisters_with_latest_wasm(_version: String, individua
 }
 
 
-async fn recharge_and_upgrade(user_canister_id: Principal, user_principal_id: Principal, version_number: u64, configuration: Configuration, version: String, individual_user_wasm: Vec<u8>) -> Result<Principal, (Principal, String)> {
-    recharge_canister_if_below_threshold(&user_canister_id).await?;
+async fn recharge_and_upgrade(user_canister_id: Principal, user_principal_id: Principal, version_number: u64, configuration: Configuration, version: String, individual_user_wasm: Vec<u8>) -> Result<(Principal, Principal), ((Principal, Principal), String)> {
+    recharge_canister_if_below_threshold(&user_canister_id).await.map_err(|e| ((user_principal_id, user_canister_id), e))?;
 
-    upgrade_user_canister(&user_principal_id, &user_canister_id, version_number, &configuration, version, individual_user_wasm).await.map_err(|s| (user_principal_id, s))?;
+    upgrade_user_canister(&user_principal_id, &user_canister_id, version_number, &configuration, version, individual_user_wasm).await.map_err(|s| ((user_principal_id, user_canister_id), s))?;
 
-    Ok(user_principal_id)
+    Ok((user_principal_id, user_canister_id))
 }
 
 async fn upgrade_user_canister(

src/canister/user_index/src/util/canister_management.rs

@@ -103,15 +103,15 @@ pub async fn upgrade_individual_user_canister(
     start_canister(CanisterIdRecord {canister_id}).await
 }
 
-pub async fn recharge_canister_if_below_threshold(canister_id: &Principal) -> Result<Principal, (Principal, String)> {
+pub async fn recharge_canister_if_below_threshold(canister_id: &Principal) -> Result<(), String> {
 
     let is_canister_below_threshold_balance = is_canister_below_threshold_balance(&canister_id).await;
 
     if is_canister_below_threshold_balance {
-        recharge_canister(canister_id).await.map_err(|s| (*canister_id, s))?;    
+        recharge_canister(canister_id).await?;    
     }
 
-    Ok(*canister_id)
+    Ok(())
 }