feat(cloudflare): add opt-in apply-event deduplication (#542)

Co-authored-by: vahidlazio <vahidlazio@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: vahidlazio

confidence-cloudflare-resolver/deployer/README.md

@@ -68,6 +68,7 @@ The deployer automatically:
 | `ENABLE_STICKY_ASSIGNMENTS`          | Set to create a KV namespace and enable sticky assignments for experiments. Requires a [KV store](https://developers.cloudflare.com/kv/platform/pricing/) |
 | `MATERIALIZATION_TTL_SECONDS`        | TTL in seconds for sticky assignment KV entries. Omit for no expiration |
 | `FORCE_APPLY`                        | Defaults to `true`: every resolve is treated as `apply=true` and assignments are logged at resolve time. Set to `false` to respect the `apply` value sent by SDKs (deferred-apply flow via `flags:apply`) |
+| `ENABLE_APPLY_DEDUP`                 | Defaults to `false`: set to `true` to enable apply-event deduplication — repeated identical assignments within a 120s window are logged once |
 
 ### Extending Wrangler Configuration
 

confidence-cloudflare-resolver/deployer/script.sh

@@ -22,6 +22,7 @@ WRANGLER_DEPLOY_TAG=${WRANGLER_DEPLOY_TAG:=}
 WRANGLER_DEPLOY_MESSAGE=${WRANGLER_DEPLOY_MESSAGE:=}
 ENABLE_STICKY_ASSIGNMENTS=${ENABLE_STICKY_ASSIGNMENTS:=}
 FORCE_APPLY=${FORCE_APPLY:=}
+ENABLE_APPLY_DEDUP=${ENABLE_APPLY_DEDUP:=}
 INITIAL_WORKDIR="$(pwd)"
 
 # CDN base URL for fetching resolver state
@@ -550,16 +551,26 @@ if [ -n "$FORCE_APPLY" ]; then
     fi
 fi
 
+# Validate ENABLE_APPLY_DEDUP if provided (worker defaults to false when unset)
+if [ -n "$ENABLE_APPLY_DEDUP" ]; then
+    ENABLE_APPLY_DEDUP=$(printf '%s' "$ENABLE_APPLY_DEDUP" | tr '[:upper:]' '[:lower:]')
+    if [ "$ENABLE_APPLY_DEDUP" != "true" ] && [ "$ENABLE_APPLY_DEDUP" != "false" ]; then
+        echo "❌ ENABLE_APPLY_DEDUP must be \"true\" or \"false\", got: $ENABLE_APPLY_DEDUP" >&2
+        exit 1
+    fi
+fi
+
 # Update [vars] table with ALLOWED_ORIGIN, RESOLVER_STATE_ETAG and RESOLVER_VERSION, without duplicating the table
-if [ -n "$ALLOWED_ORIGIN_TOML" ] || [ -n "$ETAG_TOML" ] || [ -n "$DEPLOYER_VERSION" ] || [ -n "$CLIENT_SECRET_TOML" ] || [ -n "$FORCE_APPLY" ]; then
+if [ -n "$ALLOWED_ORIGIN_TOML" ] || [ -n "$ETAG_TOML" ] || [ -n "$DEPLOYER_VERSION" ] || [ -n "$CLIENT_SECRET_TOML" ] || [ -n "$FORCE_APPLY" ] || [ -n "$ENABLE_APPLY_DEDUP" ]; then
     # Remove any existing definitions to avoid duplicates
     sed -i.tmp '/^ALLOWED_ORIGIN *= *.*$/d' wrangler.toml || true
     sed -i.tmp '/^RESOLVER_STATE_ETAG *= *.*$/d' wrangler.toml || true
     sed -i.tmp '/^RESOLVER_VERSION *= *.*$/d' wrangler.toml || true
     sed -i.tmp '/^DEPLOYER_VERSION *= *.*$/d' wrangler.toml || true
     sed -i.tmp '/^CONFIDENCE_CLIENT_SECRET *= *.*$/d' wrangler.toml || true
     sed -i.tmp '/^FORCE_APPLY *= *.*$/d' wrangler.toml || true
-    awk -v allowed="${ALLOWED_ORIGIN_TOML}" -v etag="${ETAG_TOML}" -v version="${DEPLOYER_VERSION}" -v client_secret="${CLIENT_SECRET_TOML}" -v force_apply="${FORCE_APPLY}" '
+    sed -i.tmp '/^ENABLE_APPLY_DEDUP *= *.*$/d' wrangler.toml || true
+    awk -v allowed="${ALLOWED_ORIGIN_TOML}" -v etag="${ETAG_TOML}" -v version="${DEPLOYER_VERSION}" -v client_secret="${CLIENT_SECRET_TOML}" -v force_apply="${FORCE_APPLY}" -v enable_apply_dedup="${ENABLE_APPLY_DEDUP}" '
         BEGIN{inserted=0}
         {
             print $0
@@ -569,6 +580,7 @@ if [ -n "$ALLOWED_ORIGIN_TOML" ] || [ -n "$ETAG_TOML" ] || [ -n "$DEPLOYER_VERSI
                 if (version != "") print "DEPLOYER_VERSION = \"" version "\""
                 if (client_secret != "") print "CONFIDENCE_CLIENT_SECRET = \"" client_secret "\""
                 if (force_apply != "") print "FORCE_APPLY = \"" force_apply "\""
+                if (enable_apply_dedup != "") print "ENABLE_APPLY_DEDUP = \"" enable_apply_dedup "\""
                 inserted=1
             }
         }

confidence-cloudflare-resolver/src/lib.rs

@@ -1,6 +1,7 @@
 mod materialization;
 
 use confidence_resolver::{
+    apply_dedup::ApplyDedup,
     assign_logger, flag_logger,
     proto::{confidence, google::Struct},
     resolve_logger,
@@ -15,7 +16,7 @@ use bytes::Bytes;
 use prost::Message;
 use serde_json::from_slice;
 use serde_json::json;
-use std::cell::RefCell;
+use std::cell::{Cell, RefCell};
 use wasm_bindgen::JsCast;
 
 use confidence::flags::resolver::v1::{ApplyFlagsRequest, ApplyFlagsResponse, ResolveFlagsRequest};
@@ -48,11 +49,16 @@ thread_local! {
     // Side channel for the `Host` logging callbacks, which are static methods
     // with no way to reach their caller. Only ever `Some` inside `with_log`.
     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) };
 }
 
-/// Queues one request's flag log. Called via `Context::wait_until`, so it runs
-/// after the response has been returned.
+/// Queues one request's flag log and sweeps the apply-dedup map. Called via
+/// `Context::wait_until`, so both run after the response has been returned.
 async fn queue_flag_log(log: WriteFlagLogsRequest) {
+    if APPLY_DEDUP_ENABLED.with(|c| c.get()) {
+        APPLY_DEDUP.with(|d| d.borrow_mut().sweep((js_sys::Date::now() / 1000.0) as i64));
+    }
     match serde_json::to_string(&log) {
         Ok(json) => {
             if let Some(queue) = FLAGS_LOGS_QUEUE.get() {
@@ -164,14 +170,32 @@ impl Host for H {
         client: &Client,
         sdk: &Option<Sdk>,
     ) {
+        if !assigned_flags.is_empty() && APPLY_DEDUP_ENABLED.with(|c| c.get()) {
+            let now_seconds = (js_sys::Date::now() / 1000.0) as i64;
+            let result = APPLY_DEDUP.with(|dedup| {
+                dedup.borrow_mut().filter_duplicates(assigned_flags, now_seconds)
+            });
+            if result.is_empty() {
+                return;
+            }
+            if result.kept_count() < assigned_flags.len() {
+                let filtered = result.collect(assigned_flags);
+                FLAG_LOG.with(|f| {
+                    if let Some(req) = f.borrow_mut().as_mut() {
+                        req.flag_assigned
+                            .push(assign_logger::build_flag_assigned(
+                                resolve_id, &filtered, client, sdk,
+                            ));
+                    }
+                });
+                return;
+            }
+        }
         FLAG_LOG.with(|f| {
             if let Some(req) = f.borrow_mut().as_mut() {
                 req.flag_assigned
                     .push(assign_logger::build_flag_assigned(
-                        resolve_id,
-                        assigned_flags,
-                        client,
-                        sdk,
+                        resolve_id, assigned_flags, client, sdk,
                     ));
             }
         });
@@ -290,6 +314,12 @@ pub async fn main(req: Request, env: Env, ctx: Context) -> Result<Response> {
         .map(|var| var.to_string())
         .unwrap_or_default();
 
+    let enable_apply_dedup = env
+        .var("ENABLE_APPLY_DEDUP")
+        .map(|var| var.to_string().trim().eq_ignore_ascii_case("true"))
+        .unwrap_or(false);
+    APPLY_DEDUP_ENABLED.with(|c| c.set(enable_apply_dedup));
+
     if req.method() == Method::Options {
         return Response::ok("")?.with_cors_headers(&allowed_origin_env);
     }