feat: record a couple more state changes, particularly Delivered (#853)

and generalize report_handler's error reporting

Closes SYNC-4644

---------

Co-authored-by: JR Conlin <jr+git@mozilla.com>

Author: pjenvey

autoconnect/autoconnect-common/src/protocol.rs

@@ -85,11 +85,13 @@ impl FromStr for ClientMessage {
 ///
 #[derive(Debug, Deserialize)]
 pub struct ClientAck {
-    // The channel_id which received messages
+    /// The channel_id which received messages
     #[serde(rename = "channelID")]
     pub channel_id: Uuid,
-    // The corresponding version number for the message.
+    /// The corresponding version number for the message.
     pub version: String,
+    /// An optional code categorizing the status of the ACK
+    pub code: Option<u16>,
 }
 
 #[derive(Debug, Serialize)]

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/identified/mod.rs

@@ -237,8 +237,20 @@ impl WebPushClient {
         let uaid = self.uaid;
         let connected_at = self.connected_at;
         rt::spawn(async move {
+            #[cfg(not(feature = "reliable_report"))]
             app_state.db.save_messages(&uaid, notifs).await?;
-            // XXX: record reliability
+            #[cfg(feature = "reliable_report")]
+            {
+                app_state.db.save_messages(&uaid, notifs.clone()).await?;
+                for mut notif in notifs {
+                    notif
+                        .record_reliability(
+                            &app_state.reliability,
+                            autopush_common::reliability::ReliabilityState::Stored,
+                        )
+                        .await;
+                }
+            }
             debug!("Finished saving unacked direct notifs, checking for reconnect");
             let Some(user) = app_state.db.get_user(&uaid).await? else {
                 return Err(SMErrorKind::Internal(format!(
@@ -342,6 +354,10 @@ struct AckState {
     unacked_direct_notifs: Vec<Notification>,
     /// List of unAck'd sent notifications from storage
     unacked_stored_notifs: Vec<Notification>,
+    /// List of Ack'd timestamp notifications from storage, cleared
+    /// via `increment_storage`
+    #[cfg(feature = "reliable_report")]
+    acked_stored_timestamp_notifs: Vec<Notification>,
     /// Either the `current_timestamp` value in storage (returned from
     /// `fetch_messages`) or the last unAck'd timestamp Message's
     /// `sortkey_timestamp` (returned from `fetch_timestamp_messages`).

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/identified/on_client_msg.rs

@@ -214,16 +214,15 @@ impl WebPushClient {
                     "version" => &notif.version
                 );
                 // Get the stored notification record.
-                let n = &self.ack_state.unacked_stored_notifs[pos];
+                let n = &mut self.ack_state.unacked_stored_notifs[pos];
+                let is_topic = n.topic.is_some();
                 debug!("✅ Ack notif: {:?}", &n);
-                // TODO: Record "ack'd" reliability_id, if present.
                 // Only force delete Topic messages, since they don't have a timestamp.
                 // Other messages persist in the database, to be, eventually, cleaned up by their
                 // TTL. We will need to update the `CurrentTimestamp` field for the channel
                 // record. Use that field to set the baseline timestamp for when to pull messages
                 // in the future.
-                // Topic/legacy messages have no sortkey_timestamp
-                if n.sortkey_timestamp.is_none() {
+                if is_topic {
                     debug!(
                         "✅ WebPushClient:ack removing Stored, sort_key: {}",
                         &n.chidmessageid()
@@ -232,8 +231,21 @@ impl WebPushClient {
                         .db
                         .remove_message(&self.uaid, &n.chidmessageid())
                         .await?;
+                    // NOTE: timestamp messages may still be in state of flux: they're not fully
+                    // ack'd (removed/unable to be resurrected) until increment_storage is called,
+                    // so their reliability is recorded there
+                    #[cfg(feature = "reliable_report")]
+                    n.record_reliability(
+                        &self.app_state.reliability,
+                        autopush_common::reliability::ReliabilityState::Delivered,
+                    )
+                    .await;
+                }
+                let n = self.ack_state.unacked_stored_notifs.remove(pos);
+                #[cfg(feature = "reliable_report")]
+                if !is_topic {
+                    self.ack_state.acked_stored_timestamp_notifs.push(n);
                 }
-                self.ack_state.unacked_stored_notifs.remove(pos);
                 self.stats.stored_acked += 1;
                 continue;
             };

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/identified/on_server_notif.rs

@@ -1,3 +1,5 @@
+use std::mem;
+
 use cadence::{Counted, CountedExt};
 
 use autoconnect_common::protocol::{ServerMessage, ServerNotification};
@@ -116,24 +118,34 @@ impl WebPushClient {
         // Filter out TTL expired messages
         let now_sec = sec_since_epoch();
         // Topic messages require immediate deletion from the db
-        let mut expired_topic_sort_keys = vec![];
+        let mut expired_messages = vec![];
+        // NOTE: Vec::extract_if (stabilizing soon) can negate the need for the
+        // inner msg.clone()
         messages.retain(|msg| {
             if !msg.expired(now_sec) {
                 return true;
             }
             if msg.sortkey_timestamp.is_none() {
-                expired_topic_sort_keys.push(msg.chidmessageid());
+                expired_messages.push(msg.clone());
             }
             // XXX: record ReliabilityState::Expired?
             false
         });
         // TODO: A batch remove_messages would be nicer
-        for sort_key in expired_topic_sort_keys {
-            trace!("🉑 removing expired topic sort key: {sort_key}");
+        #[allow(unused_mut)]
+        for mut msg in expired_messages {
+            let chidmessageid = msg.chidmessageid();
+            trace!("🉑 removing expired topic chidmessageid: {chidmessageid}");
             self.app_state
                 .db
-                .remove_message(&self.uaid, &sort_key)
+                .remove_message(&self.uaid, &chidmessageid)
                 .await?;
+            #[cfg(feature = "reliable_report")]
+            msg.record_reliability(
+                &self.app_state.reliability,
+                autopush_common::reliability::ReliabilityState::Expired,
+            )
+            .await;
         }
 
         self.flags.increment_storage = !include_topic && timestamp.is_some();
@@ -311,6 +323,15 @@ impl WebPushClient {
             .db
             .increment_storage(&self.uaid, timestamp)
             .await?;
+        #[cfg(feature = "reliable_report")]
+        {
+            let mut notifs = mem::take(&mut self.ack_state.acked_stored_timestamp_notifs);
+            self.record_state(
+                &mut notifs,
+                autopush_common::reliability::ReliabilityState::Delivered,
+            )
+            .await;
+        }
         self.flags.increment_storage = false;
         Ok(())
     }

autoendpoint/src/routes/reliability.rs

@@ -1,9 +1,14 @@
 use actix_web::{web::Data, HttpResponse};
 
-use crate::server::AppState;
+use crate::{
+    error::{ApiErrorKind, ApiResult},
+    server::AppState,
+};
 
-pub async fn report_handler(app_state: Data<AppState>) -> HttpResponse {
+pub async fn report_handler(app_state: Data<AppState>) -> ApiResult<HttpResponse> {
     let reliability = &app_state.reliability;
 
-    autopush_common::reliability::report_handler(reliability).await
+    autopush_common::reliability::report_handler(reliability)
+        .await
+        .map_err(|e| ApiErrorKind::General(format!("Reliability report error: {e}")).into())
 }

autopush-common/src/db/client.rs

@@ -86,7 +86,7 @@ pub trait DbClient: Send + Sync {
     async fn increment_storage(&self, uaid: &Uuid, timestamp: u64) -> DbResult<()>;
 
     /// Delete a notification
-    async fn remove_message(&self, uaid: &Uuid, sort_key: &str) -> DbResult<()>;
+    async fn remove_message(&self, uaid: &Uuid, chidmessageid: &str) -> DbResult<()>;
 
     /// Check if the router table exists
     async fn router_table_exists(&self) -> DbResult<bool>;

autopush-common/src/db/mock.rs

@@ -91,8 +91,8 @@ impl DbClient for Arc<MockDbClient> {
         Arc::as_ref(self).increment_storage(uaid, timestamp).await
     }
 
-    async fn remove_message(&self, uaid: &Uuid, sort_key: &str) -> DbResult<()> {
-        Arc::as_ref(self).remove_message(uaid, sort_key).await
+    async fn remove_message(&self, uaid: &Uuid, chidmessageid: &str) -> DbResult<()> {
+        Arc::as_ref(self).remove_message(uaid, chidmessageid).await
     }
 
     #[cfg(feature = "reliable_report")]

autopush-common/src/notification.rs

@@ -28,6 +28,7 @@ pub struct Notification {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub reliability_id: Option<String>,
     #[cfg(feature = "reliable_report")]
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub reliable_state: Option<crate::reliability::ReliabilityState>,
 }
 

autopush-common/src/reliability.rs

@@ -285,30 +285,12 @@ pub fn gen_report(values: HashMap<String, i32>) -> Result<String> {
     Ok(encoded)
 }
 
-pub async fn report_handler(reliability: &Arc<PushReliability>) -> HttpResponse {
-    if let Err(err) = reliability.gc().await {
-        error!("🔍🟥 Reporting, Error {:?}", &err);
-        return HttpResponse::InternalServerError()
-            .content_type("text/plain")
-            .body(format!("# ERROR: {err}\n"));
-    };
-    match reliability.report().await {
-        Ok(values) => match gen_report(values) {
-            Ok(report) => HttpResponse::Ok()
-                .content_type("application/openmetrics-text; version=1.0.0; charset=utf-8")
-                .body(report),
-            Err(e) => HttpResponse::InternalServerError()
-                .content_type("text/plain")
-                .body(format!("# ERROR: {e}\n")),
-        },
-        Err(e) => {
-            error!("🔍🟥 Reporting, Error {:?}", &e);
-            // NOTE: This will NOT be read by Prometheus, but serves as a diagnostic message.
-            HttpResponse::InternalServerError()
-                .content_type("text/plain")
-                .body(format!("# ERROR: {e}\n"))
-        }
-    }
+pub async fn report_handler(reliability: &Arc<PushReliability>) -> Result<HttpResponse> {
+    reliability.gc().await?;
+    let report = gen_report(reliability.report().await?)?;
+    Ok(HttpResponse::Ok()
+        .content_type("application/openmetrics-text; version=1.0.0; charset=utf-8")
+        .body(report))
 }
 
 #[cfg(test)]