feat: Clean up terminal states (#899)

jrconlin · web-flow · commit 791e8751d9f3 · 2025-05-23T13:24:54.000-07:00
Address potential problem where final message states may grow without bounds. This also updates the `reliability_report.py` to write to a Google Storage Bucket. Closes: #PUSH-12 / #895
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -71,6 +71,12 @@ executors:
         auth:
           username: $DOCKER_USER
           password: $DOCKER_PASS
+  build-reliability-cron:
+    docker:
+      - image: docker:18.03.0-ce
+        auth:
+          username: $DOCKER_USER
+          password: $DOCKER_PASS
   build-test-container-executor:
     docker:
       - image: cimg/base:2025.02
@@ -363,6 +369,39 @@ jobs:
           paths:
             - <<parameters.image>>.tar
 
+  # Create the reliability cron docker image. This is a singleton that does some
+  # clean-up and reporting for the Push Reliability task.
+  build-reliability-cron:
+    executor: build-reliability-cron
+    resource_class: small
+    working_directory: /app
+    parameters:
+      image:
+        type: string
+      tag:
+        type: string
+    steps:
+      # Install these packages before checkout because git may not exist or work
+      - run:
+          name: Install Docker build dependencies
+          command: apk add --no-cache openssh-client git clang
+      - checkout
+      - setup_remote_docker
+      - docker_login
+      - run:
+          name: Build cron docker image
+          command: |
+            docker build --tag <<parameters.image>>:<<parameters.tag>> -f ./scripts/reliability/Dockerfile ./scripts/reliability
+      - run:
+          name: Save cron docker image
+          command: |
+            mkdir -p /tmp/cron_cache
+            docker save -o /tmp/cron_cache/<<parameters.image>>.tag <<parameters.image>>
+      - persist_to_workspace:
+          root: /tmp/cron_cache
+          paths:
+            - <<parameters.image>>.tag
+
   build-test-container:
     executor: build-test-container-executor
     parameters:
@@ -515,6 +554,17 @@ workflows:
           filters:
             tags:
               only: /.*/
+
+      - build-reliability-cron:
+          name: build-reliability-cron
+          image: autopush-reliability-cron
+          tag: latest
+          filters:
+            tags:
+              only: /.*/
+            # branches:
+            #  only: master
+
       - build-test-container:
           name: Build Load Test Image
           image: autopush-load-tests
diff --git a/Cargo.toml b/Cargo.toml
@@ -108,5 +108,8 @@ autoconnect_ws = { path = "./autoconnect/autoconnect-ws" }
 autoconnect_ws_clientsm = { path = "./autoconnect/autoconnect-ws/autoconnect-ws-clientsm" }
 autopush_common = { path = "./autopush-common", features = ["bigtable"] }
 
+[workspace.lints.clippy]
+result_large_err = "allow" # 1.87.0 will flag ApiError as too large.
+
 [profile.release]
 debug = 1
diff --git a/autoconnect/autoconnect-settings/Cargo.toml b/autoconnect/autoconnect-settings/Cargo.toml
@@ -26,3 +26,6 @@ autopush_common.workspace = true
 bigtable = ["autopush_common/bigtable"]
 emulator = ["bigtable"]
 reliable_report = ["autopush_common/reliable_report"]
+
+[lints]
+workspace = true
diff --git a/autoconnect/autoconnect-ws/autoconnect-ws-sm/Cargo.toml b/autoconnect/autoconnect-ws/autoconnect-ws-sm/Cargo.toml
@@ -33,3 +33,6 @@ autoconnect_common = { workspace = true, features = ["test-support"] }
 
 [features]
 reliable_report = []
+
+[lints]
+workspace = true
diff --git a/autoendpoint/Cargo.toml b/autoendpoint/Cargo.toml
@@ -68,3 +68,6 @@ stub = []
 log_vapid = []
 
 reliable_report = ["autopush_common/reliable_report"]
+
+[lints]
+workspace = true
diff --git a/autoendpoint/src/routers/apns/router.rs b/autoendpoint/src/routers/apns/router.rs
@@ -501,7 +501,7 @@ impl Router for ApnsRouter {
             // mutable, but we are also essentially consuming the
             // notification nothing else should modify it.
             notification
-                .record_reliability(&self.reliability, ReliabilityState::Transmitted)
+                .record_reliability(&self.reliability, ReliabilityState::BridgeTransmitted)
                 .await;
         }
 
diff --git a/autoendpoint/src/routers/fcm/router.rs b/autoendpoint/src/routers/fcm/router.rs
@@ -199,7 +199,7 @@ impl Router for FcmRouter {
         // mutable, but we are also essentially consuming the
         // notification nothing else should modify it.
         notification
-            .record_reliability(&self.reliability, ReliabilityState::Transmitted)
+            .record_reliability(&self.reliability, ReliabilityState::BridgeTransmitted)
             .await;
         // Sent successfully, update metrics and make response
         trace!("Send request was successful");
diff --git a/autoendpoint/src/settings.rs b/autoendpoint/src/settings.rs
@@ -99,6 +99,7 @@ impl Default for Settings {
             stub: StubSettings::default(),
             #[cfg(feature = "reliable_report")]
             reliability_dsn: None,
+            #[cfg(feature = "reliable_report")]
             reliability_retry_count: autopush_common::redis_util::MAX_TRANSACTION_LOOP,
         }
     }
diff --git a/autopush-common/src/reliability.rs b/autopush-common/src/reliability.rs
@@ -14,8 +14,7 @@ use deadpool_redis::Config;
 use prometheus_client::{
     encoding::text::encode, metrics::family::Family, metrics::gauge::Gauge, registry::Registry,
 };
-use redis::aio::ConnectionLike;
-use redis::AsyncCommands;
+use redis::{aio::ConnectionLike, AsyncCommands};
 
 use crate::db::client::DbClient;
 use crate::errors::{ApcErrorKind, Result};
@@ -46,18 +45,36 @@ const CONNECTION_EXPIRATION: TimeDelta = TimeDelta::seconds(10);
 #[serde(rename_all = "snake_case")]
 #[strum(serialize_all = "snake_case")]
 pub enum ReliabilityState {
-    Received,        // Message was received by the Push Server
-    Stored,          // Message was stored because it could not be delivered immediately
-    Retrieved,       // Message was taken from storage for delivery
-    IntTransmitted,  // Message was handed off between autoendpoint and autoconnect
-    IntAccepted,     // Message was accepted by autoconnect from autopendpoint
-    Transmitted,     // Message was handed off for delivery to the UA
-    Accepted,        // Message was accepted for delivery by the UA
-    Delivered,       // Message was provided to the WebApp recipient by the UA
-    DecryptionError, // Message was provided to the UA and it reported a decryption error
-    NotDelivered,    // Message was provided to the UA and it reported a not delivered error
-    Expired,         // Message expired naturally (e.g. TTL=0)
-    Errored,         // Message resulted in an Error state and failed to be delivered.
+    Received,          // Message was received by the Push Server
+    Stored,            // Message was stored because it could not be delivered immediately
+    Retrieved,         // Message was taken from storage for delivery
+    IntTransmitted,    // Message was handed off between autoendpoint and autoconnect
+    IntAccepted,       // Message was accepted by autoconnect from autopendpoint
+    BridgeTransmitted, // Message was handed off to a mobile bridge for eventual delivery
+    Transmitted,       // Message was handed off for delivery to the UA
+    Accepted,          // Message was accepted for delivery by the UA
+    Delivered,         // Message was provided to the WebApp recipient by the UA
+    DecryptionError,   // Message was provided to the UA and it reported a decryption error
+    NotDelivered,      // Message was provided to the UA and it reported a not delivered error
+    Expired,           // Message expired naturally (e.g. TTL=0)
+    Errored,           // Message resulted in an Error state and failed to be delivered.
+}
+
+impl ReliabilityState {
+    ///  Has the message reached a state where no further transitions should be possible?
+    pub fn is_terminal(self) -> bool {
+        // NOTE: this list should match the daily snapshot captured by `reliability_cron.py`
+        // which will trim these counts after the max message TTL has expired.
+        matches!(
+            self,
+            ReliabilityState::DecryptionError
+                | ReliabilityState::BridgeTransmitted
+                | ReliabilityState::Delivered
+                | ReliabilityState::Errored
+                | ReliabilityState::Expired
+                | ReliabilityState::NotDelivered
+        )
+    }
 }
 
 #[derive(Clone)]
@@ -157,10 +174,18 @@ impl PushReliability {
         };
         // Errors are not fatal, and should not impact message flow, but
         // we should record them somewhere.
-        pipeline.zadd(EXPIRY, format!("{}#{}", new, id), expr.unwrap_or_default());
-        let _ = pipeline.exec_async(conn).await.inspect_err(|e| {
-            warn!("🔍 Failed to write to storage: {:?}", e);
-        });
+        if !new.is_terminal() {
+            // Write the expiration only if the state is non-terminal. Otherwise we run the risk of
+            // messages reporting a false "expired" state even if they were "successful".
+            let cc = pipeline
+                .zadd(EXPIRY, format!("{}#{}", new, id), expr.unwrap_or_default())
+                .exec_async(conn)
+                .await
+                .inspect_err(|e| {
+                    warn!("🔍 Failed to write to storage: {:?}", e);
+                });
+            trace!("🔍 internal record result: {:?}", cc);
+        }
     }
 
     /// Perform a garbage collection cycle on a reliability object.
