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/kv_dc_relay.rs

@@ -6,16 +6,21 @@
 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, ModelAlias, ModelAliasError, ModelTarget, PoolIdentitySources,
+};

lib/llm/src/kv_dc_relay/actor.rs

@@ -12,13 +12,15 @@ use std::sync::atomic::{AtomicU64, Ordering};
 #[cfg(feature = "ckf-diagnostics")]
 use std::time::Instant;
 
+#[cfg(test)]
+use dynamo_kv_router::indexer::cuckoo::CkfConfig;
 #[cfg(any(test, feature = "ckf-diagnostics"))]
 use dynamo_kv_router::indexer::cuckoo::DcCkfStats;
 #[cfg(feature = "ckf-diagnostics")]
 use dynamo_kv_router::indexer::cuckoo::PublisherEmitOutcome;
 use dynamo_kv_router::indexer::cuckoo::{
-    CkfConfig, CkfFailureAction, CkfFailureDisposition, CkfFailurePoint, DcCkfDelta,
-    DcCkfDeltaSink, DcCkfPublisher, DcCkfSnapshot, DcCkfState, LaneLease, ProducerIdentity,
+    CkfFailureAction, CkfFailureDisposition, CkfFailurePoint, DcCkfDelta, DcCkfDeltaSink,
+    DcCkfPublisher, DcCkfSnapshot, DcCkfState, LaneLease, ProducerIdentity,
 };
 use dynamo_kv_router::protocols::{
     DpRank, ExternalSequenceBlockHash, KvCacheEventData, KvCacheEventError, RouterEvent,
@@ -32,7 +34,6 @@ use tokio_util::sync::CancellationToken;
 use crate::kv_router::indexer::{RecoveryResetReason, RecoveryTarget, SourceEpoch};
 
 use super::host::KvDcRelayError;
-use super::resolution::PoolBinding;
 
 const DEFAULT_MAILBOX_CAPACITY: usize = 256;
 const DEFAULT_PENDING_BLOCK_PERMITS: usize = 65_536;
@@ -253,11 +254,23 @@ fn actor_fault_category(disposition: CkfFailureDisposition) -> ActorFaultCategor
     }
 }
 
+async fn send_actor_fault(
+    sender: &mpsc::Sender<ActorFault>,
+    fence: &CancellationToken,
+    fault: ActorFault,
+) -> bool {
+    tokio::select! {
+        biased;
+        _ = fence.cancelled() => false,
+        result = sender.send(fault) => result.is_ok(),
+    }
+}
+
 #[derive(Debug, Clone)]
 pub(super) struct StreamScope {
     pub(super) process_incarnation: u64,
     pub(super) layout_generation: u64,
-    pub(super) pool_binding: PoolBinding,
+    pub(super) pool_id: dynamo_kv_router::identity::PoolId,
 }
 
 #[derive(Debug, Clone)]
