Method execute_deployment_task

app/src/deployment_manager/mod.rs:411–529

We have a deployment and a task to execute, we execute the task, business as usual

(
        &mut self,
        mut deployment: DeploymentContext,
        mut task: TaskContext,
        mut upstream: Box<UpstreamGatewayContext>,
    )

    /// We have a deployment and a task to execute, we execute the task, business as usual
    async fn execute_deployment_task(
        &mut self,
        mut deployment: DeploymentContext,
        mut task: TaskContext,
        mut upstream: Box<UpstreamGatewayContext>,
    ) -> (DeploymentManagerState, Option<Duration>) {
        info!("Starting to execute deployment task");

        loop {
            let task_is_terminated = task.task.is_terminated();
            let mut await_task_termination = task.task.await_terminated();

            tokio::select! {
                biased;

                // If there is no task on-going for this deployment, we wait at max 15sec to receive a new one
                _ = tokio::time::sleep(self.deadline_for_new_task), if task_is_terminated => {
                    info!("No new message after {}s, assuming deployment is terminated", self.deadline_for_new_task.as_secs());
                    task.terminate_task().await;
                    deployment.terminate_deployment().await;
                    upstream.await_termination().await;

                    return (DeploymentManagerState::SeekingNewDeployment {}, None);
                }

                // Task is terminated, we re-loop to check if there is a new task for this deployment
                // and to set the timeout correctly
                _ = await_task_termination.recv(), if !task_is_terminated => {
                    info!("Engine Task terminated");
                }

                // We lost the connection with gateway to forward engine message, trying to reconnect
                _ = upstream.close_upstream_tx.closed() => {
                    info!("EngineEvent forwarder to gateway has been close, trying to resume connection");
                    let next_step = DeploymentManagerState::ResumingDeploymentTask {
                        deployment,
                        task,
                    };
                    return (next_step, None);
                }


                // We wait to receive a new message from the gateway
                // In case of error, we return to try to resume the current deployment.
                // The server will let us know if the deployment is still valid
                msg = upstream.msg_stream.next() => {
                    match msg {
                        Some(Ok(msg)) => {
                            match msg.request {
                                Some(engine_message_rx::Request::DeploymentRequest(payload)) => {
                                    info!("Received new deployment task: {}", payload);
                                    let new_task = (self.mk_engine_task)(payload, &deployment.deployment_info, &self.engine_client, upstream.logger(), upstream.metrics_registry(), upstream.log_file_writer().clone());
                                    match new_task {
                                        Ok(new_task) => {
                                            task.await_task_termination().await;
                                            task = TaskContext::spawn_new_task(new_task);
                                        }
                                        Err(err) => {