feat(protocol): drop the telemetry queue when the portal toggle is off

CHANGELOG.md

@@ -132,6 +132,15 @@ the client's answers were wrong on a device rather than merely different:
   through, and the queue now waits for a good one. `telemetryStats().dropped`
   counts events and only events, so a queued batch that cannot be opened on
   load is warned about rather than added to it in a different unit.
+- **A 403 `telemetry_disabled` drops the queue.** The ingest answers with that
+  `error` code, rather than the `forbidden` a bad key draws, while the
+  application's telemetry toggle is off in the developer portal. The pipe
+  drops the refused batch and everything queued behind it, discards what it
+  collects afterwards instead of queuing it, and sends nothing more until the
+  next `enableTelemetry`, so a toggled-off stretch stays a gap on the
+  dashboard rather than filling in from the device's queue once the toggle
+  comes back. Every other 401 or 403 keeps its queue as above, and so does a
+  403 from an ingest that predates the code, which answers `forbidden`.
 - **Telemetry session boundaries follow the process on Android**, through the
   set of started activities rather than `onHostPause`. `onHostPause` is
   `Activity.onPause`, which fires for a runtime permission dialog (including

bindings/python/offline_protocol_sdk/protocol_manager.py

@@ -527,8 +527,10 @@ def telemetry_stats(self) -> TelemetryStats | None:
 
         ``accepted_events`` is what the ingest reported accepting, which is
         what an invoice is reconciled against; ``dropped`` counts events
-        lost to the ring buffer, the queue caps, the six-day expiry or a
-        permanent rejection; ``last_error`` is the most recent send failure.
+        lost to the ring buffer, the queue caps, the six-day expiry, a
+        permanent rejection, or everything queued or collected after the
+        ingest reported the application's telemetry toggle off;
+        ``last_error`` is the most recent send failure.
         """
         return self._protocol.telemetry_stats()
 

bindings/react-native/src/index.ts

@@ -1955,10 +1955,11 @@ export class OfflineProtocol {
    * The telemetry pipe's counters, or `null` while telemetry is not
    * enabled. `acceptedEvents` is what the ingest reported accepting, which
    * is what an invoice is reconciled against; `dropped` counts events lost
-   * to the ring buffer, the durable queue's caps, the six-day expiry, or a
-   * permanent rejection; `lastError` is the most recent send failure, and
-   * clears once a batch is accepted, so it reports the current state rather
-   * than the high-water mark of a recovered outage.
+   * to the ring buffer, the durable queue's caps, the six-day expiry, a
+   * permanent rejection, or everything queued or collected after the ingest
+   * reported the application's telemetry toggle off; `lastError` is the most
+   * recent send failure, and clears once a batch is accepted, so it reports
+   * the current state rather than the high-water mark of a recovered outage.
    */
   async telemetryStats(): Promise<TelemetryStats | null> {
     const stats = await OfflineProtocolNativeModule.telemetryStats();

bindings/react-native/src/types.ts

@@ -3081,8 +3081,10 @@ export interface TelemetryConfig {
  * `sentEvents` counts events in batches the ingest answered 2xx;
  * `acceptedEvents` sums the `accepted` count the ingest reported, which is
  * what an invoice is reconciled against. `dropped` counts events lost to
- * the ring buffer, the durable queue's caps, the six-day expiry, or a
- * permanent rejection. `lastFlushAtMs` is when a batch was last accepted.
+ * the ring buffer, the durable queue's caps, the six-day expiry, a
+ * permanent rejection, or everything queued or collected after the ingest
+ * reported the application's telemetry toggle off. `lastFlushAtMs` is when
+ * a batch was last accepted.
  */
 export interface TelemetryStats {
   buffered: number;

crates/offline-protocol-uniffi/src/offline_protocol.udl

@@ -934,7 +934,8 @@ dictionary TelemetryConfig {
 // in batches the ingest answered 2xx; `accepted_events` sums the `accepted`
 // count the ingest reported, which is what an invoice is reconciled against.
 // `dropped` counts events lost to the ring buffer, the durable queue's caps,
-// the six-day expiry, or a permanent rejection.
+// the six-day expiry, a permanent rejection, or everything queued or
+// collected after the ingest reported the application's telemetry toggle off.
 dictionary TelemetryStats {
     u64 buffered;
     u64 sent_events;

crates/offline-protocol/src/telemetry/pipe/mod.rs

@@ -118,7 +118,9 @@ pub struct TelemetryStats {
     /// Events the ingest reported accepted (the `accepted` count of each
     /// 202 body, summed; a replay counts nothing new).
     pub accepted_events: u64,
-    /// Events lost: ring overflow, queue caps, TTL expiry, permanent 4xx.
+    /// Events lost: ring overflow, queue caps, TTL expiry, permanent 4xx,
+    /// and everything queued or collected once the developer portal switched
+    /// telemetry off for the application.
     ///
     /// Events, never records. This number is what reconciles an invoice
     /// against the device, so it counts in the unit the ingest meters. A
@@ -188,6 +190,16 @@ pub(crate) struct PipeShared {
     /// `-1` until a batch has been accepted.
     last_flush_at_ms: AtomicI64,
     last_error: Mutex<Option<String>>,
+    /// Set once the ingest has answered `telemetry_disabled`: the
+    /// application's toggle is off in the developer portal. From then on the
+    /// ring is emptied into nothing at every cycle rather than cut into
+    /// batches, and a durable queue adopted afterwards is emptied as it is
+    /// adopted, so nothing the pipe holds or adopts is left on disk for a
+    /// later pipe to send, and the gap the toggle promises holds at the
+    /// device. Read on the cycle and at adoption, never on the emit path,
+    /// which keeps its one atomic load. Cleared only by a fresh pipe, which is
+    /// what the next `enable_telemetry` builds.
+    server_disabled: AtomicBool,
 }
 
 fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
@@ -229,7 +241,13 @@ impl PipeShared {
         self.adopt_pending_backend();
 
         let drained = lock(&self.pipeline).drain();
-        if !drained.is_empty() {
+        if self.server_disabled.load(Ordering::Relaxed) {
+            // Counted as dropped, in events, the unit the rest of `dropped`
+            // uses: the toggle is off, and these are the events it says are
+            // not kept.
+            self.uploader_dropped
+                .fetch_add(drained.len() as u64, Ordering::Relaxed);
+        } else if !drained.is_empty() {
             let device_id = if self.include_device_id {
                 lock(&self.device_id).clone()
             } else {
@@ -280,6 +298,15 @@ impl PipeShared {
         if report.batches_sent > 0 {
             self.last_flush_at_ms.store(now, Ordering::Relaxed);
         }
+        if report.disabled && !self.server_disabled.swap(true, Ordering::Relaxed) {
+            tracing::warn!(
+                target: PIPE_LOG_TARGET,
+                dropped_events = report.dropped_events,
+                "hosted telemetry is switched off for this application in the developer \
+                 portal; the queue was dropped and nothing more is collected until \
+                 enable_telemetry"
+            );
+        }
         // Mirrored, not accumulated: the uploader clears its own error when a
         // batch is accepted, and a stats reader asking "is telemetry working"
         // after a recovered outage must not still be shown the outage. A
@@ -305,6 +332,20 @@ impl PipeShared {
             Attach::Busy(backend) => Some((backend, true)),
             Attach::Failed(backend) => Some((backend, false)),
         };
+        if refused.is_none() && self.server_disabled.load(Ordering::Relaxed) {
+            // The toggle is off, and adoption is the one way a batch can still
+            // enter the store: the cycle no longer cuts any, but adopting
+            // loads whatever the queue on that storage already holds, an
+            // earlier launch's batches included. The halted uploader returns
+            // before it reads the queue, so a batch kept here would sit on
+            // disk until the next `enable_telemetry` resent it, which is the
+            // backfill the toggle promises not to make.
+            let mut dropped = 0u64;
+            while let Some(batch) = store.pop_front() {
+                dropped += u64::from(batch.event_count);
+            }
+            self.uploader_dropped.fetch_add(dropped, Ordering::Relaxed);
+        }
         store.flush_index();
         self.durable.store(store.is_durable(), Ordering::Relaxed);
         self.store_dropped
@@ -606,6 +647,7 @@ impl TelemetryPipe {
             store_dropped: AtomicU64::new(0),
             last_flush_at_ms: AtomicI64::new(-1),
             last_error: Mutex::new(None),
+            server_disabled: AtomicBool::new(false),
         });
         // The host's state at enable time, recorded rather than reported:
         // there is no session to close yet, and the seed exists so that a

crates/offline-protocol/src/telemetry/pipe/tests/mod.rs

@@ -38,18 +38,22 @@ impl FakeClock {
     }
 }
 
-/// A client that captures every body and answers with a fixed status.
+/// A client that captures every body and answers with a fixed status. A 2xx
+/// carries the ingest's accepted count; anything else carries the ingest's
+/// error shape, with `error_code` as the code (`forbidden` when unset).
 #[derive(Clone)]
 pub(crate) struct CapturingClient {
     pub(crate) bodies: Arc<Mutex<Vec<String>>>,
     pub(crate) status: Arc<AtomicI64>,
+    pub(crate) error_code: Arc<Mutex<Option<String>>>,
 }
 
 impl CapturingClient {
     pub(crate) fn accepting() -> Self {
         Self {
             bodies: Arc::default(),
             status: Arc::new(AtomicI64::new(202)),
+            error_code: Arc::default(),
         }
     }
 
@@ -74,10 +78,21 @@ impl HttpClient for CapturingClient {
             .ok()
             .and_then(|v| v["events"].as_array().map(|e| e.len()))
             .unwrap_or(0);
+        let body = if (200..300).contains(&status) {
+            format!(r#"{{"accepted":{accepted},"rejected":0,"rejected_reasons":[]}}"#)
+        } else {
+            let code = self
+                .error_code
+                .lock()
+                .unwrap()
+                .clone()
+                .unwrap_or_else(|| "forbidden".to_string());
+            format!(r#"{{"error":"{code}","message":"{code}"}}"#)
+        };
         Ok(Response {
             status: status as u16,
             retry_after: None,
-            body: format!(r#"{{"accepted":{accepted},"rejected":0,"rejected_reasons":[]}}"#),
+            body,
         })
     }
 }

crates/offline-protocol/src/telemetry/pipe/tests/scenarios.rs

@@ -556,3 +556,174 @@ fn a_detached_worker_finishing_an_upload_cannot_orphan_its_replacements_batch()
         "the batch the ingest accepted was queued again: {sessions:?}"
     );
 }
+
+/// The toggle's gap holds at the device. A `403` whose `error` code is
+/// `telemetry_disabled` drops the refused batch and everything queued behind
+/// it, what is collected afterwards is discarded rather than queued, and a
+/// fresh pipe over the same store has nothing from the off period to resend.
+/// Contrast `a_batch_refused_for_a_bad_key_is_still_there_when_the_key_is_fixed`:
+/// a bad key keeps its queue, because that is a fault the developer fixes.
+#[test]
+fn a_403_telemetry_disabled_drops_the_queue_and_discards_what_follows() {
+    let clock = FakeClock::at(1_000);
+    let storage = Arc::new(MemoryStorage::default());
+    let client = CapturingClient::accepting();
+    client.status.store(0, Ordering::SeqCst); // offline, so two batches queue up
+    let pipe = inline_pipe(&test_config(), &clock, client.clone(), sealed(&storage));
+    emit_failed(&pipe, 1);
+    pipe.flush();
+    clock.set(1_500); // inside the backoff: the second batch is cut, not sent
+    emit_failed(&pipe, 2);
+    pipe.flush();
+    let queued = |pipe: &TelemetryPipe| crate::telemetry::pipe::lock(&pipe.shared().store).len();
+    assert_eq!(queued(&pipe), 2);
+    assert!(client.bodies.lock().unwrap().is_empty());
+
+    // The developer switches the toggle off in the portal.
+    client.status.store(403, Ordering::SeqCst);
+    *client.error_code.lock().unwrap() = Some("telemetry_disabled".into());
+    clock.set(3_000);
+    pipe.flush();
+    assert_eq!(client.bodies.lock().unwrap().len(), 1, "one attempt");
+    assert_eq!(
+        pipe.stats().last_error.as_deref(),
+        Some("ingest responded 403 (telemetry_disabled)")
+    );
+    assert_eq!(
+        queued(&pipe),
+        0,
+        "the refused batch and the one queued behind it are gone"
+    );
+    assert_eq!(pipe.stats().dropped, 2);
+
+    // Collected while off: discarded at the next flush, never queued or sent.
+    emit_failed(&pipe, 3);
+    clock.set(4_000);
+    pipe.flush();
+    assert_eq!(queued(&pipe), 0);
+    assert_eq!(pipe.stats().dropped, 3);
+    assert_eq!(
+        client.bodies.lock().unwrap().len(),
+        1,
+        "nothing more is sent"
+    );
+    pipe.stop(FINAL_FLUSH_BUDGET);
+
+    // The toggle comes back on and the app relaunches: a fresh pipe over the
+    // same protocol-state store has nothing from the off period to resend.
+    client.status.store(202, Ordering::SeqCst);
+    *client.error_code.lock().unwrap() = None;
+    let again = inline_pipe(&test_config(), &clock, client.clone(), sealed(&storage));
+    // Without this, a fresh pipe that failed to adopt the store would run in
+    // memory, and "no backfill" below would hold with nothing proven.
+    assert!(
+        again.is_durable(),
+        "the fresh pipe adopted the durable queue"
+    );
+    again.flush();
+    assert_eq!(client.bodies.lock().unwrap().len(), 1, "no backfill");
+    emit_failed(&again, 4);
+    again.flush();
+    let bodies = client.bodies.lock().unwrap();
+    assert_eq!(bodies.len(), 2, "and new events flow again");
+    assert!(
+        bodies[1].contains("protocol.message.failed"),
+        "{}",
+        bodies[1]
+    );
+}
+
+/// A queue adopted after the refusal is emptied as it is adopted. The engine
+/// starts every pipe in memory and hands the protocol-state store over once
+/// `initialize_mls` has it, so a flush can reach the ingest first. The halted
+/// uploader never reads the queue again, so without the sweep at adoption the
+/// batches an earlier launch left on disk would stay there until the next
+/// `enable_telemetry` resent them.
+#[test]
+fn a_queue_adopted_after_a_telemetry_disabled_refusal_is_dropped_too() {
+    let clock = FakeClock::at(1_000);
+    let storage = Arc::new(MemoryStorage::default());
+    let client = CapturingClient::accepting();
+
+    // An earlier launch, offline throughout, leaves one batch on disk.
+    client.status.store(0, Ordering::SeqCst);
+    let earlier = inline_pipe(&test_config(), &clock, client.clone(), sealed(&storage));
+    emit_failed(&earlier, 1);
+    earlier.flush();
+    earlier.stop(FINAL_FLUSH_BUDGET);
+    assert!(client.bodies.lock().unwrap().is_empty());
+
+    // This launch starts in memory, as `enable_telemetry` does, and reaches
+    // the ingest before storage attaches. The toggle is off.
+    client.status.store(403, Ordering::SeqCst);
+    *client.error_code.lock().unwrap() = Some("telemetry_disabled".into());
+    let pipe = inline_pipe(&test_config(), &clock, client.clone(), Backend::Memory);
+    emit_failed(&pipe, 2);
+    pipe.flush();
+    assert_eq!(client.bodies.lock().unwrap().len(), 1, "one attempt");
+    assert_eq!(pipe.stats().dropped, 1);
+
+    // `initialize_mls` attaches the store, and the pipe adopts the queue the
+    // earlier launch left there.
+    let Backend::Sealed {
+        storage: backing,
+        cipher,
+    } = sealed(&storage)
+    else {
+        unreachable!("sealed() builds a sealed backend");
+    };
+    pipe.attach_storage(backing, cipher);
+    // Without this, an adoption that never happened would leave the store
+    // empty too, and the assertions below would hold with nothing proven.
+    assert!(pipe.is_durable(), "the queue on disk was adopted");
+    let queued = crate::telemetry::pipe::lock(&pipe.shared().store).len();
+    assert_eq!(queued, 0, "and emptied as it was adopted");
+    assert_eq!(
+        pipe.stats().dropped,
+        2,
+        "the adopted batch counts as dropped"
+    );
+    pipe.stop(FINAL_FLUSH_BUDGET);
+
+    // The toggle comes back on: the next launch has nothing to resend.
+    client.status.store(202, Ordering::SeqCst);
+    *client.error_code.lock().unwrap() = None;
+    let again = inline_pipe(&test_config(), &clock, client.clone(), sealed(&storage));
+    assert!(
+        again.is_durable(),
+        "the fresh pipe adopted the durable queue"
+    );
+    again.flush();
+    assert_eq!(client.bodies.lock().unwrap().len(), 1, "no backfill");
+}
+
+/// A refusal on the final flush drops the queue the same way. `stop` runs the
+/// same cycle, and a batch it left on disk would be resent by the next
+/// `enable_telemetry`.
+#[test]
+fn a_telemetry_disabled_refusal_on_the_final_flush_drops_the_queue_too() {
+    let clock = FakeClock::at(1_000);
+    let storage = Arc::new(MemoryStorage::default());
+    let client = CapturingClient::accepting();
+    client.status.store(403, Ordering::SeqCst);
+    *client.error_code.lock().unwrap() = Some("telemetry_disabled".into());
+    let pipe = inline_pipe(&test_config(), &clock, client.clone(), sealed(&storage));
+    emit_failed(&pipe, 1);
+    pipe.stop(FINAL_FLUSH_BUDGET);
+    assert_eq!(
+        client.bodies.lock().unwrap().len(),
+        1,
+        "the final flush reached the ingest"
+    );
+    assert_eq!(pipe.stats().dropped, 1);
+
+    client.status.store(202, Ordering::SeqCst);
+    *client.error_code.lock().unwrap() = None;
+    let again = inline_pipe(&test_config(), &clock, client.clone(), sealed(&storage));
+    assert!(
+        again.is_durable(),
+        "the fresh pipe adopted the durable queue"
+    );
+    again.flush();
+    assert_eq!(client.bodies.lock().unwrap().len(), 1, "no backfill");
+}