@@ -276,12 +289,12 @@ impl DcCkfDeltaSink for BroadcastDeltaSink {
 #[derive(Debug, Clone)]
 pub(crate) struct KvDcRelayHandle {
     sender: mpsc::Sender<ActorCommand>,
+    identity: ProducerIdentity,
     payload_permits: Arc<Semaphore>,
     fence: CancellationToken,
     stopped: CancellationToken,
     #[cfg(feature = "ckf-diagnostics")]
     pub(super) diagnostics: ActorDiagnosticsHandle,
-    pub(super) scope: StreamScope,
 }
 
 impl KvDcRelayHandle {
@@ -298,6 +311,7 @@ impl KvDcRelayHandle {
         )
     }
 
+    #[cfg(test)]
     pub(super) fn spawn_with_publication_delay(
         config: CkfConfig,
         scope: StreamScope,
@@ -311,6 +325,19 @@ impl KvDcRelayHandle {
         )
     }
 
+    pub(super) fn spawn_with_state_and_publication_delay(
+        state: DcCkfState,
+        scope: StreamScope,
+        publication_delay: Duration,
+    ) -> (Self, mpsc::Receiver<ActorFault>) {
+        Self::spawn_with_state_capacity_and_delay(
+            state,
+            scope,
+            DEFAULT_MAILBOX_CAPACITY,
+            publication_delay,
+        )
+    }
+
     #[cfg(test)]
     fn spawn_with_capacity(
         config: CkfConfig,
@@ -320,17 +347,32 @@ impl KvDcRelayHandle {
         Self::spawn_with_capacity_and_delay(config, scope, capacity, DEFAULT_PUBLICATION_DELAY)
     }
 
+    #[cfg(test)]
     fn spawn_with_capacity_and_delay(
         config: CkfConfig,
         scope: StreamScope,
         capacity: usize,
         publication_delay: Duration,
     ) -> Result<(Self, mpsc::Receiver<ActorFault>), KvDcRelayError> {
         let state = DcCkfState::new(config)?;
+        Ok(Self::spawn_with_state_capacity_and_delay(
+            state,
+            scope,
+            capacity,
+            publication_delay,
+        ))
+    }
+
+    fn spawn_with_state_capacity_and_delay(
+        state: DcCkfState,
+        scope: StreamScope,
+        capacity: usize,
+        publication_delay: Duration,
+    ) -> (Self, mpsc::Receiver<ActorFault>) {
         let (sender, receiver) = mpsc::channel(capacity);
         let (publication_tx, _) = broadcast::channel(DEFAULT_PUBLICATION_CAPACITY);
         let identity = ProducerIdentity::new(
-            scope.pool_binding.pool_id(),
+            scope.pool_id,
             scope.process_incarnation,
             scope.layout_generation,
             state.format(),
@@ -356,18 +398,22 @@ impl KvDcRelayHandle {
             fence.clone(),
             stopped.clone(),
         ));
-        Ok((
+        (
             Self {
                 sender,
+                identity,
                 payload_permits: Arc::new(Semaphore::new(DEFAULT_PENDING_BLOCK_PERMITS)),
                 fence,
                 stopped,
                 #[cfg(feature = "ckf-diagnostics")]
                 diagnostics,
-                scope,
             },
             fault_rx,
-        ))
+        )
+    }
+
+    pub(super) const fn identity(&self) -> ProducerIdentity {
+        self.identity
     }
 
     async fn submit<T>(
@@ -930,19 +976,22 @@ async fn run_actor(
                         source_epoch.get()
                     );
                     diagnostics.record_error(&message);
-                    if fault_tx
-                        .send(ActorFault {
+                    if !send_actor_fault(
+                        &fault_tx,
+                        &fence,
+                        ActorFault {
                             worker_id,
                             dp_rank,
                             source_epoch,
                             event_id: Some(event_id),
                             category: actor_fault_category(disposition),
                             disposition,
                             message,
-                        })
-                        .await
-                        .is_err()
+                        },
+                    )
+                    .await
                     {
+                        discard_tail = fence.is_cancelled();
                         break;
                     }
                     diagnostics.finish_command();
@@ -1023,19 +1072,22 @@ async fn run_actor(
                     let message = error.to_string();
                     let category = actor_fault_category(disposition);
                     diagnostics.record_error(&message);
-                    if fault_tx
-                        .send(ActorFault {
+                    if !send_actor_fault(
+                        &fault_tx,
+                        &fence,
+                        ActorFault {
                             worker_id,
                             dp_rank,
                             source_epoch,
                             event_id: Some(event_id),
                             category,
                             disposition,
                             message,
-                        })
-                        .await
-                        .is_err()
+                        },
+                    )
+                    .await
                     {
+                        discard_tail = fence.is_cancelled();
                         break;
                     }
                 }
@@ -1314,14 +1366,10 @@ mod tests {
     use dynamo_kv_router::protocols::{
         KvCacheEvent, KvCacheStoreData, KvCacheStoredBlockData, LocalBlockHash,
     };
-    use dynamo_runtime::protocols::EndpointId;
 
     use super::*;
-    use crate::kv_dc_relay::resolution::EndpointLocator;
 
-    fn scope(name: &str) -> StreamScope {
-        let endpoint = format!("ns.worker.{name}");
-        let endpoint_id = EndpointId::from(endpoint.as_str());
+    fn scope(_name: &str) -> StreamScope {
         let dc_id = DcId::new(2);
         let domain = IndexerDomainId::new(
             CacheSemanticsId::new([1; 16], IdentitySource::Explicit),
@@ -1330,11 +1378,7 @@ mod tests {
         StreamScope {
             process_incarnation: 1,
             layout_generation: 1,
-            pool_binding: PoolBinding::new(
-                PoolId::new(domain, dc_id),
-                EndpointLocator::new(dc_id, endpoint_id),
-                None,
-            ),
+            pool_id: PoolId::new(domain, dc_id),
         }
     }
 
@@ -1399,10 +1443,7 @@ mod tests {
             snapshot.buckets.len(),
             snapshot.identity.format().bucket_count()
         );
-        assert_eq!(
-            actor_health(&handle).mailbox_capacity,
-            DEFAULT_MAILBOX_CAPACITY
-        );
+        assert_eq!(handle.mailbox_capacity(), DEFAULT_MAILBOX_CAPACITY);
         assert!(
             handle
                 .diagnostics
@@ -1597,6 +1638,37 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn producer_fence_interrupts_a_full_fault_channel() {
+        let worker = WorkerWithDpRank::new(1, 0);
+        let (handle, faults) =
+            KvDcRelayHandle::spawn(CkfConfig::new(32), scope("fault-backpressure")).unwrap();
+        handle
+            .admit_event(SourceEpoch::new(0), stored(worker, 1, &[1]))
+            .await
+            .unwrap();
+        handle.flush().await.unwrap();
+
+        for event_id in 2..=(DEFAULT_FAULT_CAPACITY as u64 + 2) {
+            handle
+                .admit_event(SourceEpoch::new(1), stored(worker, event_id, &[event_id]))
+                .await
+                .unwrap();
+        }
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while faults.len() != DEFAULT_FAULT_CAPACITY || handle.mailbox_depth() != 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("actor must block after filling the fault channel");
+
+        tokio::time::timeout(Duration::from_secs(1), handle.fence())
+            .await
+            .expect("fence must interrupt a blocked fault send")
+            .unwrap();
+    }
+
     #[tokio::test]
     async fn cadence_advances_on_duplicate_events_without_acknowledging_mutation() {
         let worker = WorkerWithDpRank::new(1, 0);