rovenue/exposure/
mod.rs

1use std::collections::HashSet;
2use std::sync::{Arc, Mutex};
3
4use serde::Serialize;
5
6use crate::cache::ExposureRepo;
7use crate::identity::IdentityManager;
8use crate::remote_config::types::ExperimentAssignment;
9use crate::time::Clock;
10use crate::transport::api::ApiEnvelope;
11use crate::transport::http_client::HttpClient;
12use crate::transport::types::HttpPostRequest;
13
14/// Backend `error.code` (409) for an exposure on an experiment that is not
15/// RUNNING (`packages/shared/src/error-catalog.ts`).
16pub const EXPERIMENT_NOT_RUNNING_CODE: &str = "EXPERIMENT_NOT_RUNNING";
17
18#[derive(Serialize)]
19struct ExposeBody<'a> {
20    #[serde(rename = "variantId")]
21    variant_id: &'a str,
22    #[serde(rename = "subscriberId")]
23    subscriber_id: &'a str,
24}
25
26/// Best-effort, deduped experiment-exposure reporter. Reads stay non-blocking:
27/// the POST runs on a spawned thread and only marks the persistent dedup ledger
28/// on success, so a failed report is retried on the next read.
29///
30/// `pending` is an in-memory set of `"scope\0experiment_id\0variant_id"` keys
31/// that are currently in-flight or already persisted. It prevents spawning
32/// duplicate threads for rapid back-to-back reads before the first thread's
33/// HTTP round-trip has completed.
34pub struct ExposureTracker {
35    repo: ExposureRepo,
36    http: Option<Arc<HttpClient>>,
37    clock: Option<Arc<dyn Clock>>,
38    identity: Arc<IdentityManager>,
39    /// In-memory coalesce guard: tracks keys with an in-flight POST so rapid
40    /// repeated reads don't fan-out N threads for the same (scope, exp, var).
41    pending: Mutex<HashSet<String>>,
42}
43
44impl ExposureTracker {
45    pub fn new(
46        repo: ExposureRepo,
47        http: Option<Arc<HttpClient>>,
48        clock: Option<Arc<dyn Clock>>,
49        identity: Arc<IdentityManager>,
50    ) -> Arc<Self> {
51        Arc::new(Self {
52            repo,
53            http,
54            clock,
55            identity,
56            pending: Mutex::new(HashSet::new()),
57        })
58    }
59
60    /// Fire-and-forget exposure report for `assignment`. If http or clock are
61    /// absent (unconfigured SDK), does nothing. Deduplicates via the persistent
62    /// `ExposureRepo` ledger (checked synchronously before spawning) and via an
63    /// in-memory `pending` set (blocks duplicate spawns for the same triple
64    /// while the first thread's POST is in flight).
65    pub fn maybe_track(self: &Arc<Self>, assignment: &ExperimentAssignment) {
66        let (http, clock) = match (self.http.as_ref(), self.clock.as_ref()) {
67            (Some(h), Some(c)) => (Arc::clone(h), Arc::clone(c)),
68            _ => return,
69        };
70
71        // `scope` namespaces the local dedup ledger/coalesce key; `wire_id` is
72        // the stable rovenue_id sent to the server (resolved as a rovenueId).
73        let scope = self.identity.current_user_scope();
74        let wire_id = self.identity.rovenue_id();
75        let experiment_id = assignment.experiment_id.clone();
76        let variant_id = assignment.variant_id.clone();
77
78        // Build the coalesce key: NUL-separated to avoid collisions.
79        let coalesce_key = format!("{scope}\0{experiment_id}\0{variant_id}");
80
81        // Fast path: check in-memory pending set first (avoids SQLite read
82        // for the common case of back-to-back reads after the first spawn).
83        {
84            let mut pending = match self.pending.lock() {
85                Ok(g) => g,
86                Err(_) => return,
87            };
88            if pending.contains(&coalesce_key) {
89                return;
90            }
91            // Also check the persistent ledger while we hold the lock so there
92            // is no window between the check and the insert.
93            if self
94                .repo
95                .is_exposed(&scope, &experiment_id, &variant_id)
96                .unwrap_or(false)
97            {
98                // Already persisted — insert into pending so future reads skip
99                // even the SQLite check.
100                pending.insert(coalesce_key);
101                return;
102            }
103            // Reserve the slot: mark in-flight before releasing the lock.
104            pending.insert(coalesce_key.clone());
105        }
106
107        let this = Arc::clone(self);
108        std::thread::spawn(move || {
109            let path = format!("/v1/experiments/{experiment_id}/expose");
110            let body = ExposeBody {
111                variant_id: &variant_id,
112                subscriber_id: &wire_id,
113            };
114            // Use serde_json::Value as the response data type so that a 202 with
115            // an empty or non-JSON body doesn't cause a deserialization failure.
116            let res = http.post_json::<ExposeBody, ApiEnvelope<serde_json::Value>>(
117                HttpPostRequest::new(&path).user_scope(&wire_id),
118                &body,
119            );
120            // A 409 EXPERIMENT_NOT_RUNNING is the server's final word on this
121            // exposure (draft/paused/completed experiment): recorded as sent,
122            // like a success, so every later read does not re-POST it.
123            let settled = match &res {
124                Ok(_) => true,
125                Err(e) => e.server_code.as_deref() == Some(EXPERIMENT_NOT_RUNNING_CODE),
126            };
127            if settled {
128                let _ = this
129                    .repo
130                    .mark(&scope, &experiment_id, &variant_id, clock.now_unix_ms());
131                // Leave the key in `pending` — it acts as a permanent in-memory
132                // sentinel so subsequent reads never even hit SQLite.
133            } else {
134                // On failure, remove the key from pending so the next read
135                // retries the POST (dedup set only marks success).
136                if let Ok(mut guard) = this.pending.lock() {
137                    guard.remove(&coalesce_key);
138                }
139            }
140        });
141    }
142}
rovenue/exposure/mod.rs

rovenue/exposure/
mod.rs
