feat: rust backend core

rust/Cargo.toml

@@ -3,6 +3,7 @@ resolver = "3"
 members = [
   "crates/adc-sdk",
   "crates/adc-differ",
+  "crates/adc-backend-core",
   "crates/adc-sync-bench",
   "crates/adc-mock-server",
 ]

rust/crates/adc-backend-core/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "adc-backend-core"
+version = "0.1.0"
+edition.workspace = true
+publish.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+adc-sdk = { path = "../adc-sdk" }
+reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] }
+futures = "0.3"
+tokio = { version = "1", features = ["time"] }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "time"] }
+axum = "0.8"
+serde_json = { workspace = true }

rust/crates/adc-backend-core/src/client.rs

@@ -0,0 +1,100 @@
+use std::time::Duration;
+
+use adc_sdk::BackendError;
+use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderValue};
+use reqwest::{Method, RequestBuilder, Response, Url};
+
+use crate::tls::TlsConfig;
+
+/// Everything needed to stand up a backend's HTTP client: where it lives,
+/// how to authenticate, and how to connect. Doesn't include backend-specific
+/// concerns like a gateway group name — those live in each backend crate.
+#[derive(Debug, Clone)]
+pub struct HttpClientConfig {
+    pub server: String,
+    pub token: String,
+    pub timeout: Option<Duration>,
+    pub tls: TlsConfig,
+}
+
+/// A backend's HTTP client: a `reqwest::Client` pre-configured with the
+/// `X-API-KEY` auth header apisix/api7 both expect, TLS settings, and a base
+/// URL, plus request/response handling that classifies failures into
+/// `BackendError` uniformly. Connection pooling (the Node `agentkeepalive`
+/// equivalent) comes for free from `reqwest::Client`'s own pool — nothing to
+/// configure for that.
+pub struct HttpClient {
+    inner: reqwest::Client,
+    base_url: Url,
+}
+
+impl HttpClient {
+    pub fn new(config: HttpClientConfig) -> Result<Self, BackendError> {
+        let base_url = Url::parse(&config.server)
+            .map_err(|e| BackendError::Other(format!("invalid server URL {:?}: {e}", config.server).into()))?;
+
+        let mut headers = HeaderMap::new();
+        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        let mut token = HeaderValue::from_str(&config.token)
+            .map_err(|e| BackendError::Other(format!("invalid token: {e}").into()))?;
+        token.set_sensitive(true);
+        headers.insert("X-API-KEY", token);
+
+        let mut builder = reqwest::Client::builder().default_headers(headers);
+        if let Some(timeout) = config.timeout {
+            builder = builder.timeout(timeout);
+        }
+        builder = config.tls.apply(builder)?;
+
+        let inner = builder
+            .build()
+            .map_err(|e| BackendError::Other(format!("failed to build HTTP client: {e}").into()))?;
+
+        Ok(Self { inner, base_url })
+    }
+
+    /// Starts a request against `path`, resolved relative to the configured
+    /// server URL (e.g. `/apisix/admin/routes`).
+    pub fn request(&self, method: Method, path: &str) -> Result<RequestBuilder, BackendError> {
+        let url = self
+            .base_url
+            .join(path)
+            .map_err(|e| BackendError::Other(format!("invalid request path {path:?}: {e}").into()))?;
+        Ok(self.inner.request(method, url))
+    }
+
+    /// Sends a request built via [`HttpClient::request`], classifying
+    /// transport failures and non-2xx responses into `BackendError`. Callers
+    /// decode the body themselves (`.json()`, `.text()`, or just read
+    /// headers) since the right shape depends on the endpoint.
+    pub async fn send(&self, builder: RequestBuilder) -> Result<Response, BackendError> {
+        let (client, request) = builder.build_split();
+        let request = request.map_err(|e| BackendError::Other(format!("failed to build request: {e}").into()))?;
+        let method = request.method().clone();
+        let url = request.url().clone();
+
+        let response = client.execute(request).await.map_err(|e| classify_transport_error(&e, &method, &url))?;
+
+        let status = response.status();
+        if status.is_success() {
+            return Ok(response);
+        }
+
+        let message = response.text().await.unwrap_or_default();
+        Err(match status.as_u16() {
+            401 | 403 => BackendError::Auth(message),
+            404 => BackendError::NotFound(format!("{method} {url}")),
+            _ => BackendError::Api { status: status.as_u16(), message },
+        })
+    }
+}
+
+fn classify_transport_error(error: &reqwest::Error, method: &Method, url: &Url) -> BackendError {
+    if error.is_timeout() {
+        BackendError::Transport(format!(
+            "request \"{method} {url}\" timed out. Consider increasing the timeout."
+        ))
+    } else {
+        BackendError::Transport(format!("request \"{method} {url}\" failed: {error}"))
+    }
+}

