feat: add apply dedup and flush delivery telemetry

confidence-cloudflare-resolver/src/lib.rs

@@ -1,7 +1,7 @@
 mod materialization;
 
 use confidence_resolver::{
-    apply_dedup::ApplyDedup,
+    apply_dedup::{ApplyDedup, ApplyDedupSnapshot},
     assign_logger, flag_logger,
     proto::{confidence, google::Struct},
     resolve_logger,
@@ -51,6 +51,22 @@ thread_local! {
     static FLAG_LOG: RefCell<Option<WriteFlagLogsRequest>> = const { RefCell::new(None) };
     static APPLY_DEDUP: RefCell<ApplyDedup> = RefCell::new(ApplyDedup::new(120, 100_000));
     static APPLY_DEDUP_ENABLED: Cell<bool> = const { Cell::new(false) };
+    static LAST_DEDUP_SNAPSHOT: RefCell<ApplyDedupSnapshot> =
+        RefCell::new(ApplyDedupSnapshot::default());
+}
+
+fn dedup_telemetry_delta() -> Option<confidence::flags::resolver::v1::telemetry_data::ApplyDedupTelemetry> {
+    if !APPLY_DEDUP_ENABLED.with(|c| c.get()) {
+        return None;
+    }
+    let current = APPLY_DEDUP.with(|d| d.borrow().telemetry_snapshot());
+    let prev = LAST_DEDUP_SNAPSHOT.with(|s| std::mem::replace(&mut *s.borrow_mut(), current.clone()));
+    let delta = current.to_proto_delta(&prev);
+    if delta.applies_total > 0 || current.map_size > 0 {
+        Some(delta)
+    } else {
+        None
+    }
 }
 
 /// Queues one request's flag log and sweeps the apply-dedup map. Called via
@@ -497,6 +513,7 @@ pub async fn main(req: Request, env: Env, ctx: Context) -> Result<Response> {
 
                         let mut td = telemetry::build_request_telemetry(elapsed_us, &reasons);
                         td.sdk = Some(sdk_info());
+                        td.apply_dedup = dedup_telemetry_delta();
                         log.telemetry_data = Some(td);
                         event_ctx.wait_until(queue_flag_log(log));
 
@@ -540,9 +557,10 @@ pub async fn main(req: Request, env: Env, ctx: Context) -> Result<Response> {
                                 Response::error(msg, 500)?.with_cors_headers(&allowed_origin)
                             }
                         };
-                        // Unlike resolve there is no telemetry to attach, so
-                        // skip queueing when the apply logged nothing (an
-                        // errored apply).
+                        if let Some(dedup_delta) = dedup_telemetry_delta() {
+                            let td = log.telemetry_data.get_or_insert_with(Default::default);
+                            td.apply_dedup = Some(dedup_delta);
+                        }
                         if log != WriteFlagLogsRequest::default() {
                             event_ctx.wait_until(queue_flag_log(log));
                         }
@@ -704,11 +722,6 @@ async fn consume_flag_logs(
 
         let req = flag_logger::aggregate_batch(logs);
 
-        // Accumulate telemetry deltas into KV-backed cumulative snapshot for /metrics.
-        if let Ok(kv) = env.kv("CONFIDENCE_METRICS_KV") {
-            update_prometheus_kv(&kv, &req).await;
-        }
-
         let client_secret = CONFIDENCE_CLIENT_SECRET.get().unwrap().as_str();
         let account_id = CDN_STATE_REQUEST.account_id.as_str();
         let destinations = &*LOG_DESTINATIONS;
@@ -719,13 +732,15 @@ async fn consume_flag_logs(
             (destinations[0], None)
         };
 
-        if let Err(reason) = deliver_flag_logs(client_secret, account_id, &req, primary).await {
+        let delivered = if let Err(reason) =
+            deliver_flag_logs(client_secret, account_id, &req, primary).await
+        {
             console_log!(
                 "flag log delivery to {:?} failed ({}), trying fallback",
                 primary,
                 reason
             );
-            let fallback_delivered = match fallback {
+            match fallback {
                 Some(fb) => match deliver_flag_logs(client_secret, account_id, &req, fb).await {
                     Ok(()) => true,
                     Err(fb_reason) => {
@@ -738,16 +753,19 @@ async fn consume_flag_logs(
                     }
                 },
                 None => false,
-            };
-            if !fallback_delivered {
-                // Returning Err makes Cloudflare Queues redeliver the batch,
-                // so a delivery outage doesn't silently drop logs. The
-                // telemetry KV update above may run again on redelivery —
-                // acceptable for metrics.
-                return Err(worker::Error::RustError(
-                    "flag log delivery failed on all destinations".to_string(),
-                ));
             }
+        } else {
+            true
+        };
+
+        if let Ok(kv) = env.kv("CONFIDENCE_METRICS_KV") {
+            update_kv_snapshot(&kv, req.telemetry_data.as_ref(), Some(delivered), None).await;
+        }
+
+        if !delivered {
+            return Err(worker::Error::RustError(
+                "flag log delivery failed on all destinations".to_string(),
+            ));
         }
     }
 
@@ -774,22 +792,51 @@ async fn deliver_flag_logs(
     }
 }
 
-/// Accumulate telemetry deltas from all isolates into a cumulative
-/// `TelemetrySnapshot` stored in KV, then write its Prometheus text
-/// representation for the /metrics endpoint.
+/// Single read-modify-write of the KV-backed cumulative telemetry snapshot.
 ///
-/// Note: concurrent queue consumer invocations can race on KV read-modify-write.
-/// Acceptable for metrics — at worst one batch's deltas are lost, not cumulative state.
-async fn update_prometheus_kv(kv: &kv::KvStore, req: &WriteFlagLogsRequest) {
+/// Merges flag-log telemetry deltas, flush delivery results, and event
+/// delivery results into one KV update — avoiding cross-queue races that
+/// would occur if flag-log and event consumers each did independent
+/// read-modify-write cycles on the same "snapshot" key.
+///
+/// Note: concurrent invocations of the *same* queue can still race on
+/// KV read-modify-write. Acceptable for metrics — at worst one batch's
+/// deltas are lost, not cumulative state.
+async fn update_kv_snapshot(
+    kv: &kv::KvStore,
+    telemetry_delta: Option<&confidence_resolver::proto::confidence::flags::resolver::v1::TelemetryData>,
+    flush_result: Option<bool>,
+    event_result: Option<(u64, bool)>,
+) {
     let mut cumulative = match kv.get("snapshot").text().await {
         Ok(Some(text)) => serde_json::from_str::<TelemetrySnapshot>(&text).unwrap_or_default(),
         _ => TelemetrySnapshot::default(),
     };
 
-    if let Some(td) = &req.telemetry_data {
+    if let Some(td) = telemetry_delta {
         cumulative.accumulate_delta(td);
     }
 
+    match flush_result {
+        Some(true) => {
+            cumulative.flush.succeeded = cumulative.flush.succeeded.wrapping_add(1);
+        }
+        Some(false) => {
+            cumulative.flush.failed = cumulative.flush.failed.wrapping_add(1);
+        }
+        None => {}
+    }
+
+    if let Some((event_count, succeeded)) = event_result {
+        if succeeded {
+            cumulative.events.published = cumulative.events.published.wrapping_add(event_count);
+            cumulative.events.batches_succeeded =
+                cumulative.events.batches_succeeded.wrapping_add(1);
+        } else {
+            cumulative.events.batches_failed = cumulative.events.batches_failed.wrapping_add(1);
+        }
+    }
+
     let prom_text = cumulative.to_prometheus(
         "cf-resolver",
         &confidence_resolver::telemetry::PrometheusConfig::default(),
@@ -970,7 +1017,7 @@ fn build_publish_events_request(
 
 async fn consume_events_queue(
     message_batch: MessageBatch<String>,
-    _env: Env,
+    env: Env,
 ) -> Result<()> {
     let messages = message_batch.messages()?;
     let raw: Vec<String> = messages.iter().map(|m| m.body().clone()).collect();
@@ -980,6 +1027,8 @@ async fn consume_events_queue(
         return Ok(());
     }
 
+    let event_count = all_events.len() as u64;
+
     let client_secret = CONFIDENCE_CLIENT_SECRET
         .get()
         .ok_or_else(|| worker::Error::RustError("client secret not configured".into()))?;
@@ -991,17 +1040,32 @@ async fn consume_events_queue(
         &now.as_string().unwrap_or_default(),
     );
 
-    let resp = send_events(&publish_request).await?;
-    if resp.status_code() >= 400 {
-        return Err(worker::Error::RustError(format!(
-            "events delivery failed: HTTP {}",
-            resp.status_code()
-        )));
+    let delivered = match send_events(&publish_request).await {
+        Ok(resp) if resp.status_code() < 400 => true,
+        Ok(resp) => {
+            console_log!("events delivery failed: HTTP {}", resp.status_code());
+            false
+        }
+        Err(e) => {
+            console_log!("events delivery error: {:?}", e);
+            false
+        }
+    };
+
+    if let Ok(kv) = env.kv("CONFIDENCE_METRICS_KV") {
+        update_kv_snapshot(&kv, None, None, Some((event_count, delivered))).await;
+    }
+
+    if !delivered {
+        return Err(worker::Error::RustError(
+            "events delivery failed".to_string(),
+        ));
     }
 
     Ok(())
 }
 
+
 async fn send_events(body: &serde_json::Value) -> Result<Response> {
     let mut init = RequestInit::new();
     let headers = Headers::new();

confidence-cloudflare-resolver/wrangler-test.toml

@@ -0,0 +1,32 @@
+name = "confidence-telemetry-test"
+main = "build/worker/shim.mjs"
+compatibility_date = "2025-01-28"
+
+[observability]
+enabled = true
+
+[[queues.consumers]]
+queue = "telemetry-test-flag-logs"
+max_batch_size = 100
+max_batch_timeout = 10
+
+[[queues.producers]]
+queue = "telemetry-test-flag-logs"
+binding = "flag_logs_queue"
+
+[[queues.consumers]]
+queue = "telemetry-test-events"
+max_batch_size = 100
+max_batch_timeout = 10
+
+[[queues.producers]]
+queue = "telemetry-test-events"
+binding = "events_queue"
+
+[[kv_namespaces]]
+binding = "CONFIDENCE_METRICS_KV"
+id = "05a77023635a4dbaa4744a20d6419a53"
+
+[vars]
+CONFIDENCE_CLIENT_SECRET = "mkjJruAATQWjeY7foFIWfVAcBWnci2YF"
+ENABLE_APPLY_DEDUP = "true"

confidence-resolver/protos/confidence/flags/resolver/v1/internal_api.proto

@@ -132,6 +132,34 @@ message TelemetryData {
 
   repeated ProviderInitRate provider_init_rate = 9;
 
+  ApplyDedupTelemetry apply_dedup = 10;
+
+  FlushTelemetry flush = 11;
+  EventsTelemetry events = 12;
+
+  message FlushTelemetry {
+    uint32 succeeded = 1;
+    uint32 failed = 2;
+  }
+
+  message EventsTelemetry {
+    uint32 published = 1;
+    uint32 batches_succeeded = 2;
+    uint32 batches_failed = 3;
+  }
+
+  message ApplyDedupTelemetry {
+    // Delta counters (since last flush):
+    uint32 applies_total = 1;
+    uint32 applies_deduped = 2;
+    uint32 apply_dedup_overflow = 3;
+    uint32 sweeps = 4;
+
+    // Gauges (point-in-time):
+    uint32 map_size = 5;
+    uint32 map_capacity = 6;
+  }
+
   message ProviderInitRate {
     uint32 count = 1;
     reserved 2; // status — tbd