feat(etcd): ConfigProvider trait + watch supervisor by moonming · Pull Request #4 · api7/aisix

moonming · 2026-04-17T05:42:10Z

Summary

Introduces `aisix-etcd` — the single long-running task that owns the
etcd subscription and keeps the data-plane `AisixSnapshot` current.

ConfigProvider (`async_trait`) — the seam between the supervisor
and the backing store. Lets unit tests plug in an in-memory provider
so the watch loop is exercised without a container.
EtcdConfigProvider — spec §2 connect policy (5s × 5 attempts),
prefix range read via `get`, watch adapter that maps compaction to
`ProviderError::Compacted`. `ConnectPolicy` exposes the retry knobs.
ExpBackoff — 1s → 60s exponential ladder used between reconnects.
key::parse — `{prefix}/{kind}/{id}` splitter with typed errors.
loader::build_snapshot — validates entries against the aisix-core
JSON Schemas, skips bad rows, returns `BuildStats` for metrics.
Supervisor — `load_once` publishes the initial snapshot;
`run` owns the cancellation-aware main loop with copy-on-write
snapshot replacement on every event so reads stay lock-free.

Also loosens `SnapshotHandle` Clone bound — the handle only
Arc-clones its internal cell, so `S` no longer needs to be Clone.

Test plan

`cargo test -p aisix-etcd` — 26 tests pass

`cargo test -p aisix-core` — 42 tests still pass after Clone change

`cargo clippy --all-targets -- -D warnings` clean

`cargo fmt --check` clean

CI green across all 6 jobs

Introduces aisix-etcd — the single long-running task that owns the etcd subscription and keeps the data-plane AisixSnapshot current. - ConfigProvider trait (async_trait) as the seam between the supervisor and the backing store; lets unit tests plug in an in-memory provider. - EtcdConfigProvider: 5s×5 bootstrap retry (spec §2), prefix range read, watch stream adapter that maps compaction to ProviderError::Compacted. ConnectPolicy exposes the retry knobs for tests. - ExpBackoff: 1s → 60s exponential ladder for reconnect (spec §2). - key::parse: `{prefix}/{kind}/{id}` key splitter with typed errors for prefix mismatch, missing suffix, and empty segments. - loader::build_snapshot: validates raw entries against the JSON Schemas from aisix-core, skips bad rows (never aborts the batch), and returns BuildStats for metrics. - Supervisor: load-once publishes the initial snapshot, the run-loop uses tokio::select! for cancellation, copy-on-write snapshot replacement keeps the read path lock-free. Also loosens `SnapshotHandle<S>` Clone bound — the handle only Arc- clones internals, so S no longer needs to be Clone. 26 unit tests (all sync) covering key parsing, backoff sequence, loader skip-not-abort, supervisor apply_put/delete/resync, and the run-loop cancellation path.

Copilot

Pull request overview

Adds a new aisix-etcd crate that owns the etcd subscription and continuously publishes an up-to-date AisixSnapshot into a lock-free SnapshotHandle, plus a small aisix-core improvement to remove an unnecessary Clone bound from SnapshotHandle<S>.

Changes:

Introduces ConfigProvider + EtcdConfigProvider for loading and watching config from etcd with retry/backoff semantics.
Implements a cancellation-aware Supervisor that applies watch events via copy-on-write snapshot replacement.
Adds key parsing and snapshot building/validation utilities (schema-validated loader, exponential backoff), and relaxes SnapshotHandle’s Clone bound in aisix-core.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
crates/aisix-etcd/src/lib.rs	Exposes the new crate’s modules and public API surface.
crates/aisix-etcd/src/provider.rs	Defines the `ConfigProvider` abstraction, raw entry/event types, and provider errors.
crates/aisix-etcd/src/etcd_provider.rs	Implements the real etcd-backed provider and adapts etcd watch responses into `WatchEvent`s.
crates/aisix-etcd/src/supervisor.rs	Implements the long-running supervisor loop that maintains `SnapshotHandle<AisixSnapshot>`.
crates/aisix-etcd/src/loader.rs	Builds typed snapshots from raw etcd entries with schema validation and stats reporting.
crates/aisix-etcd/src/key.rs	Parses canonical etcd keys into `(kind, id)` with typed errors.
crates/aisix-etcd/src/backoff.rs	Implements the exponential backoff ladder for reconnect behavior.
crates/aisix-core/src/snapshot.rs	Makes `SnapshotHandle<S>` clonable without requiring `S: Clone`.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-04-17T05:46:09Z

+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // We use a simple one-event-at-a-time strategy: on every poll
+        // we ask the underlying stream for the next WatchResponse, then
+        // emit its events back-to-back by storing leftovers… but to keep
+        // this crate small the first event of the batch is emitted and
+        // the rest arrive on subsequent responses, since etcd in practice
+        // batches per key and our prefix produces one-entry batches.
+        match self.inner.poll_next_unpin(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Ready(Some(Err(err))) => {
+                let msg = err.to_string();
+                if msg.contains("required revision has been compacted")
+                    || msg.contains("mvcc: required revision")
+                {
+                    Poll::Ready(Some(Err(ProviderError::Compacted)))
+                } else {
+                    Poll::Ready(Some(Err(ProviderError::Watch(msg))))
+                }
+            }
+            Poll::Ready(Some(Ok(resp))) => {
+                if resp.compact_revision() > 0 {
+                    return Poll::Ready(Some(Err(ProviderError::Compacted)));
+                }
+                // Emit the first event. If a single response has multiple
+                // events, they will be received on subsequent polls by
+                // etcd's own batching — good enough for small clusters
+                // and correct under heavy load (we never drop events,
+                // we only smear them over wakeups).
+                if let Some(ev) = resp.events().first() {
+                    let item = match ev.event_type() {
+                        EventType::Put => ev.kv().map(|kv| {
+                            WatchEvent::Put(RawEntry {
+                                key: String::from_utf8_lossy(kv.key()).into_owned(),
+                                value: kv.value().to_vec(),
+                                revision: kv.mod_revision(),
+                            })
+                        }),
+                        EventType::Delete => ev.kv().map(|kv| WatchEvent::Delete {
+                            key: String::from_utf8_lossy(kv.key()).into_owned(),
+                            revision: kv.mod_revision(),
+                        }),
+                    };
+                    if let Some(item) = item {
+                        return Poll::Ready(Some(Ok(item)));
+                    }
+                }
+                // Empty response (e.g. header-only): tell the runtime
+                // to poll us again rather than stalling.
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            }


EtcdWatchStream::poll_next only emits resp.events().first() and ignores the rest of the events in the same WatchResponse, which will drop watch events under load/batching. Buffer remaining events from a response (e.g., keep a VecDeque of pending events) and drain them across polls so every event is surfaced to the supervisor.

Copilot · 2026-04-17T05:46:10Z

+pub fn parse<'a>(prefix: &str, key: &'a str) -> Result<ResourceKey<'a>, KeyError> {
+    // Accept both `/aisix` and `/aisix/` prefixes transparently.
+    let trimmed_prefix = prefix.trim_end_matches('/');
+    let rest = key
+        .strip_prefix(trimmed_prefix)
+        .ok_or_else(|| KeyError::PrefixMismatch {
+            key: key.to_string(),
+            prefix: prefix.to_string(),
+        })?;
+    let rest = rest.trim_start_matches('/');


key::parse uses strip_prefix(trimmed_prefix), which will treat keys like /aisix2/models/x as matching prefix /aisix (because it’s a raw string prefix, not a path-segment prefix). This can misroute or accidentally accept out-of-scope keys; ensure the match is either exact (key == prefix) or key starts with prefix + "/", otherwise return KeyError::PrefixMismatch.

Copilot · 2026-04-17T05:46:10Z

+    let (kind, id) = rest
+        .split_once('/')
+        .ok_or_else(|| KeyError::MissingSuffix(key.to_string()))?;
+
+    if kind.is_empty() || id.is_empty() {
+        return Err(KeyError::EmptySegment(key.to_string()));
+    }
+
+    Ok(ResourceKey { kind, id })


key::parse accepts IDs containing additional / characters (e.g. /aisix/models/a/b parses as kind=models, id=a/b). If the canonical shape is exactly {prefix}/{kind}/{id} with id as a single segment, reject keys where id contains / to avoid ambiguous routing and delete/apply mismatches.

Copilot · 2026-04-17T05:46:10Z

+/// Raw (key, value, revision) triple as returned by etcd ranges / watches.
+/// Values are `serde_json::Value` so callers can run schema validation
+/// before typed deserialisation.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RawEntry {
+    pub key: String,
+    pub value: Vec<u8>,
+    pub revision: i64,


The doc comment for RawEntry says values are serde_json::Value, but the field type is Vec<u8>. Update the comment to match the actual representation (raw bytes) and where JSON parsing/validation happens (in loader).

Copilot · 2026-04-17T05:46:11Z

+/// Shallow clone of every [`Arc<ResourceEntry>`] — fast and, importantly,
+/// it doesn't materialise a deep copy of the `T` payload.
+fn clone_snapshot(src: &AisixSnapshot) -> AisixSnapshot {
+    let out = AisixSnapshot::new();
+    for e in src.models.entries() {
+        out.models.insert(clone_entry(&e));
+    }
+    for e in src.apikeys.entries() {
+        out.apikeys.insert(clone_entry(&e));
+    }
+    out
+}
+
+fn clone_entry<T: Clone>(src: &Arc<aisix_core::ResourceEntry<T>>) -> aisix_core::ResourceEntry<T> {
+    aisix_core::ResourceEntry {
+        id: src.id.clone(),
+        value: src.value.clone(),
+        revision: src.revision,
+    }


The clone_snapshot comment claims this is a “shallow clone” that doesn’t materialize a deep copy of the payload, but clone_entry requires T: Clone and clones value, which can be a deep copy. Either adjust the comment to reflect reality, or consider extending ResourceTable to support inserting existing Arc<ResourceEntry<T>> so copy-on-write updates can reuse arcs and avoid cloning the inner payload.

Copilot · 2026-04-17T05:46:11Z

+            // Schema passed but serde refused — usually a deny_unknown_fields
+            // mismatch. Treat as schema-rejected for stats purposes.
+            tracing::warn!(key = %key, error = %err, "serde parse failed after schema pass");
+            stats.parse_rejected += 1;


In validate_and_parse, the comment says serde failures after a schema pass should be treated as “schema-rejected for stats purposes”, but the code increments stats.parse_rejected. Either change the counter you increment or update the comment so metrics remain interpretable.

Suggested change

stats.parse_rejected += 1;

stats.schema_rejected += 1;

…ped reads, redact 5xx message, Vertex content-type guard Five concrete fixes from the Copilot inline review on PR #323. Two stale comments (#3, #4 — already fixed in commit 3) are skipped. **#1+#7 — Azure OpenAI-compatible code preservation.** Azure's envelope omits `error.type` and carries only `error.code`. The bridge previously put the upstream code into `view.kind` and left `view.code` as `None`. For OpenAI-compat tokens Azure inherits unchanged (e.g. `rate_limit_exceeded`), this meant downstream OpenAI clients received `error.type=rate_limit_exceeded` but `error.code=null` — exactly the SDK-retry break issue #322 is about. Fix: - Azure parser populates BOTH `view.kind` AND `view.code` from the upstream `error.code` field. - `render_openai_envelope`'s AzureOpenAI branch now prefers the translation-table-derived code (so explicit Azure tokens like `DeploymentNotFound` → `model_not_found` still win), falling back to `view.code` for OpenAI-compat pass-through. **#2 — Drain the response stream after hitting the cap.** `read_body_capped` previously broke out of the read loop the moment `limit` bytes were buffered. With reqwest/hyper that leaves unread bytes in the response and prevents connection reuse — during a burst of upstream errors the gateway would churn TCP connections instead of recycling the keep-alive pool. Fix: keep iterating the stream, discarding chunks past the cap. Memory stays bounded by `limit`. **#5 — Redact upstream `error.message` on 5xx.** The 5xx branch of `render_bridge_upstream_envelope` was forwarding `BridgeError::UpstreamStatus.message` verbatim — which for OpenAI / Anthropic comes from the parsed upstream `error.message`. Upstream 5xx bodies routinely embed operator-internal detail (engine names, shard ids, queue depth). Fix: on 5xx, emit a canned `"upstream returned {status}"` message; the full upstream body remains in operator logs via tracing. **#6 — Stale "follow-up" comment.** The docstring on `render_bridge_upstream_envelope` claimed cross-wire translation would ship in a follow-up, but it already shipped in commit 2. Rewrite the comment to describe current behaviour (4xx → `error_translate`; 5xx → canned envelope; `Unknown` wire → legacy generic envelope). **#8 — Content-type guard on Vertex (and Azure, while at it).** `capture_upstream_error_http` already gates serde parsing on `Content-Type: application/json` so a 64 KB HTML error page from a fronting WAF doesn't waste CPU on a doomed JSON parse. The Vertex and Azure bridges call serde directly because they need a custom parse path (canned message for redaction) — same guard now applies. Promoted `content_type_is_json` and added a `response_is_json` helper to the gateway's public surface; both bridges call it before `parse_*_error_*`. New tests: - `upstream_openai_5xx_with_json_envelope_collapses_and_redacts_message` pins the 5xx redaction (asserts `engine offline` / `shard 47` / `engine_overloaded` don't reach the customer envelope). - `chat_429_preserves_openai_compatible_code_for_sdk_retry` (Azure) pins that `parsed.code` carries the OpenAI-compat upstream code. - `chat_400_non_json_body_skips_envelope_parse` (Azure) and `chat_gemini_non_json_body_skips_envelope_parse` (Vertex) pin the new content-type guard. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…he snippet Addresses Copilot inline `3274962628` on PR #344 round-5. Round-5's trim collapsed the Option A/B labels and put the "pick one, not both" warning AFTER the jq snippet. Readers following the page top-to-bottom would reach the jq snippet before the alternation framing — risking a second POST to `/admin/v1/provider_keys` that hits a 409 duplicate `display_name` (silent rejection per the stealth-cost rule locked in Lesson #4). Restructure (net ~10 words shorter than round-5): - Lead-in combines the no-jq path ("copy it by eye from the response above") with the jq-capturing form, made explicit as an alternative via "**instead of** the curl above". - 409 stealth-cost warning is now inline in the lead-in parenthetical — BEFORE the snippet, not after — so the reader has the alternation framing in hand before reaching the action that would cause harm. - "Same pattern applies to model id and API-key id captures" follow-up preserved. The structural improvement is more rigorous than round-3's Option A/B form (which had the warning trailing the snippets) and round-5 (which had the warning trailing the single snippet). This round-6 form lands the alternation explicitly before the action — the correct order for stealth-cost warnings on tutorial-shaped artifacts. Substantive preservation: - duplicate-`display_name` 409 named-consequence: preserved (now in the lead-in parenthetical, before the snippet). - jq snippet itself: preserved verbatim. - no-jq path: now explicitly named ("copy it by eye from the response above") — was implicit in round-5. - "Same pattern applies..." follow-up: preserved.

…dor consts Round-3 audit follow-up + scope expansion confirmed by user. Cuts the last remnants of the pre-#302 vendor-enumeration design that the prior clean-cut PR left as soft-deprecated: #1 Provider enum - Delete the enum + its Adapter mapping + the `every_provider_variant_has_as_str_and_adapter` test. - Drop the `Provider` re-export from `aisix_core::{lib, models}`. #2 OpenAiBridge::with_name + `name` field - Drop the builder method and the per-instance `name` field; the bridge is the singular OpenAI family bridge and reports `"openai"` unconditionally. - Same surgery on AnthropicBridge for symmetry (its `with_name` was unused at every call site). #3 Per-vendor default-base consts + match arms - Delete DEEPSEEK_DEFAULT_BASE, GOOGLE_DEFAULT_BASE, COHERE_DEFAULT_BASE plus the 11 long-tail consts (GROQ/MISTRAL/TOGETHERAI/FIREWORKS/ PERPLEXITY/MOONSHOTAI/ALIBABA/ZHIPUAI/BASETEN/HUGGINGFACE/CEREBRAS). - Replace OpenAiBridge::default_base() with the hardcoded OpenAI bare-host fallback inside resolve_base; the safety guard now returns a Config error for any non-`"openai"` vendor whose PK reaches the bridge with an empty api_base, preventing a silent credential leak to api.openai.com. #4 normalize_canonical_deepseek + normalize_canonical_cohere - Delete both helpers + DEEPSEEK_CANONICAL_HOSTS / COHERE_CANONICAL_HOSTS. - normalize_api_base loses its `provider` parameter; the canonical `/v1` synthesis only applies to api.openai.com. Operators paste the documented URL for every other vendor (cp-api populates it from adapter_map's `default_base_url`). Compat shim: build_hub() keeps `register_specialized("openai", …)` + `register_specialized("anthropic", …)` so pre-Phase-A ProviderKeys that carry `provider` but not `adapter` still dispatch through the two-tier lookup. cp-api admits every catalog vendor post-Phase-A with `adapter` populated, so the family bridge above covers them without any specialized entry needed. Tests deleted: every test that exercised the deleted surface (deepseek/cohere canonical-host normalization, gemini/cohere/long-tail with_name vendor-default targeting, with_name x-aisix-bridge header variant). Kept a single non-OpenAI host pass-through test that pins the family bridge's verbatim treatment of any non-openai api_base (corporate proxies, alternative deployments). Doc updates: stale `Provider::default_base_url` / `Provider::Cohere` / `OpenAiBridge::with_name` references in dispatch.rs, rerank.rs, provider_key.rs, azure-openai/lib.rs replaced with the post-clean-cut language. Net: −464 LOC. Workspace cargo test + clippy clean.

Copilot AI review requested due to automatic review settings April 17, 2026 05:42

Copilot started reviewing on behalf of moonming April 17, 2026 05:42 View session

moonming merged commit 5ae2136 into main Apr 17, 2026
10 checks passed

moonming deleted the feat/etcd-watch branch April 17, 2026 05:45

Copilot AI reviewed Apr 17, 2026

View reviewed changes

moonming mentioned this pull request May 16, 2026

tracking: engineering & docs quality improvements — config schema / provider doc templates / architecture deep-dives / AGENTS.md / concept abstraction #304

Open

This was referenced May 18, 2026

docs: fix new-developer onboarding gaps across overview, quickstart, and configuration pages #326

Merged

docs: Configuration reference corrections across admin-api, api-keys, caching, and models pages #347

Closed

This was referenced May 21, 2026

ProviderKey schema rejects adapter family extra-config fields (aws_region, gcp_project, azure_resource_name) #361

Closed

test(etcd): pin ProviderKey loader behavior for adapter-family payloads #362

Merged

moonming mentioned this pull request May 26, 2026

Azure AAD: no production-grade token endpoint override (blocks live e2e against mock) #413

Closed

juzhiyuan mentioned this pull request May 29, 2026

security: enforce guardrails through a unified request policy pipeline #448

Closed

20 tasks

jarvis9443 mentioned this pull request Jun 2, 2026

fix: broaden guardrail/quota coverage for tool calls, cache, oversized tokens & /v1/messages (#448) #490

Merged

moonming mentioned this pull request Aug 3, 2026

build(deps): clear all 8 open dependabot alerts #867

Open

Navigation Menu

Uh oh!

feat(etcd): ConfigProvider trait + watch supervisor - #4

feat(etcd): ConfigProvider trait + watch supervisor#4
moonming merged 1 commit into
mainfrom
feat/etcd-watch

moonming commented Apr 17, 2026

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Apr 17, 2026

Uh oh!

Copilot AI Apr 17, 2026

Uh oh!

Copilot AI Apr 17, 2026

Uh oh!

Copilot AI Apr 17, 2026

Uh oh!

Copilot AI Apr 17, 2026

Uh oh!

Copilot AI Apr 17, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

moonming commented Apr 17, 2026

Summary

Test plan

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 17, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants