refactor(push): retire the push service delivery experiment (#3000)

fluxer_push/src/delivery.rs

@@ -387,7 +387,6 @@ async fn run_message_job(
         guild_id = %job.guild_id,
         channel_id = %job.channel_id,
         message_id = %job.message_id,
-        config_version = job.config_version,
         recipients = job.user_ids.len(),
         subscriptions = summary.subscriptions,
         accepted = summary.accepted,
@@ -432,7 +431,6 @@ async fn run_clear_job(state: &AppState, sends: &Semaphore, job: ClearJob) -> an
         user_id = %job.user_id,
         channel_id = %job.channel_id,
         message_id = %job.message_id,
-        config_version = job.config_version,
         recipients = 1,
         subscriptions = summary.subscriptions,
         accepted = summary.accepted,
@@ -455,7 +453,6 @@ async fn run_ring_job(state: &AppState, sends: &Semaphore, job: RingJob) -> anyh
             user_id = %job.user_id,
             channel_id = %job.channel_id,
             message_id = %job.message_id,
-            config_version = job.config_version,
             expires_at_ms = job.expires_at_ms,
             "push ring dropped past its ring window"
         );
@@ -487,7 +484,6 @@ async fn run_ring_job(state: &AppState, sends: &Semaphore, job: RingJob) -> anyh
         user_id = %job.user_id,
         channel_id = %job.channel_id,
         message_id = %job.message_id,
-        config_version = job.config_version,
         recipients = 1,
         subscriptions = summary.subscriptions,
         accepted = summary.accepted,

fluxer_push/src/job.rs

@@ -14,7 +14,6 @@ const DIRECT_MESSAGE_GUILD_ID: &str = "0";
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
 pub struct MessageJob {
     pub v: u8,
-    pub config_version: u64,
     pub guild_id: String,
     pub channel_id: String,
     pub message_id: String,
@@ -38,7 +37,6 @@ pub struct NotificationFields {
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
 pub struct ClearJob {
     pub v: u8,
-    pub config_version: u64,
     pub user_id: String,
     pub channel_id: String,
     pub message_id: String,
@@ -47,7 +45,6 @@ pub struct ClearJob {
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
 pub struct RingJob {
     pub v: u8,
-    pub config_version: u64,
     pub user_id: String,
     pub channel_id: String,
     pub message_id: String,
@@ -99,3 +96,38 @@ fn supported(version: u8) -> Result<(), JobError> {
     }
     Err(JobError::UnsupportedVersion(version))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::{Value, json};
+
+    fn clear_job(config_version: Option<u64>) -> Vec<u8> {
+        let mut job = json!({
+            "v": 1,
+            "user_id": "1",
+            "channel_id": "2",
+            "message_id": "3",
+        });
+        if let Some(version) = config_version {
+            job["config_version"] = Value::from(version);
+        }
+        job.to_string().into_bytes()
+    }
+
+    #[test]
+    fn a_job_from_a_gateway_that_sends_a_config_version_decodes() {
+        assert_eq!(
+            decode_clear(&clear_job(Some(7))).expect("decodes").user_id,
+            "1"
+        );
+    }
+
+    #[test]
+    fn a_job_without_a_config_version_decodes() {
+        assert_eq!(
+            decode_clear(&clear_job(None)).expect("decodes").user_id,
+            "1"
+        );
+    }
+}

fluxer_push/src/lib.rs

@@ -11,9 +11,9 @@ mod metrics;
 mod payload;
 mod providers;
 mod relay;
+mod relay_consent;
 mod resolver;
 mod retry;
-mod rollout;
 mod rpc;
 mod secret;
 pub mod server;

fluxer_push/src/metrics.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use crate::relay::reject::{REASON_COUNT, Reason};
-use crate::rollout::{RolloutOutcome, RolloutSnapshot};
+use crate::relay_consent::ConsentUpdate;
 use fluxer_svc::metrics::now_ms;
 use std::fmt::{self, Write as _};
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -343,7 +343,7 @@ const BUCKET_KEY_COUNT: usize = BucketKey::ALL.len();
 const PAYLOAD_SHRINK_COUNT: usize = PayloadShrink::ALL.len();
 const DELIVERY_ROUTE_COUNT: usize = DeliveryRoute::ALL.len();
 const RPC_OUTCOME_COUNT: usize = RpcOutcome::ALL.len();
-const ROLLOUT_OUTCOME_COUNT: usize = RolloutOutcome::ALL.len();
+const CONSENT_UPDATE_COUNT: usize = ConsentUpdate::ALL.len();
 
 pub struct Metrics {
     jobs_received: [AtomicU64; JOB_KIND_COUNT],
@@ -357,7 +357,7 @@ pub struct Metrics {
     own_relay_shortcuts: AtomicU64,
     auth_tokens_minted: [AtomicU64; AUTH_PROVIDER_COUNT],
     rpc_requests: [[AtomicU64; RPC_OUTCOME_COUNT]; RPC_METHOD_COUNT],
-    rollout_updates: [AtomicU64; ROLLOUT_OUTCOME_COUNT],
+    relay_consent_updates: [AtomicU64; CONSENT_UPDATE_COUNT],
     delivery_routes: [[AtomicU64; SEND_RESULT_COUNT]; DELIVERY_ROUTE_COUNT],
     relay_served: [[AtomicU64; RELAY_RESULT_COUNT]; RELAY_LEG_COUNT],
     relay_vendor_requests: [[AtomicU64; RELAY_RESULT_COUNT]; RELAY_LEG_COUNT],
@@ -368,9 +368,7 @@ pub struct Metrics {
     rpc_duration: [Histogram; RPC_METHOD_COUNT],
     send_duration: [Histogram; PROVIDER_COUNT],
     queue_depth: AtomicU64,
-    rollout_enabled: AtomicU64,
-    rollout_basis_points: AtomicU64,
-    rollout_config_version: AtomicU64,
+    relay_consent_accepted: AtomicU64,
     start_ms: i64,
 }
 
@@ -389,7 +387,7 @@ impl Metrics {
             auth_tokens_minted: [const { AtomicU64::new(0) }; AUTH_PROVIDER_COUNT],
             rpc_requests: [const { [const { AtomicU64::new(0) }; RPC_OUTCOME_COUNT] };
                 RPC_METHOD_COUNT],
-            rollout_updates: [const { AtomicU64::new(0) }; ROLLOUT_OUTCOME_COUNT],
+            relay_consent_updates: [const { AtomicU64::new(0) }; CONSENT_UPDATE_COUNT],
             delivery_routes: [const { [const { AtomicU64::new(0) }; SEND_RESULT_COUNT] };
                 DELIVERY_ROUTE_COUNT],
             relay_served: [const { [const { AtomicU64::new(0) }; RELAY_RESULT_COUNT] };
@@ -403,9 +401,7 @@ impl Metrics {
             rpc_duration: [const { Histogram::new() }; RPC_METHOD_COUNT],
             send_duration: [const { Histogram::new() }; PROVIDER_COUNT],
             queue_depth: AtomicU64::new(0),
-            rollout_enabled: AtomicU64::new(0),
-            rollout_basis_points: AtomicU64::new(0),
-            rollout_config_version: AtomicU64::new(0),
+            relay_consent_accepted: AtomicU64::new(0),
             start_ms: now_ms(),
         }
     }
@@ -457,17 +453,13 @@ impl Metrics {
         self.rpc_duration[method as usize].observe(duration_ms);
     }
 
-    pub fn record_rollout_update(&self, outcome: RolloutOutcome) {
-        self.rollout_updates[outcome as usize].fetch_add(1, ORDERING);
+    pub fn record_relay_consent_update(&self, outcome: ConsentUpdate) {
+        self.relay_consent_updates[outcome as usize].fetch_add(1, ORDERING);
     }
 
-    pub fn record_rollout_snapshot(&self, snapshot: &RolloutSnapshot) {
-        self.rollout_enabled
-            .store(u64::from(snapshot.enabled), ORDERING);
-        self.rollout_basis_points
-            .store(u64::from(snapshot.rollout_basis_points), ORDERING);
-        self.rollout_config_version
-            .store(snapshot.config_version, ORDERING);
+    pub fn record_relay_consent_accepted(&self, accepted: bool) {
+        self.relay_consent_accepted
+            .store(u64::from(accepted), ORDERING);
     }
 
     pub fn record_delivery_route(&self, route: DeliveryRoute, result: SendResult) {
@@ -575,10 +567,10 @@ impl Metrics {
 
         render_labelled_counter(
             out,
-            "fluxer_push_rollout_updates_total",
+            "fluxer_push_relay_consent_updates_total",
             "result",
-            RolloutOutcome::ALL.map(RolloutOutcome::label),
-            &self.rollout_updates,
+            ConsentUpdate::ALL.map(ConsentUpdate::label),
+            &self.relay_consent_updates,
         )?;
         render_labelled_histogram(
             out,
@@ -642,16 +634,10 @@ impl Metrics {
             &self.rings_suppressed,
         )?;
         render_gauge(out, "fluxer_push_queue_depth", &self.queue_depth)?;
-        render_gauge(out, "fluxer_push_rollout_enabled", &self.rollout_enabled)?;
         render_gauge(
             out,
-            "fluxer_push_rollout_basis_points",
-            &self.rollout_basis_points,
-        )?;
-        render_gauge(
-            out,
-            "fluxer_push_rollout_config_version",
-            &self.rollout_config_version,
+            "fluxer_push_relay_consent_accepted",
+            &self.relay_consent_accepted,
         )?;
 
         writeln!(out, "# TYPE fluxer_push_uptime_seconds gauge")?;

fluxer_push/src/payload.rs

@@ -605,7 +605,6 @@ mod tests {
     fn message_job(image_url: Option<&str>) -> MessageJob {
         MessageJob {
             v: 1,
-            config_version: 7,
             guild_id: "0".to_owned(),
             channel_id: CHANNEL_ID.to_owned(),
             message_id: MESSAGE_ID.to_owned(),
@@ -626,7 +625,6 @@ mod tests {
     fn clear_job() -> ClearJob {
         ClearJob {
             v: 1,
-            config_version: 7,
             user_id: USER_ID.to_owned(),
             channel_id: CHANNEL_ID.to_owned(),
             message_id: MESSAGE_ID.to_owned(),

fluxer_push/src/providers/mod.rs

@@ -118,18 +118,10 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen
 }
 
 fn relay_consent_missing(state: &AppState, endpoint: &str) -> bool {
-    !relay_consent_accepted(state)
+    !state.relay_consent.accepted()
         && own_relay::is_managed(endpoint, &state.cfg.managed_relay_hosts)
 }
 
-fn relay_consent_accepted(state: &AppState) -> bool {
-    state.cfg.relay_consent_accepted
-        || state
-            .rollout
-            .snapshot()
-            .is_some_and(|held| held.relay_consent_accepted)
-}
-
 fn in_process_hop(endpoint: &str, hosts: &[String]) -> Option<own_relay::Hop> {
     own_relay::parse(endpoint, hosts).filter(|hop| matches!(hop.leg, own_relay::Leg::Apns))
 }
@@ -276,8 +268,7 @@ mod consent_tests {
     #[tokio::test]
     async fn a_notice_accepted_in_the_instance_config_lets_the_send_through() {
         let state = state(false);
-        state.rollout.update(&serde_json::json!({
-            "enabled": true,
+        state.relay_consent.update(&serde_json::json!({
             "config_version": 1,
             "relay_consent_accepted": true,
         }));
@@ -290,8 +281,7 @@ mod consent_tests {
     #[tokio::test]
     async fn an_instance_config_that_has_not_accepted_still_refuses_the_send() {
         let state = state(false);
-        state.rollout.update(&serde_json::json!({
-            "enabled": true,
+        state.relay_consent.update(&serde_json::json!({
             "config_version": 1,
             "relay_consent_accepted": false,
         }));

fluxer_push/src/relay_consent.rs

@@ -0,0 +1,314 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use crate::metrics::Metrics;
+use crate::rpc::RpcClient;
+use fluxer_svc::transport::{Transport, TransportMessage, TransportSubscriber};
+use serde_json::Value;
+use std::sync::RwLock;
+use std::time::Duration;
+use tokio::time::{Instant, MissedTickBehavior};
+use tracing::{info, warn};
+
+pub const RECONCILE_INTERVAL: Duration = Duration::from_secs(30);
+
+const SUBJECT: &str = "config.push.delivery";
+const MESSAGE_TYPE: &str = "push_service_delivery_config";
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[repr(usize)]
+pub enum ConsentUpdate {
+    Updated,
+    Unchanged,
+    Stale,
+    Rejected,
+}
+
+impl ConsentUpdate {
+    pub const ALL: [Self; 4] = [Self::Updated, Self::Unchanged, Self::Stale, Self::Rejected];
+
+    pub fn label(self) -> &'static str {
+        match self {
+            Self::Updated => "updated",
+            Self::Unchanged => "unchanged",
+            Self::Stale => "stale",
+            Self::Rejected => "rejected",
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+struct HeldConsent {
+    accepted: bool,
+    config_version: u64,
+}
+
+impl HeldConsent {
+    fn parse(config: &Value) -> Option<Self> {
+        let accepted = match config.get("relay_consent_accepted") {
+            None | Some(Value::Null) => false,
+            Some(Value::Bool(flag)) => *flag,
+            Some(_) => return None,
+        };
+        let config_version = match config.get("config_version") {
+            None | Some(Value::Null) => 0,
+            Some(value) => value.as_u64()?,
+        };
+        Some(Self {
+            accepted,
+            config_version,
+        })
+    }
+}
+
+pub struct RelayConsentStore {
+    env_accepted: bool,
+    held: RwLock<Option<HeldConsent>>,
+}
+
+impl RelayConsentStore {
+    pub fn new(env_accepted: bool) -> Self {
+        Self {
+            env_accepted,
+            held: RwLock::new(None),
+        }
+    }
+
+    pub fn accepted(&self) -> bool {
+        self.env_accepted || self.held().is_some_and(|held| held.accepted)
+    }
+
+    fn held(&self) -> Option<HeldConsent> {
+        *self.held.read().expect("relay consent lock poisoned")
+    }
+
+    fn apply(&self, payload: &[u8]) -> ConsentUpdate {
+        let Ok(value) = serde_json::from_slice::<Value>(payload) else {
+            warn!("relay consent payload is not JSON");
+            return ConsentUpdate::Rejected;
+        };
+        let Some(config) = config_object(&value) else {
+            warn!("relay consent payload has no config object of its type");
+            return ConsentUpdate::Rejected;
+        };
+        self.update(config)
+    }
+
+    pub(crate) fn update(&self, config: &Value) -> ConsentUpdate {
+        let Some(offered) = HeldConsent::parse(config) else {
+            warn!("relay consent config rejected as invalid");
+            return ConsentUpdate::Rejected;
+        };
+        let mut held = self.held.write().expect("relay consent lock poisoned");
+        match *held {
+            Some(current) if offered.config_version < current.config_version => {
+                warn!(
+                    highest = current.config_version,
+                    offered = offered.config_version,
+                    "relay consent ignored a lower config_version"
+                );
+                return ConsentUpdate::Stale;
+            }
+            Some(current) if current == offered => return ConsentUpdate::Unchanged,
+            _ => {}
+        }
+        info!(
+            accepted = offered.accepted,
+            config_version = offered.config_version,
+            "relay consent updated"
+        );
+        *held = Some(offered);
+        ConsentUpdate::Updated
+    }
+}
+
+pub async fn run_subscriber<T: Transport>(
+    transport: T,
+    rpc: &RpcClient,
+    store: &RelayConsentStore,
+    metrics: &Metrics,
+    reconcile_every: Duration,
+) {
+    loop {
+        let mut subscriber = match transport.subscribe(SUBJECT).await {
+            Ok(subscriber) => subscriber,
+            Err(error) => {
+                warn!(error = %error, subject = SUBJECT, "relay consent subscribe failed");
+                transport.wait_for_reconnect().await;
+                continue;
+            }
+        };
+        info!(subject = SUBJECT, "listening for relay consent updates");
+        let outcome = fetch(rpc, store, metrics).await;
+        info!(
+            outcome = outcome.label(),
+            "relay consent read after subscribing"
+        );
+        let mut reconcile =
+            tokio::time::interval_at(Instant::now() + reconcile_every, reconcile_every);
+        reconcile.set_missed_tick_behavior(MissedTickBehavior::Delay);
+        loop {
+            tokio::select! {
+                message = subscriber.next() => {
+                    let Some(message) = message else {
+                        break;
+                    };
+                    let outcome = store.apply(message.payload());
+                    record(metrics, store, outcome);
+                }
+                _ = reconcile.tick() => {
+                    fetch(rpc, store, metrics).await;
+                }
+            }
+        }
+        warn!(
+            subject = SUBJECT,
+            "relay consent subscription ended, will re-subscribe"
+        );
+    }
+}
+
+async fn fetch(rpc: &RpcClient, store: &RelayConsentStore, metrics: &Metrics) -> ConsentUpdate {
+    let outcome = match rpc.push_service_delivery_config().await {
+        Ok(config) => store.update(&config),
+        Err(error) => {
+            warn!(error = %error, "relay consent read failed");
+            ConsentUpdate::Rejected
+        }
+    };
+    record(metrics, store, outcome);
+    outcome
+}
+
+fn record(metrics: &Metrics, store: &RelayConsentStore, outcome: ConsentUpdate) {
+    metrics.record_relay_consent_update(outcome);
+    metrics.record_relay_consent_accepted(store.accepted());
+}
+
+fn config_object(value: &Value) -> Option<&Value> {
+    if value
+        .get("type")
+        .is_some_and(|found| found.as_str() != Some(MESSAGE_TYPE))
+    {
+        return None;
+    }
+    if let Some(config) = value.get("config").filter(|config| config.is_object()) {
+        return Some(config);
+    }
+    value.is_object().then_some(value)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    fn legacy_document(config_version: u64, accepted: bool) -> Value {
+        json!({
+            "enabled": true,
+            "rollout_basis_points": 10000,
+            "rollout_salt": "push-service-delivery-v1",
+            "included_user_ids": [],
+            "excluded_user_ids": [],
+            "config_version": config_version,
+            "relay_consent_accepted": accepted,
+            "relay_consent_accepted_at": null,
+            "relay_consent_accepted_by": null,
+        })
+    }
+
+    #[test]
+    fn the_operator_relay_consent_is_read_off_the_legacy_document() {
+        let store = RelayConsentStore::new(false);
+        assert_eq!(
+            store.update(&legacy_document(3, true)),
+            ConsentUpdate::Updated
+        );
+        assert!(store.accepted());
+    }
+
+    #[test]
+    fn a_document_without_the_consent_field_has_not_consented() {
+        let store = RelayConsentStore::new(false);
+        assert_eq!(
+            store.update(&json!({"config_version": 3})),
+            ConsentUpdate::Updated
+        );
+        assert!(!store.accepted());
+    }
+
+    #[test]
+    fn a_consent_field_that_is_not_a_boolean_is_refused() {
+        let store = RelayConsentStore::new(false);
+        assert_eq!(
+            store.update(&json!({"relay_consent_accepted": "yes"})),
+            ConsentUpdate::Rejected
+        );
+        assert!(store.held().is_none());
+    }
+
+    #[test]
+    fn a_config_version_that_is_not_an_integer_is_refused() {
+        let store = RelayConsentStore::new(false);
+        assert_eq!(
+            store.update(&json!({"config_version": -1, "relay_consent_accepted": true})),
+            ConsentUpdate::Rejected
+        );
+        assert!(!store.accepted());
+    }
+
+    #[test]
+    fn an_older_config_version_never_overwrites_a_newer_one() {
+        let store = RelayConsentStore::new(false);
+        store.update(&legacy_document(5, true));
+        assert_eq!(
+            store.update(&legacy_document(4, false)),
+            ConsentUpdate::Stale
+        );
+        assert!(store.accepted());
+        assert_eq!(
+            store.update(&legacy_document(6, false)),
+            ConsentUpdate::Updated
+        );
+        assert!(!store.accepted());
+    }
+
+    #[test]
+    fn the_same_document_again_is_unchanged() {
+        let store = RelayConsentStore::new(false);
+        store.update(&legacy_document(5, true));
+        assert_eq!(
+            store.update(&legacy_document(5, true)),
+            ConsentUpdate::Unchanged
+        );
+    }
+
+    #[test]
+    fn the_env_override_accepts_whatever_the_document_says() {
+        let store = RelayConsentStore::new(true);
+        assert!(store.accepted());
+        store.update(&legacy_document(1, false));
+        assert!(store.accepted());
+    }
+
+    #[test]
+    fn a_nats_message_is_unwrapped_from_its_envelope() {
+        let store = RelayConsentStore::new(false);
+        let message = json!({"type": MESSAGE_TYPE, "config": legacy_document(2, true)});
+        assert_eq!(
+            store.apply(message.to_string().as_bytes()),
+            ConsentUpdate::Updated
+        );
+        assert!(store.accepted());
+    }
+
+    #[test]
+    fn a_nats_message_of_another_type_is_refused() {
+        let store = RelayConsentStore::new(false);
+        let message = json!({"type": "something_else", "config": legacy_document(2, true)});
+        assert_eq!(
+            store.apply(message.to_string().as_bytes()),
+            ConsentUpdate::Rejected
+        );
+        assert!(!store.accepted());
+    }
+}

fluxer_push/src/rollout.rs

@@ -1,326 +0,0 @@
-// SPDX-License-Identifier: AGPL-3.0-or-later
-
-use crate::metrics::{Metrics, RpcMethod};
-use crate::rpc::RpcClient;
-use fluxer_svc::transport::{Transport, TransportMessage, TransportSubscriber};
-use serde_json::Value;
-use std::fmt::Debug;
-use std::sync::{Arc, RwLock};
-use std::time::Duration;
-use tokio::time::{Instant, MissedTickBehavior};
-use tracing::{info, warn};
-
-pub const RECONCILE_INTERVAL: Duration = Duration::from_secs(30);
-
-const MAX_BASIS_POINTS: u64 = 10_000;
-
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
-#[repr(usize)]
-pub enum RolloutOutcome {
-    Updated,
-    Unchanged,
-    Stale,
-    Rejected,
-}
-
-impl RolloutOutcome {
-    pub const ALL: [Self; 4] = [Self::Updated, Self::Unchanged, Self::Stale, Self::Rejected];
-
-    pub fn label(self) -> &'static str {
-        match self {
-            Self::Updated => "updated",
-            Self::Unchanged => "unchanged",
-            Self::Stale => "stale",
-            Self::Rejected => "rejected",
-        }
-    }
-}
-
-pub trait RolloutConfig: Debug + Eq + Send + Sync + Sized + 'static {
-    const NAME: &'static str;
-    const SUBJECT: &'static str;
-    const MESSAGE_TYPE: &'static str;
-    const RPC_METHOD: RpcMethod;
-
-    fn parse(config: &Value) -> Option<Self>;
-    fn enabled(&self) -> bool;
-    fn config_version(&self) -> u64;
-    fn record_update(metrics: &Metrics, outcome: RolloutOutcome);
-    fn record_held(&self, metrics: &Metrics);
-}
-
-#[derive(Clone, Debug, Default, Eq, PartialEq)]
-pub struct RolloutSnapshot {
-    pub enabled: bool,
-    pub config_version: u64,
-    pub rollout_basis_points: u32,
-    pub relay_consent_accepted: bool,
-}
-
-impl RolloutConfig for RolloutSnapshot {
-    const NAME: &'static str = "push delivery";
-    const SUBJECT: &'static str = "config.push.delivery";
-    const MESSAGE_TYPE: &'static str = "push_service_delivery_config";
-    const RPC_METHOD: RpcMethod = RpcMethod::GetPushServiceDeliveryConfig;
-
-    fn parse(config: &Value) -> Option<Self> {
-        let rollout_basis_points = match config.get("rollout_basis_points") {
-            None | Some(Value::Null) => 0,
-            Some(value) => {
-                let points = value.as_u64()?;
-                if points > MAX_BASIS_POINTS {
-                    return None;
-                }
-                u32::try_from(points).ok()?
-            }
-        };
-        Some(Self {
-            enabled: parse_enabled(config)?,
-            config_version: parse_config_version(config)?,
-            rollout_basis_points,
-            relay_consent_accepted: parse_flag(config, "relay_consent_accepted")?,
-        })
-    }
-
-    fn enabled(&self) -> bool {
-        self.enabled
-    }
-
-    fn config_version(&self) -> u64 {
-        self.config_version
-    }
-
-    fn record_update(metrics: &Metrics, outcome: RolloutOutcome) {
-        metrics.record_rollout_update(outcome);
-    }
-
-    fn record_held(&self, metrics: &Metrics) {
-        metrics.record_rollout_snapshot(self);
-    }
-}
-
-pub fn parse_enabled(config: &Value) -> Option<bool> {
-    parse_flag(config, "enabled")
-}
-
-fn parse_flag(config: &Value, key: &str) -> Option<bool> {
-    match config.get(key) {
-        None | Some(Value::Null) => Some(false),
-        Some(Value::Bool(flag)) => Some(*flag),
-        Some(_) => None,
-    }
-}
-
-pub fn parse_config_version(config: &Value) -> Option<u64> {
-    match config.get("config_version") {
-        None | Some(Value::Null) => Some(0),
-        Some(value) => value.as_u64(),
-    }
-}
-
-struct Current<C> {
-    held: Option<Arc<C>>,
-    highest_version: u64,
-}
-
-pub struct RolloutStore<C> {
-    current: RwLock<Current<C>>,
-}
-
-impl<C> Default for RolloutStore<C> {
-    fn default() -> Self {
-        Self {
-            current: RwLock::new(Current {
-                held: None,
-                highest_version: 0,
-            }),
-        }
-    }
-}
-
-impl<C: RolloutConfig> RolloutStore<C> {
-    pub fn new() -> Self {
-        Self::default()
-    }
-
-    pub fn snapshot(&self) -> Option<Arc<C>> {
-        self.current
-            .read()
-            .expect("rollout config lock poisoned")
-            .held
-            .clone()
-    }
-
-    fn apply(&self, payload: &[u8]) -> RolloutOutcome {
-        let Ok(value) = serde_json::from_slice::<Value>(payload) else {
-            warn!(config = C::NAME, "rollout config payload is not JSON");
-            return RolloutOutcome::Rejected;
-        };
-        let Some(config) = config_object(&value, C::MESSAGE_TYPE) else {
-            warn!(
-                config = C::NAME,
-                "rollout config payload has no config object of its type"
-            );
-            return RolloutOutcome::Rejected;
-        };
-        self.update(config)
-    }
-
-    pub(crate) fn update(&self, config: &Value) -> RolloutOutcome {
-        let Some(offered) = C::parse(config) else {
-            warn!(config = C::NAME, "rollout config rejected as invalid");
-            return RolloutOutcome::Rejected;
-        };
-        let mut current = self.current.write().expect("rollout config lock poisoned");
-        match current.held.as_deref() {
-            Some(_) if offered.enabled() && offered.config_version() < current.highest_version => {
-                warn!(
-                    config = C::NAME,
-                    highest = current.highest_version,
-                    offered = offered.config_version(),
-                    "rollout config ignored a lower config_version"
-                );
-                return RolloutOutcome::Stale;
-            }
-            Some(held) if *held == offered => return RolloutOutcome::Unchanged,
-            _ => {}
-        }
-        info!(config = C::NAME, held = ?offered, "rollout config updated");
-        current.highest_version = current.highest_version.max(offered.config_version());
-        current.held = Some(Arc::new(offered));
-        RolloutOutcome::Updated
-    }
-}
-
-pub async fn run_rollout_subscriber<T: Transport, C: RolloutConfig>(
-    transport: T,
-    rpc: &RpcClient,
-    store: &RolloutStore<C>,
-    metrics: &Metrics,
-    reconcile_every: Duration,
-) {
-    loop {
-        let mut subscriber = match transport.subscribe(C::SUBJECT).await {
-            Ok(subscriber) => subscriber,
-            Err(error) => {
-                warn!(
-                    config = C::NAME,
-                    error = %error,
-                    subject = C::SUBJECT,
-                    "rollout config subscribe failed"
-                );
-                transport.wait_for_reconnect().await;
-                continue;
-            }
-        };
-        info!(
-            config = C::NAME,
-            subject = C::SUBJECT,
-            "listening for rollout config updates"
-        );
-        let outcome = fetch_config(rpc, store, metrics).await;
-        info!(
-            config = C::NAME,
-            outcome = outcome.label(),
-            "rollout config read after subscribing"
-        );
-        let mut reconcile =
-            tokio::time::interval_at(Instant::now() + reconcile_every, reconcile_every);
-        reconcile.set_missed_tick_behavior(MissedTickBehavior::Delay);
-        loop {
-            tokio::select! {
-                message = subscriber.next() => {
-                    let Some(message) = message else {
-                        break;
-                    };
-                    let outcome = store.apply(message.payload());
-                    record_outcome(metrics, store, outcome);
-                }
-                _ = reconcile.tick() => {
-                    fetch_config(rpc, store, metrics).await;
-                }
-            }
-        }
-        warn!(
-            config = C::NAME,
-            subject = C::SUBJECT,
-            "rollout config subscription ended, will re-subscribe"
-        );
-    }
-}
-
-async fn fetch_config<C: RolloutConfig>(
-    rpc: &RpcClient,
-    store: &RolloutStore<C>,
-    metrics: &Metrics,
-) -> RolloutOutcome {
-    match rpc.rollout_config(C::RPC_METHOD).await {
-        Ok(config) => {
-            let outcome = store.update(&config);
-            record_outcome(metrics, store, outcome);
-            outcome
-        }
-        Err(error) => {
-            warn!(config = C::NAME, error = %error, "rollout config read failed");
-            C::record_update(metrics, RolloutOutcome::Rejected);
-            RolloutOutcome::Rejected
-        }
-    }
-}
-
-fn record_outcome<C: RolloutConfig>(
-    metrics: &Metrics,
-    store: &RolloutStore<C>,
-    outcome: RolloutOutcome,
-) {
-    C::record_update(metrics, outcome);
-    if outcome == RolloutOutcome::Rejected {
-        return;
-    }
-    if let Some(held) = store.snapshot() {
-        held.record_held(metrics);
-    }
-}
-
-fn config_object<'a>(value: &'a Value, message_type: &str) -> Option<&'a Value> {
-    if value
-        .get("type")
-        .is_some_and(|found| found.as_str() != Some(message_type))
-    {
-        return None;
-    }
-    if let Some(config) = value.get("config").filter(|config| config.is_object()) {
-        return Some(config);
-    }
-    value.is_object().then_some(value)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn the_operator_relay_consent_is_read_off_the_instance_config() {
-        let config = json!({
-            "enabled": true,
-            "config_version": 3,
-            "relay_consent_accepted": true,
-        });
-        let snapshot = RolloutSnapshot::parse(&config).expect("the config parses");
-        assert!(snapshot.relay_consent_accepted);
-    }
-
-    #[test]
-    fn an_instance_config_without_the_consent_field_has_not_consented() {
-        let config = json!({"enabled": true, "config_version": 3});
-        let snapshot = RolloutSnapshot::parse(&config).expect("the config parses");
-        assert!(!snapshot.relay_consent_accepted);
-    }
-
-    #[test]
-    fn a_consent_field_that_is_not_a_boolean_is_refused() {
-        let config = json!({"enabled": true, "relay_consent_accepted": "yes"});
-        assert!(RolloutSnapshot::parse(&config).is_none());
-    }
-}

fluxer_push/src/rpc.rs

@@ -62,8 +62,13 @@ impl RpcClient {
         }
     }
 
-    pub async fn rollout_config(&self, method: RpcMethod) -> Result<Value, RpcError> {
-        let data: RolloutConfigData = self.call(method, &json!({"type": method.label()})).await?;
+    pub async fn push_service_delivery_config(&self) -> Result<Value, RpcError> {
+        let data: PushServiceDeliveryConfigData = self
+            .call(
+                RpcMethod::GetPushServiceDeliveryConfig,
+                &json!({"type": "get_push_service_delivery_config"}),
+            )
+            .await?;
         Ok(data.config)
     }
 
@@ -190,7 +195,7 @@ impl RpcClient {
 }
 
 #[derive(Deserialize)]
-struct RolloutConfigData {
+struct PushServiceDeliveryConfigData {
     config: Value,
 }
 

fluxer_push/src/server.rs

@@ -4,7 +4,7 @@ use crate::config::{Config, DeliveryConfig};
 use crate::delivery;
 use crate::metrics::Metrics;
 use crate::relay;
-use crate::rollout::{self, RolloutSnapshot, RolloutStore};
+use crate::relay_consent::{self, RelayConsentStore};
 use crate::rpc::RpcClient;
 use crate::secret::SecretString;
 use crate::tokens::TokenCache;
@@ -91,7 +91,7 @@ pub struct AppState {
     pub(crate) cfg: DeliveryConfig,
     pub(crate) metrics: Arc<Metrics>,
     pub(crate) sidecar: Arc<Sidecar>,
-    pub(crate) rollout: RolloutStore<RolloutSnapshot>,
+    pub(crate) relay_consent: RelayConsentStore,
     pub(crate) rpc: RpcClient,
     pub(crate) http: reqwest::Client,
     pub(crate) web_push_http: reqwest::Client,
@@ -103,10 +103,12 @@ pub struct AppState {
 impl AppState {
     pub(crate) fn try_new(cfg: DeliveryConfig) -> anyhow::Result<Self> {
         let metrics = Arc::new(Metrics::new());
+        let relay_consent = RelayConsentStore::new(cfg.relay_consent_accepted);
+        metrics.record_relay_consent_accepted(relay_consent.accepted());
         let http = vendor::http_client()?;
         Ok(Self {
             rpc: RpcClient::new(&cfg.rpc, http.clone(), Arc::clone(&metrics)),
-            rollout: RolloutStore::new(),
+            relay_consent,
             sidecar: Arc::new(Sidecar::new(Arc::clone(&metrics))),
             web_push_http: vendor::web_push_http_client()?,
             apns_http: vendor::apns_http_client()?,
@@ -144,12 +146,12 @@ async fn run_delivery(cfg: DeliveryConfig) -> anyhow::Result<()> {
         let transport = transport.clone();
         let state = Arc::clone(&state);
         async move {
-            rollout::run_rollout_subscriber(
+            relay_consent::run_subscriber(
                 transport,
                 &state.rpc,
-                &state.rollout,
+                &state.relay_consent,
                 &state.metrics,
-                rollout::RECONCILE_INTERVAL,
+                relay_consent::RECONCILE_INTERVAL,
             )
             .await
         }