feat(llm): add pool-scoped KV DC Relay runtime

components/src/dynamo/kv_dc_relay/README.md

@@ -5,14 +5,16 @@ SPDX-License-Identifier: Apache-2.0
 
 # DC KV Relay
 
-The DC KV Relay aggregates exact KV-cache ownership inside one data center and publishes a compact
-Cuckoo-filter (CKF) projection for multi-DC routing. It discovers workers through the Dynamo
-runtime, consumes their ordered KV events, and supervises one actor-owned producer for each local
-routing pool.
+The DC KV Relay discovers Dynamo inference pools, consumes their ordered KV events, and supervises
+one actor-owned Cuckoo-filter (CKF) producer for each local pool.
 
-A pool is one logical indexer domain in one DC. The domain captures cache compatibility and routing
-isolation; the DC identity remains stable across Relay restarts and endpoint replacement. Runtime
-endpoints are bindings for a pool rather than part of the CKF publication identity.
+A pool is one atomic Dynamo indexer domain in one data center. Its domain captures cache
+compatibility and routing isolation. The Relay does not merge KV state from independent endpoints
+or deployments into one actor, even when they serve the same canonical model.
+
+Canonical model names are request-facing bindings. One model can bind to multiple independent
+pools, and each pool keeps its own KV stream. LoRA registrations remain attached to the pool of
+their backing base model.
 
 For each pool, the Relay:
 
@@ -23,26 +25,13 @@ For each pool, the Relay:
 - Publishes barrier snapshots and sequenced deltas containing absolute packed-bucket images.
 
 The full hashes and refcounts stay in the Relay because a CKF fingerprint is lossy, can collide,
-and has no owner identity. The global consumer needs only the compact projection required for
-cross-DC prefix search.
+and has no owner identity.
 
 ## Recovery boundaries
 
-Recovery has two stages:
-
-1. **Worker to Relay:** The Relay shares the normal Dynamo indexer's worker-query recovery path.
-   Ordered KV events handle live mutations; gaps and source replacement recover exact rank state
-   before the new source generation becomes active.
-2. **Relay to global consumer:** A new or reconnected lane installs a barrier snapshot, then
-   continues with sequenced absolute bucket-image deltas. A missing delta retires that lane and
-   requires another snapshot.
-
-The current component uses the producer lifecycle and exposes local diagnostics. An in-process
-adapter exercises the complete producer/consumer protocol today. Non-local gRPC transport and
-cross-DC request forwarding are separate global-router integration work.
-
-For the complete architecture, pool model, consistency contract, and recovery flow, see
-[Multi-DC KV Routing and the DC Relay](../../../../docs/fern/components/router/multi-dc-kv-routing.md).
+The Relay shares the normal Dynamo indexer's worker-query recovery path. Ordered KV events handle
+live mutations; gaps and source replacement recover exact rank state before the new source epoch
+becomes active. A fenced pool is withdrawn before its actor stops.
 
 ## Usage
 

lib/llm/src/discovery/kv_source_membership.rs

@@ -182,6 +182,28 @@ impl<S> KvSourceMembershipView<S> {
             KvStateEndpointResolution::Ambiguous { .. } => None,
         }
     }
+
+    /// Whether runtime fields that determine this source binding still match the view.
+    ///
+    /// Metadata-only runtime changes do not require waiting for another source-membership
+    /// publication. Endpoint mapping and the logical worker/rank set do.
+    pub(crate) fn matches_binding_inputs(
+        &self,
+        runtime_configs: &HashMap<WorkerId, ModelRuntimeConfig>,
+    ) -> bool {
+        if self.endpoint_resolution
+            != resolve_kv_state_endpoint(&self.serving_endpoint, runtime_configs.values())
+        {
+            return false;
+        }
+
+        let mut worker_count = 0usize;
+        let workers_match = expected_workers(runtime_configs).all(|(worker, _)| {
+            worker_count = worker_count.saturating_add(1);
+            self.sources.contains_key(&worker)
+        });
+        workers_match && worker_count == self.sources.len()
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Error)]
@@ -356,28 +378,13 @@ where
     ) -> KvSourceMembershipView<S> {
         let endpoint_resolution =
             resolve_kv_state_endpoint(serving_endpoint, runtime_configs.values());
-        let workers: Vec<_> = runtime_configs
-            .iter()
-            .flat_map(|(&worker_id, config)| {
-                (0..config.data_parallel_size).filter_map(move |offset| {
-                    config
-                        .data_parallel_start_rank
-                        .checked_add(offset)
-                        .map(|dp_rank| {
-                            (
-                                WorkerWithDpRank::new(worker_id, dp_rank),
-                                config.enable_local_indexer,
-                            )
-                        })
-                })
-            })
-            .collect();
+        let workers: HashMap<_, _> = expected_workers(runtime_configs).collect();
 
         let sources: HashMap<WorkerWithDpRank, KvSourceStatus<S>> = match &endpoint_resolution {
             KvStateEndpointResolution::Resolved(kv_state_endpoint) => workers
-                .iter()
+                .keys()
                 .copied()
-                .map(|(worker, _)| {
+                .map(|worker| {
                     let key = KvSourceKey::new(kv_state_endpoint.clone(), worker);
                     (worker, self.status(&key))
                 })
@@ -387,15 +394,12 @@ where
                     endpoints: endpoints.clone(),
                 };
                 workers
-                    .iter()
+                    .keys()
                     .copied()
-                    .map(|(worker, _)| (worker, KvSourceStatus::Ambiguous(ambiguity.clone())))
+                    .map(|worker| (worker, KvSourceStatus::Ambiguous(ambiguity.clone())))
                     .collect()
             }
         };
-        let recovery_expected = workers
-            .into_iter()
-            .collect::<HashMap<WorkerWithDpRank, bool>>();
         let kv_event_publishing_enabled = runtime_configs
             .iter()
             .map(|(&worker_id, config)| (worker_id, config.kv_event_publishing_enabled))
@@ -405,13 +409,31 @@ where
             serving_endpoint: serving_endpoint.clone(),
             endpoint_resolution,
             lifecycle_generations: sources.keys().map(|worker| (*worker, 0)).collect(),
-            recovery_expected,
+            recovery_expected: workers,
             kv_event_publishing_enabled,
             sources,
         }
     }
 }
 
