feat: Add redis based reliability reporting

Makefile

@@ -5,7 +5,11 @@ CARGO = cargo
 # Let's be very explicit about it for now.
 TESTS_DIR := `pwd`/tests
 TEST_RESULTS_DIR ?= workspace/test-results
-PYTEST_ARGS ?= $(if $(SKIP_SENTRY),-m "not sentry") $(if $(TEST_STUB),,-m "not stub") # Stub tests do not work in CI
+# NOTE: Do not be clever.
+# The integration tests (and a few others) use pytest markers to control
+# the tests that are being run. These markers are set and defined within
+# the `./pyproject.toml`. That is the single source of truth.
+PYTEST_ARGS := ${PYTEST_ARGS}
 INTEGRATION_TEST_DIR := $(TESTS_DIR)/integration
 INTEGRATION_TEST_FILE := $(INTEGRATION_TEST_DIR)/test_integration_all_rust.py
 NOTIFICATION_TEST_DIR := $(TESTS_DIR)/notification
@@ -46,17 +50,17 @@ integration-test-clean:
 	$(DOCKER_COMPOSE) -f $(INTEGRATION_TEST_DIR)/docker-compose.yml down
 	docker rm integration-tests
 
-integration-test-legacy:
+integration-test-legacy: ## pytest markers are stored in `tests/pytest.ini`
 	$(POETRY) -V
 	$(POETRY) install --without dev,load,notification --no-root
 	$(POETRY) run pytest $(INTEGRATION_TEST_FILE) \
 		--junit-xml=$(TEST_RESULTS_DIR)/integration_test_legacy_results.xml \
 		-v $(PYTEST_ARGS)
 
-integration-test-local:
+integration-test-local: ## pytest markers are stored in `tests/pytest.ini`
 	$(POETRY) -V
 	$(POETRY) install --without dev,load,notification --no-root
-		$(POETRY) run pytest $(INTEGRATION_TEST_FILE) \
+	$(POETRY) run pytest $(INTEGRATION_TEST_FILE) \
 		--junit-xml=$(TEST_RESULTS_DIR)/integration_test_results.xml \
 		-v $(PYTEST_ARGS)
 

autoconnect/Cargo.toml

@@ -53,7 +53,13 @@ actix-service = "2.0"
 docopt = "1.1"
 
 [features]
-default = ["bigtable"]
+default = ["bigtable", "reliable_report"]
 bigtable = ["autopush_common/bigtable", "autoconnect_settings/bigtable"]
 emulator = ["bigtable"]
 log_vapid = []
+reliable_report = [
+    "autoconnect_settings/reliable_report",
+    "autoconnect_web/reliable_report",
+    "autoconnect_ws/reliable_report",
+    "autopush_common/reliable_report",
+]

autoconnect/autoconnect-common/src/broadcast.rs

@@ -199,11 +199,11 @@ impl BroadcastChangeTracker {
             }
             *ver = broadcast.version;
         } else {
-            trace!("📢 Not found: {}", &b_id);
+            trace!("📢 Not found: {b_id}");
             return Err(ApcErrorKind::BroadcastError("Broadcast not found".into()).into());
         }
 
-        trace!("📢 New version of {}", &b_id);
+        trace!("📢 New version of {b_id}");
         // Check to see if this broadcast has been updated since initialization
         let bcast_index = self
             .broadcast_list

autoconnect/autoconnect-settings/Cargo.toml

@@ -25,3 +25,4 @@ autopush_common.workspace = true
 # specify the default via the calling crate, in order to simplify default chains.
 bigtable = ["autopush_common/bigtable"]
 emulator = ["bigtable"]
+reliable_report = ["autopush_common/reliable_report"]

autoconnect/autoconnect-settings/src/app_state.rs

@@ -12,6 +12,8 @@ use autoconnect_common::{
     registry::ClientRegistry,
 };
 use autopush_common::db::{client::DbClient, DbSettings, StorageType};
