feat: fastly compute service to perform assignments (FF-3552)

.github/workflows/ci.yml

@@ -23,6 +23,12 @@ jobs:
           submodules: true
       - run: npm ci
       - run: rustup update ${{ matrix.toolchain }} && rustup default ${{ matrix.toolchain }}
-      - run: cargo build --verbose --all-targets
-      - run: cargo test --verbose
+      # Add WASM target
+      - run: rustup target add wasm32-wasi
+      # Build non-WASM targets
+      - run: cargo build --verbose --all-targets --workspace --exclude fastly-edge-assignments
+      # Build WASM target separately
+      - run: cargo build --verbose -p fastly-edge-assignments --target wasm32-wasi
+      # Run tests (excluding WASM package)
+      - run: cargo test --verbose --workspace --exclude fastly-edge-assignments
       - run: cargo doc --verbose

Cargo.toml

@@ -1,10 +1,11 @@
 [workspace]
 resolver = "2"
 members = [
-  "eppo_core",
-  "rust-sdk",
-  "python-sdk",
-  "ruby-sdk/ext/eppo_client",
+    "eppo_core",
+    "rust-sdk",
+    "python-sdk",
+    "ruby-sdk/ext/eppo_client",
+    "fastly-edge-assignments",
 ]
 
 [patch.crates-io]

Makefile

@@ -1,28 +1,47 @@
-# Make settings - @see https://tech.davis-hansson.com/p/make/
 SHELL := bash
 .ONESHELL:
 .SHELLFLAGS := -eu -o pipefail -c
 .DELETE_ON_ERROR:
 MAKEFLAGS += --warn-undefined-variables
 MAKEFLAGS += --no-builtin-rules
 
-# Log levels
-DEBUG := $(shell printf "\e[2D\e[35m")
-INFO  := $(shell printf "\e[2D\e[36m🔵 ")
-OK    := $(shell printf "\e[2D\e[32m🟢 ")
-WARN  := $(shell printf "\e[2D\e[33m🟡 ")
-ERROR := $(shell printf "\e[2D\e[31m🔴 ")
-END   := $(shell printf "\e[0m")
+WASM_TARGET=wasm32-wasi
+FASTLY_PACKAGE=fastly-edge-assignments
+BUILD_DIR=target/$(WASM_TARGET)/release
+WASM_FILE=$(BUILD_DIR)/$(FASTLY_PACKAGE).wasm
 
-.PHONY: default
-default: help
-
-## help - Print help message.
+# Help target for easy documentation
 .PHONY: help
-help: Makefile
-	@echo "usage: make <target>"
-	@sed -n 's/^##//p' $<
+help:
+	@echo "Available targets:"
+	@echo "  all                           - Default target (build workspace)"
+	@echo "  workspace-build               - Build the entire workspace excluding the Fastly package"
+	@echo "  workspace-test                - Test the entire workspace excluding the Fastly package"
+	@echo "  fastly-edge-assignments-build - Build only the Fastly package for WASM"
+	@echo "  fastly-edge-assignments-test  - Test only the Fastly package"
+	@echo "  clean                         - Clean all build artifacts"
 
 .PHONY: test
 test: ${testDataDir}
 	npm test
+
+# Build the entire workspace excluding the `fastly-edge-assignments` package
+.PHONY: workspace-build
+workspace-build:
+	cargo build --workspace --exclude $(FASTLY_PACKAGE)
+
+# Run tests for the entire workspace excluding the `fastly-edge-assignments` package
+.PHONY: workspace-test
+workspace-test:
+	cargo test --workspace --exclude $(FASTLY_PACKAGE)
+
+# Build only the `fastly-edge-assignments` package for WASM
+.PHONY: fastly-edge-assignments-build
+fastly-edge-assignments-build:
+	rustup target add $(WASM_TARGET)
+	cargo build --release --target $(WASM_TARGET) --package $(FASTLY_PACKAGE)
+
+# Test only the `fastly-edge-assignments` package
+.PHONY: fastly-edge-assignments-test
+fastly-edge-assignments-test:
+	cargo test --target $(WASM_TARGET) --package $(FASTLY_PACKAGE)