rust/crates/adc-backend-core/src/concurrency.rs

@@ -0,0 +1,20 @@
+use std::future::Future;
+
+use futures::stream::{self, StreamExt};
+
+/// Runs `f` over `items` with at most `concurrency` in flight at once.
+/// `None` means unbounded — every item starts immediately, matching RxJS
+/// `mergeMap(fn)` called with no concurrency argument.
+///
+/// Results come back in completion order, not input order (same as
+/// `mergeMap`'s emission order): fine for callers like `Backend::sync`,
+/// where each output already carries its own identity (the `Event` it came
+/// from) rather than relying on positional correlation with `items`.
+pub async fn concurrent_map<T, U, F, Fut>(items: Vec<T>, concurrency: Option<usize>, f: F) -> Vec<U>
+where
+    F: FnMut(T) -> Fut,
+    Fut: Future<Output = U>,
+{
+    let concurrency = concurrency.unwrap_or(items.len()).max(1);
+    stream::iter(items).map(f).buffer_unordered(concurrency).collect().await
+}

rust/crates/adc-backend-core/src/lib.rs

@@ -0,0 +1,18 @@
+//! Shared HTTP client plumbing for the backend integrations
+//! (`adc-backend-apisix`, `adc-backend-api7`, `adc-backend-apisix-standalone`):
+//! building a `reqwest::Client` with auth headers and TLS settings,
+//! classifying request failures into `adc_sdk::BackendError` uniformly, and
+//! the retry/concurrency helpers each backend's operator/fetcher builds
+//! request fan-out on top of.
+
+mod client;
+mod concurrency;
+mod retry;
+mod tls;
+
+pub use client::{HttpClient, HttpClientConfig};
+pub use concurrency::concurrent_map;
+pub use retry::RetryPolicy;
+pub use tls::TlsConfig;
+
+pub use reqwest::{Method, Response};

rust/crates/adc-backend-core/src/retry.rs

@@ -0,0 +1,38 @@
+use std::future::Future;
+use std::time::Duration;
+
+/// A fixed-delay retry policy: on failure, wait `delay` and try again, up to
+/// `retries` additional attempts (so `retries + 1` attempts total).
+#[derive(Debug, Clone, Copy)]
+pub struct RetryPolicy {
+    pub retries: usize,
+    pub delay: Duration,
+}
+
+impl Default for RetryPolicy {
+    /// Matches the apisix operator's hardcoded `retry({ count: 3, delay: 100 })`
+    /// for mutating requests (PUT/DELETE against `/apisix/admin/*`).
+    fn default() -> Self {
+        Self { retries: 3, delay: Duration::from_millis(100) }
+    }
+}
+
+impl RetryPolicy {
+    pub async fn run<T, E, F, Fut>(&self, mut f: F) -> Result<T, E>
+    where
+        F: FnMut() -> Fut,
+        Fut: Future<Output = Result<T, E>>,
+    {
+        let mut attempt = 0;
+        loop {
+            match f().await {
+                Ok(value) => return Ok(value),
+                Err(_) if attempt < self.retries => {
+                    attempt += 1;
+                    tokio::time::sleep(self.delay).await;
+                }
+                Err(err) => return Err(err),
+            }
+        }
+    }
+}

