statsig_rust/
spec_store.rs

1use crate::data_store_interface::{get_data_adapter_dcs_key, DataStoreTrait};
2use crate::global_configs::GlobalConfigs;
3use crate::id_lists_adapter::{IdList, IdListsUpdateListener};
4use crate::observability::observability_client_adapter::{MetricType, ObservabilityEvent};
5use crate::observability::ops_stats::{OpsStatsForInstance, OPS_STATS};
6use crate::observability::sdk_errors_observer::ErrorBoundaryEvent;
7use crate::spec_types::{SpecsResponse, SpecsResponseFull};
8use crate::{
9    log_d, log_e, log_error_to_statsig_and_console, SpecsInfo, SpecsSource, SpecsUpdate,
10    SpecsUpdateListener, StatsigErr, StatsigRuntime,
11};
12use chrono::Utc;
13use serde::Serialize;
14use std::collections::HashMap;
15use std::sync::{Arc, RwLock};
16
17#[derive(Clone, Serialize)]
18pub struct SpecStoreData {
19    pub source: SpecsSource,
20    pub time_received_at: Option<u64>,
21    pub values: SpecsResponseFull,
22
23    pub id_lists: HashMap<String, IdList>,
24}
25
26const TAG: &str = stringify!(SpecStore);
27
28pub struct SpecStore {
29    pub hashed_sdk_key: String,
30    pub data: Arc<RwLock<SpecStoreData>>,
31    pub data_store: Option<Arc<dyn DataStoreTrait>>,
32    pub statsig_runtime: Option<Arc<StatsigRuntime>>,
33    ops_stats: Arc<OpsStatsForInstance>,
34    global_configs: Arc<GlobalConfigs>,
35}
36
37impl SpecsUpdateListener for SpecStore {
38    fn did_receive_specs_update(&self, update: SpecsUpdate) -> Result<(), StatsigErr> {
39        self.set_values(update)
40    }
41
42    fn get_current_specs_info(&self) -> SpecsInfo {
43        match self.data.read() {
44            Ok(data) => SpecsInfo {
45                lcut: Some(data.values.time),
46                checksum: data.values.checksum.clone(),
47                source: data.source.clone(),
48            },
49            Err(e) => {
50                log_e!(TAG, "Failed to acquire read lock: {}", e);
51                SpecsInfo {
52                    lcut: None,
53                    checksum: None,
54                    source: SpecsSource::Error,
55                }
56            }
57        }
58    }
59}
60
61impl IdListsUpdateListener for SpecStore {
62    fn get_current_id_list_metadata(
63        &self,
64    ) -> HashMap<String, crate::id_lists_adapter::IdListMetadata> {
65        match self.data.read() {
66            Ok(data) => data
67                .id_lists
68                .iter()
69                .map(|(key, list)| (key.clone(), list.metadata.clone()))
70                .collect(),
71            Err(e) => {
72                log_e!(TAG, "Failed to acquire read lock: {}", e);
73                HashMap::new()
74            }
75        }
76    }
77
78    fn did_receive_id_list_updates(
79        &self,
80        updates: HashMap<String, crate::id_lists_adapter::IdListUpdate>,
81    ) {
82        let mut data = match self.data.write() {
83            Ok(data) => data,
84            Err(e) => {
85                log_e!(TAG, "Failed to acquire write lock: {}", e);
86                return;
87            }
88        };
89
90        // delete any id_lists that are not in the updates
91        data.id_lists.retain(|name, _| updates.contains_key(name));
92
93        for (list_name, update) in updates {
94            if let Some(entry) = data.id_lists.get_mut(&list_name) {
95                // update existing
96                entry.apply_update(&update);
97            } else {
98                // add new
99                let mut list = IdList::new(update.new_metadata.clone());
100                list.apply_update(&update);
101                data.id_lists.insert(list_name, list);
102            }
103        }
104    }
105}
106
107impl Default for SpecStore {
108    fn default() -> Self {
109        let sdk_key = String::new();
110        Self::new(&sdk_key, sdk_key.to_string(), None, None)
111    }
112}
113
114impl SpecStore {
115    #[must_use]
116    pub fn new(
117        sdk_key: &str,
118        hashed_sdk_key: String,
119        data_store: Option<Arc<dyn DataStoreTrait>>,
120        statsig_runtime: Option<Arc<StatsigRuntime>>,
121    ) -> SpecStore {
122        SpecStore {
123            hashed_sdk_key,
124            data: Arc::new(RwLock::new(SpecStoreData {
125                values: SpecsResponseFull::blank(),
126                time_received_at: None,
127                source: SpecsSource::Uninitialized,
128                id_lists: HashMap::new(),
129            })),
130            data_store,
131            statsig_runtime,
132            ops_stats: OPS_STATS.get_for_instance(sdk_key),
133            global_configs: GlobalConfigs::get_instance(sdk_key),
134        }
135    }
136
137    pub fn set_source(&self, source: SpecsSource) {
138        if let Ok(mut mut_values) = self.data.write() {
139            mut_values.source = source;
140            log_d!(TAG, "SpecStore - Source Changed ({:?})", mut_values.source);
141        }
142    }
143
144    pub fn set_values(&self, values: SpecsUpdate) -> Result<(), StatsigErr> {
145        let parsed = serde_json::from_str::<SpecsResponse>(&values.data);
146        let dcs = match parsed {
147            Ok(SpecsResponse::Full(full)) => {
148                if !full.has_updates {
149                    self.log_no_update(values.source);
150                    return Ok(());
151                }
152
153                log_d!(
154                    TAG,
155                    "SpecStore Full Update: {} - [gates({}), configs({}), layers({})]",
156                    full.time,
157                    full.feature_gates.len(),
158                    full.dynamic_configs.len(),
159                    full.layer_configs.len(),
160                );
161
162                full
163            }
164            Ok(SpecsResponse::NoUpdates(no_updates)) => {
165                if !no_updates.has_updates {
166                    self.log_no_update(values.source);
167                    return Ok(());
168                }
169                log_error_to_statsig_and_console!(
170                    self.ops_stats,
171                    TAG,
172                    "Empty response with has_updates = true {:?}",
173                    values.source
174                );
175                return Err(StatsigErr::JsonParseError(
176                    "SpecsResponse".to_owned(),
177                    "Parse failure. 'has_update' is true, but failed to deserialize to response format 'dcs-v2'".to_owned(),
178                ));
179            }
180            Err(e) => {
181                // todo: Handle bad parsing
182                log_error_to_statsig_and_console!(
183                    self.ops_stats,
184                    TAG,
185                    "{:?}, {:?}",
186                    e,
187                    values.source
188                );
189                return Err(StatsigErr::JsonParseError(
190                    "config_spec".to_string(),
191                    e.to_string(),
192                ));
193            }
194        };
195
196        if let Some(diagnostics) = &dcs.diagnostics {
197            self.global_configs
198                .set_diagnostics_sampling_rates(diagnostics.clone());
199        }
200        if let Some(sdk_configs) = &dcs.sdk_configs {
201            self.global_configs.set_sdk_configs(sdk_configs.clone());
202        }
203
204        if let Ok(mut mut_values) = self.data.write() {
205            let cached_time_is_newer =
206                mut_values.values.time > 0 && mut_values.values.time > dcs.time;
207            let checksums_match =
208                mut_values
209                    .values
210                    .checksum
211                    .as_ref()
212                    .is_some_and(|cached_checksum| {
213                        dcs.checksum
214                            .as_ref()
215                            .is_some_and(|new_checksum| cached_checksum == new_checksum)
216                    });
217
218            if cached_time_is_newer || checksums_match {
219                log_d!(
220                    TAG,
221                    "SpecStore - Received values for [time: {}, checksum: {}], but currently has values for [time: {}, checksum: {}]. Ignoring values.",
222                    dcs.time,
223                    dcs.checksum.unwrap_or(String::new()),
224                    mut_values.values.time,
225                    mut_values.values.checksum.clone().unwrap_or(String::new()),
226                );
227                return Ok(());
228            }
229            let curr_time = Some(Utc::now().timestamp_millis() as u64);
230            let prev_source = mut_values.source.clone();
231            mut_values.values = *dcs;
232            mut_values.time_received_at = curr_time;
233            mut_values.source = values.source.clone();
234            if self.data_store.is_some() && mut_values.source == SpecsSource::Network {
235                if let Some(data_store) = self.data_store.clone() {
236                    let hashed_key = self.hashed_sdk_key.clone();
237                    self.statsig_runtime.clone().map(|rt| {
238                        let copy = curr_time;
239                        rt.spawn("update data adapter", move |_| async move {
240                            let _ = data_store
241                                .set(&get_data_adapter_dcs_key(&hashed_key), &values.data, copy)
242                                .await;
243                        })
244                    });
245                }
246            }
247            self.log_processing_config(
248                mut_values.values.time,
249                mut_values.source.clone(),
250                prev_source,
251            );
252        }
253
254        Ok(())
255    }
256
257    fn log_processing_config(&self, lcut: u64, source: SpecsSource, prev_source: SpecsSource) {
258        let delay = Utc::now().timestamp_millis() as u64 - lcut;
259        log_d!(TAG, "SpecStore - Updated ({:?})", source);
260
261        if prev_source != SpecsSource::Uninitialized && prev_source != SpecsSource::Loading {
262            self.ops_stats.log(ObservabilityEvent::new_event(
263                MetricType::Dist,
264                "config_propogation_diff".to_string(),
265                delay as f64,
266                Some(HashMap::from([("source".to_string(), source.to_string())])),
267            ));
268        }
269    }
270
271    fn log_no_update(&self, source: SpecsSource) {
272        log_d!(TAG, "SpecStore - No Updates");
273        self.ops_stats.log(ObservabilityEvent::new_event(
274            MetricType::Increment,
275            "config_no_update".to_string(),
276            1.0,
277            Some(HashMap::from([("source".to_string(), source.to_string())])),
278        ));
279    }
280}
281
282impl SpecsResponseFull {
283    fn blank() -> Self {
284        SpecsResponseFull {
285            feature_gates: Default::default(),
286            dynamic_configs: Default::default(),
287            layer_configs: Default::default(),
288            condition_map: Default::default(),
289            experiment_to_layer: Default::default(),
290            has_updates: true,
291            time: 0,
292            checksum: None,
293            default_environment: None,
294            app_id: None,
295            sdk_keys_to_app_ids: None,
296            hashed_sdk_keys_to_app_ids: None,
297            diagnostics: None,
298            param_stores: None,
299            sdk_configs: None,
300            cmab_configs: None,
301            overrides: None,
302            override_rules: None,
303        }
304    }
305}
statsig_rust/spec_store.rs

statsig_rust/
spec_store.rs