eppo_core/Cargo.toml

@@ -26,11 +26,12 @@ log = { version = "0.4.21", features = ["kv", "kv_serde"] }
 md5 = "0.7.0"
 rand = "0.8.5"
 regex = "1.10.4"
-reqwest = { version = "0.12.4", features = ["blocking", "json"] }
+reqwest = { version = "0.12.4", features = ["json"] }
 semver = { version = "1.0.22", features = ["serde"] }
 serde = { version = "1.0.198", features = ["derive", "rc"] }
 serde_json = "1.0.116"
 thiserror = "1.0.60"
+tokio = { version = "1.34.0", features = ["rt", "time"] }
 url = "2.5.0"
 
 # pyo3 dependencies

eppo_core/src/configuration_fetcher.rs

@@ -20,7 +20,7 @@ const BANDIT_ENDPOINT: &'static str = "/flag-config/v1/bandits";
 /// A client that fetches Eppo configuration from the server.
 pub struct ConfigurationFetcher {
     // Client holds a connection pool internally, so we're reusing the client between requests.
-    client: reqwest::blocking::Client,
+    client: reqwest::Client,
     config: ConfigurationFetcherConfig,
     /// If we receive a 401 Unauthorized error during a request, it means the API key is not
     /// valid. We cache this error so we don't issue additional requests to the server.
@@ -29,7 +29,7 @@ pub struct ConfigurationFetcher {
 
 impl ConfigurationFetcher {
     pub fn new(config: ConfigurationFetcherConfig) -> ConfigurationFetcher {
-        let client = reqwest::blocking::Client::new();
+        let client = reqwest::Client::new();
 
         ConfigurationFetcher {
             client,
@@ -38,24 +38,24 @@ impl ConfigurationFetcher {
         }
     }
 
-    pub fn fetch_configuration(&mut self) -> Result<Configuration> {
+    pub async fn fetch_configuration(&mut self) -> Result<Configuration> {
         if self.unauthorized {
             return Err(Error::Unauthorized);
         }
 
-        let ufc = self.fetch_ufc_configuration()?;
+        let ufc = self.fetch_ufc_configuration().await?;
 
         let bandits = if ufc.compiled.flag_to_bandit_associations.is_empty() {
             // We don't need bandits configuration if there are no bandits.
             None
         } else {
-            Some(self.fetch_bandits_configuration()?)
+            Some(self.fetch_bandits_configuration().await?)
         };
 
         Ok(Configuration::from_server_response(ufc, bandits))
     }
 
-    fn fetch_ufc_configuration(&mut self) -> Result<UniversalFlagConfig> {
+    async fn fetch_ufc_configuration(&mut self) -> Result<UniversalFlagConfig> {
         let url = Url::parse_with_params(
             &format!("{}{}", self.config.base_url, UFC_ENDPOINT),
             &[
@@ -68,7 +68,7 @@ impl ConfigurationFetcher {
         .map_err(|err| Error::InvalidBaseUrl(err))?;
 
         log::debug!(target: "eppo", "fetching UFC flags configuration");
-        let response = self.client.get(url).send()?;
+        let response = self.client.get(url).send().await?;
 
         let response = response.error_for_status().map_err(|err| {
             if err.status() == Some(StatusCode::UNAUTHORIZED) {
@@ -82,15 +82,17 @@ impl ConfigurationFetcher {
             }
         })?;
 
-        let configuration =
-            UniversalFlagConfig::from_json(self.config.sdk_metadata, response.bytes()?.into())?;
+        let configuration = UniversalFlagConfig::from_json(
+            self.config.sdk_metadata,
+            response.bytes().await?.into(),
+        )?;
 
         log::debug!(target: "eppo", "successfully fetched UFC flags configuration");
 
         Ok(configuration)
     }
 
-    fn fetch_bandits_configuration(&mut self) -> Result<BanditResponse> {
+    async fn fetch_bandits_configuration(&mut self) -> Result<BanditResponse> {
         let url = Url::parse_with_params(
             &format!("{}{}", self.config.base_url, BANDIT_ENDPOINT),
             &[
@@ -103,7 +105,7 @@ impl ConfigurationFetcher {
         .map_err(|err| Error::InvalidBaseUrl(err))?;
 
         log::debug!(target: "eppo", "fetching UFC bandits configuration");
-        let response = self.client.get(url).send()?;
+        let response = self.client.get(url).send().await?;
 
         let response = response.error_for_status().map_err(|err| {
             if err.status() == Some(StatusCode::UNAUTHORIZED) {
@@ -117,7 +119,7 @@ impl ConfigurationFetcher {
             }
         })?;
 
-        let configuration = response.json()?;
+        let configuration = response.json().await?;
 
         log::debug!(target: "eppo", "successfully fetched UFC bandits configuration");
 

eppo_core/src/eval/eval_assignment.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 
 use chrono::{DateTime, Utc};
 
@@ -91,6 +91,56 @@ pub fn get_assignment_details(
     (result_with_details, event)
 }
 
+pub fn get_subject_assignments(
+    configuration: Option<&Configuration>,
+    subject_key: &Str,
+    subject_attributes: &Arc<Attributes>,
+    now: DateTime<Utc>,
+) -> HashMap<String, Result<Option<Assignment>, EvaluationError>> {
+    get_assignments_with_visitor(
+        configuration,
+        &mut NoopEvalVisitor,
+        subject_key,
+        subject_attributes,
+        now,
+    )
+}
+
+/**
+ * Evaluate all flags for the given subject and return a map of flag keys to assignments.
+ *
+ * If an error occurs while evaluating a flag,
+ * it is returned for the resulting flag key.
+ */
+pub(super) fn get_assignments_with_visitor<V: EvalAssignmentVisitor>(
+    configuration: Option<&Configuration>,
+    visitor: &mut V,
+    subject_key: &Str,
+    subject_attributes: &Arc<Attributes>,
+    now: DateTime<Utc>,
+) -> HashMap<String, Result<Option<Assignment>, EvaluationError>> {
+    configuration.map_or_else(HashMap::new, |config| {
+        config
+            .flags
+            .compiled
+            .flags
+            .keys()
+            .map(|flag_key| {
+                let result = get_assignment_with_visitor(
+                    Some(&config),
+                    visitor,
+                    flag_key,
+                    subject_key,
+                    subject_attributes,
+                    None,
+                    now,
+                );
+                (flag_key.clone(), result)
+            })
+            .collect()
+    })
+}
+
 // Exposed for use in bandit evaluation.
 pub(super) fn get_assignment_with_visitor<V: EvalAssignmentVisitor>(
     configuration: Option<&Configuration>,
@@ -317,7 +367,7 @@ mod tests {
             eval_details::{
                 AllocationEvaluationCode, AllocationEvaluationDetails, FlagEvaluationCode,
             },
-            get_assignment, get_assignment_details,
+            get_assignment, get_assignment_details, get_subject_assignments,
         },
         ufc::{RuleWire, UniversalFlagConfig, ValueWire, VariationType},
         Attributes, Configuration, SdkMetadata, Str,
@@ -451,6 +501,8 @@ mod tests {
 
             for subject in test_file.subjects {
                 print!("test subject {:?} ... ", subject.subject_key);
+
+                // Verify that get_assignment returns the desired result.
                 let result = get_assignment(
                     Some(&config),
                     &test_file.flag,
@@ -461,15 +513,33 @@ mod tests {
                 )
                 .unwrap_or(None);
 
-                let result_assingment = result
+                let result_assignment = result
                     .as_ref()
                     .map(|assignment| &assignment.value)
                     .unwrap_or(&default_assignment);
                 let expected_assignment = to_value(subject.assignment)
                     .into_assignment_value(test_file.variation_type)
                     .unwrap();
 
-                assert_eq!(result_assingment, &expected_assignment);
+                assert_eq!(result_assignment, &expected_assignment);
+
+                // Verify that get_subject_assignments returns the same result.
+                let subject_assignments = get_subject_assignments(
+                    Some(&config),
+                    &subject.subject_key,
+                    &subject.subject_attributes,
+                    now,
+                );
+                let subject_assignment_for_flag = subject_assignments
+                    .get(&test_file.flag)
+                    .and_then(|result| result.as_ref().ok())
+                    .and_then(|opt| opt.as_ref())
+                    .map(|a| &a.value)
+                    .unwrap_or(&default_assignment);
+
+                // Compare against the original expected assignment instead of result_assignment
+                assert_eq!(subject_assignment_for_flag, &expected_assignment);
+
                 println!("ok");
             }
         }

eppo_core/src/eval/evaluator.rs

@@ -12,7 +12,7 @@ use crate::{
 use super::{
     eval_details::{EvaluationDetails, EvaluationResultWithDetails},
     get_assignment, get_assignment_details, get_bandit_action, get_bandit_action_details,
-    BanditResult,
+    get_subject_assignments, BanditResult,
 };
 
 pub struct EvaluatorConfig {
@@ -49,6 +49,20 @@ impl Evaluator {
         )
     }
 
+    pub fn get_subject_assignments(
+        &self,
+        subject_key: &Str,
+        subject_attributes: &Arc<Attributes>,
+    ) -> HashMap<String, Result<Option<Assignment>, EvaluationError>> {
+        let config = self.get_configuration();
+        get_subject_assignments(
+            config.as_ref().map(AsRef::as_ref),
+            subject_key,
+            subject_attributes,
+            Utc::now(),
+        )
+    }
+
     pub fn get_assignment_details(
         &self,
         flag_key: &str,

eppo_core/src/eval/mod.rs

@@ -7,6 +7,6 @@ mod evaluator;
 
 pub mod eval_details;
 
-pub use eval_assignment::{get_assignment, get_assignment_details};
+pub use eval_assignment::{get_assignment, get_assignment_details, get_subject_assignments};
 pub use eval_bandits::{get_bandit_action, get_bandit_action_details, BanditResult};
 pub use evaluator::{Evaluator, EvaluatorConfig};

eppo_core/src/poller_thread.rs

@@ -131,9 +131,13 @@ impl PollerThread {
             std::thread::Builder::new()
                 .name("eppo-poller".to_owned())
                 .spawn(move || {
+                    let runtime = tokio::runtime::Builder::new_current_thread()
+                        .enable_all()
+                        .build()
+                        .unwrap();
                     loop {
                         log::debug!(target: "eppo", "fetching new configuration");
-                        let result = fetcher.fetch_configuration();
+                        let result = runtime.block_on(fetcher.fetch_configuration());
                         match result {
                             Ok(configuration) => {
                                 store.set_configuration(Arc::new(configuration));

fastly-edge-assignments/.cargo/config.toml

@@ -0,0 +1,2 @@
+[build]
+target = "wasm32-wasi"

fastly-edge-assignments/.gitignore

@@ -0,0 +1,2 @@
+pkg/
+bin/

fastly-edge-assignments/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "fastly-edge-assignments"
+version = "0.1.0"
+edition = "2021"
+# Remove this line if you want to be able to publish this crate as open source on crates.io.
+# Otherwise, `publish = false` prevents an accidental `cargo publish` from revealing private source.
+publish = false
+
+[dependencies]
+chrono = "0.4.19"
+eppo_core = { version = "=4.1.0", path = "../eppo_core" }
+fastly = "0.11.0"
+serde_json = "1.0.132"
+serde = "1.0.192"

fastly-edge-assignments/README.md

@@ -0,0 +1,3 @@
+# Eppo Assignments on Fastly Compute@Edge
+
+TODO: Add a description