15441100167

Committed 04 Jun 2025 11:28AM UTC coverage: 78.265% (+10.3%) from 67.995%

Build # 15441100167

Build Type

Pull #970

github

Committed by

web-flow

Commit Message

Merge 8ad457ed6 into 34ad3228b

Pull Request Pull Request #970: task(rust): Update Rust version to 1.87.0

Run Details

10140 of 12956 relevant lines covered (78.26%)

158.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.96

/server/src/feature_cache.rs

use crate::types::EdgeToken;
use dashmap::DashMap;
use tokio::sync::broadcast;
use unleash_types::client_features::ClientFeaturesDelta;
use unleash_types::{
    Deduplicate,
    client_features::{ClientFeature, ClientFeatures, Segment},
};

#[derive(Debug, Clone)]
pub enum UpdateType {
    Full(String),
    Update(String),
    Deletion,
}

#[derive(Debug, Clone)]
pub struct FeatureCache {
    features: DashMap<String, ClientFeatures>,
    update_sender: broadcast::Sender<UpdateType>,
}

impl FeatureCache {
    pub fn new(features: DashMap<String, ClientFeatures>) -> Self {
        let (tx, _rx) = tokio::sync::broadcast::channel::<UpdateType>(16);
        Self {
            features,
            update_sender: tx,
        }
    }

    pub fn len(&self) -> usize {
        self.features.len()
    }

    pub fn subscribe(&self) -> broadcast::Receiver<UpdateType> {
        self.update_sender.subscribe()
    }
    pub fn get(&self, key: &str) -> Option<dashmap::mapref::one::Ref<'_, String, ClientFeatures>> {
        self.features.get(key)
    }

    pub fn insert(&self, key: String, features: ClientFeatures) -> Option<ClientFeatures> {
        let v = self.features.insert(key.clone(), features);
        self.send_full_update(key);
        v
    }

    pub fn send_full_update(&self, cache_key: String) {
        let _ = self.update_sender.send(UpdateType::Full(cache_key));
    }

    pub fn remove(&self, key: &str) -> Option<(String, ClientFeatures)> {
        let v = self.features.remove(key);
        self.send_full_update(key.to_string());
        v
    }

    pub fn modify(&self, key: String, token: &EdgeToken, features: ClientFeatures) {
        self.features
            .entry(key.clone())
            .and_modify(|existing_features| {
                let updated = update_client_features(token, existing_features, &features);
                *existing_features = updated;
            })
            .or_insert(features);
        self.send_full_update(key);
    }

    pub fn apply_delta(&self, key: String, delta: &ClientFeaturesDelta) {
        self.features
            .entry(key.clone())
            .and_modify(|existing_features| {
                existing_features.apply_delta(delta);
            })
            .or_insert(ClientFeatures::create_from_delta(delta));
        self.send_full_update(key);
    }

    pub fn is_empty(&self) -> bool {
        self.features.is_empty()
    }

    pub fn iter(&self) -> dashmap::iter::Iter<'_, String, ClientFeatures> {
        self.features.iter()
    }
}

impl Default for FeatureCache {
    fn default() -> Self {
        FeatureCache::new(DashMap::default())
    }
}

fn update_client_features(
    token: &EdgeToken,
    old: &ClientFeatures,
    update: &ClientFeatures,
) -> ClientFeatures {
    let mut updated_features =
        update_projects_from_feature_update(token, &old.features, &update.features);
    updated_features.sort();
    let segments = merge_segments_update(old.segments.clone(), update.segments.clone());
    ClientFeatures {
        version: old.version.max(update.version),
        features: updated_features,
        segments: segments.map(|mut s| {
            s.sort();
            s
        }),
        query: old.query.clone().or(update.query.clone()),
        meta: old.meta.clone().or(update.meta.clone()),
    }
}

pub(crate) fn update_projects_from_feature_update(
    token: &EdgeToken,
    original: &[ClientFeature],
    updated: &[ClientFeature],
) -> Vec<ClientFeature> {
    let projects_to_update = &token.projects;
    if projects_to_update.contains(&"*".into()) {
        updated.into()
    } else {
        let mut to_keep: Vec<ClientFeature> = original
            .iter()
            .filter(|toggle| {
                let p = toggle.project.clone().unwrap_or_else(|| "default".into());
                !projects_to_update.contains(&p)
            })
            .cloned()
            .collect();
        to_keep.extend(updated.iter().cloned());
        to_keep
    }
}

fn merge_segments_update(
    segments: Option<Vec<Segment>>,
    updated_segments: Option<Vec<Segment>>,
) -> Option<Vec<Segment>> {
    match (segments, updated_segments) {
        (Some(s), Some(mut o)) => {
            o.extend(s);
            Some(o.deduplicate())
        }
        (Some(s), None) => Some(s),
        (None, Some(o)) => Some(o),
        (None, None) => None,
    }
}

