From 9e368cff4a289c753c717a2af21ad32c15258fb3 Mon Sep 17 00:00:00 2001 From: Anna Tchernych Date: Mon, 26 Oct 2026 10:16:49 -0700 Subject: [PATCH 1/8] feat(global-router): add a crate for types the Global Router components share dynamo-global-common holds the types that more than one Global Router component needs, so each is defined once: - a pool's identity and location (PoolKey, PoolId and its deriver, PoolLocation), from the shared crate on the global-router-v2 test branch; - the JSON of the Global Discovery Service's HTTP API: the POST /pools body and its lease reply, and the pool and router lists. Today the service and its client each define these; - how a router names a PoolRelay on a trusted network (address_identity), which the service writes into each pool's relay_identity; - Global Router replicas, and list-and-watch events and snapshots, which the service and the client in #17208 both define today. The JSON does not change. PoolKey now reads and writes {"site_id", "namespace", "dgd_name"} and checks each field while reading. A lease reply parses only when the heartbeat interval is above zero and below the lease, as in the client in #17208, and the lease is at most a day. Co-authored-by: Sachal Malick Co-authored-by: Jacob Ioffe Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Anna Tchernych --- .github/codeowners/areas.yaml | 1 + CODEOWNERS | 1 + Cargo.lock | 11 ++ Cargo.toml | 2 + lib/global-common/Cargo.toml | 21 +++ lib/global-common/src/discovery.rs | 225 +++++++++++++++++++++++++++++ lib/global-common/src/lib.rs | 34 +++++ lib/global-common/src/pool.rs | 224 ++++++++++++++++++++++++++++ lib/global-common/src/replicas.rs | 23 +++ lib/global-common/src/watch.rs | 33 +++++ 10 files changed, 575 insertions(+) create mode 100644 lib/global-common/Cargo.toml create mode 100644 lib/global-common/src/discovery.rs create mode 100644 lib/global-common/src/lib.rs create mode 100644 lib/global-common/src/pool.rs create mode 100644 lib/global-common/src/replicas.rs create mode 100644 lib/global-common/src/watch.rs diff --git a/.github/codeowners/areas.yaml b/.github/codeowners/areas.yaml index 57f8b743f723..6770cac05260 100644 --- a/.github/codeowners/areas.yaml +++ b/.github/codeowners/areas.yaml @@ -76,6 +76,7 @@ areas: - docs/fern/pages/kubernetes/kv-aware-routing/kv-dc-relay.md - docs/fern/pages/reference/components/kv-dc-relay-configuration.md - lib/bindings/python/rust/llm/kv_dc_relay.rs + - lib/global-common/ - lib/llm/src/kv_dc_relay.rs - lib/llm/src/kv_dc_relay/ - lib/llm/tests/fixtures/relay-sources-canonical.json diff --git a/CODEOWNERS b/CODEOWNERS index 17aec522c0ad..f4e56fe9c444 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -88,6 +88,7 @@ /components/src/dynamo/thunderagent_router/ @ai-dynamo/dynamo-router-codeowners # === global-router (@ai-dynamo/dynamo-global-router-codeowners) === +/lib/global-common/ @ai-dynamo/dynamo-global-router-codeowners /components/src/dynamo/kv_dc_relay/ @ai-dynamo/dynamo-global-router-codeowners # === backend-vllm (@ai-dynamo/dynamo-backend-vllm-codeowners) === diff --git a/Cargo.lock b/Cargo.lock index 8a16c3cb80d5..807d20a3e58e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2475,6 +2475,17 @@ dependencies = [ "validator", ] +[[package]] +name = "dynamo-global-common" +version = "1.7.0" +dependencies = [ + "blake3", + "serde", + "serde_json", + "thiserror 2.0.18", + "url", +] + [[package]] name = "dynamo-kv-hashing" version = "1.7.0" diff --git a/Cargo.toml b/Cargo.toml index 8f6ee4d717f7..7f4e1bfc3b10 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ members = [ "lib/kv-hashing", "lib/mocker", "lib/kv-router", + "lib/global-common", "lib/router-plugins/catalog", "lib/router-plugins/builtin", "examples/router/custom-policy-example/soft-pin-repin", @@ -64,6 +65,7 @@ dynamo-tokens = { path = "lib/tokens", version = "1.7.0" } dynamo-truthy = { path = "lib/truthy", version = "1.7.0" } dynamo-data-gen = { path = "lib/data-gen", version = "1.7.0" } dynamo-kv-hashing = { path = "lib/kv-hashing", version = "1.7.0" } +dynamo-global-common = { path = "lib/global-common", version = "1.7.0" } dynamo-memory = { path = "lib/memory", version = "1.7.0" } dynamo-mocker = { path = "lib/mocker", version = "1.7.0" } dynamo-kv-router = { path = "lib/kv-router", version = "1.7.0" } diff --git a/lib/global-common/Cargo.toml b/lib/global-common/Cargo.toml new file mode 100644 index 000000000000..97bd89948b93 --- /dev/null +++ b/lib/global-common/Cargo.toml @@ -0,0 +1,21 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "dynamo-global-common" +description = "Types shared by the Global Router components" +version.workspace = true +edition.workspace = true +authors.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true + +[dependencies] +blake3 = { workspace = true } +serde = { workspace = true } +thiserror = { workspace = true } +url = { workspace = true } + +[dev-dependencies] +serde_json = { workspace = true } diff --git a/lib/global-common/src/discovery.rs b/lib/global-common/src/discovery.rs new file mode 100644 index 000000000000..3a35e777708a --- /dev/null +++ b/lib/global-common/src/discovery.rs @@ -0,0 +1,225 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! The JSON of the Global Discovery Service's HTTP API. A reader ignores +//! fields it does not know, so the service can add a field before its clients +//! use it. + +use std::net::IpAddr; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use url::Url; + +use crate::pool::{PoolKey, PoolLocation}; +use crate::replicas::ReplicaEndpoint; + +/// `POST /pools` body. A PoolRelay sends it on a timer, and each call is also +/// the pool's heartbeat. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PoolRegistration { + pub key: PoolKey, + pub location: PoolLocation, + /// The model or checkpoint the pool serves. + pub model: String, + #[serde(default)] + pub capabilities: Vec, + /// Local frontends the Global Router sends requests to. + #[serde(default)] + pub frontends: Vec, + #[serde(default)] + pub capacity: Vec, +} + +/// The workers of one hardware SKU. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +pub struct Capacity { + pub sku: String, + pub count: u32, +} + +/// `POST /pools` reply: register again after the heartbeat interval. If the +/// lease passes with no successful registration, the pool drops out of +/// `GET /pools`. +/// +/// Both are whole seconds, the interval is above zero and below the lease, +/// and the lease is at most [`Lease::MAX_DURATION_S`]. A reply that breaks +/// this does not parse, so it cannot make a PoolRelay register in a tight +/// loop, and every `Lease` can be written as JSON and read back. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(try_from = "LeaseFields")] +pub struct Lease { + heartbeat_interval_s: u64, + lease_duration_s: u64, +} + +impl Lease { + /// One day. + pub const MAX_DURATION_S: u64 = 86_400; + + pub fn new(heartbeat_interval_s: u64, lease_duration_s: u64) -> Result { + if heartbeat_interval_s == 0 + || heartbeat_interval_s >= lease_duration_s + || lease_duration_s > Self::MAX_DURATION_S + { + return Err(LeaseError { + heartbeat_interval_s, + lease_duration_s, + }); + } + Ok(Self { + heartbeat_interval_s, + lease_duration_s, + }) + } + + pub fn heartbeat_interval(&self) -> Duration { + Duration::from_secs(self.heartbeat_interval_s) + } + + pub fn lease_duration(&self) -> Duration { + Duration::from_secs(self.lease_duration_s) + } +} + +/// A lease that [`Lease::new`] refuses. +#[derive(Debug, Error, PartialEq, Eq)] +#[error( + "heartbeat interval {heartbeat_interval_s}s must be above zero and below the \ + {lease_duration_s}s lease, which may be at most a day" +)] +pub struct LeaseError { + pub heartbeat_interval_s: u64, + pub lease_duration_s: u64, +} + +#[derive(Deserialize)] +struct LeaseFields { + heartbeat_interval_s: u64, + lease_duration_s: u64, +} + +impl TryFrom for Lease { + type Error = LeaseError; + + fn try_from(fields: LeaseFields) -> Result { + Self::new(fields.heartbeat_interval_s, fields.lease_duration_s) + } +} + +/// `GET /pools` reply. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PoolsResponse { + pub pools: Vec, + /// True when the service lists every pool, so a new router can tell a + /// full list from a partial one. + pub complete: bool, +} + +/// One pool in `GET /pools`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PoolListing { + pub key: PoolKey, + pub location: PoolLocation, + /// `None` when the service does not know the pool's model. + pub model: Option, + #[serde(default)] + pub capabilities: Vec, + /// Local frontends the Global Router sends requests to. + #[serde(default)] + pub frontends: Vec, + #[serde(default)] + pub capacity: Vec, + /// The only PoolRelay allowed to stream this pool's state, when the + /// service knows it. On a trusted network it is [`address_identity`]. + #[serde(default)] + pub relay_identity: Option, +} + +/// Names a PoolRelay by its address on a trusted network, as +/// `ip:
`. An IPv4-mapped IPv6 address is written as IPv4, as +/// EndpointSlices list it. A router matches a connecting PoolRelay against +/// [`PoolListing::relay_identity`] with this. +pub fn address_identity(address: IpAddr) -> String { + format!("ip:{}", address.to_canonical()) +} + +/// `GET /routers` reply: every Global Router replica. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RoutersResponse { + pub routers: Vec, +} + +#[cfg(test)] +mod tests { + use serde_json::{Value, json}; + + use super::*; + use crate::replicas::ReplicaId; + + /// Reads `json` as `T` and writes it back unchanged. + fn round_trip Deserialize<'de>>(json: Value) -> T { + let parsed: T = serde_json::from_value(json.clone()).unwrap(); + assert_eq!(serde_json::to_value(&parsed).unwrap(), json); + parsed + } + + #[test] + fn json_matches_the_service_and_its_clients() { + let key = + json!({"site_id": "aws-west", "namespace": "dynamo", "dgd_name": "mocker-west-1"}); + let location = json!({ + "region": "us-west-2", + "availability_zone": null, + "cluster": "aws-west", + "datacenter": null + }); + round_trip::(json!({ + "key": key, + "location": location, + "model": "Qwen/Qwen3-0.6B", + "capabilities": ["aggregated"], + "frontends": ["http://10.43.1.2:8000/"], + "capacity": [{"sku": "H100", "count": 8}] + })); + let pools = round_trip::(json!({ + "pools": [{ + "key": key, + "location": location, + "model": null, + "capabilities": [], + "frontends": [], + "capacity": [], + "relay_identity": "ip:10.43.66.175" + }], + "complete": true + })); + assert!(pools.complete); + let routers = round_trip::(json!({ + "routers": [{"replica": "aws-ohio/global-router-0", "url": "http://10.42.0.5:50051/"}] + })); + assert_eq!( + routers.routers[0].replica, + ReplicaId("aws-ohio/global-router-0".to_owned()) + ); + let mapped: IpAddr = "::ffff:10.43.66.175".parse().unwrap(); + assert_eq!(address_identity(mapped), "ip:10.43.66.175"); + assert_eq!(address_identity("fd00::2".parse().unwrap()), "ip:fd00::2"); + } + + #[test] + fn a_lease_that_would_spin_or_overflow_does_not_parse() { + let lease = round_trip::(json!({"heartbeat_interval_s": 5, "lease_duration_s": 15})); + assert_eq!(lease, Lease::new(5, 15).unwrap()); + assert_eq!(lease.heartbeat_interval(), Duration::from_secs(5)); + let too_long = Lease::MAX_DURATION_S + 1; + for (interval, lease) in [(0, 15), (15, 15), (20, 15), (5, too_long)] { + let body = json!({"heartbeat_interval_s": interval, "lease_duration_s": lease}); + assert!( + serde_json::from_value::(body).is_err(), + "{interval}s / {lease}s" + ); + } + } +} diff --git a/lib/global-common/src/lib.rs b/lib/global-common/src/lib.rs new file mode 100644 index 000000000000..153d9bec1157 --- /dev/null +++ b/lib/global-common/src/lib.rs @@ -0,0 +1,34 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Types that more than one Global Router component needs, defined once. +//! +//! - A pool's identity and location: [`PoolKey`], [`PoolId`] and its deriver +//! [`V1PoolIdDeriver`], and [`PoolLocation`]. +//! - The JSON of the Global Discovery Service's HTTP API, which the service +//! writes and its clients read: +//! +//! | Call | Body | Reply | +//! |---|---|---| +//! | `POST /pools` | [`PoolRegistration`] | [`Lease`] | +//! | `GET /pools` | | [`PoolsResponse`] | +//! | `GET /routers` | | [`RoutersResponse`] | +//! +//! A router matches a connecting PoolRelay against a listing's +//! `relay_identity` with [`address_identity`]. +//! - Global Router replicas ([`ReplicaEndpoint`]), and list-and-watch events +//! and snapshots ([`WatchEvent`], [`Snapshot`]), which the service's watch +//! contract and its clients share. + +mod discovery; +mod pool; +mod replicas; +mod watch; + +pub use discovery::{ + Capacity, Lease, LeaseError, PoolListing, PoolRegistration, PoolsResponse, RoutersResponse, + address_identity, +}; +pub use pool::{PoolId, PoolIdDeriver, PoolKey, PoolKeyError, PoolLocation, V1PoolIdDeriver}; +pub use replicas::{ReplicaEndpoint, ReplicaEvent, ReplicaId}; +pub use watch::{Completeness, Snapshot, WatchEvent}; diff --git a/lib/global-common/src/pool.rs b/lib/global-common/src/pool.rs new file mode 100644 index 000000000000..fc475013dcef --- /dev/null +++ b/lib/global-common/src/pool.rs @@ -0,0 +1,224 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Runtime-independent identities for pools in a Global View. +//! +//! A routing pool is one DGD in the initial deployment. Its identity is stable +//! across process and relay restarts, while relay and KV producer generations +//! are tracked separately by the source adapter. + +use std::fmt; + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +/// An opaque routing identity. Callers must not parse its representation. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct PoolId(String); + +impl PoolId { + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Display for PoolId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +/// Stable deployment coordinates used by the POC derivation. +/// +/// `namespace` is the Kubernetes namespace containing the DGD, not the +/// Dynamo runtime namespace derived by the operator. In JSON it is +/// `{"site_id", "namespace", "dgd_name"}`, and reading it checks each field as +/// [`PoolKey::new`] does. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(try_from = "PoolKeyFields", into = "PoolKeyFields")] +pub struct PoolKey { + site_id: String, + namespace: String, + dgd_name: String, +} + +#[derive(Serialize, Deserialize)] +struct PoolKeyFields { + site_id: String, + namespace: String, + dgd_name: String, +} + +impl TryFrom for PoolKey { + type Error = PoolKeyError; + + fn try_from(fields: PoolKeyFields) -> Result { + Self::new(fields.site_id, fields.namespace, fields.dgd_name) + } +} + +impl From for PoolKeyFields { + fn from(key: PoolKey) -> Self { + Self { + site_id: key.site_id, + namespace: key.namespace, + dgd_name: key.dgd_name, + } + } +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum PoolKeyError { + #[error("{field} must be non-empty without surrounding whitespace or control characters")] + InvalidField { field: &'static str }, +} + +impl PoolKey { + pub fn new( + site_id: impl Into, + namespace: impl Into, + dgd_name: impl Into, + ) -> Result { + let site_id = site_id.into(); + let namespace = namespace.into(); + let dgd_name = dgd_name.into(); + validate(&site_id, "site_id")?; + validate(&namespace, "namespace")?; + validate(&dgd_name, "dgd_name")?; + Ok(Self { + site_id, + namespace, + dgd_name, + }) + } + + pub fn site_id(&self) -> &str { + &self.site_id + } + + pub fn namespace(&self) -> &str { + &self.namespace + } + + pub fn dgd_name(&self) -> &str { + &self.dgd_name + } +} + +fn validate(value: &str, field: &'static str) -> Result<(), PoolKeyError> { + if value.is_empty() || value.trim() != value || value.chars().any(char::is_control) { + return Err(PoolKeyError::InvalidField { field }); + } + Ok(()) +} + +/// Global View owns assignment of routing pool IDs. Implementations can +/// change without leaking derivation details into catalog or routing code. +pub trait PoolIdDeriver: Send + Sync { + fn derive(&self, key: &PoolKey) -> PoolId; +} + +/// Initial deterministic derivation, shared by every Global Router replica. +/// +/// Changing this algorithm changes pool IDs and requires a deployment +/// migration. The API continues treating IDs as opaque strings. +#[derive(Clone, Copy, Debug, Default)] +pub struct V1PoolIdDeriver; + +impl PoolIdDeriver for V1PoolIdDeriver { + fn derive(&self, key: &PoolKey) -> PoolId { + let mut hasher = blake3::Hasher::new_derive_key("dynamo.global_view.pool_id.v1"); + for part in [&key.site_id, &key.namespace, &key.dgd_name] { + let bytes = part.as_bytes(); + hasher.update(&(bytes.len() as u64).to_be_bytes()); + hasher.update(bytes); + } + PoolId(format!("pool-v1-{}", hasher.finalize().to_hex())) + } +} + +/// Where a pool runs. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PoolLocation { + pub region: String, + pub availability_zone: Option, + pub cluster: Option, + pub datacenter: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(site: &str, namespace: &str, dgd: &str) -> PoolKey { + PoolKey::new(site, namespace, dgd).unwrap() + } + + #[test] + fn identity_survives_deriver_restart() { + let pool = key("ohio", "dynamo", "mocker-1"); + assert_eq!(V1PoolIdDeriver.derive(&pool), V1PoolIdDeriver.derive(&pool)); + // Pinned: changing the derivation changes every deployed pool's ID. + assert_eq!( + V1PoolIdDeriver.derive(&pool).as_str(), + "pool-v1-2d83e849329ee9362567e384ac43e4d13db27a2b6a00cb79a8be9024504c56ae" + ); + } + + #[test] + fn moving_or_renaming_a_dgd_changes_identity() { + let deriver = V1PoolIdDeriver; + let original = deriver.derive(&key("ohio", "dynamo", "mocker-1")); + for changed in [ + key("west", "dynamo", "mocker-1"), + key("ohio", "other", "mocker-1"), + key("ohio", "dynamo", "mocker-2"), + ] { + assert_ne!(original, deriver.derive(&changed)); + } + } + + #[test] + fn component_boundaries_cannot_alias() { + let deriver = V1PoolIdDeriver; + assert_ne!( + deriver.derive(&key("ab", "c", "d")), + deriver.derive(&key("a", "bc", "d")), + ); + } + + #[test] + fn rejects_invalid_coordinates() { + assert_eq!( + PoolKey::new("", "dynamo", "mocker-1"), + Err(PoolKeyError::InvalidField { field: "site_id" }) + ); + assert_eq!( + PoolKey::new("ohio", " dynamo", "mocker-1"), + Err(PoolKeyError::InvalidField { field: "namespace" }) + ); + assert_eq!( + PoolKey::new("ohio", "dynamo", "mocker\n1"), + Err(PoolKeyError::InvalidField { field: "dgd_name" }) + ); + } + + #[test] + fn api_representation_is_an_opaque_string() { + let id = V1PoolIdDeriver.derive(&key("ohio", "dynamo", "mocker-1")); + assert_eq!(serde_json::to_string(&id).unwrap(), format!("\"{id}\"")); + } + + #[test] + fn key_json_is_checked_when_read() { + let pool = key("ohio", "dynamo", "mocker-1"); + let json = + serde_json::json!({"site_id": "ohio", "namespace": "dynamo", "dgd_name": "mocker-1"}); + assert_eq!(serde_json::to_value(&pool).unwrap(), json); + assert_eq!(serde_json::from_value::(json).unwrap(), pool); + let invalid = + serde_json::json!({"site_id": " ", "namespace": "dynamo", "dgd_name": "mocker-1"}); + assert!(serde_json::from_value::(invalid).is_err()); + } +} diff --git a/lib/global-common/src/replicas.rs b/lib/global-common/src/replicas.rs new file mode 100644 index 000000000000..15cbc9082d52 --- /dev/null +++ b/lib/global-common/src/replicas.rs @@ -0,0 +1,23 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Global Router replicas. Every replica needs every pool's state, so a +//! PoolRelay opens one connection per replica. + +use serde::{Deserialize, Serialize}; +use url::Url; + +use crate::watch::WatchEvent; + +/// In JSON, a plain string. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct ReplicaId(pub String); + +/// A replica's address as seen from workload clusters. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ReplicaEndpoint { + pub replica: ReplicaId, + pub url: Url, +} + +pub type ReplicaEvent = WatchEvent; diff --git a/lib/global-common/src/watch.rs b/lib/global-common/src/watch.rs new file mode 100644 index 000000000000..e0884f015eb1 --- /dev/null +++ b/lib/global-common/src/watch.rs @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! List-and-watch, as `dynamo_runtime::discovery::Discovery` does it: a watch +//! starts with one `Added` per current item, then one `Resync` with the same +//! set, then changes. A `Resync` replaces all earlier state. + +#[derive(Clone, Debug)] +pub enum WatchEvent { + /// A new or changed item. It replaces any earlier item with the same key. + Added(T), + /// An item went away. It can name a key the watcher never got, for an + /// item that went away just before the watch started; ignore those. + Removed(K), + Resync(Snapshot), +} + +/// A list, and whether it is complete. +#[derive(Clone, Debug)] +pub struct Snapshot { + pub items: Vec, + pub completeness: Completeness, +} + +/// Whether a snapshot lists every item, so a new router replica can tell a +/// complete list from a partial one. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Completeness { + /// Read from a shared store: every existing item is listed. + Exact, + /// Built from what this process has heard so far. + Estimated, +} From 008d8f2488494c9c1eb6375c628707d7fc6fbae4 Mon Sep 17 00:00:00 2001 From: Anna Tchernych Date: Mon, 26 Oct 2026 16:40:10 -0700 Subject: [PATCH 2/8] refactor(global-router): drop doc comments that restate their items Review feedback: the comments on `round_trip`, `PoolLocation` and `Snapshot` repeated the name or the fields right below them. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Anna Tchernych --- lib/global-common/src/discovery.rs | 1 - lib/global-common/src/pool.rs | 1 - lib/global-common/src/watch.rs | 1 - 3 files changed, 3 deletions(-) diff --git a/lib/global-common/src/discovery.rs b/lib/global-common/src/discovery.rs index 3a35e777708a..342736329e9e 100644 --- a/lib/global-common/src/discovery.rs +++ b/lib/global-common/src/discovery.rs @@ -158,7 +158,6 @@ mod tests { use super::*; use crate::replicas::ReplicaId; - /// Reads `json` as `T` and writes it back unchanged. fn round_trip Deserialize<'de>>(json: Value) -> T { let parsed: T = serde_json::from_value(json.clone()).unwrap(); assert_eq!(serde_json::to_value(&parsed).unwrap(), json); diff --git a/lib/global-common/src/pool.rs b/lib/global-common/src/pool.rs index fc475013dcef..5b44c7e387ba 100644 --- a/lib/global-common/src/pool.rs +++ b/lib/global-common/src/pool.rs @@ -138,7 +138,6 @@ impl PoolIdDeriver for V1PoolIdDeriver { } } -/// Where a pool runs. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct PoolLocation { pub region: String, diff --git a/lib/global-common/src/watch.rs b/lib/global-common/src/watch.rs index e0884f015eb1..d8d486677b3f 100644 --- a/lib/global-common/src/watch.rs +++ b/lib/global-common/src/watch.rs @@ -15,7 +15,6 @@ pub enum WatchEvent { Resync(Snapshot), } -/// A list, and whether it is complete. #[derive(Clone, Debug)] pub struct Snapshot { pub items: Vec, From 71f89c9e13ad17c49bc8560377d394b06d16dcf7 Mon Sep 17 00:00:00 2001 From: Anna Tchernych Date: Mon, 26 Oct 2026 17:17:41 -0700 Subject: [PATCH 3/8] fix(global-router): check a registration's model and correct three doc comments - A `POST /pools` body with an empty or blank `model` no longer parses, as with an invalid pool key, so the PoolRelay gets a 422 instead of registering a pool that no request can match. - The crate docs said the service writes the API's JSON and its clients read it; the registration body goes the other way. - `pool.rs` no longer refers to a source adapter that is not on main, or calls the V1 derivation the POC one. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Anna Tchernych --- lib/global-common/src/discovery.rs | 21 ++++++++++++++++----- lib/global-common/src/lib.rs | 2 +- lib/global-common/src/pool.rs | 7 +++---- 3 files changed, 20 insertions(+), 10 deletions(-) diff --git a/lib/global-common/src/discovery.rs b/lib/global-common/src/discovery.rs index 342736329e9e..4380af7c7a12 100644 --- a/lib/global-common/src/discovery.rs +++ b/lib/global-common/src/discovery.rs @@ -8,11 +8,11 @@ use std::net::IpAddr; use std::time::Duration; -use serde::{Deserialize, Serialize}; +use serde::{Deserialize, Deserializer, Serialize}; use thiserror::Error; use url::Url; -use crate::pool::{PoolKey, PoolLocation}; +use crate::pool::{self, PoolKey, PoolLocation}; use crate::replicas::ReplicaEndpoint; /// `POST /pools` body. A PoolRelay sends it on a timer, and each call is also @@ -21,7 +21,9 @@ use crate::replicas::ReplicaEndpoint; pub struct PoolRegistration { pub key: PoolKey, pub location: PoolLocation, - /// The model or checkpoint the pool serves. + /// The model or checkpoint the pool serves. Reading it checks it as + /// [`PoolKey::new`] checks the key's fields. + #[serde(deserialize_with = "checked_model")] pub model: String, #[serde(default)] pub capabilities: Vec, @@ -32,6 +34,12 @@ pub struct PoolRegistration { pub capacity: Vec, } +fn checked_model<'de, D: Deserializer<'de>>(deserializer: D) -> Result { + let model = String::deserialize(deserializer)?; + pool::validate(&model, "model").map_err(serde::de::Error::custom)?; + Ok(model) +} + /// The workers of one hardware SKU. #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] pub struct Capacity { @@ -174,14 +182,17 @@ mod tests { "cluster": "aws-west", "datacenter": null }); - round_trip::(json!({ + let mut registration = json!({ "key": key, "location": location, "model": "Qwen/Qwen3-0.6B", "capabilities": ["aggregated"], "frontends": ["http://10.43.1.2:8000/"], "capacity": [{"sku": "H100", "count": 8}] - })); + }); + round_trip::(registration.clone()); + registration["model"] = json!(" "); + assert!(serde_json::from_value::(registration).is_err()); let pools = round_trip::(json!({ "pools": [{ "key": key, diff --git a/lib/global-common/src/lib.rs b/lib/global-common/src/lib.rs index 153d9bec1157..b481ea73e9c8 100644 --- a/lib/global-common/src/lib.rs +++ b/lib/global-common/src/lib.rs @@ -6,7 +6,7 @@ //! - A pool's identity and location: [`PoolKey`], [`PoolId`] and its deriver //! [`V1PoolIdDeriver`], and [`PoolLocation`]. //! - The JSON of the Global Discovery Service's HTTP API, which the service -//! writes and its clients read: +//! and its clients exchange: //! //! | Call | Body | Reply | //! |---|---|---| diff --git a/lib/global-common/src/pool.rs b/lib/global-common/src/pool.rs index 5b44c7e387ba..ee7d8d1d821f 100644 --- a/lib/global-common/src/pool.rs +++ b/lib/global-common/src/pool.rs @@ -4,8 +4,7 @@ //! Runtime-independent identities for pools in a Global View. //! //! A routing pool is one DGD in the initial deployment. Its identity is stable -//! across process and relay restarts, while relay and KV producer generations -//! are tracked separately by the source adapter. +//! across process and relay restarts. use std::fmt; @@ -29,7 +28,7 @@ impl fmt::Display for PoolId { } } -/// Stable deployment coordinates used by the POC derivation. +/// Stable deployment coordinates that a [`PoolIdDeriver`] turns into a [`PoolId`]. /// /// `namespace` is the Kubernetes namespace containing the DGD, not the /// Dynamo runtime namespace derived by the operator. In JSON it is @@ -106,7 +105,7 @@ impl PoolKey { } } -fn validate(value: &str, field: &'static str) -> Result<(), PoolKeyError> { +pub(crate) fn validate(value: &str, field: &'static str) -> Result<(), PoolKeyError> { if value.is_empty() || value.trim() != value || value.chars().any(char::is_control) { return Err(PoolKeyError::InvalidField { field }); } From 8e30f3c5e5044d693de3048378b0c19354f745aa Mon Sep 17 00:00:00 2001 From: Anna Tchernych Date: Mon, 26 Oct 2026 20:22:59 -0700 Subject: [PATCH 4/8] fix(global-router): let a client follow any lease the service grants The one-day cap made the shared `Lease` stricter than the client in #17208: once that client reads the shared type, a reply with a longer lease would fail to parse. The cap only protects the service's own clock arithmetic, so the service now bounds the lease it grants (#17371), and `Lease` keeps the checks a client needs: the interval is above zero and below the lease. Also drops `LeaseError`'s doc comment, which restated its use. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Anna Tchernych --- lib/global-common/src/discovery.rs | 31 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/lib/global-common/src/discovery.rs b/lib/global-common/src/discovery.rs index 4380af7c7a12..ef3d753cdb0e 100644 --- a/lib/global-common/src/discovery.rs +++ b/lib/global-common/src/discovery.rs @@ -51,10 +51,11 @@ pub struct Capacity { /// lease passes with no successful registration, the pool drops out of /// `GET /pools`. /// -/// Both are whole seconds, the interval is above zero and below the lease, -/// and the lease is at most [`Lease::MAX_DURATION_S`]. A reply that breaks -/// this does not parse, so it cannot make a PoolRelay register in a tight -/// loop, and every `Lease` can be written as JSON and read back. +/// Both are whole seconds, and the interval is above zero and below the +/// lease. A reply that breaks this does not parse, so it cannot make a +/// PoolRelay register in a tight loop, and every `Lease` can be written as +/// JSON and read back. There is no upper bound: a service bounds the lease it +/// grants, and a client follows any lease it is given. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(try_from = "LeaseFields")] pub struct Lease { @@ -63,14 +64,8 @@ pub struct Lease { } impl Lease { - /// One day. - pub const MAX_DURATION_S: u64 = 86_400; - pub fn new(heartbeat_interval_s: u64, lease_duration_s: u64) -> Result { - if heartbeat_interval_s == 0 - || heartbeat_interval_s >= lease_duration_s - || lease_duration_s > Self::MAX_DURATION_S - { + if heartbeat_interval_s == 0 || heartbeat_interval_s >= lease_duration_s { return Err(LeaseError { heartbeat_interval_s, lease_duration_s, @@ -91,11 +86,10 @@ impl Lease { } } -/// A lease that [`Lease::new`] refuses. #[derive(Debug, Error, PartialEq, Eq)] #[error( "heartbeat interval {heartbeat_interval_s}s must be above zero and below the \ - {lease_duration_s}s lease, which may be at most a day" + {lease_duration_s}s lease" )] pub struct LeaseError { pub heartbeat_interval_s: u64, @@ -219,12 +213,17 @@ mod tests { } #[test] - fn a_lease_that_would_spin_or_overflow_does_not_parse() { + fn a_lease_parses_unless_it_would_make_a_relay_spin() { let lease = round_trip::(json!({"heartbeat_interval_s": 5, "lease_duration_s": 15})); assert_eq!(lease, Lease::new(5, 15).unwrap()); assert_eq!(lease.heartbeat_interval(), Duration::from_secs(5)); - let too_long = Lease::MAX_DURATION_S + 1; - for (interval, lease) in [(0, 15), (15, 15), (20, 15), (5, too_long)] { + // 48 hours, and the longest a client could be given. + for (interval, lease) in [(60, 172_800), (1, u64::MAX)] { + round_trip::( + json!({"heartbeat_interval_s": interval, "lease_duration_s": lease}), + ); + } + for (interval, lease) in [(0, 15), (15, 15), (20, 15)] { let body = json!({"heartbeat_interval_s": interval, "lease_duration_s": lease}); assert!( serde_json::from_value::(body).is_err(), From 8b6c23ea5db170dbe80b15d5555ad4e27db1544f Mon Sep 17 00:00:00 2001 From: Anna Tchernych Date: Mon, 26 Oct 2026 20:57:47 -0700 Subject: [PATCH 5/8] feat(global-router): name a pool's PoolRelay by every address of its pod A pool's listing named its PoolRelay by one address. On a dual-stack cluster the PoolRelay's pod has two, and it connects to a router from whichever family the router's address uses, so a router matching one address could turn the right PoolRelay away. `relay_identities` now lists each of them. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Anna Tchernych --- lib/global-common/src/discovery.rs | 14 ++++++++------ lib/global-common/src/lib.rs | 2 +- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/lib/global-common/src/discovery.rs b/lib/global-common/src/discovery.rs index ef3d753cdb0e..c08b881201bd 100644 --- a/lib/global-common/src/discovery.rs +++ b/lib/global-common/src/discovery.rs @@ -133,16 +133,18 @@ pub struct PoolListing { pub frontends: Vec, #[serde(default)] pub capacity: Vec, - /// The only PoolRelay allowed to stream this pool's state, when the - /// service knows it. On a trusted network it is [`address_identity`]. + /// The only PoolRelay allowed to stream this pool's state, by each + /// identity it can connect with. On a trusted network these are its pod's + /// addresses, as [`address_identity`] writes them: two on a dual-stack + /// cluster. Empty when the service does not know the PoolRelay. #[serde(default)] - pub relay_identity: Option, + pub relay_identities: Vec, } -/// Names a PoolRelay by its address on a trusted network, as +/// Names a PoolRelay by one of its addresses on a trusted network, as /// `ip:
`. An IPv4-mapped IPv6 address is written as IPv4, as /// EndpointSlices list it. A router matches a connecting PoolRelay against -/// [`PoolListing::relay_identity`] with this. +/// [`PoolListing::relay_identities`] with this. pub fn address_identity(address: IpAddr) -> String { format!("ip:{}", address.to_canonical()) } @@ -195,7 +197,7 @@ mod tests { "capabilities": [], "frontends": [], "capacity": [], - "relay_identity": "ip:10.43.66.175" + "relay_identities": ["ip:10.43.66.175", "ip:fd00::5"] }], "complete": true })); diff --git a/lib/global-common/src/lib.rs b/lib/global-common/src/lib.rs index b481ea73e9c8..2ab384a5d8c6 100644 --- a/lib/global-common/src/lib.rs +++ b/lib/global-common/src/lib.rs @@ -15,7 +15,7 @@ //! | `GET /routers` | | [`RoutersResponse`] | //! //! A router matches a connecting PoolRelay against a listing's -//! `relay_identity` with [`address_identity`]. +//! `relay_identities` with [`address_identity`]. //! - Global Router replicas ([`ReplicaEndpoint`]), and list-and-watch events //! and snapshots ([`WatchEvent`], [`Snapshot`]), which the service's watch //! contract and its clients share. From c7ee161c56d79475d98539349990a4c8ff65337f Mon Sep 17 00:00:00 2001 From: Anna Tchernych Date: Tue, 27 Oct 2026 09:04:24 -0700 Subject: [PATCH 6/8] test(global-router): drop checks the JSON round trips already make round_trip fails unless the parsed value serializes back to the same JSON, so the field assertions after it could not catch anything more. A lease of u64::MAX seconds already shows that a lease has no upper bound, so the 48-hour case goes too. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Anna Tchernych --- lib/global-common/src/discovery.rs | 18 ++++-------------- 1 file changed, 4 insertions(+), 14 deletions(-) diff --git a/lib/global-common/src/discovery.rs b/lib/global-common/src/discovery.rs index c08b881201bd..a5b6dab1b94d 100644 --- a/lib/global-common/src/discovery.rs +++ b/lib/global-common/src/discovery.rs @@ -160,7 +160,6 @@ mod tests { use serde_json::{Value, json}; use super::*; - use crate::replicas::ReplicaId; fn round_trip Deserialize<'de>>(json: Value) -> T { let parsed: T = serde_json::from_value(json.clone()).unwrap(); @@ -189,7 +188,7 @@ mod tests { round_trip::(registration.clone()); registration["model"] = json!(" "); assert!(serde_json::from_value::(registration).is_err()); - let pools = round_trip::(json!({ + round_trip::(json!({ "pools": [{ "key": key, "location": location, @@ -201,14 +200,9 @@ mod tests { }], "complete": true })); - assert!(pools.complete); - let routers = round_trip::(json!({ + round_trip::(json!({ "routers": [{"replica": "aws-ohio/global-router-0", "url": "http://10.42.0.5:50051/"}] })); - assert_eq!( - routers.routers[0].replica, - ReplicaId("aws-ohio/global-router-0".to_owned()) - ); let mapped: IpAddr = "::ffff:10.43.66.175".parse().unwrap(); assert_eq!(address_identity(mapped), "ip:10.43.66.175"); assert_eq!(address_identity("fd00::2".parse().unwrap()), "ip:fd00::2"); @@ -219,12 +213,8 @@ mod tests { let lease = round_trip::(json!({"heartbeat_interval_s": 5, "lease_duration_s": 15})); assert_eq!(lease, Lease::new(5, 15).unwrap()); assert_eq!(lease.heartbeat_interval(), Duration::from_secs(5)); - // 48 hours, and the longest a client could be given. - for (interval, lease) in [(60, 172_800), (1, u64::MAX)] { - round_trip::( - json!({"heartbeat_interval_s": interval, "lease_duration_s": lease}), - ); - } + // The longest lease a client could be given. + round_trip::(json!({"heartbeat_interval_s": 1, "lease_duration_s": u64::MAX})); for (interval, lease) in [(0, 15), (15, 15), (20, 15)] { let body = json!({"heartbeat_interval_s": interval, "lease_duration_s": lease}); assert!( From 4323f2ab719fbba9f116c5531a4fb382b7b902ab Mon Sep 17 00:00:00 2001 From: Anna Tchernych Date: Wed, 28 Oct 2026 18:28:23 -0700 Subject: [PATCH 7/8] refactor(global-router): rename the shared types as reviewers asked From Neelay Shah's review: - PoolKey's first field is cluster_id, not site_id, in Rust and in JSON. It is the cluster ID the pool's PoolRelay runs with (--cluster-id, DYN_CLUSTER_ID), so a pool key and its relay share one name. Where MCS also reports the pool, that ID must be the cluster's name in the ClusterSet, and the doc says so. Pool IDs do not change: the deriver hashes the values, not the field names. - PoolLocation drops cluster, which the key carries. - ReplicaEndpoint, ReplicaId and ReplicaEvent become RouterReplica, RouterReplicaId and RouterReplicaEvent, since relays and workers have replicas too. Their JSON does not change. - WatchEvent::Added becomes Upserted, Lease::lease_duration() becomes duration(), and Capacity.count becomes workers, in JSON too. Lease is now #[must_use], as the client's copy was, so the client keeps that check when it moves onto this type. From the review threads: a test, from Dmitry Tokarev's review, that a reader skips unknown fields and fills missing lists, and two duplicate assertions removed. A pre-push review added a test that a reply without its list does not parse, so an empty body never reads as "no routers". Co-authored-by: Dmitry Tokarev <191198572+dmitry-tokarev-nv@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Anna Tchernych --- lib/global-common/src/discovery.rs | 65 +++++++++++++++++++++++------- lib/global-common/src/lib.rs | 4 +- lib/global-common/src/pool.rs | 49 +++++++++++----------- lib/global-common/src/replicas.rs | 10 ++--- lib/global-common/src/watch.rs | 7 ++-- 5 files changed, 88 insertions(+), 47 deletions(-) diff --git a/lib/global-common/src/discovery.rs b/lib/global-common/src/discovery.rs index a5b6dab1b94d..b838cb1c236d 100644 --- a/lib/global-common/src/discovery.rs +++ b/lib/global-common/src/discovery.rs @@ -13,7 +13,7 @@ use thiserror::Error; use url::Url; use crate::pool::{self, PoolKey, PoolLocation}; -use crate::replicas::ReplicaEndpoint; +use crate::replicas::RouterReplica; /// `POST /pools` body. A PoolRelay sends it on a timer, and each call is also /// the pool's heartbeat. @@ -44,7 +44,7 @@ fn checked_model<'de, D: Deserializer<'de>>(deserializer: D) -> Result Duration { + pub fn duration(&self) -> Duration { Duration::from_secs(self.lease_duration_s) } } @@ -152,7 +153,7 @@ pub fn address_identity(address: IpAddr) -> String { /// `GET /routers` reply: every Global Router replica. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RoutersResponse { - pub routers: Vec, + pub routers: Vec, } #[cfg(test)] @@ -170,20 +171,16 @@ mod tests { #[test] fn json_matches_the_service_and_its_clients() { let key = - json!({"site_id": "aws-west", "namespace": "dynamo", "dgd_name": "mocker-west-1"}); - let location = json!({ - "region": "us-west-2", - "availability_zone": null, - "cluster": "aws-west", - "datacenter": null - }); + json!({"cluster_id": "aws-west", "namespace": "dynamo", "dgd_name": "mocker-west-1"}); + let location = + json!({"region": "us-west-2", "availability_zone": null, "datacenter": null}); let mut registration = json!({ "key": key, "location": location, "model": "Qwen/Qwen3-0.6B", "capabilities": ["aggregated"], "frontends": ["http://10.43.1.2:8000/"], - "capacity": [{"sku": "H100", "count": 8}] + "capacity": [{"sku": "H100", "workers": 8}] }); round_trip::(registration.clone()); registration["model"] = json!(" "); @@ -211,10 +208,12 @@ mod tests { #[test] fn a_lease_parses_unless_it_would_make_a_relay_spin() { let lease = round_trip::(json!({"heartbeat_interval_s": 5, "lease_duration_s": 15})); - assert_eq!(lease, Lease::new(5, 15).unwrap()); assert_eq!(lease.heartbeat_interval(), Duration::from_secs(5)); + assert_eq!(lease.duration(), Duration::from_secs(15)); // The longest lease a client could be given. - round_trip::(json!({"heartbeat_interval_s": 1, "lease_duration_s": u64::MAX})); + let longest = + round_trip::(json!({"heartbeat_interval_s": 1, "lease_duration_s": u64::MAX})); + assert_eq!(longest.duration(), Duration::from_secs(u64::MAX)); for (interval, lease) in [(0, 15), (15, 15), (20, 15)] { let body = json!({"heartbeat_interval_s": interval, "lease_duration_s": lease}); assert!( @@ -223,4 +222,42 @@ mod tests { ); } } + + #[test] + fn a_reply_without_its_list_does_not_parse() { + // `{}` must not read as "no routers", or a PoolRelay would drop every + // router connection. + assert!(serde_json::from_value::(json!({})).is_err()); + assert!(serde_json::from_value::(json!({"complete": true})).is_err()); + } + + #[test] + fn a_reader_skips_unknown_fields_and_fills_missing_lists() { + // Either side can add a field, at any depth, before the other reads it. + let key = json!({"cluster_id": "aws-west", "namespace": "dynamo", "dgd_name": "mocker-west-1", "new": 1}); + let location = + json!({"region": "us-west-2", "availability_zone": null, "datacenter": null, "new": 1}); + serde_json::from_value::(json!({ + "key": key, "location": location, "model": "Qwen/Qwen3-0.6B", "new": 1 + })) + .unwrap(); + serde_json::from_value::(json!({ + "pools": [{ + "key": key, "location": location, "model": null, + "capacity": [{"sku": "H100", "workers": 8, "new": 1}], "new": 1 + }], + "complete": true, + "new": 1 + })) + .unwrap(); + let _: Lease = serde_json::from_value( + json!({"heartbeat_interval_s": 5, "lease_duration_s": 15, "new": 1}), + ) + .unwrap(); + serde_json::from_value::(json!({ + "routers": [{"replica": "aws-ohio/global-router-0", "url": "http://10.42.0.5:50051/", "new": 1}], + "new": 1 + })) + .unwrap(); + } } diff --git a/lib/global-common/src/lib.rs b/lib/global-common/src/lib.rs index f78299d923f4..7adb7a080391 100644 --- a/lib/global-common/src/lib.rs +++ b/lib/global-common/src/lib.rs @@ -16,7 +16,7 @@ //! //! A router matches a connecting PoolRelay against a listing's //! `relay_identities` with [`address_identity`]. -//! - Global Router replicas ([`ReplicaEndpoint`]), and list-and-watch events +//! - Global Router replicas ([`RouterReplica`]), and list-and-watch events //! and snapshots ([`WatchEvent`], [`Snapshot`]), which the service's watch //! contract and its clients share. @@ -30,5 +30,5 @@ pub use discovery::{ address_identity, }; pub use pool::{PoolId, PoolIdDeriver, PoolKey, PoolKeyError, PoolLocation, V1PoolIdDeriver}; -pub use replicas::{ReplicaEndpoint, ReplicaEvent, ReplicaId}; +pub use replicas::{RouterReplica, RouterReplicaEvent, RouterReplicaId}; pub use watch::{Completeness, Snapshot, WatchEvent}; diff --git a/lib/global-common/src/pool.rs b/lib/global-common/src/pool.rs index ee7d8d1d821f..9de5616acf29 100644 --- a/lib/global-common/src/pool.rs +++ b/lib/global-common/src/pool.rs @@ -30,21 +30,24 @@ impl fmt::Display for PoolId { /// Stable deployment coordinates that a [`PoolIdDeriver`] turns into a [`PoolId`]. /// -/// `namespace` is the Kubernetes namespace containing the DGD, not the -/// Dynamo runtime namespace derived by the operator. In JSON it is -/// `{"site_id", "namespace", "dgd_name"}`, and reading it checks each field as -/// [`PoolKey::new`] does. +/// `cluster_id` is the cluster ID the pool's PoolRelay runs with +/// (`--cluster-id`, `DYN_CLUSTER_ID`). Where MCS also reports the pool, that +/// ID must be the cluster's name in the ClusterSet, so a registration and the +/// pool MCS reports have one key. `namespace` is the Kubernetes namespace +/// containing the DGD, not the Dynamo runtime namespace derived by the +/// operator. In JSON it is `{"cluster_id", "namespace", "dgd_name"}`, and +/// reading it checks each field as [`PoolKey::new`] does. #[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(try_from = "PoolKeyFields", into = "PoolKeyFields")] pub struct PoolKey { - site_id: String, + cluster_id: String, namespace: String, dgd_name: String, } #[derive(Serialize, Deserialize)] struct PoolKeyFields { - site_id: String, + cluster_id: String, namespace: String, dgd_name: String, } @@ -53,14 +56,14 @@ impl TryFrom for PoolKey { type Error = PoolKeyError; fn try_from(fields: PoolKeyFields) -> Result { - Self::new(fields.site_id, fields.namespace, fields.dgd_name) + Self::new(fields.cluster_id, fields.namespace, fields.dgd_name) } } impl From for PoolKeyFields { fn from(key: PoolKey) -> Self { Self { - site_id: key.site_id, + cluster_id: key.cluster_id, namespace: key.namespace, dgd_name: key.dgd_name, } @@ -75,25 +78,25 @@ pub enum PoolKeyError { impl PoolKey { pub fn new( - site_id: impl Into, + cluster_id: impl Into, namespace: impl Into, dgd_name: impl Into, ) -> Result { - let site_id = site_id.into(); + let cluster_id = cluster_id.into(); let namespace = namespace.into(); let dgd_name = dgd_name.into(); - validate(&site_id, "site_id")?; + validate(&cluster_id, "cluster_id")?; validate(&namespace, "namespace")?; validate(&dgd_name, "dgd_name")?; Ok(Self { - site_id, + cluster_id, namespace, dgd_name, }) } - pub fn site_id(&self) -> &str { - &self.site_id + pub fn cluster_id(&self) -> &str { + &self.cluster_id } pub fn namespace(&self) -> &str { @@ -128,7 +131,7 @@ pub struct V1PoolIdDeriver; impl PoolIdDeriver for V1PoolIdDeriver { fn derive(&self, key: &PoolKey) -> PoolId { let mut hasher = blake3::Hasher::new_derive_key("dynamo.global_view.pool_id.v1"); - for part in [&key.site_id, &key.namespace, &key.dgd_name] { + for part in [&key.cluster_id, &key.namespace, &key.dgd_name] { let bytes = part.as_bytes(); hasher.update(&(bytes.len() as u64).to_be_bytes()); hasher.update(bytes); @@ -137,11 +140,11 @@ impl PoolIdDeriver for V1PoolIdDeriver { } } +/// The cluster a pool runs in is [`PoolKey::cluster_id`], not a field here. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct PoolLocation { pub region: String, pub availability_zone: Option, - pub cluster: Option, pub datacenter: Option, } @@ -149,14 +152,13 @@ pub struct PoolLocation { mod tests { use super::*; - fn key(site: &str, namespace: &str, dgd: &str) -> PoolKey { - PoolKey::new(site, namespace, dgd).unwrap() + fn key(cluster: &str, namespace: &str, dgd: &str) -> PoolKey { + PoolKey::new(cluster, namespace, dgd).unwrap() } #[test] fn identity_survives_deriver_restart() { let pool = key("ohio", "dynamo", "mocker-1"); - assert_eq!(V1PoolIdDeriver.derive(&pool), V1PoolIdDeriver.derive(&pool)); // Pinned: changing the derivation changes every deployed pool's ID. assert_eq!( V1PoolIdDeriver.derive(&pool).as_str(), @@ -190,7 +192,9 @@ mod tests { fn rejects_invalid_coordinates() { assert_eq!( PoolKey::new("", "dynamo", "mocker-1"), - Err(PoolKeyError::InvalidField { field: "site_id" }) + Err(PoolKeyError::InvalidField { + field: "cluster_id" + }) ); assert_eq!( PoolKey::new("ohio", " dynamo", "mocker-1"), @@ -211,12 +215,11 @@ mod tests { #[test] fn key_json_is_checked_when_read() { let pool = key("ohio", "dynamo", "mocker-1"); - let json = - serde_json::json!({"site_id": "ohio", "namespace": "dynamo", "dgd_name": "mocker-1"}); + let json = serde_json::json!({"cluster_id": "ohio", "namespace": "dynamo", "dgd_name": "mocker-1"}); assert_eq!(serde_json::to_value(&pool).unwrap(), json); assert_eq!(serde_json::from_value::(json).unwrap(), pool); let invalid = - serde_json::json!({"site_id": " ", "namespace": "dynamo", "dgd_name": "mocker-1"}); + serde_json::json!({"cluster_id": " ", "namespace": "dynamo", "dgd_name": "mocker-1"}); assert!(serde_json::from_value::(invalid).is_err()); } } diff --git a/lib/global-common/src/replicas.rs b/lib/global-common/src/replicas.rs index 15cbc9082d52..4bea24c2fd14 100644 --- a/lib/global-common/src/replicas.rs +++ b/lib/global-common/src/replicas.rs @@ -11,13 +11,13 @@ use crate::watch::WatchEvent; /// In JSON, a plain string. #[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] -pub struct ReplicaId(pub String); +pub struct RouterReplicaId(pub String); -/// A replica's address as seen from workload clusters. +/// A Global Router replica and the address workload clusters reach it at. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct ReplicaEndpoint { - pub replica: ReplicaId, +pub struct RouterReplica { + pub replica: RouterReplicaId, pub url: Url, } -pub type ReplicaEvent = WatchEvent; +pub type RouterReplicaEvent = WatchEvent; diff --git a/lib/global-common/src/watch.rs b/lib/global-common/src/watch.rs index d8d486677b3f..12ad773fd94b 100644 --- a/lib/global-common/src/watch.rs +++ b/lib/global-common/src/watch.rs @@ -2,13 +2,14 @@ // SPDX-License-Identifier: Apache-2.0 //! List-and-watch, as `dynamo_runtime::discovery::Discovery` does it: a watch -//! starts with one `Added` per current item, then one `Resync` with the same -//! set, then changes. A `Resync` replaces all earlier state. +//! starts with one `Upserted` per current item (the runtime's `Added`), then +//! one `Resync` with the same set, then changes. A `Resync` replaces all +//! earlier state. #[derive(Clone, Debug)] pub enum WatchEvent { /// A new or changed item. It replaces any earlier item with the same key. - Added(T), + Upserted(T), /// An item went away. It can name a key the watcher never got, for an /// item that went away just before the watch started; ignore those. Removed(K), From 5aadafa11eb432eb17425595969f887d71754f9d Mon Sep 17 00:00:00 2001 From: Anna Tchernych Date: Wed, 28 Oct 2026 18:28:23 -0700 Subject: [PATCH 8/8] refactor(global-router): build the discovery client on the shared types Neelay Shah and Jacob Ioffe asked on #17208 that its client move onto dynamo-global-common in this PR, so two PoolKey definitions never sit on main together. lib/global-discovery-client drops its own copies of the JSON types, router replicas and watch events. It re-exports the shared types its calls send and return, with their error types; the watch types come from dynamo-global-common. Its calls and errors do not change. What it sends follows the shared types: the key's cluster_id, a location without cluster, and capacity in workers. A PoolKey is built with PoolKey::new, which checks its fields. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Anna Tchernych --- Cargo.lock | 1 + lib/global-discovery-client/Cargo.toml | 1 + lib/global-discovery-client/src/client.rs | 6 +- lib/global-discovery-client/src/lib.rs | 14 +-- lib/global-discovery-client/src/replicas.rs | 22 ---- lib/global-discovery-client/src/watch.rs | 28 ----- lib/global-discovery-client/src/wire.rs | 117 ------------------ .../tests/relay_calls.rs | 30 ++--- 8 files changed, 19 insertions(+), 200 deletions(-) delete mode 100644 lib/global-discovery-client/src/replicas.rs delete mode 100644 lib/global-discovery-client/src/watch.rs delete mode 100644 lib/global-discovery-client/src/wire.rs diff --git a/Cargo.lock b/Cargo.lock index 7958d81bac26..dbe6b17e27ab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2492,6 +2492,7 @@ name = "dynamo-global-discovery-client" version = "1.7.0" dependencies = [ "axum 0.8.4", + "dynamo-global-common", "reqwest 0.12.28", "serde", "serde_json", diff --git a/lib/global-discovery-client/Cargo.toml b/lib/global-discovery-client/Cargo.toml index 31d5920254e9..0445b966a691 100644 --- a/lib/global-discovery-client/Cargo.toml +++ b/lib/global-discovery-client/Cargo.toml @@ -12,6 +12,7 @@ homepage.workspace = true repository.workspace = true [dependencies] +dynamo-global-common = { workspace = true } reqwest = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/lib/global-discovery-client/src/client.rs b/lib/global-discovery-client/src/client.rs index 0c6b77b76a69..672616103e24 100644 --- a/lib/global-discovery-client/src/client.rs +++ b/lib/global-discovery-client/src/client.rs @@ -3,14 +3,12 @@ use std::time::Duration; +use dynamo_global_common::{Lease, PoolRegistration, RouterReplica, RoutersResponse}; use reqwest::StatusCode; use serde::de::DeserializeOwned; use thiserror::Error; use url::Url; -use crate::replicas::ReplicaEndpoint; -use crate::wire::{Lease, PoolRegistration, RoutersResponse}; - #[derive(Debug, Error)] #[non_exhaustive] pub enum DiscoveryClientError { @@ -75,7 +73,7 @@ impl GlobalDiscoveryClient { /// /// Each call returns the full current set. If any entry does not parse, /// the whole call fails, so a caller never acts on a partial list. - pub async fn routers(&self) -> Result, DiscoveryClientError> { + pub async fn routers(&self) -> Result, DiscoveryClientError> { let url = &self.routers_url; let response = self.send(self.http.get(url.clone()), url).await?; let body: RoutersResponse = read_json(response, url).await?; diff --git a/lib/global-discovery-client/src/lib.rs b/lib/global-discovery-client/src/lib.rs index de0081896992..404703d2d050 100644 --- a/lib/global-discovery-client/src/lib.rs +++ b/lib/global-discovery-client/src/lib.rs @@ -6,15 +6,13 @@ //! ([`GlobalDiscoveryClient::routers`]) and registering its pool //! ([`GlobalDiscoveryClient::register_pool`]). //! -//! The crate also defines the router replica types and the list-and-watch -//! events. +//! The types it sends and receives are `dynamo-global-common`'s, re-exported +//! here, so the client reads and writes the JSON as the service does. mod client; -mod replicas; -mod watch; -mod wire; pub use client::{DiscoveryClientError, GlobalDiscoveryClient}; -pub use replicas::{ReplicaEndpoint, ReplicaEvent, ReplicaId}; -pub use watch::{Completeness, Snapshot, WatchEvent}; -pub use wire::{Capacity, Lease, PoolKey, PoolLocation, PoolRegistration}; +pub use dynamo_global_common::{ + Capacity, Lease, LeaseError, PoolKey, PoolKeyError, PoolLocation, PoolRegistration, + RouterReplica, RouterReplicaId, +}; diff --git a/lib/global-discovery-client/src/replicas.rs b/lib/global-discovery-client/src/replicas.rs deleted file mode 100644 index b34b5e526a1d..000000000000 --- a/lib/global-discovery-client/src/replicas.rs +++ /dev/null @@ -1,22 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -//! Global Router replicas a PoolRelay connects to. Every replica needs every -//! pool's state, so a PoolRelay opens one connection per replica. - -use serde::{Deserialize, Serialize}; -use url::Url; - -use crate::watch::WatchEvent; - -#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] -pub struct ReplicaId(pub String); - -/// A replica's address as seen from workload clusters. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct ReplicaEndpoint { - pub replica: ReplicaId, - pub url: Url, -} - -pub type ReplicaEvent = WatchEvent; diff --git a/lib/global-discovery-client/src/watch.rs b/lib/global-discovery-client/src/watch.rs deleted file mode 100644 index 362ebd5fa292..000000000000 --- a/lib/global-discovery-client/src/watch.rs +++ /dev/null @@ -1,28 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -//! List-and-watch events. A `Resync` replaces all earlier state. - -#[derive(Clone, Debug)] -pub enum WatchEvent { - /// A new or changed item. It replaces any earlier item with the same key. - Added(T), - Removed(K), - Resync(Snapshot), -} - -#[derive(Clone, Debug)] -pub struct Snapshot { - pub items: Vec, - pub completeness: Completeness, -} - -/// Whether a snapshot lists every item, so a new router replica can tell a -/// complete list from a partial one. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum Completeness { - /// Read from a shared store, so it lists every existing item. - Exact, - /// Built from the updates this replica has received so far. - Estimated, -} diff --git a/lib/global-discovery-client/src/wire.rs b/lib/global-discovery-client/src/wire.rs deleted file mode 100644 index 2f1a550a6c6e..000000000000 --- a/lib/global-discovery-client/src/wire.rs +++ /dev/null @@ -1,117 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -//! JSON exchanged with the Global Discovery Service. The client ignores -//! response fields it does not know, so the service can add fields first. - -use std::time::Duration; - -use serde::{Deserialize, Serialize}; -use url::Url; - -use crate::replicas::ReplicaEndpoint; - -#[derive(Debug, Deserialize)] -pub(crate) struct RoutersResponse { - pub routers: Vec, -} - -/// `POST /pools` reply: the service's timing for registrations. Register again -/// after `heartbeat_interval`. If `lease_duration` passes with no successful -/// registration, the pool drops out of `GET /pools`. A missing body, a zero -/// interval, or an interval not below the lease does not parse, so a bad reply -/// cannot make a PoolRelay register in a tight loop. -#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)] -#[serde(try_from = "LeaseBody")] -#[must_use = "wait `heartbeat_interval` before the next `register_pool` call"] -pub struct Lease { - pub heartbeat_interval: Duration, - pub lease_duration: Duration, -} - -#[derive(Deserialize)] -struct LeaseBody { - heartbeat_interval_s: u64, - lease_duration_s: u64, -} - -impl TryFrom for Lease { - type Error = String; - - fn try_from(body: LeaseBody) -> Result { - if body.heartbeat_interval_s == 0 || body.heartbeat_interval_s >= body.lease_duration_s { - return Err(format!( - "heartbeat interval {}s must be above zero and below the {}s lease", - body.heartbeat_interval_s, body.lease_duration_s - )); - } - Ok(Self { - heartbeat_interval: Duration::from_secs(body.heartbeat_interval_s), - lease_duration: Duration::from_secs(body.lease_duration_s), - }) - } -} - -/// `POST /pools` request body. A PoolRelay sends it on every registration. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct PoolRegistration { - pub key: PoolKey, - pub location: PoolLocation, - /// The model or checkpoint the pool serves. - pub model: String, - pub capabilities: Vec, - /// Local frontends the Global Router dispatches requests to. - pub frontends: Vec, - pub capacity: Vec, -} - -/// `namespace` is the Kubernetes namespace containing the DGD, not the -/// Dynamo runtime namespace derived by the operator. -#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] -pub struct PoolKey { - pub site_id: String, - pub namespace: String, - pub dgd_name: String, -} - -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct PoolLocation { - pub region: String, - pub availability_zone: Option, - pub cluster: Option, - pub datacenter: Option, -} - -/// The workers of one hardware SKU. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct Capacity { - pub sku: String, - pub count: u32, -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn a_lease_that_would_spin_does_not_parse() { - let lease: Lease = - serde_json::from_str(r#"{ "heartbeat_interval_s": 5, "lease_duration_s": 15 }"#) - .unwrap(); - assert_eq!( - lease, - Lease { - heartbeat_interval: Duration::from_secs(5), - lease_duration: Duration::from_secs(15), - } - ); - for body in [ - "", - r#"{ "heartbeat_interval_s": 0, "lease_duration_s": 15 }"#, - r#"{ "heartbeat_interval_s": 15, "lease_duration_s": 15 }"#, - r#"{ "heartbeat_interval_s": 20, "lease_duration_s": 15 }"#, - ] { - assert!(serde_json::from_str::(body).is_err(), "{body:?}"); - } - } -} diff --git a/lib/global-discovery-client/tests/relay_calls.rs b/lib/global-discovery-client/tests/relay_calls.rs index b59a004af1ec..740c518f44f0 100644 --- a/lib/global-discovery-client/tests/relay_calls.rs +++ b/lib/global-discovery-client/tests/relay_calls.rs @@ -11,7 +11,7 @@ use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; use dynamo_global_discovery_client::{ Capacity, DiscoveryClientError, GlobalDiscoveryClient, Lease, PoolKey, PoolLocation, - PoolRegistration, ReplicaEndpoint, ReplicaId, + PoolRegistration, RouterReplica, RouterReplicaId, }; use serde_json::{Value, json}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -77,24 +77,19 @@ async fn serve_register( Json(json!({ "heartbeat_interval_s": 5, "lease_duration_s": 15 })) } -fn replica(id: &str, url: &str) -> ReplicaEndpoint { - ReplicaEndpoint { - replica: ReplicaId(id.into()), +fn replica(id: &str, url: &str) -> RouterReplica { + RouterReplica { + replica: RouterReplicaId(id.into()), url: url.parse().unwrap(), } } fn registration() -> PoolRegistration { PoolRegistration { - key: PoolKey { - site_id: "aws-west".into(), - namespace: "dynamo".into(), - dgd_name: "mocker-west-1".into(), - }, + key: PoolKey::new("aws-west", "dynamo", "mocker-west-1").unwrap(), location: PoolLocation { region: "us-west-2".into(), availability_zone: Some("usw2-az1".into()), - cluster: Some("aws-west".into()), datacenter: None, }, model: "Qwen/Qwen3-0.6B".into(), @@ -102,7 +97,7 @@ fn registration() -> PoolRegistration { frontends: vec!["http://10.1.22.20:8000".parse().unwrap()], capacity: vec![Capacity { sku: "H100".into(), - count: 8, + workers: 8, }], } } @@ -188,29 +183,22 @@ async fn registers_a_pool() { let lease = client.register_pool(®istration()).await.unwrap(); - assert_eq!( - lease, - Lease { - heartbeat_interval: Duration::from_secs(5), - lease_duration: Duration::from_secs(15), - } - ); + assert_eq!(lease, Lease::new(5, 15).unwrap()); // The field names are the contract with the service, so compare the JSON // it received rather than decoding it back into the same struct. assert_eq!( fake.0.lock().unwrap().registrations, [json!({ - "key": { "site_id": "aws-west", "namespace": "dynamo", "dgd_name": "mocker-west-1" }, + "key": { "cluster_id": "aws-west", "namespace": "dynamo", "dgd_name": "mocker-west-1" }, "location": { "region": "us-west-2", "availability_zone": "usw2-az1", - "cluster": "aws-west", "datacenter": null }, "model": "Qwen/Qwen3-0.6B", "capabilities": ["aggregated"], "frontends": ["http://10.1.22.20:8000/"], - "capacity": [{ "sku": "H100", "count": 8 }] + "capacity": [{ "sku": "H100", "workers": 8 }] })] ); }