statsig_rust/
spec_store.rs

1use arc_swap::ArcSwap;
2use chrono::Utc;
3use parking_lot::{Mutex, MutexGuard};
4use std::collections::HashMap;
5use std::sync::{Arc, OnceLock};
6use std::time::Instant;
7
8use crate::data_store_interface::{DataStoreCacheKeys, DataStoreTrait, RequestPath};
9use crate::evaluation::evaluator::SpecType;
10use crate::gcir::evaluation_plan::GcirEvaluationPlan;
11use crate::global_configs::GlobalConfigs;
12use crate::hashing::HashUtil;
13use crate::id_lists_adapter::{IdList, IdListsUpdateListener};
14use crate::interned_string::InternedString;
15use crate::interned_values::InternedStore;
16use crate::macros::LOCK_TIMEOUT;
17use crate::networking::ResponseData;
18use crate::observability::observability_client_adapter::{MetricType, ObservabilityEvent};
19use crate::observability::ops_stats::{OpsStatsForInstance, OPS_STATS};
20use crate::observability::sdk_errors_observer::ErrorBoundaryEvent;
21use crate::sdk_event_emitter::{SdkEvent, SdkEventEmitter};
22use crate::specs_response::proto_specs::{deserialize_protobuf_for_store, ProtobufUpdate};
23use crate::specs_response::spec_types::{SpecsResponseFull, SpecsResponseNoUpdates};
24use crate::specs_response::specs_hash_map::{track_spec_decodes, SpecDecodeStats};
25use crate::utils::{get_loggable_sdk_key, try_release_unused_heap_memory};
26use crate::{
27    log_d, log_e, log_error_to_statsig_and_console, log_w, SpecsFormat, SpecsInfo, SpecsSource,
28    SpecsUpdate, SpecsUpdateListener, StatsigErr, StatsigOptions, StatsigRuntime,
29};
30
31#[derive(Clone)]
32pub struct SpecStoreData {
33    pub source: SpecsSource,
34    pub source_api: Option<String>,
35    pub time_received_at: Option<u64>,
36    pub snapshot: Arc<SpecsResponseFull>,
37    pub id_lists: Arc<HashMap<String, IdList>>,
38    pub gcir_evaluation_plan: Arc<OnceLock<GcirEvaluationPlan>>,
39    sync_cursor: ConfigSyncCursor,
40    // Reused as the starting point for delta telemetry so unchanged maps can use native cloning.
41    spec_decode_stats: SpecDecodeStats,
42}
43
44#[derive(Clone, Default)]
45struct ConfigSyncCursor {
46    lcut: u64,
47    checksum: Option<String>,
48}
49
50impl SpecStoreData {
51    pub(crate) fn gcir_evaluation_plan(&self, hashing: &HashUtil) -> &GcirEvaluationPlan {
52        self.gcir_evaluation_plan
53            .get_or_init(|| GcirEvaluationPlan::new(self.snapshot.as_ref(), hashing))
54    }
55
56    pub(crate) fn lcut(&self) -> u64 {
57        self.sync_cursor.lcut
58    }
59
60    fn checksum(&self) -> Option<&str> {
61        self.sync_cursor.checksum.as_deref()
62    }
63}
64
65fn cursor_is_stale_or_duplicate(data: &SpecStoreData, lcut: u64, checksum: &str) -> bool {
66    lcut < data.lcut() || (lcut == data.lcut() && data.checksum() == Some(checksum))
67}
68
69fn proto_response_lcut(data: &ResponseData) -> Option<u64> {
70    data.get_header_ref("x-since-time")?.parse().ok()
71}
72
73fn proto_response_matches_cursor(data: &ResponseData, current: &SpecStoreData) -> bool {
74    let Some(lcut) = proto_response_lcut(data) else {
75        return false;
76    };
77    let Some(checksum) = data.get_header_ref("x-checksum") else {
78        return false;
79    };
80
81    lcut == current.lcut() && current.checksum() == Some(checksum)
82}
83
84const TAG: &str = stringify!(SpecStore);
85const RESPONSE_TYPE_TAG: &str = "response_type";
86const DELTAS_USED_HEADER: &str = "x-deltas-used";
87
88#[derive(Clone, Copy)]
89enum ConfigResponseType {
90    Delta,
91    Full,
92    NoUpdate,
93}
94
95impl ConfigResponseType {
96    fn from_applied_response_data(data: &ResponseData) -> Self {
97        if data.get_header_ref(DELTAS_USED_HEADER).is_some() {
98            Self::Delta
99        } else {
100            Self::Full
101        }
102    }
103
104    fn as_str(self) -> &'static str {
105        match self {
106            Self::Delta => "delta",
107            Self::Full => "full",
108            Self::NoUpdate => "no_update",
109        }
110    }
111}
112
113const CONFIG_PROTO_UPDATE_COUNT_METRIC: &str = "config_proto_update.count";
114const CONFIG_PROTO_UPDATE_LATENCY_METRIC: &str = "config_proto_update.latency";
115const CONFIG_PROTO_UPDATE_OUTCOME_TAG: &str = "outcome";
116const CONFIG_PROTO_UPDATE_CURSOR_ONLY: &str = "cursor_only";
117const CONFIG_PROTO_UPDATE_DUPLICATE: &str = "duplicate";
118const CONFIG_PROTO_UPDATE_MATERIALIZED: &str = "materialized";
119const INTERNED_MMAP_SPEC_DECODE_COUNT_METRIC: &str = "interned_mmap.spec_decode.count";
120
121pub struct SpecStore {
122    data: ArcSwap<SpecStoreData>,
123    update_lock: Mutex<()>,
124
125    data_store_keys: DataStoreCacheKeys,
126    data_store: Option<Arc<dyn DataStoreTrait>>,
127    statsig_runtime: Arc<StatsigRuntime>,
128    ops_stats: Arc<OpsStatsForInstance>,
129    global_configs: Arc<GlobalConfigs>,
130    event_emitter: Arc<SdkEventEmitter>,
131    loggable_sdk_key: String,
132}
133
134impl SpecStore {
135    #[must_use]
136    pub fn new(
137        sdk_key: &str,
138        data_store_key: String,
139        statsig_runtime: Arc<StatsigRuntime>,
140        event_emitter: Arc<SdkEventEmitter>,
141        options: Option<&StatsigOptions>,
142    ) -> SpecStore {
143        let mut data_store = None;
144        if let Some(options) = options {
145            data_store = options.data_store.clone();
146        }
147
148        let sdk_instance_id = options
149            .map(|opts| opts.get_sdk_instance_id(sdk_key))
150            .unwrap_or(sdk_key);
151
152        SpecStore {
153            data_store_keys: DataStoreCacheKeys::from_selected_key(&data_store_key),
154            data: ArcSwap::from_pointee(SpecStoreData {
155                snapshot: Arc::new(SpecsResponseFull::default()),
156                time_received_at: None,
157                source: SpecsSource::Uninitialized,
158                source_api: None,
159                id_lists: Arc::new(HashMap::new()),
160                gcir_evaluation_plan: Arc::new(OnceLock::new()),
161                sync_cursor: ConfigSyncCursor::default(),
162                spec_decode_stats: SpecDecodeStats::default(),
163            }),
164            update_lock: Mutex::new(()),
165            event_emitter,
166            data_store,
167            statsig_runtime,
168            ops_stats: OPS_STATS.get_for_instance(sdk_instance_id),
169            global_configs: GlobalConfigs::get_instance(sdk_instance_id),
170            loggable_sdk_key: get_loggable_sdk_key(sdk_key),
171        }
172    }
173
174    pub fn set_source(&self, source: SpecsSource) {
175        {
176            let Some(_update_guard) = self.try_lock_for_update("set_source") else {
177                return;
178            };
179
180            let mut next_data = self.load_data().as_ref().clone();
181            next_data.source = source.clone();
182            self.publish_data(next_data);
183        }
184
185        log_d!(TAG, "Source Changed ({:?})", source);
186    }
187
188    pub fn get_current_values(&self) -> Option<SpecsResponseFull> {
189        let data = self.load_data();
190        let json = serde_json::to_string(data.snapshot.as_ref()).ok()?;
191        let mut values = serde_json::from_str::<SpecsResponseFull>(&json).ok()?;
192        values.time = data.lcut();
193        values.checksum = data.sync_cursor.checksum.clone();
194        Some(values)
195    }
196
197    pub fn get_fields_used_for_entity(
198        &self,
199        entity_name: &str,
200        entity_type: SpecType,
201    ) -> Vec<String> {
202        let data = self.load_data();
203        let entities = match entity_type {
204            SpecType::Gate => &data.snapshot.feature_gates,
205            SpecType::DynamicConfig | SpecType::Experiment => &data.snapshot.dynamic_configs,
206            SpecType::Layer => &data.snapshot.layer_configs,
207            SpecType::ParameterStore => return vec![],
208        };
209
210        let entity_name = InternedString::from_str_ref(entity_name);
211        let entity = entities.get(&entity_name);
212
213        entity
214            .map(|entity| entity.view().fields_used())
215            .unwrap_or_default()
216    }
217
218    pub fn unperformant_keys_entity_filter(
219        &self,
220        top_level_key: &str,
221        entity_type: &str,
222    ) -> Vec<String> {
223        let data = self.load_data();
224        if top_level_key == "param_stores" {
225            match &data.snapshot.param_stores {
226                Some(param_stores) => {
227                    return param_stores
228                        .keys()
229                        .map(|k| k.unperformant_to_string())
230                        .collect()
231                }
232                None => return vec![],
233            }
234        }
235
236        let values = match top_level_key {
237            "feature_gates" => &data.snapshot.feature_gates,
238            "dynamic_configs" => &data.snapshot.dynamic_configs,
239            "layer_configs" => &data.snapshot.layer_configs,
240            _ => {
241                log_e!(TAG, "Invalid top level key: {}", top_level_key);
242                return vec![];
243            }
244        };
245
246        if entity_type == "*" {
247            return values.keys().map(|k| k.unperformant_to_string()).collect();
248        }
249
250        values
251            .iter()
252            .filter(|(_, v)| v.view().entity().as_str() == entity_type)
253            .map(|(k, _)| k.unperformant_to_string())
254            .collect()
255    }
256
257    pub fn set_values(&self, mut specs_update: SpecsUpdate) -> Result<(), StatsigErr> {
258        let update_started_at = Instant::now();
259        // Updating the spec store is a three step process:
260        // 1. Prep (serialized writer path). Deserialize and compare to the current snapshot.
261        // 2. Apply (serialized writer path). Publish a new immutable snapshot.
262        // 3. Notify (no writer lock). Emit SDK events and update the data store.
263        let locked_result = {
264            let Some(_update_guard) = self.try_lock_for_update("set_values") else {
265                return Err(StatsigErr::LockFailure(
266                    "Failed to acquire spec store update lock for set_values".to_string(),
267                ));
268            };
269
270            let prep_result = self.specs_update_prep(&mut specs_update)?;
271
272            match prep_result {
273                PrepResult::HasUpdates {
274                    values,
275                    response_format,
276                    is_delta,
277                    spec_decode_stats,
278                } => {
279                    let apply_result =
280                        self.specs_update_apply(values, &specs_update, spec_decode_stats)?;
281                    Ok(LockedSetValuesResult::Applied(
282                        response_format,
283                        apply_result,
284                        ConfigResponseType::from_applied_response_data(&specs_update.data),
285                        is_delta,
286                    ))
287                }
288                PrepResult::CursorOnly { lcut, checksum } => {
289                    self.specs_cursor_update_apply(lcut, checksum, &specs_update);
290                    Ok(LockedSetValuesResult::NoSemanticUpdate(
291                        specs_update.source.clone(),
292                        specs_update.source_api.clone(),
293                        ConfigResponseType::Delta,
294                    ))
295                }
296                PrepResult::CurrentValuesNewer => Ok(LockedSetValuesResult::CurrentValuesNewer),
297                PrepResult::Duplicate => Ok(LockedSetValuesResult::Duplicate),
298                PrepResult::NoUpdates => Ok(LockedSetValuesResult::NoSemanticUpdate(
299                    specs_update.source.clone(),
300                    specs_update.source_api.clone(),
301                    ConfigResponseType::NoUpdate,
302                )),
303            }
304        }
305        .map_err(|e: StatsigErr| {
306            log_error_to_statsig_and_console!(self.ops_stats, TAG, e);
307            e
308        })?;
309
310        let (response_format, apply_result, response_type, is_delta) = match locked_result {
311            LockedSetValuesResult::Applied(
312                response_format,
313                apply_result,
314                response_type,
315                is_delta,
316            ) => {
317                if is_delta {
318                    self.ops_stats_log_proto_update(CONFIG_PROTO_UPDATE_MATERIALIZED);
319                }
320                (response_format, apply_result, response_type, is_delta)
321            }
322            LockedSetValuesResult::NoSemanticUpdate(source, source_api, response_type) => {
323                if matches!(response_type, ConfigResponseType::Delta) {
324                    self.ops_stats_log_proto_update(CONFIG_PROTO_UPDATE_CURSOR_ONLY);
325                    self.ops_stats_log_proto_update_latency(
326                        CONFIG_PROTO_UPDATE_CURSOR_ONLY,
327                        update_started_at.elapsed().as_secs_f64() * 1000.0,
328                    );
329                }
330                self.ops_stats_log_no_update(source, source_api, response_type);
331                return Ok(());
332            }
333            LockedSetValuesResult::CurrentValuesNewer => return Ok(()),
334            LockedSetValuesResult::Duplicate => {
335                self.ops_stats_log_proto_update(CONFIG_PROTO_UPDATE_DUPLICATE);
336                return Ok(());
337            }
338        };
339
340        self.ops_stats_log_interned_mmap_spec_decode(apply_result.spec_decode_stats);
341
342        try_release_unused_heap_memory();
343
344        // --- Notify ---
345
346        let notify_result = self
347            .specs_update_notify(response_format, response_type, specs_update, apply_result)
348            .map_err(|e| {
349                log_error_to_statsig_and_console!(self.ops_stats, TAG, e);
350                e
351            });
352
353        if is_delta {
354            self.ops_stats_log_proto_update_latency(
355                CONFIG_PROTO_UPDATE_MATERIALIZED,
356                update_started_at.elapsed().as_secs_f64() * 1000.0,
357            );
358        }
359
360        notify_result
361    }
362}
363
364// -------------------------------------------------------------------------------------------- [ Private ]
365
366enum PrepResult {
367    HasUpdates {
368        values: Box<SpecsResponseFull>,
369        response_format: SpecsFormat,
370        is_delta: bool,
371        spec_decode_stats: SpecDecodeStats,
372    },
373    CursorOnly {
374        lcut: u64,
375        checksum: String,
376    },
377    Duplicate,
378    NoUpdates,
379    CurrentValuesNewer,
380}
381
382enum LockedSetValuesResult {
383    Applied(SpecsFormat, ApplyResult, ConfigResponseType, bool),
384    NoSemanticUpdate(SpecsSource, Option<String>, ConfigResponseType),
385    Duplicate,
386    CurrentValuesNewer,
387}
388
389enum DeserializedSpecs {
390    Materialized {
391        values: Box<SpecsResponseFull>,
392        is_delta: bool,
393    },
394    CursorOnly {
395        lcut: u64,
396        checksum: String,
397    },
398}
399
400struct ApplyResult {
401    prev_source: SpecsSource,
402    prev_lcut: u64,
403    time_received_at: u64,
404    notification: SpecUpdateNotification,
405    spec_decode_stats: SpecDecodeStats,
406}
407
408struct SpecUpdateNotification {
409    source: SpecsSource,
410    source_api: Option<String>,
411    values: Arc<SpecsResponseFull>,
412    lcut: u64,
413    checksum: Option<String>,
414}
415
416impl SpecStore {
417    pub(crate) fn load_data(&self) -> Arc<SpecStoreData> {
418        self.data.load_full()
419    }
420
421    fn publish_data(&self, data: SpecStoreData) {
422        self.data.store(Arc::new(data));
423    }
424
425    fn try_lock_for_update(&self, operation: &str) -> Option<MutexGuard<'_, ()>> {
426        match self.update_lock.try_lock_for(LOCK_TIMEOUT) {
427            Some(guard) => Some(guard),
428            None => {
429                log_e!(
430                    TAG,
431                    "Failed to acquire spec store update lock for {}",
432                    operation
433                );
434                None
435            }
436        }
437    }
438
439    fn specs_update_prep(&self, specs_update: &mut SpecsUpdate) -> Result<PrepResult, StatsigErr> {
440        if specs_update.has_updates == Some(false) {
441            return Ok(PrepResult::NoUpdates);
442        }
443
444        let response_format = self.get_spec_response_format(specs_update);
445        let read_data = self.load_data();
446        if matches!(response_format, SpecsFormat::Protobuf)
447            && proto_response_lcut(&specs_update.data).is_some_and(|lcut| lcut < read_data.lcut())
448        {
449            return Ok(PrepResult::CurrentValuesNewer);
450        }
451        if matches!(response_format, SpecsFormat::Protobuf)
452            && proto_response_matches_cursor(&specs_update.data, &read_data)
453        {
454            return Ok(PrepResult::Duplicate);
455        }
456        // First, try a full or delta specs response deserialization
457        let first_deserialize_result =
458            self.deserialize_specs_data(&read_data, &response_format, &mut specs_update.data);
459
460        let first_deserialize_error = match first_deserialize_result {
461            Ok((
462                DeserializedSpecs::Materialized {
463                    values: next_values,
464                    is_delta,
465                },
466                spec_decode_stats,
467            )) => {
468                if self.are_current_values_newer(&read_data, &next_values) {
469                    return Ok(PrepResult::CurrentValuesNewer);
470                }
471
472                if next_values.has_updates {
473                    return Ok(PrepResult::HasUpdates {
474                        values: next_values,
475                        response_format,
476                        is_delta,
477                        spec_decode_stats,
478                    });
479                }
480
481                None
482            }
483            Ok((DeserializedSpecs::CursorOnly { lcut, checksum }, _)) => {
484                if cursor_is_stale_or_duplicate(&read_data, lcut, &checksum) {
485                    return Ok(PrepResult::CurrentValuesNewer);
486                }
487
488                return Ok(PrepResult::CursorOnly { lcut, checksum });
489            }
490            Err(e) => Some(e),
491        };
492
493        // Second, try a No Updates deserialization
494        let second_deserialize_result = specs_update
495            .data
496            .deserialize_into::<SpecsResponseNoUpdates>();
497
498        let second_deserialize_error = match second_deserialize_result {
499            Ok(result) => {
500                if !result.has_updates {
501                    return Ok(PrepResult::NoUpdates);
502                }
503
504                None
505            }
506            Err(e) => Some(e),
507        };
508
509        let error = first_deserialize_error
510            .or(second_deserialize_error)
511            .unwrap_or_else(|| {
512                StatsigErr::JsonParseError("SpecsResponse".to_string(), "Unknown error".to_string())
513            });
514
515        Err(error)
516    }
517
518    fn specs_update_apply(
519        &self,
520        next_values: Box<SpecsResponseFull>,
521        specs_update: &SpecsUpdate,
522        spec_decode_stats: SpecDecodeStats,
523    ) -> Result<ApplyResult, StatsigErr> {
524        // DANGER: try_update_global_configs contains its own locks
525        self.try_update_global_configs(&next_values);
526
527        let data = self.load_data();
528        let prev_source = data.source.clone();
529        let prev_lcut = data.lcut();
530        let time_received_at = Utc::now().timestamp_millis() as u64;
531        let values: Arc<SpecsResponseFull> = Arc::from(next_values);
532        let sync_cursor = ConfigSyncCursor {
533            lcut: values.time,
534            checksum: values.checksum.clone(),
535        };
536        let notification = SpecUpdateNotification {
537            source: specs_update.source.clone(),
538            source_api: specs_update.source_api.clone(),
539            lcut: values.time,
540            checksum: values
541                .checksum
542                .as_ref()
543                .map(|value| value.as_str().to_string()),
544            values: values.clone(),
545        };
546
547        self.publish_data(SpecStoreData {
548            source: specs_update.source.clone(),
549            source_api: specs_update.source_api.clone(),
550            time_received_at: Some(time_received_at),
551            snapshot: values,
552            id_lists: data.id_lists.clone(),
553            gcir_evaluation_plan: Arc::new(OnceLock::new()),
554            sync_cursor,
555            spec_decode_stats,
556        });
557
558        Ok(ApplyResult {
559            prev_source,
560            prev_lcut,
561            time_received_at,
562            notification,
563            spec_decode_stats,
564        })
565    }
566
567    fn specs_cursor_update_apply(&self, lcut: u64, checksum: String, specs_update: &SpecsUpdate) {
568        let data = self.load_data();
569        let mut next_data = data.as_ref().clone();
570        next_data.source = specs_update.source.clone();
571        next_data.source_api = specs_update.source_api.clone();
572        next_data.time_received_at = Some(Utc::now().timestamp_millis() as u64);
573        next_data.sync_cursor = ConfigSyncCursor {
574            lcut,
575            checksum: Some(checksum),
576        };
577        self.publish_data(next_data);
578    }
579
580    fn specs_update_notify(
581        &self,
582        response_format: SpecsFormat,
583        response_type: ConfigResponseType,
584        specs_update: SpecsUpdate,
585        apply_result: ApplyResult,
586    ) -> Result<(), StatsigErr> {
587        let SpecsUpdate { data, .. } = specs_update;
588        let ApplyResult {
589            prev_source,
590            prev_lcut,
591            time_received_at,
592            notification,
593            spec_decode_stats: _,
594        } = apply_result;
595        let SpecUpdateNotification {
596            source,
597            source_api,
598            values,
599            lcut,
600            checksum,
601        } = notification;
602
603        self.emit_specs_updated_sdk_event(&source, &source_api, values.as_ref());
604        if let Some(sdk_configs) = &values.sdk_configs {
605            self.emit_internal_sdk_configs_updated_sdk_event(sdk_configs);
606        }
607
608        // Data store writes now support both legacy JSON payloads and statsig-br protobuf bytes.
609        self.try_update_data_store(
610            &source,
611            data,
612            time_received_at,
613            checksum,
614            matches!(response_format, SpecsFormat::Protobuf),
615        );
616
617        self.ops_stats_log_config_propagation_diff(
618            lcut,
619            prev_lcut,
620            &source,
621            &prev_source,
622            source_api,
623            response_format,
624            response_type,
625        );
626
627        Ok(())
628    }
629
630    fn deserialize_specs_data(
631        &self,
632        current_data: &SpecStoreData,
633        response_format: &SpecsFormat,
634        response_data: &mut ResponseData,
635    ) -> Result<(DeserializedSpecs, SpecDecodeStats), StatsigErr> {
636        let (result, spec_decode_stats) = track_spec_decodes(|| {
637            let mut next_values = Box::new(SpecsResponseFull::default());
638
639            match response_format {
640                SpecsFormat::Protobuf => {
641                    let update = deserialize_protobuf_for_store(
642                        &self.ops_stats,
643                        current_data.snapshot.as_ref(),
644                        current_data.spec_decode_stats,
645                        next_values.as_mut(),
646                        response_data,
647                    )?;
648                    match update {
649                        ProtobufUpdate::Materialized { is_delta } => {
650                            Ok(DeserializedSpecs::Materialized {
651                                values: next_values,
652                                is_delta,
653                            })
654                        }
655                        ProtobufUpdate::CursorOnly { lcut, checksum } => {
656                            Ok(DeserializedSpecs::CursorOnly { lcut, checksum })
657                        }
658                    }
659                }
660                SpecsFormat::Json => {
661                    response_data.deserialize_in_place(next_values.as_mut())?;
662                    Ok(DeserializedSpecs::Materialized {
663                        values: next_values,
664                        is_delta: false,
665                    })
666                }
667            }
668        });
669
670        Ok((result?, spec_decode_stats))
671    }
672
673    fn emit_specs_updated_sdk_event(
674        &self,
675        source: &SpecsSource,
676        source_api: &Option<String>,
677        values: &SpecsResponseFull,
678    ) {
679        self.event_emitter.emit(SdkEvent::SpecsUpdated {
680            source,
681            source_api,
682            values,
683        });
684    }
685
686    fn emit_internal_sdk_configs_updated_sdk_event(
687        &self,
688        sdk_configs: &HashMap<String, crate::DynamicValue>,
689    ) {
690        self.event_emitter
691            .emit(SdkEvent::InternalSdkConfigsUpdated { sdk_configs });
692    }
693
694    fn get_spec_response_format(&self, update: &SpecsUpdate) -> SpecsFormat {
695        let content_type = update.data.get_header_ref("content-type");
696        if content_type.map(|s| s.as_str().contains("application/octet-stream")) != Some(true) {
697            return SpecsFormat::Json;
698        }
699
700        let content_encoding = update.data.get_header_ref("content-encoding");
701        if content_encoding.map(|s| s.as_str().contains("statsig-br")) != Some(true) {
702            return SpecsFormat::Json;
703        }
704
705        SpecsFormat::Protobuf
706    }
707
708    fn try_update_global_configs(&self, dcs: &SpecsResponseFull) {
709        if let Some(diagnostics) = &dcs.diagnostics {
710            self.global_configs
711                .set_diagnostics_sampling_rates(diagnostics.clone());
712        }
713
714        if let Some(sdk_configs) = &dcs.sdk_configs {
715            self.global_configs.set_sdk_configs(sdk_configs.clone());
716        }
717
718        if let Some(sdk_flags) = &dcs.sdk_flags {
719            self.global_configs.set_sdk_flags(sdk_flags.clone());
720        }
721    }
722
723    fn try_update_data_store(
724        &self,
725        source: &SpecsSource,
726        mut data: ResponseData,
727        now: u64,
728        checksum: Option<String>,
729        is_protobuf: bool,
730    ) {
731        if source != &SpecsSource::Network {
732            return;
733        }
734
735        if data.get_header_ref("x-deltas-used").is_some() {
736            log_d!(
737                TAG,
738                "Skipping data store write for delta response identified by x-deltas-used header"
739            );
740            return;
741        }
742
743        let data_store = match &self.data_store {
744            Some(data_store) => data_store.clone(),
745            None => return,
746        };
747
748        let data_store_key = if is_protobuf {
749            self.data_store_keys.statsig_br.clone()
750        } else {
751            self.data_store_keys.plain_text.clone()
752        };
753
754        let spawn_result = self.statsig_runtime.spawn(
755            "spec_store_update_data_store",
756            move |_shutdown_notif| async move {
757                let data_bytes = match data.read_to_bytes() {
758                    Ok(bytes) => bytes,
759                    Err(e) => {
760                        log_e!(TAG, "Failed to read data as bytes: {}", e);
761                        return;
762                    }
763                };
764
765                write_specs_to_data_store(
766                    data_store,
767                    data_store_key,
768                    data_bytes,
769                    checksum,
770                    now,
771                    is_protobuf,
772                )
773                .await;
774            },
775        );
776
777        if let Err(e) = spawn_result {
778            log_e!(
779                TAG,
780                "Failed to spawn spec store update data store task: {e}"
781            );
782        }
783    }
784
785    fn are_current_values_newer(
786        &self,
787        data: &SpecStoreData,
788        next_values: &SpecsResponseFull,
789    ) -> bool {
790        let curr_checksum = data.checksum().unwrap_or_default();
791        let new_checksum = next_values.checksum.as_deref().unwrap_or_default();
792
793        let cached_time_is_newer = data.lcut() > 0 && data.lcut() > next_values.time;
794        let checksums_match = !curr_checksum.is_empty() && curr_checksum == new_checksum;
795
796        if cached_time_is_newer || checksums_match {
797            log_d!(
798                TAG,
799                "Received values for [time: {}, checksum: {}], but currently has values for [time: {}, checksum: {}]. Ignoring values.",
800                next_values.time,
801                new_checksum,
802                data.lcut(),
803                curr_checksum,
804            );
805            return true;
806        }
807
808        false
809    }
810}
811
812async fn write_specs_to_data_store(
813    data_store: Arc<dyn DataStoreTrait>,
814    data_store_key: String,
815    data_bytes: Vec<u8>,
816    checksum: Option<String>,
817    now: u64,
818    is_protobuf: bool,
819) {
820    match data_store
821        .set_bytes(&data_store_key, &data_bytes, Some(now), checksum)
822        .await
823    {
824        Ok(()) => return,
825        Err(e @ StatsigErr::BytesNotImplemented) if is_protobuf => {
826            if data_store
827                .support_polling_updates_for(RequestPath::RulesetsV2)
828                .await
829            {
830                log_w!(
831                    TAG,
832                    "Failed to write protobuf specs to data store as bytes. Protobuf specs cannot fall back to string writes: {}",
833                    e
834                );
835            }
836            return;
837        }
838        Err(e @ StatsigErr::BytesNotImplemented) => {
839            log_w!(
840                TAG,
841                "Data store bytes write is not implemented. Falling back to string write: {}",
842                e
843            );
844        }
845        Err(e) => {
846            log_w!(TAG, "Failed to write specs to data store as bytes: {}", e);
847            return;
848        }
849    }
850
851    let data_string = match String::from_utf8(data_bytes) {
852        Ok(s) => s,
853        Err(e) => {
854            log_w!(
855                TAG,
856                "Skipping data store string write because payload is not valid UTF-8: {}",
857                e
858            );
859            return;
860        }
861    };
862
863    if let Err(e) = data_store
864        .set(&data_store_key, &data_string, Some(now))
865        .await
866    {
867        log_w!(TAG, "Failed to write specs to data store as string: {}", e);
868    }
869}
870
871// -------------------------------------------------------------------------------------------- [ OpsStats Helpers ]
872
873impl SpecStore {
874    fn ops_stats_log_interned_mmap_spec_decode(&self, stats: SpecDecodeStats) {
875        if stats.total == 0 {
876            return;
877        }
878
879        let (source, reason) = if stats.mmap == stats.total {
880            ("mmap", "preloaded")
881        } else if stats.mmap > 0 {
882            ("mixed", "partial_match")
883        } else if InternedStore::has_preloaded_mmap_v2() {
884            ("owned", "no_match")
885        } else {
886            ("owned", "spec_preload_unavailable")
887        };
888
889        self.ops_stats.log(ObservabilityEvent::new_event(
890            MetricType::Increment,
891            INTERNED_MMAP_SPEC_DECODE_COUNT_METRIC.to_string(),
892            1.0,
893            Some(HashMap::from([
894                ("source".to_string(), source.to_string()),
895                ("reason".to_string(), reason.to_string()),
896                ("sdk_key".to_string(), self.loggable_sdk_key.clone()),
897            ])),
898        ));
899    }
900
901    fn ops_stats_log_no_update(
902        &self,
903        source: SpecsSource,
904        source_api: Option<String>,
905        response_type: ConfigResponseType,
906    ) {
907        log_d!(TAG, "No Updates");
908        self.ops_stats.log(ObservabilityEvent::new_event(
909            MetricType::Increment,
910            "config_no_update".to_string(),
911            1.0,
912            Some(HashMap::from([
913                ("source".to_string(), source.to_string()),
914                ("source_api".to_string(), source_api.unwrap_or_default()),
915                (
916                    RESPONSE_TYPE_TAG.to_string(),
917                    response_type.as_str().to_string(),
918                ),
919            ])),
920        ));
921    }
922
923    fn ops_stats_log_proto_update(&self, outcome: &str) {
924        self.ops_stats.log(ObservabilityEvent::new_event(
925            MetricType::Increment,
926            CONFIG_PROTO_UPDATE_COUNT_METRIC.to_string(),
927            1.0,
928            Some(HashMap::from([(
929                CONFIG_PROTO_UPDATE_OUTCOME_TAG.to_string(),
930                outcome.to_string(),
931            )])),
932        ));
933    }
934
935    fn ops_stats_log_proto_update_latency(&self, outcome: &str, duration_ms: f64) {
936        self.ops_stats.log(ObservabilityEvent::new_event(
937            MetricType::Dist,
938            CONFIG_PROTO_UPDATE_LATENCY_METRIC.to_string(),
939            duration_ms,
940            Some(HashMap::from([(
941                CONFIG_PROTO_UPDATE_OUTCOME_TAG.to_string(),
942                outcome.to_string(),
943            )])),
944        ));
945    }
946
947    #[allow(clippy::too_many_arguments)]
948    fn ops_stats_log_config_propagation_diff(
949        &self,
950        lcut: u64,
951        prev_lcut: u64,
952        source: &SpecsSource,
953        prev_source: &SpecsSource,
954        source_api: Option<String>,
955        response_format: SpecsFormat,
956        response_type: ConfigResponseType,
957    ) {
958        let delay = (Utc::now().timestamp_millis() as u64).saturating_sub(lcut);
959        log_d!(TAG, "Updated ({:?})", source);
960
961        if *prev_source == SpecsSource::Uninitialized || *prev_source == SpecsSource::Loading {
962            return;
963        }
964
965        self.ops_stats.log(ObservabilityEvent::new_event(
966            MetricType::Dist,
967            "config_propagation_diff".to_string(),
968            delay as f64,
969            Some(HashMap::from([
970                ("source".to_string(), source.to_string()),
971                ("lcut".to_string(), lcut.to_string()),
972                ("prev_lcut".to_string(), prev_lcut.to_string()),
973                ("source_api".to_string(), source_api.unwrap_or_default()),
974                ("sdk_key".to_string(), self.loggable_sdk_key.clone()),
975                (
976                    "response_format".to_string(),
977                    Into::<&str>::into(&response_format).to_string(),
978                ),
979                (
980                    RESPONSE_TYPE_TAG.to_string(),
981                    response_type.as_str().to_string(),
982                ),
983            ])),
984        ));
985    }
986}
987
988// -------------------------------------------------------------------------------------------- [Impl SpecsUpdateListener]
989
990impl SpecsUpdateListener for SpecStore {
991    fn did_receive_specs_update(&self, update: SpecsUpdate) -> Result<(), StatsigErr> {
992        self.set_values(update)
993    }
994
995    fn get_current_specs_info(&self) -> SpecsInfo {
996        let data = self.load_data();
997        SpecsInfo {
998            lcut: Some(data.lcut()),
999            checksum: data.sync_cursor.checksum.clone(),
1000            source: data.source.clone(),
1001            source_api: data.source_api.clone(),
1002        }
1003    }
1004}
1005
1006// -------------------------------------------------------------------------------------------- [Impl IdListsUpdateListener]
1007
1008impl IdListsUpdateListener for SpecStore {
1009    fn get_current_id_list_metadata(
1010        &self,
1011    ) -> HashMap<String, crate::id_lists_adapter::IdListMetadata> {
1012        let data = self.load_data();
1013        data.id_lists
1014            .iter()
1015            .map(|(key, list)| (key.clone(), list.metadata.clone()))
1016            .collect()
1017    }
1018
1019    fn did_receive_id_list_updates(
1020        &self,
1021        updates: HashMap<String, crate::id_lists_adapter::IdListUpdate>,
1022    ) {
1023        let Some(_update_guard) = self.try_lock_for_update("did_receive_id_list_updates") else {
1024            return;
1025        };
1026
1027        let data = self.load_data();
1028        let mut id_lists = data.id_lists.as_ref().clone();
1029
1030        // delete any id_lists that are not in the updates
1031        id_lists.retain(|name, _| updates.contains_key(name));
1032
1033        for (list_name, update) in updates {
1034            if let Some(entry) = id_lists.get_mut(&list_name) {
1035                // update existing
1036                entry.apply_update(update);
1037            } else {
1038                // add new
1039                let mut list = IdList::new(update.new_metadata.clone());
1040                list.apply_update(update);
1041                id_lists.insert(list_name, list);
1042            }
1043        }
1044
1045        let mut next_data = data.as_ref().clone();
1046        next_data.id_lists = Arc::new(id_lists);
1047        self.publish_data(next_data);
1048    }
1049}
1050
1051#[cfg(test)]
1052mod tests {
1053    use super::{
1054        ConfigResponseType, ConfigSyncCursor, SpecDecodeStats, SpecStoreData, DELTAS_USED_HEADER,
1055    };
1056    use crate::hashing::HashUtil;
1057    use crate::networking::ResponseData;
1058    use crate::specs_response::spec_types::SpecsResponseFull;
1059    use crate::SpecsSource;
1060    use std::collections::HashMap;
1061    use std::sync::{Arc, OnceLock};
1062
1063    #[test]
1064    fn applied_response_type_uses_delta_header() {
1065        let delta_data = ResponseData::from_bytes_with_headers(
1066            Vec::new(),
1067            Some(HashMap::from([(
1068                DELTAS_USED_HEADER.to_string(),
1069                "true".to_string(),
1070            )])),
1071        );
1072        let full_data = ResponseData::from_bytes(Vec::new());
1073
1074        assert_eq!(
1075            ConfigResponseType::from_applied_response_data(&delta_data).as_str(),
1076            "delta"
1077        );
1078        assert_eq!(
1079            ConfigResponseType::from_applied_response_data(&full_data).as_str(),
1080            "full"
1081        );
1082    }
1083
1084    #[test]
1085    fn spec_store_data_reuses_gcir_evaluation_plan_cache_across_clones() {
1086        let data = SpecStoreData {
1087            source: SpecsSource::Network,
1088            source_api: Some("/v2/download_config_specs".to_string()),
1089            time_received_at: Some(1),
1090            snapshot: Arc::new(SpecsResponseFull::default()),
1091            id_lists: Arc::new(HashMap::new()),
1092            gcir_evaluation_plan: Arc::new(OnceLock::new()),
1093            sync_cursor: ConfigSyncCursor::default(),
1094            spec_decode_stats: SpecDecodeStats::default(),
1095        };
1096        let hashing = HashUtil::new();
1097
1098        let first = data.gcir_evaluation_plan(&hashing);
1099        let cloned = data.clone();
1100        let second = cloned.gcir_evaluation_plan(&hashing);
1101
1102        assert!(std::ptr::eq(first, second));
1103    }
1104}
statsig_rust/spec_store.rs

statsig_rust/
spec_store.rs