rust/crates/adc-backend-core/src/tls.rs

@@ -0,0 +1,42 @@
+use adc_sdk::BackendError;
+
+/// TLS options for connecting to a backend's admin API. Mirrors what the CLI
+/// previously built into per-request Node `http(s).Agent` instances (custom
+/// CA, mTLS client identity, skip-verify) — here they're baked directly into
+/// the `reqwest::Client` at construction time instead.
+///
+/// Takes PEM bytes directly rather than file paths: reading config off disk
+/// is the CLI/config layer's job, not this HTTP client's — keeps this crate
+/// free of filesystem I/O and lets callers source certs from anywhere
+/// (files, env vars, a secrets manager).
+#[derive(Debug, Clone, Default)]
+pub struct TlsConfig {
+    pub ca_cert_pem: Option<Vec<u8>>,
+    pub client_cert_pem: Option<Vec<u8>>,
+    pub client_key_pem: Option<Vec<u8>>,
+    pub skip_verify: bool,
+}
+
+impl TlsConfig {
+    pub(crate) fn apply(&self, mut builder: reqwest::ClientBuilder) -> Result<reqwest::ClientBuilder, BackendError> {
+        if self.skip_verify {
+            builder = builder.danger_accept_invalid_certs(true);
+        }
+
+        if let Some(pem) = &self.ca_cert_pem {
+            let cert = reqwest::Certificate::from_pem(pem)
+                .map_err(|e| BackendError::Other(format!("invalid CA certificate: {e}").into()))?;
+            builder = builder.add_root_certificate(cert);
+        }
+
+        if let (Some(cert_pem), Some(key_pem)) = (&self.client_cert_pem, &self.client_key_pem) {
+            let mut pem = cert_pem.clone();
+            pem.extend(key_pem);
+            let identity = reqwest::Identity::from_pem(&pem)
+                .map_err(|e| BackendError::Other(format!("invalid client certificate/key: {e}").into()))?;
+            builder = builder.identity(identity);
+        }
+
+        Ok(builder)
+    }
+}

rust/crates/adc-backend-core/tests/concurrency.rs

@@ -0,0 +1,54 @@
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Duration;
+
+use adc_backend_core::concurrent_map;
+
+#[tokio::test]
+async fn respects_the_concurrency_bound() {
+    let current = AtomicUsize::new(0);
+    let peak = AtomicUsize::new(0);
+
+    let items: Vec<u32> = (0..20).collect();
+    let results = concurrent_map(items.clone(), Some(4), |item| {
+        let current = &current;
+        let peak = &peak;
+        async move {
+            let now = current.fetch_add(1, Ordering::SeqCst) + 1;
+            peak.fetch_max(now, Ordering::SeqCst);
+            tokio::time::sleep(Duration::from_millis(5)).await;
+            current.fetch_sub(1, Ordering::SeqCst);
+            item
+        }
+    })
+    .await;
+
+    assert_eq!(results.len(), items.len());
+    assert!(peak.load(Ordering::SeqCst) <= 4, "peak concurrency was {}", peak.load(Ordering::SeqCst));
+
+    let mut sorted = results;
+    sorted.sort();
+    assert_eq!(sorted, items);
+}
+
+#[tokio::test]
+async fn unbounded_when_concurrency_is_none() {
+    let current = AtomicUsize::new(0);
+    let peak = AtomicUsize::new(0);
+
+    let items: Vec<u32> = (0..10).collect();
+    let results = concurrent_map(items.clone(), None, |item| {
+        let current = &current;
+        let peak = &peak;
+        async move {
+            let now = current.fetch_add(1, Ordering::SeqCst) + 1;
+            peak.fetch_max(now, Ordering::SeqCst);
+            tokio::time::sleep(Duration::from_millis(5)).await;
+            current.fetch_sub(1, Ordering::SeqCst);
+            item
+        }
+    })
+    .await;
+
+    assert_eq!(results.len(), items.len());
+    assert_eq!(peak.load(Ordering::SeqCst), items.len());
+}