fn spawn(

        &self,

        container_chain_para_id: ParaId,

        start_collation: bool,

    ) -> Pin<Box<dyn Future<Output = ()> + Send>> {

        let (

            orchestrator_chain_interface,

            orchestrator_client,

            orchestrator_tx_pool,

            mut container_chain_cli,

            tokio_handle,

            chain_type,

            relay_chain,

            relay_chain_interface,

            collator_key,

            sync_keystore,

            orchestrator_para_id,

            validator,

            spawn_handle,

            state,

        ) = (

            self.orchestrator_chain_interface.clone(),

            self.orchestrator_client.clone(),

            self.orchestrator_tx_pool.clone(),

            self.container_chain_cli.clone(),

            self.tokio_handle.clone(),

            self.chain_type.clone(),

            self.relay_chain.clone(),

            self.relay_chain_interface.clone(),

            self.collator_key.clone(),

            self.sync_keystore.clone(),

            self.orchestrator_para_id,

            self.validator,

            self.spawn_handle.clone(),

            self.state.clone(),

        );

        let state2 = state.clone();

        // This closure is used to emulate a try block, it enables using the `?` operator inside

        let try_closure = move || async move {

            // Preload genesis data from orchestrator chain storage.

            // The preload must finish before calling create_configuration, so any async operations

            // need to be awaited.

            // TODO: the orchestrator chain node may not be fully synced yet,

            // in that case we will be reading an old state.

            let orchestrator_chain_info = orchestrator_client.chain_info();

            log::info!(

                "Reading container chain genesis data from orchestrator chain at block #{} {}",

                orchestrator_chain_info.best_number,

                orchestrator_chain_info.best_hash,

            );

            let orchestrator_runtime_api = orchestrator_client.runtime_api();

            log::info!(

                "Detected assignment for container chain {}",

                container_chain_para_id

            );

            let genesis_data = orchestrator_runtime_api

                .genesis_data(orchestrator_chain_info.best_hash, container_chain_para_id)

                .map_err(|e| format!("Failed to call genesis_data runtime api: {}", e))?

                .ok_or_else(|| {

                    format!(

                        "No genesis data registered for container chain id {}",

                        container_chain_para_id

                    )

                })?;

            let boot_nodes_raw = orchestrator_runtime_api

                .boot_nodes(orchestrator_chain_info.best_hash, container_chain_para_id)

                .map_err(|e| format!("Failed to call boot_nodes runtime api: {}", e))?;

            if boot_nodes_raw.is_empty() {

                log::warn!(

                    "No boot nodes registered on-chain for container chain {}",

                    container_chain_para_id

                );

            }

            let boot_nodes =

                parse_boot_nodes_ignore_invalid(boot_nodes_raw, container_chain_para_id);

            if boot_nodes.is_empty() {

                log::warn!(

                    "No valid boot nodes for container chain {}",

                    container_chain_para_id

                );

            }

            container_chain_cli

                .preload_chain_spec_from_genesis_data(

                    container_chain_para_id.into(),

                    genesis_data,

                    chain_type.clone(),

                    relay_chain.clone(),

                    boot_nodes,

                )

                .map_err(|e| format!("failed to create container chain chain spec from on chain genesis data: {}", e))?;

            log::info!(

                "Loaded chain spec for container chain {}",

                container_chain_para_id

            );

            if !start_collation {

                log::info!("This is a syncing container chain, using random ports");

                // Use random ports to avoid conflicts with the other running container chain

                let random_ports = [23456, 23457, 23458];

                container_chain_cli

                    .base

                    .base

                    .prometheus_params

                    .prometheus_port = Some(random_ports[0]);

                container_chain_cli.base.base.network_params.port = Some(random_ports[1]);

                container_chain_cli.base.base.rpc_port = Some(random_ports[2]);

            }

            // Update CLI params

            container_chain_cli.base.para_id = Some(container_chain_para_id.into());

            container_chain_cli

                .base

                .base

                .import_params

                .database_params

                .database = Some(Database::ParityDb);

            let create_container_chain_cli_config = || {

                let mut container_chain_cli_config = sc_cli::SubstrateCli::create_configuration(

                    &container_chain_cli,

                    &container_chain_cli,

                    tokio_handle.clone(),

                )

                .map_err(|err| format!("Container chain argument error: {}", err))?;

                // Change database path to make it depend on container chain para id

                // So instead of the usual "db/full" we have "db/full-container-2000"

                let mut db_path = container_chain_cli_config

                    .database

                    .path()

                    .ok_or_else(|| "Failed to get database path".to_string())?

                    .to_owned();

                db_path.set_file_name(format!("full-container-{}", container_chain_para_id));

                container_chain_cli_config.database.set_path(&db_path);

                sc_service::error::Result::Ok((container_chain_cli_config, db_path))

            };

            let (_container_chain_cli_config, db_path) = create_container_chain_cli_config()?;

            let db_exists = db_path.exists();

            let db_exists_but_may_need_removal = db_exists && validator;

            if db_exists_but_may_need_removal {

                // If the database exists it may be invalid (genesis hash mismatch), so check if it is valid

                // and if not, delete it.

                // Create a new cli config because otherwise the tasks spawned in `open_and_maybe_delete_db` don't stop

                let (container_chain_cli_config, db_path) = create_container_chain_cli_config()?;

                open_and_maybe_delete_db(

                    container_chain_cli_config,

                    &db_path,

                    &orchestrator_client,

                    container_chain_para_id,

                    &container_chain_cli,

                    container_chain_cli.base.keep_db,

                )?;

                // Wait here to for the partial components created in`open_and_maybe_delete_db` to close.

                // Dropping is not enough because there is some background process that keeps the database open,

                // so we check the paritydb lock file directly.

                log::info!("Restarting container chain {}", container_chain_para_id);

                let max_restart_timeout = Duration::from_secs(60);

                wait_for_paritydb_lock(&db_path, max_restart_timeout)

                    .await

                    .map_err(|e| {

                        log::warn!(

                            "Error waiting for chain {} to release db lock: {:?}",

                            container_chain_para_id,

                            e

                        );

                        e

                    })?;

            }

            // Select appropiate sync mode. We want to use WarpSync unless the db still exists,

            // or the block number is 0 (because of a warp sync bug in that case).

            let db_still_exists = db_path.exists();

            container_chain_cli.base.base.network_params.sync = select_sync_mode(

                db_still_exists,

                &orchestrator_client,

                container_chain_para_id,

            )?;

            log::info!(

                "Container chain sync mode: {:?}",

                container_chain_cli.base.base.network_params.sync

            );

            let mut container_chain_cli_config = sc_cli::SubstrateCli::create_configuration(

                &container_chain_cli,

                &container_chain_cli,

                tokio_handle.clone(),

            )

            .map_err(|err| format!("Container chain argument error: {}", err))?;

            container_chain_cli_config.database.set_path(&db_path);

            // Start container chain node

            let (mut container_chain_task_manager, container_chain_client, container_chain_db) =

                start_node_impl_container(

                    container_chain_cli_config,

                    orchestrator_client.clone(),

                    orchestrator_tx_pool.clone(),

                    relay_chain_interface.clone(),

                    orchestrator_chain_interface.clone(),

                    collator_key.clone(),

                    sync_keystore.clone(),

                    container_chain_para_id,

                    orchestrator_para_id,

                    validator && start_collation,

                )

                .await?;

            // Signal that allows to gracefully stop a container chain

            let (signal, on_exit) = oneshot::channel::<bool>();

            let monitor_id;

            {

                let mut state = state.lock().expect("poison error");

                monitor_id = state.spawned_containers_monitor.push(SpawnedContainer {

                    id: 0,

                    para_id: container_chain_para_id,

                    start_time: Instant::now(),

                    stop_signal_time: None,

                    stop_task_manager_time: None,

                    stop_refcount_time: Default::default(),

                    backend: Arc::downgrade(&container_chain_db),

                    client: Arc::downgrade(&container_chain_client),

                });

                state.spawned_container_chains.insert(

                    container_chain_para_id,

                    ContainerChainState {

                        stop_handle: StopContainerChain {

                            signal,

                            id: monitor_id,

                        },

                        db_path: db_path.clone(),

                    },

                );

            }

            // Add the container chain task manager as a child task to the parent task manager.

            // We want to stop the node if this task manager stops, but we also want to allow a

            // graceful shutdown using the `on_exit` future.

            let name = "container-chain-task-manager";

            spawn_handle.spawn(name, None, async move {

                let mut container_chain_task_manager_future =

                    container_chain_task_manager.future().fuse();

                let mut on_exit_future = on_exit.fuse();

                futures::select! {

                    res1 = container_chain_task_manager_future => {

                        // An essential task failed or the task manager was stopped unexpectedly

                        // using `.terminate()`. This should stop the container chain but not the node.

                        if res1.is_err() {

                            log::error!("Essential task failed in container chain {} task manager. Shutting down container chain service", container_chain_para_id);

                        } else {

                            log::error!("Unexpected shutdown in container chain {} task manager. Shutting down container chain service", container_chain_para_id);

                        }

                        // Mark this container chain as "failed to stop" to avoid warning in `self.stop()`

                        let mut state = state.lock().expect("poison error");

                        state.failed_para_ids.insert(container_chain_para_id);

                        // Never delete db in this case because it is not a graceful shutdown

                    }

                    stop_unassigned = on_exit_future => {

                        // Graceful shutdown.

                        // `stop_unassigned` will be `Ok(keep_db)` if `.stop()` has been called, which means that the

                        // container chain has been unassigned, and will be `Err` if the handle has been dropped,

                        // which means that the node is stopping.

                        // Delete existing database if running as collator

                        if validator && stop_unassigned == Ok(false) && !container_chain_cli.base.keep_db {

                            delete_container_chain_db(&db_path);

                        }

                    }

                }

                let mut state = state.lock().expect("poison error");

                state

                    .spawned_containers_monitor

                    .set_stop_task_manager_time(monitor_id, Instant::now());

            });

            sc_service::error::Result::Ok(())

        };

        async move {

            match try_closure().await {

                Ok(()) => {}

                Err(e) => {

                    log::error!(

                        "Failed to start container chain {}: {}",

                    let mut state = state2.lock().expect("poison error");

                    state.failed_para_ids.insert(container_chain_para_id);

        }

        .boxed()

    }

    fn stop(&self, container_chain_para_id: ParaId, keep_db: bool) -> Option<PathBuf> {

        let mut state = self.state.lock().expect("poison error");

        let stop_handle = state

            .spawned_container_chains

            .remove(&container_chain_para_id);

        match stop_handle {

            Some(stop_handle) => {

                log::info!("Stopping container chain {}", container_chain_para_id);

                let id = stop_handle.stop_handle.id;

                state

                    .spawned_containers_monitor

                    .set_stop_signal_time(id, Instant::now());

                // Send signal to perform graceful shutdown, which will delete the db if needed

                let _ = stop_handle.stop_handle.signal.send(keep_db);

                Some(stop_handle.db_path)

                if !state.failed_para_ids.remove(&container_chain_para_id) {

                    log::warn!(

                        "Tried to stop a container chain that is not running: {}",

                }

                None

    }

    pub async fn rx_loop(mut self, mut rx: mpsc::UnboundedReceiver<CcSpawnMsg>, validator: bool) {

        // The node always starts as an orchestrator chain collator.

        // This is because the assignment is detected after importing a new block, so if all

        // collators stop at the same time, when they start again nobody will produce the new block.

        // So all nodes start as orchestrator chain collators, until the first block is imported,

        // then the real assignment is used.

        if validator {

            self.handle_update_assignment(Some(self.orchestrator_para_id), None)

                .await;

        }

        while let Some(msg) = rx.recv().await {

            match msg {

                CcSpawnMsg::UpdateAssignment { current, next } => {

                    self.handle_update_assignment(current, next).await;

        if !validator {

            let () = std::future::pending().await;

        }

    }

    async fn handle_update_assignment(&mut self, current: Option<ParaId>, next: Option<ParaId>) {

        let HandleUpdateAssignmentResult {

            chains_to_stop,

            chains_to_start,

            need_to_restart: _,

        } = handle_update_assignment_state_change(

            &mut self.state.lock().expect("poison error"),

            self.orchestrator_para_id,

            current,

            next,

        );

        if current != Some(self.orchestrator_para_id) {

            let maybe_exit_notification_receiver = self

                .collation_cancellation_constructs

                .take()

                .map(|(cancellation_token, exit_notification_receiver)| {

                    cancellation_token.cancel();

                    exit_notification_receiver

                });

            if let Some(exit_notification_receiver) = maybe_exit_notification_receiver {

                let _ = exit_notification_receiver.await;

            }

        } else if self.collation_cancellation_constructs.is_none() {

            // If assigned to orchestrator chain but the collator process is not running, start it

            self.collation_cancellation_constructs = Some((self.collate_on_tanssi)());

        }

        let mut db_paths_restart = vec![];

        for para_id in chains_to_stop {

            let keep_db = Some(para_id) == current;

            let maybe_db_path = self.stop(para_id, keep_db);

            if let Some(db_path) = maybe_db_path {

                if chains_to_start.contains(&para_id) {

                    db_paths_restart.push((para_id, db_path));

                }

            }

        if !db_paths_restart.is_empty() {

            let max_restart_timeout = Duration::from_secs(60);

            let futs = db_paths_restart

                .into_iter()

                .map(|(para_id, db_path)| async move {

                    wait_for_paritydb_lock(&db_path, max_restart_timeout)

                        .await

                        .map_err(|e| {

                            log::warn!(

                                "Error waiting for chain {} to release db lock: {:?}",

                                para_id,

                                e

                            );

                        })

                });

            futures::future::join_all(futs).await;

        }

        for para_id in chains_to_start {

            let start_collation = Some(para_id) == current;

            self.spawn(para_id, start_collation).await;

    }

        return HandleUpdateAssignmentResult {

            chains_to_stop: Default::default(),

            chains_to_start: Default::default(),

            need_to_restart: false,

        };

fn select_sync_mode(

    db_exists: bool,

    orchestrator_client: &Arc<ParachainClient>,

    container_chain_para_id: ParaId,

) -> sc_service::error::Result<SyncMode> {

    if db_exists {

        return Ok(SyncMode::Full);

    }

    // The following check is only needed because of this bug:

    // https://github.com/paritytech/polkadot-sdk/issues/1930

    let orchestrator_runtime_api = orchestrator_client.runtime_api();

    let orchestrator_chain_info = orchestrator_client.chain_info();

    let full_sync_needed = if !orchestrator_runtime_api

        .has_api::<dyn AuthorNotingApi<Block, AccountId, BlockNumber, ParaId>>(

            orchestrator_chain_info.best_hash,

        )

        .map_err(|e| format!("Failed to check if runtime has AuthorNotingApi: {}", e))?

        true

        orchestrator_runtime_api

            .latest_author(orchestrator_chain_info.best_hash, container_chain_para_id)

            .map_err(|e| format!("Failed to read latest author: {}", e))?

            .is_none()

    if full_sync_needed {

        Ok(SyncMode::Full)

        Ok(SyncMode::Warp)

}

fn open_and_maybe_delete_db(

    container_chain_cli_config: sc_service::Configuration,

    db_path: &Path,

    orchestrator_client: &Arc<ParachainClient>,

    container_chain_para_id: ParaId,

    container_chain_cli: &ContainerChainCli,

    keep_db: bool,

) -> sc_service::error::Result<()> {

    let temp_cli = NodeConfig::new_builder(&container_chain_cli_config, None)?;

    if !keep_db {

        let last_container_block_temp = temp_cli.client.chain_info().best_number;

        let orchestrator_runtime_api = orchestrator_client.runtime_api();

        let orchestrator_chain_info = orchestrator_client.chain_info();

        // Get the container chain's latest block from orchestrator chain and compare with client's one

        let last_container_block_from_orchestrator = orchestrator_runtime_api

            .latest_block_number(orchestrator_chain_info.best_hash, container_chain_para_id)

            .unwrap_or_default();

        let max_block_diff_allowed = 100u32;

        if last_container_block_from_orchestrator

            .unwrap_or(0u32)

            .abs_diff(last_container_block_temp)

            > max_block_diff_allowed

            delete_container_chain_db(db_path);

            return Ok(());

        }

    }

    let container_preloaded_genesis = container_chain_cli.preloaded_chain_spec.as_ref().unwrap();

    let block_v0: Block =

        generate_genesis_block(&**container_preloaded_genesis, sp_runtime::StateVersion::V0)

            .map_err(|e| format!("{:?}", e))?;

    let chain_spec_genesis_hash_v0 = block_v0.header().hash();

    let block_v1: Block =

        generate_genesis_block(&**container_preloaded_genesis, sp_runtime::StateVersion::V1)

            .map_err(|e| format!("{:?}", e))?;

    let chain_spec_genesis_hash_v1 = block_v1.header().hash();

    let container_client_genesis_hash = temp_cli.client.chain_info().genesis_hash;

    if container_client_genesis_hash != chain_spec_genesis_hash_v0

        && container_client_genesis_hash != chain_spec_genesis_hash_v1

        log::info!("Container genesis V0: {:?}", chain_spec_genesis_hash_v0);

        log::info!("Container genesis V1: {:?}", chain_spec_genesis_hash_v1);

        log::info!(

            "Chain spec genesis {:?} did not match with any container genesis - Restarting...",

        delete_container_chain_db(db_path);

        return Ok(());

    }

    Ok(())

}

fn delete_container_chain_db(db_path: &Path) {

    // Remove folder `full-container-2002`

    let _ = std::fs::remove_dir_all(db_path);

    if let Some(parent) = db_path.ancestors().nth(2) {

        delete_empty_folders_recursive(parent);

    }

}

fn delete_empty_folders_recursive(path: &Path) {

    let entry_iter = std::fs::read_dir(path);

    let entry_iter = match entry_iter {

        Ok(x) => x,

        Err(_e) => return,

    for entry in entry_iter {

        let entry = match entry {

            Ok(x) => x,

            Err(_e) => continue,

        let path = entry.path();

        if path.is_dir() {

            delete_empty_folders_recursive(&path);

        }

    let _ = std::fs::remove_dir(path);

}

                        "Invalid boot node in container chain {}: {}",

                        "Invalid boot node in container chain {}: {}",

async fn wait_for_paritydb_lock(db_path: &Path, max_timeout: Duration) -> Result<(), String> {

    let now = Instant::now();

    while now.elapsed() < max_timeout {

        let lock_held = check_paritydb_lock_held(db_path)

            .map_err(|e| format!("Failed to check if lock file is held: {}", e))?;

        if !lock_held {

            return Ok(());

        }

        sleep(Duration::from_secs(1)).await;

    Err("Timeout when waiting for paritydb lock".to_string())

}

fn check_paritydb_lock_held(db_path: &Path) -> Result<bool, std::io::Error> {

    if !db_path.is_dir() {

        return Ok(false);

    }

    let mut lock_path: std::path::PathBuf = db_path.to_owned();

    lock_path.push("lock");

    let lock_file = std::fs::OpenOptions::new()

        .create(true)

        .read(true)

        .write(true)

        .truncate(true)

        .open(lock_path.as_path())?;

    let lock_held = lock_file.try_lock_exclusive().is_err();

    Ok(lock_held)

}

                    "Received CollateOn message when we were already collating on this chain: {}",

                    "Received CollateOn message when we were already collating on this chain: {}",

                "tried to spawn a container chain that was already running: {}",

                    panic!(

                        "Tried to stop a container chain that is not running: {}",

                        container_chain_para_id

                    );

                        "Tried to start and stop same container chain: {}",

                        assert!(

                            !chains_to_stop.contains(para_id),

                            "Tried to start and stop same container chain: {}",