1#[cfg(test)]
2use std::path::Path;
3use std::{
4 borrow::Cow,
5 collections::{hash_map::Entry, HashMap},
6 marker::PhantomData,
7 path::PathBuf,
8 sync::{
9 atomic::{AtomicBool, Ordering},
10 Arc, OnceLock, Weak,
11 },
12 time::Instant,
13};
14
15use ahash::AHashMap;
16use fancy_regex::Regex as FancyRegex;
17use lazy_static::lazy_static;
18use parking_lot::RwLock;
19use rkyv::{
20 collections::swiss_table::{ArchivedHashMap, HashMapResolver},
21 hash::FxHasher64,
22 rancor::{Fallible, Source},
23 ser::{Allocator, Writer},
24 string::ArchivedString,
25 with::{ArchiveWith, DeserializeWith, SerializeWith, With},
26 Archive, Place, Serialize as RkyvSerialize,
27};
28use serde_json::value::RawValue;
29
30pub use super::mmap_sync::{MmapSyncCursor, MmapWriteOutcome};
31
32use crate::{
33 evaluation::{
34 dynamic_returnable::DynamicReturnableValue,
35 evaluator_value::{EvaluatorValue, EvaluatorValueInner, MemoizedEvaluatorValue},
36 rkyv_value::{ArchivedRkyvValue, RkyvValue},
37 },
38 hashing,
39 interned_string::{InternedString, InternedStringValue},
40 log_d, log_e,
41 networking::ResponseData,
42 observability::ops_stats::OpsStatsForInstance,
43 specs_adapter::{SpecsInfo, SpecsSyncTrigger, StatsigHttpSpecsAdapter},
44 specs_response::{
45 proto_specs::deserialize_protobuf,
46 spec_types::{Spec, SpecsResponseFull},
47 specs_hash_map::{SpecPointer, SpecsHashMap},
48 },
49 utils::try_release_unused_heap_memory,
50 DynamicReturnable, StatsigErr, StatsigOptions,
51};
52
53use super::mmap_data_v2::ArchivedMmapEvaluatorValue;
54
55mod mmap_artifact;
56mod mmap_manifest;
57mod mmap_reader;
58mod mmap_writer;
59
60pub use mmap_artifact::{MmapArtifactSnapshot, MmapArtifactState};
61pub use mmap_reader::MmapReaderMemorySnapshot;
62
63use mmap_manifest::open_committed_mmap_v2;
64#[cfg(all(test, any(unix, windows)))]
65pub(crate) use mmap_manifest::open_committed_mmap_v2_for_test;
66#[cfg(test)]
67pub(crate) use mmap_manifest::{
68 write_mmap_manifest_for_test, write_mmap_v2_only_manifest_for_test,
69};
70use mmap_reader::{
71 get_evaluator_value as get_archived_evaluator_value_from_mmap,
72 get_returnable as get_returnable_from_mmap, get_spec as get_mmap_spec,
73 get_string as get_string_from_mmap, MmapSpecKind,
74};
75use mmap_writer::{acquire_mmap_write_lock, write_mmap_artifacts};
76#[cfg(test)]
77pub(crate) use mmap_writer::{acquire_mmap_write_lock_for_test, write_mmap_v2_for_test};
78
79const TAG: &str = "InternedStore";
80const MMAP_DIRECTORY: &str = "statsig-interned-store";
81pub(crate) const LEGACY_MMAP_FORMAT_VERSION: u32 = 1;
82const WEAK_STORE_SWEEP_INTERVAL: usize = 4096;
83
84#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
86pub struct MmapPreloadOptions {
87 pub precompute_returnable_stable_hashes: bool,
93}
94
95#[non_exhaustive]
96#[derive(Clone, Copy, Debug, Eq, PartialEq)]
97pub struct MmapPreloadReport {
98 pub format_version: u32,
100 pub loaded: usize,
102}
103
104static IMMORTAL_DATA: OnceLock<ImmortalData> = OnceLock::new();
105static MMAP_EVALUATOR_OVERRIDE_EXISTS: AtomicBool = AtomicBool::new(false);
106
107lazy_static! {
108 static ref MUTABLE_STRINGS: MutableStore<String> = MutableStore::default();
109 static ref MUTABLE_RETURNABLES: MutableStore<HashMap<String, RkyvValue>> =
110 MutableStore::default();
111 static ref MUTABLE_EVALUATOR_VALUES: MutableStore<MemoizedEvaluatorValue> =
112 MutableStore::default();
113}
114
115type MutableEntries<T> = AHashMap<u64, Weak<T>>;
116
117struct MutableTable<T> {
118 entries: MutableEntries<T>,
119 insertions: usize,
120 next_sweep_at: usize,
121}
122
123impl<T> Default for MutableTable<T> {
124 fn default() -> Self {
125 Self {
126 entries: AHashMap::new(),
127 insertions: 0,
128 next_sweep_at: WEAK_STORE_SWEEP_INTERVAL,
129 }
130 }
131}
132
133struct MutableStore<T> {
134 table: RwLock<MutableTable<T>>,
135}
136
137impl<T> Default for MutableStore<T> {
138 fn default() -> Self {
139 Self {
140 table: RwLock::new(MutableTable::default()),
141 }
142 }
143}
144
145impl<T> MutableStore<T> {
146 fn get(&self, hash: u64) -> Option<Arc<T>> {
147 let table = self.table.read();
148 let value = table.entries.get(&hash)?.upgrade();
149 value
150 }
151
152 fn get_or_insert(&self, hash: u64, candidate: Arc<T>) -> Arc<T> {
153 let mut table = self.table.write();
154 let (value, inserted) = match table.entries.entry(hash) {
155 Entry::Occupied(mut entry) => match entry.get().upgrade() {
156 Some(value) => (value, false),
157 None => {
158 entry.insert(Arc::downgrade(&candidate));
159 (candidate, true)
160 }
161 },
162 Entry::Vacant(entry) => {
163 entry.insert(Arc::downgrade(&candidate));
164 (candidate, true)
165 }
166 };
167
168 if inserted {
169 Self::maybe_sweep(&mut table);
170 }
171 value
172 }
173
174 fn replace(&self, hash: u64, value: &Arc<T>) {
175 let mut table = self.table.write();
176 table.entries.insert(hash, Arc::downgrade(value));
177 Self::maybe_sweep(&mut table);
178 }
179
180 #[cfg(test)]
181 fn live_len(&self) -> usize {
182 self.table
183 .read()
184 .entries
185 .iter()
186 .filter(|(_, value)| value.strong_count() > 0)
187 .count()
188 }
189
190 #[cfg(test)]
191 fn stored_len(&self) -> usize {
192 self.table.read().entries.len()
193 }
194
195 #[cfg(test)]
196 fn stored_capacity(&self) -> usize {
197 self.table.read().entries.capacity()
198 }
199
200 fn take_live(&self) -> Vec<(u64, Arc<T>)> {
201 let detached = {
202 let mut table = self.table.write();
203 std::mem::take(&mut *table)
204 };
205
206 let mut values = Vec::with_capacity(detached.entries.len());
207 for (hash, value) in detached.entries {
208 if let Some(value) = value.upgrade() {
209 values.push((hash, value));
210 }
211 }
212 values
213 }
214
215 fn maybe_sweep(table: &mut MutableTable<T>) {
216 table.insertions += 1;
217 if table.insertions == table.next_sweep_at {
218 table.entries.retain(|_, value| value.strong_count() > 0);
219 table.next_sweep_at = table
220 .insertions
221 .saturating_add(table.entries.len().max(WEAK_STORE_SWEEP_INTERVAL));
222 }
223 }
224}
225
226pub(super) struct MapKVVec<A, B>(PhantomData<(A, B)>);
229
230struct WithKey<'a, K, A> {
231 key: &'a K,
232 _adapter: PhantomData<A>,
233}
234
235impl<K, A> Copy for WithKey<'_, K, A> {}
236
237impl<K, A> Clone for WithKey<'_, K, A> {
238 fn clone(&self) -> Self {
239 *self
240 }
241}
242
243impl<K: PartialEq, A> PartialEq for WithKey<'_, K, A> {
244 fn eq(&self, other: &Self) -> bool {
245 self.key == other.key
246 }
247}
248
249impl<K: Eq, A> Eq for WithKey<'_, K, A> {}
250
251impl<K: std::hash::Hash, A> std::hash::Hash for WithKey<'_, K, A> {
252 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
253 self.key.hash(state);
254 }
255}
256
257impl<K, A> Archive for WithKey<'_, K, A>
258where
259 A: ArchiveWith<K>,
260{
261 type Archived = A::Archived;
262 type Resolver = A::Resolver;
263
264 fn resolve(&self, resolver: Self::Resolver, out: Place<Self::Archived>) {
265 A::resolve_with(self.key, resolver, out);
266 }
267}
268
269impl<K, A, S> RkyvSerialize<S> for WithKey<'_, K, A>
270where
271 S: Fallible + ?Sized,
272 A: SerializeWith<K, S>,
273{
274 fn serialize(&self, serializer: &mut S) -> Result<Self::Resolver, S::Error> {
275 A::serialize_with(self.key, serializer)
276 }
277}
278
279impl<A, B, K, V> ArchiveWith<Vec<(K, V)>> for MapKVVec<A, B>
280where
281 A: ArchiveWith<K>,
282 B: ArchiveWith<V>,
283{
284 type Archived = ArchivedHashMap<A::Archived, B::Archived>;
285 type Resolver = HashMapResolver;
286
287 fn resolve_with(field: &Vec<(K, V)>, resolver: Self::Resolver, out: Place<Self::Archived>) {
288 ArchivedHashMap::resolve_from_len(field.len(), (7, 8), resolver, out);
289 }
290}
291
292impl<A, B, K, V, S> SerializeWith<Vec<(K, V)>, S> for MapKVVec<A, B>
293where
294 A: ArchiveWith<K> + SerializeWith<K, S>,
295 B: ArchiveWith<V> + SerializeWith<V, S>,
296 K: std::hash::Hash + Eq,
297 A::Archived: std::hash::Hash + Eq,
298 S: Fallible + Allocator + Writer + ?Sized,
299 S::Error: Source,
300{
301 fn serialize_with(field: &Vec<(K, V)>, serializer: &mut S) -> Result<Self::Resolver, S::Error> {
302 ArchivedHashMap::<_, _, FxHasher64>::serialize_from_iter::<
303 _,
304 _,
305 _,
306 WithKey<'_, K, A>,
307 With<V, B>,
308 S,
309 >(
310 field.iter().map(|(key, value)| {
311 (
312 WithKey {
313 key,
314 _adapter: PhantomData,
315 },
316 With::<V, B>::cast(value),
317 )
318 }),
319 (7, 8),
320 serializer,
321 )
322 }
323}
324
325impl<A, B, K, V, D> DeserializeWith<ArchivedHashMap<A::Archived, B::Archived>, Vec<(K, V)>, D>
326 for MapKVVec<A, B>
327where
328 A: ArchiveWith<K> + DeserializeWith<A::Archived, K, D>,
329 B: ArchiveWith<V> + DeserializeWith<B::Archived, V, D>,
330 D: Fallible + ?Sized,
331{
332 fn deserialize_with(
333 field: &ArchivedHashMap<A::Archived, B::Archived>,
334 deserializer: &mut D,
335 ) -> Result<Vec<(K, V)>, D::Error> {
336 let mut result = Vec::with_capacity(field.len());
337 for (key, value) in field.iter() {
338 result.push((
339 A::deserialize_with(key, deserializer)?,
340 B::deserialize_with(value, deserializer)?,
341 ));
342 }
343 Ok(result)
344 }
345}
346
347#[derive(Default)]
356struct ImmortalData {
357 strings: AHashMap<u64, &'static str>,
358 returnables: AHashMap<u64, &'static HashMap<String, RkyvValue>>,
359 evaluator_values: AHashMap<u64, &'static MemoizedEvaluatorValue>,
360 feature_gates: AHashMap<u64, &'static Spec>,
361 dynamic_configs: AHashMap<u64, &'static Spec>,
362 layer_configs: AHashMap<u64, &'static Spec>,
363}
364#[derive(Default)]
365struct MutableData {
366 strings: Vec<(u64, Arc<String>)>,
367 returnables: Vec<(u64, Arc<HashMap<String, RkyvValue>>)>,
368 evaluator_values: Vec<(u64, Arc<MemoizedEvaluatorValue>)>,
369}
370
371pub trait Internable: Sized {
372 type Input<'a>;
373 fn intern(input: Self::Input<'_>) -> Self;
374}
375
376pub struct InternedStore;
377
378impl InternedStore {
379 pub fn preload(data: &[u8]) -> Result<(), StatsigErr> {
380 Self::preload_multi(&[data])
381 }
382
383 pub fn preload_multi(data: &[&[u8]]) -> Result<(), StatsigErr> {
384 let start_time = Instant::now();
385
386 if IMMORTAL_DATA.get().is_some() {
387 log_e!(TAG, "Already preloaded");
388 return Err(StatsigErr::InvalidOperation(
389 "Already preloaded".to_string(),
390 ));
391 }
392
393 let specs_responses = data
394 .iter()
395 .map(|data| try_parse_as_json(data).or_else(|_| try_parse_as_proto(data)))
396 .collect::<Result<Vec<SpecsResponseFull>, StatsigErr>>()?;
397
398 let immortal = mutable_to_immortal(specs_responses)?;
399
400 if IMMORTAL_DATA.set(immortal).is_err() {
401 return Err(StatsigErr::LockFailure(
402 "Failed to set IMMORTAL_DATA".to_string(),
403 ));
404 }
405
406 let end_time = Instant::now();
407 log_d!(
408 TAG,
409 "Preload took {}ms",
410 end_time.duration_since(start_time).as_millis()
411 );
412
413 Ok(())
414 }
415
416 pub async fn fetch_and_write_mmap(sdk_key: &str) -> Result<(), StatsigErr> {
421 Self::fetch_and_write_mmap_with_options(sdk_key, None).await
422 }
423
424 pub async fn fetch_and_write_mmap_with_specs_url(
425 sdk_key: &str,
426 specs_url: &str,
427 ) -> Result<(), StatsigErr> {
428 let options = StatsigOptions {
429 specs_url: Some(specs_url.to_string()),
430 ..StatsigOptions::default()
431 };
432
433 Self::fetch_and_write_mmap_with_options(sdk_key, Some(&options)).await
434 }
435
436 pub async fn fetch_and_write_mmap_with_specs_url_if_changed(
437 sdk_key: &str,
438 specs_url: &str,
439 previous: Option<&MmapSyncCursor>,
440 ) -> Result<MmapWriteOutcome, StatsigErr> {
441 let options = StatsigOptions {
442 specs_url: Some(specs_url.to_string()),
443 ..StatsigOptions::default()
444 };
445
446 Self::fetch_and_write_mmap_with_options_if_changed(sdk_key, Some(&options), previous).await
447 }
448
449 pub(crate) async fn fetch_and_write_mmap_with_options(
450 sdk_key: &str,
451 options: Option<&StatsigOptions>,
452 ) -> Result<(), StatsigErr> {
453 match Self::fetch_and_write_mmap_with_options_if_changed(sdk_key, options, None).await? {
454 MmapWriteOutcome::Published(_) => Ok(()),
455 MmapWriteOutcome::NoUpdate => Err(StatsigErr::InvalidOperation(
456 "An unconditional mmap fetch did not return a publishable config snapshot"
457 .to_string(),
458 )),
459 }
460 }
461
462 pub(crate) async fn fetch_and_write_mmap_with_options_if_changed(
463 sdk_key: &str,
464 options: Option<&StatsigOptions>,
465 previous: Option<&MmapSyncCursor>,
466 ) -> Result<MmapWriteOutcome, StatsigErr> {
467 let write_lock = acquire_mmap_write_lock(&mmap_lock_path_for_sdk_key(sdk_key)).await?;
470 let adapter = StatsigHttpSpecsAdapter::new(sdk_key, options, None);
471 let mut specs_info = SpecsInfo::empty();
472 if let Some(previous) = previous {
473 specs_info.lcut = Some(previous.lcut);
474 specs_info.checksum = previous.checksum.clone();
475 }
476
477 let mut response = adapter
478 .fetch_specs_from_network(specs_info, SpecsSyncTrigger::Manual)
479 .await
480 .map_err(StatsigErr::NetworkError)?;
481 let result = write_mmap_artifacts(
482 &mut response.data,
483 previous,
484 &mmap_v2_path_for_sdk_key(sdk_key),
485 &mmap_manifest_path_for_sdk_key(sdk_key),
486 );
487
488 drop(response);
489 drop(adapter);
490 drop(write_lock);
491 try_release_unused_heap_memory();
492 result
493 }
494
495 pub fn preload_mmap(sdk_key: &str) -> Result<(), StatsigErr> {
496 Self::preload_mmap_with_options(sdk_key, &MmapPreloadOptions::default()).map(|_| ())
497 }
498
499 pub fn mmap_reader_memory_snapshot() -> Result<Option<MmapReaderMemorySnapshot>, StatsigErr> {
506 mmap_reader::memory_snapshot()
507 }
508
509 pub fn preload_mmap_with_options(
512 sdk_key: &str,
513 options: &MmapPreloadOptions,
514 ) -> Result<MmapPreloadReport, StatsigErr> {
515 let v2_path = mmap_v2_path_for_sdk_key(sdk_key);
516 let manifest_path = mmap_manifest_path_for_sdk_key(sdk_key);
517 let v2_file = open_committed_mmap_v2(
518 &manifest_path,
519 &legacy_mmap_v1_path_for_sdk_key(sdk_key),
520 &v2_path,
521 )?
522 .ok_or_else(|| {
523 StatsigErr::InvalidOperation(
524 "No committed interned mmap V2 artifact was found".to_string(),
525 )
526 })?;
527 mmap_reader::preload_v2_file(v2_file, options)?;
528 Ok(MmapPreloadReport {
529 format_version: super::mmap_data_v2::MmapDataV2::FORMAT_VERSION,
530 loaded: 1,
531 })
532 }
533}
534
535#[cfg(test)]
536pub(crate) fn preload_mmap_v2_for_test(path: &Path) -> Result<(), StatsigErr> {
537 mmap_reader::preload_v2(path, &MmapPreloadOptions::default())
538}
539
540#[cfg(test)]
541pub(crate) fn validate_mmap_v2_for_test(path: &Path) -> Result<(), StatsigErr> {
542 mmap_reader::validate_v2(path)
543}
544
545pub(crate) fn legacy_mmap_v1_path_for_sdk_key(sdk_key: &str) -> PathBuf {
546 mmap_path_for_sdk_key_version(sdk_key, 1)
547}
548
549pub(crate) fn mmap_v2_path_for_sdk_key(sdk_key: &str) -> PathBuf {
550 mmap_path_for_sdk_key_version(sdk_key, super::mmap_data_v2::MmapDataV2::FORMAT_VERSION)
551}
552
553pub(crate) fn mmap_manifest_path_for_sdk_key(sdk_key: &str) -> PathBuf {
554 std::env::temp_dir().join(MMAP_DIRECTORY).join(format!(
555 "{}_interned_store_manifest.json",
556 hashing::djb2(sdk_key),
557 ))
558}
559
560pub(crate) fn mmap_lock_path_for_sdk_key(sdk_key: &str) -> PathBuf {
561 std::env::temp_dir()
562 .join(MMAP_DIRECTORY)
563 .join(format!("{}_interned_store.lock", hashing::djb2(sdk_key)))
564}
565
566fn mmap_path_for_sdk_key_version(sdk_key: &str, version: u32) -> PathBuf {
567 std::env::temp_dir().join(MMAP_DIRECTORY).join(format!(
568 "{}_v{version}_interned_store.mmap",
569 hashing::djb2(sdk_key),
570 ))
571}
572
573impl InternedStore {
574 pub(crate) fn get_mmap_returnable_stable_hash(hash: u64) -> Option<u64> {
575 mmap_reader::get_returnable_stable_hash(hash)
576 }
577
578 pub fn get_or_intern_string<T: AsRef<str> + ToString>(value: T) -> InternedString {
579 let hash = hashing::hash_one(value.as_ref().as_bytes());
580
581 if let Some(string) = get_string_from_mmap(hash) {
582 return InternedString::from_static(hash, string);
583 }
584
585 if let Some(string) = get_string_from_shared(hash) {
586 return InternedString::from_static(hash, string);
587 }
588
589 let ptr = get_string_from_local(hash, value);
590 InternedString::from_pointer(hash, ptr)
591 }
592
593 pub fn get_or_intern_owned_string(value: String) -> InternedString {
594 let hash = hashing::hash_one(value.as_bytes());
595
596 if let Some(string) = get_string_from_mmap(hash) {
597 return InternedString::from_static(hash, string);
598 }
599
600 if let Some(string) = get_string_from_shared(hash) {
601 return InternedString::from_static(hash, string);
602 }
603
604 let ptr = get_owned_string_from_local(hash, value);
605 InternedString::from_pointer(hash, ptr)
606 }
607
608 pub fn get_or_intern_returnable(value: Cow<'_, RawValue>) -> DynamicReturnable {
609 let raw_string = value.get();
610 match raw_string {
611 "true" => return DynamicReturnable::from_bool(true),
612 "false" => return DynamicReturnable::from_bool(false),
613 "null" => return DynamicReturnable::empty(),
614 _ => {}
615 }
616
617 let hash = hashing::hash_one(raw_string.as_bytes());
618
619 if let Some(returnable) = get_returnable_from_mmap(hash) {
620 return DynamicReturnable::from_archived(hash, returnable);
621 }
622
623 if let Some(returnable) = get_returnable_from_shared(hash) {
624 return DynamicReturnable::from_static(hash, returnable);
625 }
626
627 let ptr = get_returnable_from_local(hash, value);
628 DynamicReturnable::from_pointer(hash, ptr)
629 }
630
631 pub fn get_or_intern_evaluator_value(value: Cow<'_, RawValue>) -> EvaluatorValue {
632 let raw_string = value.get();
633 let hash = hashing::hash_one(raw_string.as_bytes());
634
635 if let Some(evaluator_value) = get_evaluator_value_from_mmap(hash) {
636 return evaluator_value;
637 }
638
639 if let Some(evaluator_value) = get_evaluator_value_from_shared(hash) {
640 return EvaluatorValue::from_static(hash, evaluator_value);
641 }
642
643 let ptr = get_or_create_evaluator_value_from_local(hash, value);
644 EvaluatorValue::from_pointer(hash, ptr)
645 }
646
647 pub fn replace_evaluator_value(hash: u64, evaluator_value: Arc<MemoizedEvaluatorValue>) {
648 MUTABLE_EVALUATOR_VALUES.replace(hash, &evaluator_value);
649 }
650
651 pub(crate) fn replace_mmap_evaluator_value(
652 hash: u64,
653 evaluator_value: Arc<MemoizedEvaluatorValue>,
654 ) {
655 let has_regex = evaluator_value.regex_value.is_some();
656 MUTABLE_EVALUATOR_VALUES.replace(hash, &evaluator_value);
657 if has_regex {
658 MMAP_EVALUATOR_OVERRIDE_EXISTS.store(true, Ordering::Release);
659 }
660 }
661
662 pub fn try_get_preloaded_evaluator_value(bytes: &[u8]) -> Option<EvaluatorValue> {
663 let hash = hashing::hash_one(bytes);
664 if let Some(evaluator_value) = get_evaluator_value_from_mmap(hash) {
665 return Some(evaluator_value);
666 }
667
668 if let Some(evaluator_value) = get_evaluator_value_from_shared(hash) {
669 return Some(EvaluatorValue::from_static(hash, evaluator_value));
670 }
671
672 None
673 }
674
675 pub fn try_get_preloaded_returnable(bytes: &[u8]) -> Option<DynamicReturnable> {
676 match bytes {
677 b"true" => return Some(DynamicReturnable::from_bool(true)),
678 b"false" => return Some(DynamicReturnable::from_bool(false)),
679 b"null" => return Some(DynamicReturnable::empty()),
680 _ => {}
681 }
682
683 let hash = hashing::hash_one(bytes);
684
685 if let Some(returnable) = get_returnable_from_mmap(hash) {
686 return Some(DynamicReturnable::from_archived(hash, returnable));
687 }
688
689 if let Some(returnable) = get_returnable_from_shared(hash) {
690 return Some(DynamicReturnable::from_static(hash, returnable));
691 }
692
693 None
694 }
695
696 pub fn try_get_preloaded_dynamic_config(name: &InternedString) -> Option<SpecPointer> {
697 if let Some(spec) = get_mmap_spec(MmapSpecKind::DynamicConfig, name.hash) {
698 return Some(SpecPointer::from_mmap(spec));
699 }
700
701 match IMMORTAL_DATA.get() {
702 Some(shared) => shared
703 .dynamic_configs
704 .get(&name.hash)
705 .map(|s| SpecPointer::Static(s)),
706 None => None,
707 }
708 }
709
710 pub fn try_get_preloaded_layer_config(name: &InternedString) -> Option<SpecPointer> {
711 if let Some(spec) = get_mmap_spec(MmapSpecKind::LayerConfig, name.hash) {
712 return Some(SpecPointer::from_mmap(spec));
713 }
714
715 match IMMORTAL_DATA.get() {
716 Some(shared) => shared
717 .layer_configs
718 .get(&name.hash)
719 .map(|s| SpecPointer::Static(s)),
720 None => None,
721 }
722 }
723
724 pub fn try_get_preloaded_feature_gate(name: &InternedString) -> Option<SpecPointer> {
725 if let Some(spec) = get_mmap_spec(MmapSpecKind::FeatureGate, name.hash) {
726 return Some(SpecPointer::from_mmap(spec));
727 }
728
729 match IMMORTAL_DATA.get() {
730 Some(shared) => shared
731 .feature_gates
732 .get(&name.hash)
733 .map(|s| SpecPointer::Static(s)),
734 None => None,
735 }
736 }
737
738 pub(crate) fn try_get_preloaded_spec(
739 name: &InternedString,
740 entity: &str,
741 ) -> Option<SpecPointer> {
742 [
743 MmapSpecKind::FeatureGate,
744 MmapSpecKind::DynamicConfig,
745 MmapSpecKind::LayerConfig,
746 ]
747 .into_iter()
748 .find_map(|kind| {
749 let spec = get_mmap_spec(kind, name.hash)?;
750 (get_string_from_mmap(spec.entity.to_native()) == Some(entity))
751 .then(|| SpecPointer::from_mmap(spec))
752 })
753 }
754
755 pub(crate) fn has_preloaded_mmap_v2() -> bool {
756 mmap_reader::has_v2()
757 }
758 pub(crate) fn get_mmap_string(hash: u64) -> Option<&'static str> {
759 get_string_from_mmap(hash)
760 }
761
762 #[cfg(test)]
763 pub fn get_memoized_len() -> (
764 usize,
765 usize,
766 usize,
767 ) {
768 (
769 MUTABLE_STRINGS.live_len(),
770 MUTABLE_RETURNABLES.live_len(),
771 MUTABLE_EVALUATOR_VALUES.live_len(),
772 )
773 }
774}
775
776fn try_parse_as_json(data: &[u8]) -> Result<SpecsResponseFull, StatsigErr> {
779 serde_json::from_slice(data)
780 .map_err(|e| StatsigErr::JsonParseError(TAG.to_string(), e.to_string()))
781}
782
783fn try_parse_as_proto(data: &[u8]) -> Result<SpecsResponseFull, StatsigErr> {
784 let current = SpecsResponseFull::default();
785 let mut next = SpecsResponseFull::default();
786
787 let mut response_data = ResponseData::from_bytes_with_headers(
788 data.to_vec(),
789 Some(std::collections::HashMap::from([(
790 "content-encoding".to_string(),
791 "statsig-br".to_string(),
792 )])),
793 );
794
795 let ops_stats = OpsStatsForInstance::new();
796
797 deserialize_protobuf(&ops_stats, ¤t, &mut next, &mut response_data)?;
798
799 Ok(next)
800}
801
802fn get_string_from_shared(hash: u64) -> Option<&'static str> {
805 match IMMORTAL_DATA.get() {
806 Some(shared) => shared.strings.get(&hash).copied(),
807 None => None,
808 }
809}
810
811fn get_string_from_local<T: ToString>(hash: u64, value: T) -> Arc<String> {
812 if let Some(string) = MUTABLE_STRINGS.get(hash) {
813 return string;
814 }
815
816 MUTABLE_STRINGS.get_or_insert(hash, Arc::new(value.to_string()))
817}
818
819fn get_owned_string_from_local(hash: u64, value: String) -> Arc<String> {
820 if let Some(string) = MUTABLE_STRINGS.get(hash) {
821 return string;
822 }
823
824 MUTABLE_STRINGS.get_or_insert(hash, Arc::new(value))
825}
826
827fn get_returnable_from_shared(hash: u64) -> Option<&'static HashMap<String, RkyvValue>> {
830 match IMMORTAL_DATA.get() {
831 Some(shared) => shared.returnables.get(&hash).copied(),
832 None => None,
833 }
834}
835
836fn get_returnable_from_local(hash: u64, value: Cow<RawValue>) -> Arc<HashMap<String, RkyvValue>> {
837 if let Some(returnable) = MUTABLE_RETURNABLES.get(hash) {
838 return returnable;
839 }
840
841 let owned: HashMap<String, RkyvValue> = match serde_json::from_str(value.get()) {
842 Ok(owned) => owned,
843 Err(e) => {
844 log_e!(TAG, "Failed to parse returnable from local: {}", e);
845 return Arc::new(HashMap::new());
846 }
847 };
848
849 MUTABLE_RETURNABLES.get_or_insert(hash, Arc::new(owned))
850}
851
852fn get_evaluator_value_from_mmap(hash: u64) -> Option<EvaluatorValue> {
855 let (value, regex) = get_archived_evaluator_value_from_mmap(hash)?;
856
857 if regex.is_none() && MMAP_EVALUATOR_OVERRIDE_EXISTS.load(Ordering::Acquire) {
858 if let Some(value) =
859 get_evaluator_value_from_local(hash).filter(|value| value.regex_value.is_some())
860 {
861 return Some(EvaluatorValue::from_pointer(hash, value));
862 }
863 }
864
865 Some(EvaluatorValue::from_mmap(hash, value, regex))
866}
867
868fn get_evaluator_value_from_shared(hash: u64) -> Option<&'static MemoizedEvaluatorValue> {
869 match IMMORTAL_DATA.get() {
870 Some(shared) => shared.evaluator_values.get(&hash).copied(),
871 None => None,
872 }
873}
874
875fn get_evaluator_value_from_local(hash: u64) -> Option<Arc<MemoizedEvaluatorValue>> {
876 MUTABLE_EVALUATOR_VALUES.get(hash)
877}
878
879fn get_or_create_evaluator_value_from_local(
880 hash: u64,
881 value: Cow<'_, RawValue>,
882) -> Arc<MemoizedEvaluatorValue> {
883 if let Some(evaluator_value) = MUTABLE_EVALUATOR_VALUES.get(hash) {
884 return evaluator_value;
885 }
886
887 let ptr = Arc::new(MemoizedEvaluatorValue::from_raw_value(value));
888 MUTABLE_EVALUATOR_VALUES.get_or_insert(hash, ptr)
889}
890
891fn mutable_to_immortal(
894 specs_responses: Vec<SpecsResponseFull>,
895) -> Result<ImmortalData, StatsigErr> {
896 let mutable_data = take_mutable_data();
897 let mut immortal = ImmortalData::default();
898
899 for (hash, arc) in mutable_data.strings.into_iter() {
900 let raw = Arc::into_raw(arc);
901 let leaked: &'static str = unsafe { &*raw };
902 immortal.strings.insert(hash, leaked);
903 }
904
905 for (hash, returnable) in mutable_data.returnables.into_iter() {
906 let raw_returnable = Arc::into_raw(returnable);
907 let leaked = unsafe { &*raw_returnable };
908 immortal.returnables.insert(hash, leaked);
909 }
910
911 for (hash, evaluator_value) in mutable_data.evaluator_values.into_iter() {
912 let raw_evaluator_value = Arc::into_raw(evaluator_value);
913 let leaked = unsafe { &*raw_evaluator_value };
914 immortal.evaluator_values.insert(hash, leaked);
915 }
916
917 for response in specs_responses {
918 try_insert_specs(response.feature_gates, &mut immortal.feature_gates);
919 try_insert_specs(response.dynamic_configs, &mut immortal.dynamic_configs);
920 try_insert_specs(response.layer_configs, &mut immortal.layer_configs);
921 }
922
923 Ok(immortal)
924}
925
926fn take_mutable_data() -> MutableData {
927 MutableData {
928 strings: MUTABLE_STRINGS.take_live(),
929 returnables: MUTABLE_RETURNABLES.take_live(),
930 evaluator_values: MUTABLE_EVALUATOR_VALUES.take_live(),
931 }
932}
933
934fn try_insert_specs(source: SpecsHashMap, destination: &mut AHashMap<u64, &'static Spec>) {
935 for (name, spec_ptr) in source.0.into_iter() {
936 let Some(spec) = spec_ptr.into_pointer() else {
937 continue;
938 };
939
940 if spec.checksum.is_none() {
941 continue;
943 }
944
945 let raw_spec = Arc::into_raw(spec);
946 let spec = unsafe { &*raw_spec };
947 destination.insert(name.hash, spec);
948 }
949}
950
951impl EvaluatorValue {
954 fn from_mmap(
955 hash: u64,
956 evaluator_value: &'static ArchivedMmapEvaluatorValue,
957 regex: Option<&'static FancyRegex>,
958 ) -> Self {
959 Self {
960 hash,
961 inner: EvaluatorValueInner::Mmap(
962 crate::evaluation::evaluator_value::MmapEvaluatorValueHandle::new(
963 evaluator_value,
964 regex,
965 ),
966 ),
967 }
968 }
969
970 fn from_static(hash: u64, evaluator_value: &'static MemoizedEvaluatorValue) -> Self {
971 Self {
972 hash,
973 inner: EvaluatorValueInner::Static(evaluator_value),
974 }
975 }
976
977 fn from_pointer(hash: u64, pointer: Arc<MemoizedEvaluatorValue>) -> Self {
978 Self {
979 hash,
980 inner: EvaluatorValueInner::Pointer(pointer),
981 }
982 }
983}
984
985impl DynamicReturnable {
986 fn from_static(hash: u64, returnable: &'static HashMap<String, RkyvValue>) -> Self {
987 Self::from_interned_value(hash, DynamicReturnableValue::JsonStatic(returnable))
988 }
989
990 fn from_archived(
991 hash: u64,
992 returnable: &'static ArchivedHashMap<ArchivedString, ArchivedRkyvValue>,
993 ) -> Self {
994 Self::from_archived_value(hash, returnable)
995 }
996
997 fn from_pointer(hash: u64, pointer: Arc<HashMap<String, RkyvValue>>) -> Self {
998 Self::from_interned_value(hash, DynamicReturnableValue::JsonPointer(pointer))
999 }
1000}
1001
1002impl InternedString {
1003 pub(crate) fn from_static(hash: u64, string: &'static str) -> Self {
1004 Self {
1005 hash,
1006 value: InternedStringValue::Static(string),
1007 }
1008 }
1009
1010 fn from_pointer(hash: u64, pointer: Arc<String>) -> Self {
1011 Self {
1012 hash,
1013 value: InternedStringValue::Pointer(pointer),
1014 }
1015 }
1016}
1017
1018#[cfg(test)]
1019mod mutable_store_tests {
1020 use super::MutableStore;
1021 use std::sync::{Arc, Barrier};
1022
1023 #[test]
1024 fn concurrent_insertions_share_one_live_value() {
1025 const THREAD_COUNT: usize = 16;
1026 let store = Arc::new(MutableStore::<String>::default());
1027 let start = Arc::new(Barrier::new(THREAD_COUNT));
1028
1029 let threads = (0..THREAD_COUNT)
1030 .map(|_| {
1031 let store = Arc::clone(&store);
1032 let start = Arc::clone(&start);
1033 std::thread::spawn(move || {
1034 start.wait();
1035 store.get_or_insert(42, Arc::new("shared".to_owned()))
1036 })
1037 })
1038 .collect::<Vec<_>>();
1039 let handles = threads
1040 .into_iter()
1041 .map(|thread| thread.join().expect("interner thread should finish"))
1042 .collect::<Vec<_>>();
1043
1044 assert!(handles.iter().all(|value| Arc::ptr_eq(value, &handles[0])));
1045 }
1046
1047 #[test]
1048 fn dead_value_is_replaced() {
1049 let store = MutableStore::<String>::default();
1050 let first = store.get_or_insert(42, Arc::new("first".to_owned()));
1051 let weak_first = Arc::downgrade(&first);
1052 drop(first);
1053
1054 assert!(store.get(42).is_none());
1055
1056 let second = store.get_or_insert(42, Arc::new("second".to_owned()));
1057 assert!(weak_first.upgrade().is_none());
1058 assert_eq!(second.as_str(), "second");
1059 assert_eq!(store.live_len(), 1);
1060 }
1061
1062 #[test]
1063 fn dead_entries_are_bounded_by_periodic_sweeps() {
1064 let store = MutableStore::<String>::default();
1065
1066 for hash in 0..super::WEAK_STORE_SWEEP_INTERVAL as u64 {
1067 drop(store.get_or_insert(hash, Arc::new(hash.to_string())));
1068 }
1069
1070 assert_eq!(store.live_len(), 0);
1071 assert_eq!(store.stored_len(), 1);
1072 }
1073
1074 #[test]
1075 fn take_live_retires_the_old_table() {
1076 let store = MutableStore::<String>::default();
1077 let live = store.get_or_insert(1, Arc::new("live".to_owned()));
1078 drop(store.get_or_insert(2, Arc::new("dead".to_owned())));
1079
1080 assert!(store.stored_capacity() > 0);
1081 let taken = store.take_live();
1082
1083 assert!(Arc::ptr_eq(
1084 &taken.iter().find(|(hash, _)| *hash == 1).unwrap().1,
1085 &live
1086 ));
1087 assert!(!taken.iter().any(|(hash, _)| *hash == 2));
1088 assert_eq!(store.stored_len(), 0);
1089 assert_eq!(store.stored_capacity(), 0);
1090 }
1091
1092 #[test]
1093 fn take_live_releases_weak_metadata_before_a_batch_drop() {
1094 const BATCH_SIZE: u64 = 1024;
1095
1096 let store = MutableStore::<String>::default();
1097 let handles = (0..BATCH_SIZE)
1098 .map(|hash| store.get_or_insert(hash, Arc::new(hash.to_string())))
1099 .collect::<Vec<_>>();
1100 assert!(store.stored_capacity() >= BATCH_SIZE as usize);
1101
1102 let taken = store.take_live();
1103
1104 assert_eq!(taken.len(), BATCH_SIZE as usize);
1105 assert_eq!(store.stored_len(), 0);
1106 assert_eq!(store.stored_capacity(), 0);
1107
1108 drop(handles);
1109 assert!(taken.iter().all(|(_, value)| Arc::strong_count(value) == 1));
1110 drop(taken);
1111 assert_eq!(store.stored_len(), 0);
1112 }
1113}