+#[cfg(feature = "reliable_report")]
+use autopush_common::reliability::PushReliability;
 
 use crate::{Settings, ENV_PREFIX};
 
@@ -32,6 +34,9 @@ pub struct AppState {
     pub settings: Settings,
     pub router_url: String,
     pub endpoint_url: String,
+
+    #[cfg(feature = "reliable_report")]
+    pub reliability: Arc<PushReliability>,
 }
 
 impl AppState {
@@ -84,6 +89,13 @@ impl AppState {
                 ENV_PREFIX.to_uppercase()
             ),
         };
+
+        #[cfg(feature = "reliable_report")]
+        let reliability = Arc::new(
+            PushReliability::new(&settings.reliability_dsn, db.clone()).map_err(|e| {
+                ConfigError::Message(format!("Could not start Reliability connection: {:?}", e))
+            })?,
+        );
         let http = reqwest::Client::builder()
             .timeout(Duration::from_secs(1))
             .build()
@@ -103,6 +115,8 @@ impl AppState {
             settings,
             router_url,
             endpoint_url,
+            #[cfg(feature = "reliable_report")]
+            reliability,
         })
     }
 

autoconnect/autoconnect-settings/src/lib.rs

@@ -108,6 +108,11 @@ pub struct Settings {
     ///
     /// By default, the number of available physical CPUs is used as the worker count.
     pub actix_workers: Option<usize>,
+    #[cfg(feature = "reliable_report")]
+    /// The DNS for the reliability data store. This is normally a Redis compatible
+    /// storage system. See [Connection Parameters](https://docs.rs/redis/latest/redis/#connection-parameters)
+    /// for details.
+    pub reliability_dsn: Option<String>,
 }
 
 impl Default for Settings {
@@ -139,6 +144,8 @@ impl Default for Settings {
             msg_limit: 150,
             actix_max_connections: None,
             actix_workers: None,
+            #[cfg(feature = "reliable_report")]
+            reliability_dsn: None,
         }
     }
 }

autoconnect/autoconnect-web/Cargo.toml

@@ -36,3 +36,7 @@ tokio.workspace = true
 autoconnect_common = { workspace = true, features = ["test-support"] }
 
 [features]
+reliable_report = [
+    "autopush_common/reliable_report",
+    "autoconnect_ws/reliable_report",
+]

autoconnect/autoconnect-web/src/routes.rs

@@ -16,21 +16,45 @@ pub async fn ws_route(
 }
 
 /// Deliver a Push notification directly to a connected client
+#[allow(unused_mut)]
 pub async fn push_route(
     uaid: web::Path<Uuid>,
-    notif: web::Json<Notification>,
+    mut notif: web::Json<Notification>,
     app_state: web::Data<AppState>,
 ) -> HttpResponse {
     trace!(
-        "⏩ push_route, uaid: {} channel_id: {}",
+        "⏩ in push_route, uaid: {} channel_id: {}",
         uaid,
-        notif.channel_id
+        notif.channel_id,
     );
+    #[cfg(feature = "reliable_report")]
+    {
+        notif
+            .record_reliability(
+                &app_state.reliability,
+                autopush_common::reliability::ReliabilityState::IntAccepted,
+            )
+            .await;
+        notif
+            .record_reliability(
+                &app_state.reliability,
+                autopush_common::reliability::ReliabilityState::Transmitted,
+            )
+            .await;
+    }
+    // Attempt to send the notification to the UA using WebSocket protocol, or store on failure.
     let result = app_state
         .clients
-        .notify(uaid.into_inner(), notif.into_inner())
+        .notify(uaid.into_inner(), notif.clone())
         .await;
     if result.is_ok() {
+        #[cfg(feature = "reliable_report")]
+        notif
+            .record_reliability(
+                &app_state.reliability,
+                autopush_common::reliability::ReliabilityState::Accepted,
+            )
+            .await;
         HttpResponse::Ok().finish()
     } else {
         HttpResponse::NotFound().body("Client not available")

autoconnect/autoconnect-ws/Cargo.toml

@@ -35,3 +35,7 @@ ctor.workspace = true
 autoconnect_common = { workspace = true, features = ["test-support"] }
 
 [features]
+reliable_report = [
+    "autopush_common/reliable_report",
+    "autoconnect_ws_sm/reliable_report",
+]

autoconnect/autoconnect-ws/autoconnect-ws-sm/Cargo.toml

@@ -32,3 +32,4 @@ serde_json.workspace = true
 autoconnect_common = { workspace = true, features = ["test-support"] }
 
 [features]
+reliable_report = []

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

@@ -138,6 +138,11 @@ impl WebPushClient {
         &self.app_state.settings
     }
 
+    #[cfg(feature = "reliable_report")]
+    pub fn app_reliability(&self) -> &autopush_common::reliability::PushReliability {
+        &self.app_state.reliability
+    }
+
     /// Connect this `WebPushClient` to the `ClientRegistry`
     ///
     /// Returning a `Stream` of `ServerNotification`s from the `ClientRegistry`
@@ -233,6 +238,7 @@ impl WebPushClient {
         let connected_at = self.connected_at;
         rt::spawn(async move {
             app_state.db.save_messages(&uaid, notifs).await?;
+            // XXX: record reliability
             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!(

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

@@ -124,6 +124,7 @@ impl WebPushClient {
             if msg.sortkey_timestamp.is_none() {
                 expired_topic_sort_keys.push(msg.chidmessageid());
             }
+            // XXX: record ReliabilityState::Expired?
             false
         });
         // TODO: A batch remove_messages would be nicer
@@ -163,6 +164,22 @@ impl WebPushClient {
         Ok(smsgs)
     }
 
+    #[cfg(feature = "reliable_report")]
+    /// Record and transition the state for trackable messages.
+    async fn record_state(
+        &self,
+        messages: &mut Vec<Notification>,
+        state: autopush_common::reliability::ReliabilityState,
+    ) {
+        // *Note* because `.map()` is sync
+        // we can't call the async func without additional hoops.
+        for message in messages {
+            message
+                .record_reliability(&self.app_state.reliability, state)
+                .await;
+        }
+    }
+
     /// Read a chunk (max count 10 returned) of Notifications from storage
     ///
     /// This alternates between reading Topic Notifications and Timestamp
@@ -186,10 +203,20 @@ impl WebPushClient {
         let topic_resp = if self.flags.include_topic {
             trace!("🗄️ WebPushClient::do_check_storage: fetch_topic_messages");
             // Get the most recent max 11 messages.
-            self.app_state
+            #[allow(unused_mut)]
+            let mut messages = self
+                .app_state
                 .db
                 .fetch_topic_messages(&self.uaid, 11)
-                .await?
+                .await?;
+            #[cfg(feature = "reliable_report")]
+            // Since we pulled these from storage, mark them as "retrieved"
+            self.record_state(
+                &mut messages.messages,
+                autopush_common::reliability::ReliabilityState::Retrieved,
+            )
+            .await;
+            messages
         } else {
             Default::default()
         };
@@ -226,7 +253,8 @@ impl WebPushClient {
             "🗄️ WebPushClient::do_check_storage: fetch_timestamp_messages timestamp: {:?}",
             timestamp
         );
-        let timestamp_resp = self
+        #[allow(unused_mut)]
+        let mut timestamp_resp = self
             .app_state
             .db
             .fetch_timestamp_messages(&self.uaid, timestamp, 10)
@@ -244,6 +272,13 @@ impl WebPushClient {
                 )
                 .with_tag("topic", "false")
                 .send();
+            #[cfg(feature = "reliable_report")]
+            // Since we pulled these from storage, mark them as "retrieved"
+            self.record_state(
+                &mut timestamp_resp.messages,
+                autopush_common::reliability::ReliabilityState::Retrieved,
+            )
+            .await;
         }
 
         Ok(CheckStorageResponse {