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

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

The uploader treated every 401 and 403 alike: keep the refused batch,
halt until the next enable_telemetry, keep queuing. Right for a bad key,
which the developer fixes and re-enables through. Wrong for the
application's telemetry toggle in the developer portal: once it came back
on, a relaunch resent up to six days of the off period, the backfill the
toggle promises not to make.

The ingest now answers that refusal with the error code
telemetry_disabled. On a 403 carrying it the uploader drops the refused
batch and everything queued behind it, halts, and reports it to the pipe,
which discards the ring at every later cycle instead of cutting batches
until the next enable_telemetry. The flag is read on the cycle, never on
the emit path. Every other 401 or 403, and a 403 from an ingest that
predates the code, keeps the queue as before.

* test(protocol): make the toggle scenario prove its store was adopted

The telemetry_disabled scenario ends by building a fresh pipe over
the same sealed store and asserting that nothing is resent. That is
a negative assertion, and it passes just as happily when the fresh
pipe never adopted the store and quietly ran in memory. An empty
in-memory queue has nothing to backfill either.

So the day someone breaks release-then-adopt, this test keeps
saying "no backfill" for entirely the wrong reason.

Assert the fresh pipe is durable before trusting the count. The
bad-key sibling never needed this, because its assertion is
positive: the refused batch shows up again, which it can only do
from the store.

* refactor(protocol): match the ingest error code without cloning it

Classifying a 403 parsed the body, cloned the error code into a
fresh String, and compared that against a constant. The String
existed only to be compared and thrown away.

Ask the question directly instead: does the body name this code.
Same parse, and junk or a missing field still lands on the ordinary
halt. It only runs on a 403, so this is tidiness, not a hot-path
win, and I won't pretend otherwise.

* docs: stop promising the telemetry queue waits for re-enable

The telemetry_disabled change updated the uploader's answer table
and added an errors row, and left three other places describing the
old world. The stats table listed four sources for dropped, the
privacy page named three things that clear the queue, and the
threat model said persisted batches are kept for a later enable.
All three are wrong for an app whose portal toggle is off.

Say it in each place. dropped also counts what was queued or
collected after the ingest reported the toggle off, and the queue
is cleared when that happens rather than held for the toggle coming
back. The privacy page is the one integrators quote to their users,
so it is not the place to be out of date.

While at it, document the one sharp edge. The ingest caches a
rejection for 30 seconds, so a launch just after the toggle is
turned back on can still be refused, and that process then keeps
nothing until its next enableTelemetry. Before this change the next
launch would have backfilled it. Now it doesn't, which is the whole
point, but a developer checking the dashboard right after flipping
the switch deserves to know why that launch went missing.

* fix(protocol): drop a queue adopted after the telemetry_disabled halt

The telemetry_disabled halt promised that nothing from the off
period is left on disk for a later pipe to resend. It kept that
promise only for a pipe that was already durable when the refusal
arrived.

It turns out the engine never starts one that way.
enable_telemetry starts every pipe in memory and hands the
protocol-state store over for the worker to adopt, which on the
mobile bindings means after initialize_mls. A flush that reaches
the ingest first takes the halt with an empty in-memory queue.
Adoption then loads whatever an earlier launch persisted, and
nothing touches it again: the cycle has stopped cutting batches,
and the halted uploader returns before it reads the queue. A
re-enable that waits for the old pipe to let go of the queue gets
there the same way.

So the next enable_telemetry, with the toggle back on, sends the
off period. Which is exactly the backfill this change exists to
prevent.

Sweep the store at adoption when the flag is set. Adoption is the
one way a batch can still enter the store after the refusal, so the
check belongs there rather than in the uploader, which both a
battery-deferred cycle and the worker's adoption-only wake skip.
The swept batches count as dropped, in events, like the rest of the
halt.

The new scenario runs three launches over one store and fails with
the sweep removed.

* test(protocol): pin the telemetry_disabled drop on the final flush

stop() runs the same cycle as any other flush, so a toggle-off
refusal on the final flush already drops the queue. Nothing pinned
it, though, and a batch left on disk there would be resent by the
next enable_telemetry like any other.

Pin it: a sealed pipe whose only send is the final flush, then a
fresh pipe over the same store with nothing to resend.

* docs(bindings): list the toggle-off discard among telemetry drops

The telemetry_disabled change taught the Rust doc and
docs/telemetry.md that dropped also counts what the pipe discards
once the ingest reports the portal toggle off. The same sentence is
hand-copied into four more places: the UDL, the TypeScript type and
method docs, and the Python wrapper. All four still listed the old
sources.

Say it in all four. The UDL edit is a plain comment, which
uniffi-bindgen ignores: regenerating all three bindings from it
changes no generated file, so there is nothing to regenerate here.

Author: bahdotsh

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");
+}