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