+fn expected_workers(
+    runtime_configs: &HashMap<WorkerId, ModelRuntimeConfig>,
+) -> impl Iterator<Item = (WorkerWithDpRank, bool)> + '_ {
+    runtime_configs.iter().flat_map(|(&worker_id, config)| {
+        (0..config.data_parallel_size).filter_map(move |offset| {
+            config
+                .data_parallel_start_rank
+                .checked_add(offset)
+                .map(|dp_rank| {
+                    (
+                        WorkerWithDpRank::new(worker_id, dp_rank),
+                        config.enable_local_indexer,
+                    )
+                })
+        })
+    })
+}
+
 /// Resolve the effective KV-state endpoint advertised by active base runtime configs.
 ///
 /// An omitted mapping and an explicit mapping to `serving_endpoint` are equal after fallback.
@@ -552,6 +574,35 @@ mod tests {
         }
     }
 
+    #[test]
+    fn binding_inputs_ignore_metadata_only_runtime_changes() {
+        let serving = endpoint("generate");
+        let kv_endpoint = endpoint("kv-events");
+        let original = HashMap::from([(
+            7,
+            ModelRuntimeConfig {
+                context_length: Some(4096),
+                data_parallel_start_rank: 2,
+                data_parallel_size: 2,
+                kv_state_endpoint: Some(kv_endpoint.clone()),
+                ..Default::default()
+            },
+        )]);
+        let view = KvSourceMembership::<KvEventSource>::new().view(&serving, &original);
+
+        let mut metadata_only = original.clone();
+        metadata_only.get_mut(&7).unwrap().context_length = Some(8192);
+        assert!(view.matches_binding_inputs(&metadata_only));
+
+        let mut remapped = metadata_only.clone();
+        remapped.get_mut(&7).unwrap().kv_state_endpoint = Some(endpoint("other-kv-events"));
+        assert!(!view.matches_binding_inputs(&remapped));
+
+        let mut resized = metadata_only;
+        resized.get_mut(&7).unwrap().data_parallel_size = 3;
+        assert!(!view.matches_binding_inputs(&resized));
+    }
+
     #[test]
     fn overlapping_random_incarnations_are_ambiguous_until_one_remains() {
         let kv_endpoint = endpoint("kv-events");

lib/llm/src/kv_dc_relay.rs

@@ -6,16 +6,22 @@
 mod actor;
 mod discovery;
 mod host;
+mod identity;
+mod pool_registry;
 mod resolution;
 
 pub use host::{
     DEFAULT_EXPECTED_UNIQUE_BLOCKS, KvDcRelay, KvDcRelayConfig, KvDcRelayError, KvDcRelayHealth,
 };
-
 #[cfg(feature = "ckf-diagnostics")]
 pub use host::{
     KvDcRelayActorStats, KvDcRelayAggregationStats, KvDcRelayCacheDomainStats,
     KvDcRelayDiagnosticSnapshot, KvDcRelayEndpointStats, KvDcRelayIdentityStats,
     KvDcRelayMemberStats, KvDcRelayMemoryStats, KvDcRelayPublicationStats, KvDcRelayRecoveryStats,
     KvDcRelayStats,
 };
+pub use identity::{
+    CanonicalModelId, CanonicalModelIdError, CanonicalModelRegistration, DcPoolCatalog,
+    DcPoolDescriptor, DcRelayIdentity, ModelAlias, ModelAliasError, ModelTarget,
+    PoolIdentitySources,
+};