diff --git a/scripts/reliability/Dockerfile b/scripts/reliability/Dockerfile
@@ -0,0 +1,23 @@
+# Docker build for Reliability Reporter.
+
+FROM python:3.12-alpine
+
+ADD . /app
+WORKDIR /app
+# should this be templated? Some have a path here I presume gets loaded via a Docker build.
+# (e.g. fxa uses `/app/keys/keyfile_name.json`)
+# loaded va the webservices-infra cronjob.yaml:: containers::volumeMounts
+VOLUME [ "keys" "/app/keys" ]
+ENV GOOGLE_APPLICATION_CREDENTIALS = /app/keys/service-account-key.json
+
+# these will need to be templated in the calling k8s config.
+# ENV AUTOEND__DB_SETTINGS = '{"table_name":"projects/test/instance/test/tables/autopush"}'
+# ENV AUTOEND__DB_DSN = "grpc:://localhost:8086"
+# ENV AUTOEND__RELIABILITY_DSN = "redis:://localhost"
+
+# Generate the report
+ENV AUTOTRACK_REPORT_BUCKET_NAME = "autopush_reliability"
+ENV AUTOTRACK_OUTPUT="md json"
+RUN pip install .
+CMD ["python3", "reliability_report.py"]
+
diff --git a/scripts/reliability/README.md b/scripts/reliability/README.md
@@ -0,0 +1,12 @@
+# Push Reliability Support Cron
+
+This image runs the Push Reliability Support script. This is used by the optional Push Reliability system.
+
+Push Reliability is determined by examining the flow of messages that are generated and consumed by Mozilla, specifically the "Send Tab" control messages.
+Messages are tracked by when they reach significant points in the code (milestones). See the `autopush_common::reliability::ReliabilityState` for individual milestones.
+
+During the processing, however, there are two functions that should occur to aid in getting a clear picture of the states. The first step is to generate a daily report of the milestones. This script generates this report using the Jinja2 templates listed in `./templates`, as well as a JSON output designed to be machine processed. The other is a bit of house-keeping to ensure that milestones that have reached "terminal" (states where they will not transition to a different state later), are regularly removed. This will prevent out-dated data from collecting and producing less useful reports.
+
+This task is designed to be run as a Kubernetes Cron Job. The actual configuration for that is contained in a different, internally used repo, but all that is doing is setting a few general values (See the commented values in the `Dockerfile`.) These can optionally be defined in a `reliability.toml` configuration file. In addition the Docker image should be run with appropriate keys to the GCP data that we require for Key/Value & Bigtable access.
+
+There are a few additional config options included here to allow for future flexibility, including the `app.yaml` file, which is used by some GCP task execution functions. These may be removed later.
diff --git a/scripts/reliability/app.yaml b/scripts/reliability/app.yaml
@@ -0,0 +1,26 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+runtime: python313
+
+# this is technically a singleton, so shouldn't really need scaling.
+instance_class: B2
+
+env_variables:
+  # get these from webservices-infra configmap.yaml
+  # AUTOCONNECT__DB_DSN: "grpc://{{ .db_location }}-bigtable.googleapis.com:443"
+  # AUTOCONNECT__DB_SETTINGS: {{ .db_settings | toJson | quote }}
+  # AUTOCONNECT__RELIABILITY_DSN: {{ .reliability_dsn | quote }}
+
+  # for dev/testing us hard-coded values.
+  AUTOCONNECT__DB_SETTINGS: '{"message_family":"message","router_family":"router","table_name":"projects/autopush-nonprod/instances/autopush-nonprod-stage/tables/autopush"}'
+  AUTOCONNECT__DB_DSN: grpc://us-west1-bigtable.googleapis.com:443
+  AUTOCONNECT__RELIABILITY_DSN: "redis://autopush-redis-stage.autopush.nonprod.webservices.mozgcp.net:6379"
+
+  AUTOTRACK_BUCKET_NAME: "autopush-dev-reliability"
+  AUTOTRACK_TERMINAL_MAX_RETENTION_DAYS: 1
+
+handlers:
+- url: /gen_report
+  script: auto
diff --git a/scripts/reliability/pyproject.toml b/scripts/reliability/pyproject.toml
@@ -0,0 +1,22 @@
+[project]
+name = "autopush-reliability-report"
+description = "Generate the Autopush Reliability Report"
+version = "0.1.0"
+requires-python = ">= 3.12"
+authors = [{ name = "jr conlin", email = "src+autopush@jrconlin.com" }]
+dependencies = [
+    "aiohttp",
+    "cryptography",
+    "google-auth",
+    "google-cloud-bigtable",
+    "google-cloud-redis",
+    "google-cloud-storage",
+    "jinja2",
+    "redis",
+    "toml",
+]
+license = "MPL-2.0"
+readme = "README.md"
+
+[tool.setuptools]
+packages = []
diff --git a/scripts/reliability/reliability_report.py b/scripts/reliability/reliability_report.py
diff --git a/scripts/reliability/templates/reliable_report_template.md b/scripts/reliability/templates/reliable_report_template.md
diff --git a/scripts/reliability_cron.py b/scripts/reliability_cron.py
diff --git a/scripts/reliability_report.py b/scripts/reliability_report.py

Original file line number	Diff line number	Diff line change
`@@ -501,7 +501,7 @@ impl Router for ApnsRouter {`
`501`	`501`	`// mutable, but we are also essentially consuming the`
`502`	`502`	`// notification nothing else should modify it.`
`503`	`503`	`notification`
`504`		`- .record_reliability(&self.reliability, ReliabilityState::Transmitted)`
	`504`	`+ .record_reliability(&self.reliability, ReliabilityState::BridgeTransmitted)`
`505`	`505`	`.await;`
`506`	`506`	`}`
`507`	`507`
Original file line number	Diff line number	Diff line change
`@@ -99,6 +99,7 @@ impl Default for Settings {`
`99`	`99`	`stub: StubSettings::default(),`
`100`	`100`	`#[cfg(feature = "reliable_report")]`
`101`	`101`	`reliability_dsn: None,`
	`102`	`+ #[cfg(feature = "reliable_report")]`
`102`	`103`	`reliability_retry_count: autopush_common::redis_util::MAX_TRANSACTION_LOOP,`
`103`	`104`	`}`
`104`	`105`	`}`