feat: delta api implementation

server/Cargo.toml

@@ -95,7 +95,7 @@ tracing = { version = "0.1.41", features = ["log"] }
 tracing-subscriber = { version = "0.3.19", features = ["json", "env-filter"] }
 ulid = "1.1.3"
 unleash-types = { version = "0.15.3", features = ["openapi", "hashes"] }
-unleash-yggdrasil = { version = "0.14.3" }
+unleash-yggdrasil = { version = "0.14.5" }
 utoipa = { version = "5", features = ["actix_extras", "chrono"] }
 utoipa-swagger-ui = { version = "8", features = ["actix-web"] }
 [dev-dependencies]

server/src/builder.rs

@@ -270,6 +270,7 @@ async fn build_edge(
         Duration::seconds(args.features_refresh_interval_seconds as i64),
         refresher_mode,
         client_meta_information,
+        args.delta,
     );
     let feature_refresher = Arc::new(FeatureRefresher::new(
         unleash_client,
@@ -385,6 +386,7 @@ mod tests {
             prometheus_password: None,
             prometheus_username: None,
             streaming: false,
+            delta: false,
         };
 
         let result = build_edge(

server/src/cli.rs

@@ -217,6 +217,10 @@ pub struct EdgeArgs {
     #[clap(long, env, default_value_t = false, requires = "strict")]
     pub streaming: bool,
 
+    /// If set to true. Edge connects to upstream using delta polling instead of normal polling.
+    #[clap(long, env, default_value_t = false, conflicts_with = "streaming")]
+    pub delta: bool,
+
     /// Sets a remote write url for prometheus metrics, if this is set, prometheus metrics will be written upstream
     #[clap(long, env)]
     pub prometheus_remote_write_url: Option<String>,

server/src/client_api.rs

@@ -1012,6 +1012,7 @@ mod tests {
             strict: false,
             streaming: false,
             client_meta_information: ClientMetaInformation::test_config(),
+            delta: false,
         });
         let token_validator = Arc::new(TokenValidator {
             unleash_client: unleash_client.clone(),

server/src/feature_cache.rs

@@ -4,7 +4,7 @@ use unleash_types::{
     client_features::{ClientFeature, ClientFeatures, Segment},
     Deduplicate,
 };
-
+use unleash_types::client_features::ClientFeaturesDelta;
 use crate::types::EdgeToken;
 
 #[derive(Debug, Clone)]
@@ -67,6 +67,23 @@ impl FeatureCache {
         self.send_full_update(key);
     }
 
+    pub fn apply_delta(&self, key: String, delta: &ClientFeaturesDelta) {
+        let client_features = ClientFeatures {
+            version : 1,
+            features : delta.updated.clone(),
+            segments: delta.segments.clone(),
+            query: None,
+            meta: None,
+        };
+        self.features
+            .entry(key.clone())
+            .and_modify(|existing_features| {
+                existing_features.modify_in_place(delta);
+            })
+            .or_insert(client_features);
+        self.send_full_update(key);
+    }
+
     pub fn is_empty(&self) -> bool {
         self.features.is_empty()
     }

server/src/http/feature_refresher.rs

@@ -8,7 +8,7 @@ use eventsource_client::Client;
 use futures::TryStreamExt;
 use reqwest::StatusCode;
 use tracing::{debug, info, warn};
-use unleash_types::client_features::ClientFeatures;
+use unleash_types::client_features::{ClientFeatures, ClientFeaturesDelta};
 use unleash_types::client_metrics::{ClientApplication, MetricsMetadata};
 use unleash_yggdrasil::EngineState;
 
@@ -18,7 +18,9 @@ use crate::filters::{filter_client_features, FeatureFilterSet};
 use crate::http::headers::{
     UNLEASH_APPNAME_HEADER, UNLEASH_CLIENT_SPEC_HEADER, UNLEASH_INSTANCE_ID_HEADER,
 };
-use crate::types::{build, EdgeResult, TokenType, TokenValidationStatus};
+use crate::types::{
+    build, ClientFeaturesDeltaResponse, EdgeResult, TokenType, TokenValidationStatus,
+};
 use crate::{
     persistence::EdgePersistence,
     tokens::{cache_key, simplify},
@@ -49,6 +51,7 @@ pub struct FeatureRefresher {
     pub strict: bool,
     pub streaming: bool,
     pub client_meta_information: ClientMetaInformation,
+    pub delta: bool,
 }
 
 impl Default for FeatureRefresher {
@@ -63,6 +66,7 @@ impl Default for FeatureRefresher {
             strict: true,
             streaming: false,
             client_meta_information: Default::default(),
+            delta: false,
         }
     }
 }
@@ -100,18 +104,21 @@ pub struct FeatureRefreshConfig {
     features_refresh_interval: chrono::Duration,
     mode: FeatureRefresherMode,
     client_meta_information: ClientMetaInformation,
+    delta: bool,
 }
 
 impl FeatureRefreshConfig {
     pub fn new(
         features_refresh_interval: chrono::Duration,
         mode: FeatureRefresherMode,
         client_meta_information: ClientMetaInformation,
+        delta: bool,
     ) -> Self {
         Self {
             features_refresh_interval,
             mode,
             client_meta_information,
+            delta,
         }
     }
 }
@@ -134,6 +141,7 @@ impl FeatureRefresher {
             strict: config.mode != FeatureRefresherMode::Dynamic,
             streaming: config.mode == FeatureRefresherMode::Streaming,
             client_meta_information: config.client_meta_information,
+            delta: config.delta,
         }
     }
 
@@ -404,13 +412,21 @@ impl FeatureRefresher {
     pub async fn hydrate_new_tokens(&self) {
         let hydrations = self.get_tokens_never_refreshed();
         for hydration in hydrations {
-            self.refresh_single(hydration).await;
+            if self.delta {
+                self.refresh_single_delta(hydration).await;
+            } else {
+                self.refresh_single(hydration).await;
+            }
         }
     }
     pub async fn refresh_features(&self) {
         let refreshes = self.get_tokens_due_for_refresh();
         for refresh in refreshes {
-            self.refresh_single(refresh).await;
+            if self.delta {
+                self.refresh_single_delta(refresh).await;
+            } else {
+                self.refresh_single(refresh).await;
+            }
         }
     }
 
@@ -449,6 +465,36 @@ impl FeatureRefresher {
                         });
     }
 
+    async fn handle_client_features_delta_updated(
+        &self,
+        refresh_token: &EdgeToken,
+        delta: ClientFeaturesDelta,
+        etag: Option<EntityTag>,
+    ) {
+        debug!("Got updated client features delta. Updating features with {etag:?}");
+        let key = cache_key(refresh_token);
+        self.features_cache.apply_delta(key.clone(), &delta);
+        self.update_last_refresh(
+            refresh_token,
+            etag,
+            self.features_cache.get(&key).unwrap().features.len(),
+        );
+        self.engine_cache
+            .entry(key.clone())
+            .and_modify(|engine| {
+                engine.take_delta(&delta);
+            })
+            .or_insert_with(|| {
+                let mut new_state = EngineState::default();
+
+                let warnings = new_state.take_delta(&delta);
+                if let Some(warnings) = warnings {
+                    warn!("The following toggle failed to compile and will be defaulted to off: {warnings:?}");
+                };
+                new_state
+            });
+    }
+
     pub async fn refresh_single(&self, refresh: TokenRefresh) {
         let features_result = self
             .unleash_client
@@ -515,6 +561,72 @@ impl FeatureRefresher {
             }
         }
     }
+
+    pub async fn refresh_single_delta(&self, refresh: TokenRefresh) {
+        let features_result = self
+            .unleash_client
+            .get_client_features_delta(ClientFeaturesRequest {
+                api_key: refresh.token.token.clone(),
+                etag: refresh.etag,
+            })
+            .await;
+        match features_result {
+            Ok(delta_response) => match delta_response {
+                ClientFeaturesDeltaResponse::NoUpdate(tag) => {
+                    debug!("No update needed. Will update last check time with {tag}");
+                    self.update_last_check(&refresh.token.clone());
+                }
+                ClientFeaturesDeltaResponse::Updated(features, etag) => {
+                    self.handle_client_features_delta_updated(&refresh.token, features, etag)
+                        .await
+                }
+            },
+            Err(e) => {
+                match e {
+                    EdgeError::ClientFeaturesFetchError(fe) => {
+                        match fe {
+                            FeatureError::Retriable(status_code) => match status_code {
+                                StatusCode::INTERNAL_SERVER_ERROR
+                                | StatusCode::BAD_GATEWAY
+                                | StatusCode::SERVICE_UNAVAILABLE
+                                | StatusCode::GATEWAY_TIMEOUT => {
+                                    info!("Upstream is having some problems, increasing my waiting period");
+                                    self.backoff(&refresh.token);
+                                }
+                                StatusCode::TOO_MANY_REQUESTS => {
+                                    info!("Got told that upstream is receiving too many requests");
+                                    self.backoff(&refresh.token);
+                                }
+                                _ => {
+                                    info!("Couldn't refresh features, but will retry next go")
+                                }
+                            },
+                            FeatureError::AccessDenied => {
+                                info!("Token used to fetch features was Forbidden, will remove from list of refresh tasks");
+                                self.tokens_to_refresh.remove(&refresh.token.token);
+                                if !self.tokens_to_refresh.iter().any(|e| {
+                                    e.value().token.environment == refresh.token.environment
+                                }) {
+                                    let cache_key = cache_key(&refresh.token);
+                                    // No tokens left that access the environment of our current refresh. Deleting client features and engine cache
+                                    self.features_cache.remove(&cache_key);
+                                    self.engine_cache.remove(&cache_key);
+                                }
+                            }
+                            FeatureError::NotFound => {
+                                info!("Had a bad URL when trying to fetch features. Increasing waiting period for the token before trying again");
+                                self.backoff(&refresh.token);
+                            }
+                        }
+                    }
+                    EdgeError::ClientCacheError => {
+                        info!("Couldn't refresh features, but will retry next go")
+                    }
+                    _ => info!("Couldn't refresh features: {e:?}. Will retry next pass"),
+                }
+            }
+        }
+    }
     pub fn backoff(&self, token: &EdgeToken) {
         self.tokens_to_refresh
             .alter(&token.token, |_k, old_refresh| {
@@ -1590,4 +1702,12 @@ mod tests {
         );
         assert_eq!(updated.len(), 0);
     }
+
+    // mod delta_tests {
+    //     use super::*;
+    //     #[test]
+    //     fn delta_works() {
+    //
+    //     }
+    // }
 }