Fix bug with states not being marked as done

Author: EricGhildyal

src/adapters/backend/deploy_meta.rs

@@ -48,7 +48,6 @@ impl LockedState {
     }
 
     pub(crate) async fn mark_done(&mut self) -> Result<()> {
-        trace!("Marking state {} as done...", self.state.state_type);
         self.task_done.send(()).await.into_diagnostic()
     }
 }

src/adapters/backend/mod.rs

@@ -193,7 +193,7 @@ impl BackendClient {
             .await
             .into_diagnostic()?;
 
-        trace!("Stated successfulled marked as completed");
+        trace!("State successfully marked as complete");
         Ok(())
     }
 
@@ -242,7 +242,7 @@ impl BackendClient {
         meta: &DeploymentMetadata,
         data: Vec<StatusCode>,
     ) -> Result<()> {
-        trace!("Uploading observation to backend");
+        trace!("Uploading observations to backend");
         let mut req_waiter = JoinSet::new();
 
         for item in data {

src/adapters/ingresses/apig.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use bon::bon;
 use miette::{IntoDiagnostic as _, Result, miette};
-use tracing::debug;
+use tracing::{debug, info};
 
 use crate::{
     Shutdownable, WholePercent, subsystems::ShutdownResult, utils::load_default_aws_config,
@@ -169,7 +169,7 @@ impl Ingress for AwsApiGateway {
     }
 
     async fn set_canary_traffic(&mut self, percent: WholePercent) -> Result<()> {
-        debug!("Setting canary traffic to {percent}...");
+        info!("Setting API Gateway canary traffic to {percent}.");
         let api = self.get_api_id_by_name(&self.gateway_name).await?;
         let api_id = api.id().ok_or(miette!("Couldn't get ID of deployed API"))?;
 
@@ -196,7 +196,7 @@ impl Ingress for AwsApiGateway {
     }
 
     async fn rollback_canary(&mut self) -> Result<()> {
-        debug!("Rolling back canary deployment...");
+        info!("Rolling back canary deployment in API Gateway.");
         let api = self.get_api_id_by_name(&self.gateway_name).await?;
         let api_id = api.id().ok_or(miette!("Couldn't get ID of deployed API"))?;
 
@@ -219,7 +219,7 @@ impl Ingress for AwsApiGateway {
     }
 
     async fn promote_canary(&mut self) -> Result<()> {
-        debug!("Promoting canary deployment...");
+        info!("Promoting canary deployment in API Gateway!");
         let api = self.get_api_id_by_name(&self.gateway_name).await?;
         let api_id = api.id().ok_or(miette!("Couldn't get ID of deployed API"))?;
 

src/adapters/monitors/cloudwatch.rs

@@ -15,7 +15,7 @@ use aws_sdk_cloudwatch::{
     types::{Dimension, Metric, MetricDataQuery, MetricStat},
 };
 use aws_smithy_types::DateTime as AwsDateTime;
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, Duration, Utc};
 use miette::Result;
 
 use super::Monitor;
@@ -37,7 +37,7 @@ impl CloudWatch {
             client,
             region,
             dimensions,
-            start: Utc::now(),
+            start: Utc::now() - Duration::minutes(5),
         }
     }
 }

src/subsystems/relay/lock_mgmt.rs

@@ -64,19 +64,19 @@ impl IntoSubsystem<Report> for LockManager {
     async fn run(mut self, subsys: SubsystemHandle) -> Result<()> {
         loop {
             select! {
-                _ = subsys.on_shutdown_requested() => {
-                    // Release the lock.
-                    return self.shutdown().await;
-                }
+                // NOTE: the order matters here
                 _ = self.task_done.recv() => {
-                    // Tell the backend that the task
-                    // has been completed.
+                    // Tell the backend that the task has been completed.
                     // Don't call `shutdown` since that's for abnormal
                     // termination in this case. We don't need to release
                     // the lock on the state, since we just marked it as completed
                     // instead.
                     return self.backend.mark_state_completed(&self.meta, &self.state).await;
                 }
+                _ = subsys.on_shutdown_requested() => {
+                    // Release the lock.
+                    return self.shutdown().await;
+                }
                 // Ding! Renew the lease.
                 _ = self.timer.tick() => {
                     self.backend.refresh_lock(&self.meta, &self.state).await?;

src/subsystems/relay/mod.rs

@@ -93,8 +93,6 @@ impl IntoSubsystem<Report> for RelaySubsystem<StatusCode> {
         let mut observations = self.observations;
         loop {
             select! {
-                // TODO: We need to release the lock on
-                // any states we've locked.
                 // Besides that, we can just hang out.
                 _ = subsys.on_shutdown_requested() => {
                     subsys.wait_for_children().await;
@@ -116,7 +114,6 @@ impl IntoSubsystem<Report> for RelaySubsystem<StatusCode> {
                 // • We also need to poll the backend for new states.
                 elem = state_stream.recv() => {
                     debug!("Received new state: {:?}", &elem);
-                    let mut trigger_shutdown = false;
 
                     if let Some(state) = elem {
                         let state_id = state.id;
@@ -142,8 +139,8 @@ impl IntoSubsystem<Report> for RelaySubsystem<StatusCode> {
                                 // Ingress operation.
                                 self.ingress.promote_canary().await?;
 
-                                // If the canary is promoted, we can safely just shut down the CLI
-                                trigger_shutdown = true;
+                                locked_state.mark_done().await?;
+                                subsys.request_shutdown();
                             },
                             DeployCanary => {
                                 // First, we deploy the canary to the platform. At
@@ -156,6 +153,7 @@ impl IntoSubsystem<Report> for RelaySubsystem<StatusCode> {
                                 info!("Releasing the application...");
                                 self.ingress.release_canary(platform_id).await.inspect(|res| debug!("Result: {res:?}"))?;
                                 info!("Release successful! Beginning canarying...");
+                                locked_state.mark_done().await?;
                             },
                             SetCanaryTraffic => {
                                 let percent_traffic = if let Some(data) = locked_state.state().data.clone().flatten() {
@@ -166,27 +164,18 @@ impl IntoSubsystem<Report> for RelaySubsystem<StatusCode> {
                                 };
                                 let percent = WholePercent::try_from(percent_traffic).unwrap();
                                 self.ingress.set_canary_traffic(percent).await?;
+                                locked_state.mark_done().await?;
                             },
                             RollbackCanary => {
                                 // Set traffic to 0 immediately.
                                 self.ingress.set_canary_traffic(WholePercent::try_from(0).unwrap()).await?;
                                 // Then, yank the canary from the ingress.
                                 self.ingress.rollback_canary().await?;
 
-                                // If the canary is rolled back, we can safely just shut down the CLI
-                                trigger_shutdown = true;
+                                locked_state.mark_done().await?;
+                                subsys.request_shutdown();
                             },
                         }
-                        // Since the action completed successfully,
-                        // we can release the lock and tell the backend
-                        // that the state has been effected.
-                        locked_state.mark_done().await?;
-
-                        // This is separated so we can mark the state as done
-                        // before we shut down.
-                        if trigger_shutdown {
-                            subsys.request_shutdown();
-                        }
                     } else {
                         // The stream has been closed, so we should shutdown.
                         subsys.request_shutdown();