1	use crate::types::EdgeToken;
2	use dashmap::DashMap;
3	use tokio::sync::broadcast;
4	use unleash_types::client_features::ClientFeaturesDelta;
5	use unleash_types::{
6	Deduplicate,
7	client_features::{ClientFeature, ClientFeatures, Segment},
8	};
9
10	#[derive(Debug, Clone)]
11	pub enum UpdateType {
12	Full(String),
13	Update(String),
14	Deletion,
15	}
16
17	#[derive(Debug, Clone)]
18	pub struct FeatureCache {
19	features: DashMap<String, ClientFeatures>,
20	update_sender: broadcast::Sender<UpdateType>,
21	}
22
23	impl FeatureCache {
24	pub fn new(features: DashMap<String, ClientFeatures>) -> Self {	97✔
25	let (tx, _rx) = tokio::sync::broadcast::channel::<UpdateType>(16);	97✔
26	Self {	97✔
27	features,	97✔
28	update_sender: tx,	97✔
29	}	97✔
30	}	97✔
31
32	pub fn len(&self) -> usize {	4✔
33	self.features.len()	4✔
34	}	4✔
35
36	pub fn subscribe(&self) -> broadcast::Receiver<UpdateType> {	×
37	self.update_sender.subscribe()	×
38	}	×
39	pub fn get(&self, key: &str) -> Option<dashmap::mapref::one::Ref<'_, String, ClientFeatures>> {	63✔
40	self.features.get(key)	63✔
41	}	63✔
42
43	pub fn insert(&self, key: String, features: ClientFeatures) -> Option<ClientFeatures> {	46✔
44	let v = self.features.insert(key.clone(), features);	46✔
45	self.send_full_update(key);	46✔
46	v	46✔
47	}	46✔
48
49	pub fn send_full_update(&self, cache_key: String) {	65✔
50	let _ = self.update_sender.send(UpdateType::Full(cache_key));	65✔
51	}	65✔
52
53	pub fn remove(&self, key: &str) -> Option<(String, ClientFeatures)> {	2✔
54	let v = self.features.remove(key);	2✔
55	self.send_full_update(key.to_string());	2✔
56	v	2✔
57	}	2✔
58
59	pub fn modify(&self, key: String, token: &EdgeToken, features: ClientFeatures) {	15✔
60	self.features	15✔
61	.entry(key.clone())	15✔
62	.and_modify(\|existing_features\| {	15✔
63	let updated = update_client_features(token, existing_features, &features);	6✔
64	*existing_features = updated;	6✔
65	})	15✔
66	.or_insert(features);	15✔
67	self.send_full_update(key);	15✔
68	}	15✔
69
70	pub fn apply_delta(&self, key: String, delta: &ClientFeaturesDelta) {	2✔
71	self.features	2✔
72	.entry(key.clone())	2✔
73	.and_modify(\|existing_features\| {	2✔
74	existing_features.apply_delta(delta);	1✔
75	})	2✔
76	.or_insert(ClientFeatures::create_from_delta(delta));	2✔
77	self.send_full_update(key);	2✔
78	}	2✔
79
80	pub fn is_empty(&self) -> bool {	9✔
81	self.features.is_empty()	9✔
82	}	9✔
83
84	pub fn iter(&self) -> dashmap::iter::Iter<'_, String, ClientFeatures> {	×
85	self.features.iter()	×
86	}	×
87	}
88
89	impl Default for FeatureCache {
90	fn default() -> Self {	78✔
91	FeatureCache::new(DashMap::default())	78✔
92	}	78✔
93	}
94
95	fn update_client_features(	6✔
96	token: &EdgeToken,	6✔
97	old: &ClientFeatures,	6✔
98	update: &ClientFeatures,	6✔
99	) -> ClientFeatures {	6✔
100	let mut updated_features =	6✔
101	update_projects_from_feature_update(token, &old.features, &update.features);	6✔
102	updated_features.sort();	6✔
103	let segments = merge_segments_update(old.segments.clone(), update.segments.clone());	6✔
104	ClientFeatures {	6✔
105	version: old.version.max(update.version),	6✔
106	features: updated_features,	6✔
107	segments: segments.map(\|mut s\| {	6✔
108	s.sort();	×
109	s	×
110	}),	6✔
111	query: old.query.clone().or(update.query.clone()),	6✔
112	meta: old.meta.clone().or(update.meta.clone()),	6✔
113	}	6✔
114	}	6✔
115
116	pub(crate) fn update_projects_from_feature_update(	13✔
117	token: &EdgeToken,	13✔
118	original: &[ClientFeature],	13✔
119	updated: &[ClientFeature],	13✔
120	) -> Vec<ClientFeature> {	13✔
121	let projects_to_update = &token.projects;	13✔
122	if projects_to_update.contains(&"*".into()) {	13✔
123	updated.into()	1✔
124	} else {
125	let mut to_keep: Vec<ClientFeature> = original	12✔
126	.iter()	12✔
127	.filter(\|toggle\| {	194✔
128	let p = toggle.project.clone().unwrap_or_else(\|\| "default".into());	194✔
129	!projects_to_update.contains(&p)	194✔
130	})	194✔
131	.cloned()	12✔
132	.collect();	12✔
133	to_keep.extend(updated.iter().cloned());	12✔
134	to_keep	12✔
135	}
136	}	13✔
137
138	fn merge_segments_update(	6✔
139	segments: Option<Vec<Segment>>,	6✔
140	updated_segments: Option<Vec<Segment>>,	6✔
141	) -> Option<Vec<Segment>> {	6✔
142	match (segments, updated_segments) {	6✔
143	(Some(s), Some(mut o)) => {	×
144	o.extend(s);	×
145	Some(o.deduplicate())	×
146	}
147	(Some(s), None) => Some(s),	×
148	(None, Some(o)) => Some(o),	×
149	(None, None) => None,	6✔
150	}
151	}	6✔

Unleash / unleash-edge / 15441100167

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous