From 2037b4760a615e1ac2ff1dfd260f285144c71559 Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 10 Sep 2026 16:40:37 +0000 Subject: [PATCH 01/25] fix(discovery): fold the Qwen video-routing contract into mdcsum Two vLLM replicas of the same Qwen3-VL model whose engine-level --mm-processor-kwargs differ produce the same mdcsum(), because the function hashes only one runtime_config key. They join one cohort, the group builds its single video-routing processor from one representative card, and that processor then governs requests served by a replica that never published vllm_qwen_video_processor_contract at all. mdcsum() now absorbs the canonicalized content of that key when the card carries it. Content rather than presence, because the worker derives the contract from its installed Transformers and vLLM packages and two workers can publish different contracts. Raw JSON rather than the typed contract, because the typed form lives behind the optional mm-routing feature and mdcsum() also names the on-disk model cache directory. canonicalize_json moves from discovery::watcher to utils so both hashing paths share one copy; serde_json runs with preserve_order here, so key order would otherwise split a WorkerSet on its own. Divergent replicas now produce different fingerprints, which reconcile_group turns into GroupStatus::Conflict. Cards that do not carry this key keep a byte-identical checksum. Signed-off-by: svc-glamr@nvidia.com --- lib/llm/src/discovery/watcher.rs | 20 +------ lib/llm/src/model_card.rs | 100 +++++++++++++++++++++++++++++++ lib/llm/src/utils/mod.rs | 25 ++++++++ 3 files changed, 126 insertions(+), 19 deletions(-) diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index 79b4e1265795..6190950a225a 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -59,6 +59,7 @@ use crate::{ tensor::{NvCreateTensorRequest, NvCreateTensorResponse}, }, types::generic::realtime::{RealtimeClientEvent, RealtimeServerEvent}, + utils::canonicalize_json, worker_type::WorkerType, }; @@ -1328,25 +1329,6 @@ fn lora_projection_fingerprint(card: &ModelDeploymentCard) -> anyhow::Result { - let mut entries = std::mem::take(object).into_iter().collect::>(); - entries.sort_by(|left, right| left.0.cmp(&right.0)); - for (key, mut value) in entries { - canonicalize_json(&mut value); - object.insert(key, value); - } - } - serde_json::Value::Array(values) => { - for value in values { - canonicalize_json(value); - } - } - _ => {} - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/lib/llm/src/model_card.rs b/lib/llm/src/model_card.rs index 3da8b8e443e9..56f4b50326ee 100644 --- a/lib/llm/src/model_card.rs +++ b/lib/llm/src/model_card.rs @@ -1222,6 +1222,35 @@ impl ModelDeploymentCard { bytes_to_hash.extend_from_slice(b"\0vllm_enable_tower_connector_lora\0true"); } + // The Qwen video-routing contract decides how the frontend + // expands video placeholder tokens for exact KV routing. A + // group builds one video-routing processor from a single + // representative card and reuses it for every member, so + // members whose contract differs must not share a WorkerSet. + // + // The content is hashed, not merely the key's presence: the + // worker derives the contract from its installed Transformers + // and vLLM packages, so two workers can both publish and + // publish different contracts. Canonicalize first, because + // serde_json runs with `preserve_order` here and would + // otherwise let key order alone split a WorkerSet. + // + // Compatibility: workers older than v1.5.0 predate this key + // and hash as if the contract were absent, so a rolling + // upgrade of a Qwen video deployment across that boundary + // leaves the model unadmitted until it completes. That is + // deliberate - the alternative is serving one worker's + // requests under another worker's video token arithmetic. + // Deployments that never carry this key are unaffected. + if let Some(contract) = self.runtime_config.runtime_data.get( + crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY, + ) { + let mut contract = contract.clone(); + crate::utils::canonicalize_json(&mut contract); + bytes_to_hash.extend_from_slice(b"\0vllm_qwen_video_processor_contract\0"); + bytes_to_hash.extend(blake3::hash(contract.to_string().as_bytes()).as_bytes()); + } + // TODO: Do we want any other user_data or runtime_config? blake3::hash(&bytes_to_hash).to_string() @@ -3200,6 +3229,77 @@ mod ownership_tests { assert_eq!(missing.mdcsum(), disabled.mdcsum()); assert_ne!(missing.mdcsum(), enabled.mdcsum()); } + + /// The group builds one video-routing processor from a single + /// representative card, so every member of a WorkerSet is served with the + /// representative's Qwen video contract. Cards whose contract differs - + /// including a card that withholds it because engine-level + /// `--mm-processor-kwargs` may change the video token layout - must + /// therefore hash apart. + /// + /// Note: `mdcsum()` caches its result on first call via `OnceLock`, so each + /// case builds a fresh card rather than mutating one and re-hashing. + #[test] + fn qwen_video_processor_contract_isolates_worker_sets() { + use crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY; + + fn card_with_contract(contract: serde_json::Value) -> ModelDeploymentCard { + let mut card = ModelDeploymentCard::with_name_only("model"); + card.runtime_config.runtime_data.insert( + VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY.to_string(), + contract, + ); + card + } + + fn legacy_ceil() -> serde_json::Value { + serde_json::json!({ + "placeholder_target": "bare_video_token", + "resize_mode": "legacy_ceil", + }) + } + + let withheld = ModelDeploymentCard::with_name_only("model"); + assert_ne!( + withheld.mdcsum(), + card_with_contract(legacy_ceil()).mdcsum(), + "a worker that never published the contract must not join the WorkerSet of one that did" + ); + + assert_ne!( + card_with_contract(legacy_ceil()).mdcsum(), + card_with_contract(serde_json::json!({ + "placeholder_target": "bare_video_token", + "resize_mode": "smart_resize", + })) + .mdcsum(), + "two workers publishing different contracts must not share a WorkerSet" + ); + + // serde_json runs with `preserve_order` in this workspace, so an object + // keeps the key order it was parsed in. Two workers publishing the same + // contract in a different key order still describe the same behaviour + // and must stay in one WorkerSet. + assert_eq!( + card_with_contract(legacy_ceil()).mdcsum(), + card_with_contract(serde_json::json!({ + "resize_mode": "legacy_ceil", + "placeholder_target": "bare_video_token", + })) + .mdcsum(), + "key order within the contract must not change the checksum" + ); + + // Negative control: a deployment that never carries this key is + // unaffected, so its cache directory and reported checksum do not move. + let bare = ModelDeploymentCard::with_name_only("model"); + let mut unrelated = ModelDeploymentCard::with_name_only("model"); + unrelated + .runtime_config + .runtime_data + .insert("some_unrelated_runtime_key".to_string(), true.into()); + assert_eq!(bare.mdcsum(), unrelated.mdcsum()); + } } #[cfg(test)] diff --git a/lib/llm/src/utils/mod.rs b/lib/llm/src/utils/mod.rs index e5e3170fb9f0..e1153f0b26c8 100644 --- a/lib/llm/src/utils/mod.rs +++ b/lib/llm/src/utils/mod.rs @@ -5,3 +5,28 @@ pub mod lora; pub mod zmq; pub use lora::lora_name_to_id; + +/// Sort every JSON object in `value` by key, recursively. +/// +/// `serde_json` runs with `preserve_order` in this workspace, so an object +/// keeps the key order it was parsed in. Anything that hashes JSON as an +/// identity has to canonicalize first, or two logically identical values +/// hash apart. +pub(crate) fn canonicalize_json(value: &mut serde_json::Value) { + match value { + serde_json::Value::Object(object) => { + let mut entries = std::mem::take(object).into_iter().collect::>(); + entries.sort_by(|left, right| left.0.cmp(&right.0)); + for (key, mut value) in entries { + canonicalize_json(&mut value); + object.insert(key, value); + } + } + serde_json::Value::Array(values) => { + for value in values { + canonicalize_json(value); + } + } + _ => {} + } +} From 1018be14090e59e2255050ddfdeac6b614eddeca Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 10 Sep 2026 17:20:02 +0000 Subject: [PATCH 02/25] chore: remove factory working notes Signed-off-by: svc-glamr@nvidia.com --- lib/llm/src/model_card.rs | 28 ++++------------------------ 1 file changed, 4 insertions(+), 24 deletions(-) diff --git a/lib/llm/src/model_card.rs b/lib/llm/src/model_card.rs index 56f4b50326ee..2b3015949511 100644 --- a/lib/llm/src/model_card.rs +++ b/lib/llm/src/model_card.rs @@ -1222,26 +1222,8 @@ impl ModelDeploymentCard { bytes_to_hash.extend_from_slice(b"\0vllm_enable_tower_connector_lora\0true"); } - // The Qwen video-routing contract decides how the frontend - // expands video placeholder tokens for exact KV routing. A - // group builds one video-routing processor from a single - // representative card and reuses it for every member, so - // members whose contract differs must not share a WorkerSet. - // - // The content is hashed, not merely the key's presence: the - // worker derives the contract from its installed Transformers - // and vLLM packages, so two workers can both publish and - // publish different contracts. Canonicalize first, because - // serde_json runs with `preserve_order` here and would - // otherwise let key order alone split a WorkerSet. - // - // Compatibility: workers older than v1.5.0 predate this key - // and hash as if the contract were absent, so a rolling - // upgrade of a Qwen video deployment across that boundary - // leaves the model unadmitted until it completes. That is - // deliberate - the alternative is serving one worker's - // requests under another worker's video token arithmetic. - // Deployments that never carry this key are unaffected. + // Workers derive this contract from their own packages, so hash + // its content. Canonicalize: serde_json preserves key order. if let Some(contract) = self.runtime_config.runtime_data.get( crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY, ) { @@ -3276,10 +3258,8 @@ mod ownership_tests { "two workers publishing different contracts must not share a WorkerSet" ); - // serde_json runs with `preserve_order` in this workspace, so an object - // keeps the key order it was parsed in. Two workers publishing the same - // contract in a different key order still describe the same behaviour - // and must stay in one WorkerSet. + // serde_json preserves key order here, so the same contract written in + // a different key order must still hash the same. assert_eq!( card_with_contract(legacy_ceil()).mdcsum(), card_with_contract(serde_json::json!({ From ba585179ac8f2ee9fe82f96c9aefd14952367e9a Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 10 Sep 2026 22:35:28 +0000 Subject: [PATCH 03/25] fix(llm): resolve the Qwen video contract across the WorkerSet The Qwen video prompt-expansion contract is a per-WorkerSet routing input, not part of a worker's identity. Keep it out of `mdcsum()` so workers that publish different contracts, or none at all, still share one cohort, and resolve it across the cohort in the discovery controller instead: the group builds with the contract every member published, or with none when they disagree, which leaves exact video routing off while text serving continues. Signed-off-by: svc-glamr@nvidia.com --- lib/llm/src/discovery/controller.rs | 201 ++++++++++++++++++++++++++-- lib/llm/src/discovery/watcher.rs | 115 +++++++++++++++- lib/llm/src/model_card.rs | 50 ++----- lib/llm/src/utils/mod.rs | 25 ---- 4 files changed, 316 insertions(+), 75 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 96f2198c5e4f..3b84b474d051 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -47,6 +47,10 @@ pub(crate) struct DesiredInstance { pub(crate) group_key: GroupKey, pub(crate) fingerprint: String, pub(crate) projection_fingerprint: String, + /// Digest of the Qwen video prompt-expansion contract this worker + /// published, or `None` when it published none. Deliberately not part of + /// `fingerprint`: it is resolved across the cohort rather than per card. + pub(crate) video_contract: Option, } impl DesiredInstance { @@ -55,12 +59,47 @@ impl DesiredInstance { } } +/// The Qwen video prompt-expansion contract every member of the cohort +/// published, or `None` when they do not all publish the same one. +/// +/// A WorkerSet is served by a single video-routing processor, so it may only +/// use a contract the whole cohort agrees on. Members can legitimately +/// disagree: workers older than the contract publish nothing, and engine-level +/// `--mm-processor-kwargs` can change the video token layout between two +/// workers of the same version. +fn cohort_video_contract(members: &[DesiredInstance]) -> Option { + let mut members = members.iter(); + let agreed = members.next()?.video_contract.clone()?; + members + .all(|member| member.video_contract.as_deref() == Some(agreed.as_str())) + .then_some(agreed) +} + +/// Fold the cohort's video contract into the fingerprint the status machine +/// compares against. +/// +/// The contract is outside the card checksum, so a member joining or leaving +/// can change what the cohort agrees on without changing any member's +/// fingerprint. Folding it in here rebuilds the group on that transition +/// instead of leaving it serving with a contract one of its members never +/// published. A cohort with no agreement keeps the bare fingerprint, so +/// deployments that never carry the contract are unaffected. +fn cohort_fingerprint(fingerprint: &str, video_contract: Option<&str>) -> String { + match video_contract { + Some(contract) => format!("{fingerprint}\0video_contract\0{contract}"), + None => fingerprint.to_string(), + } +} + #[derive(Clone, Debug)] pub(crate) struct GroupSpec { pub(crate) key: GroupKey, pub(crate) fingerprint: String, pub(crate) generation: u64, pub(crate) representative: DesiredInstance, + /// The cohort-wide contract from [`cohort_video_contract`]. `None` means + /// the group must build with exact video routing disabled. + pub(crate) video_contract: Option, } #[async_trait] @@ -160,6 +199,14 @@ impl DesiredGroup { .insert(instance.key.clone()); } + /// The one cohort's per-card fingerprint and members, or `None` while the + /// group is empty or in conflict. + fn sole_cohort(&self) -> Option<(&String, &BTreeSet)> { + let mut cohorts = self.cohorts.iter(); + let cohort = cohorts.next()?; + cohorts.next().is_none().then_some(cohort) + } + fn remove(&mut self, instance: &DesiredInstance) { let Some(cohort) = self.cohorts.get_mut(&instance.fingerprint) else { return; @@ -410,6 +457,8 @@ impl ModelDiscoveryController { .map(|(fingerprint, members)| (fingerprint.clone(), members.clone())) .expect("non-empty group has one cohort"); let members = self.members(&member_keys); + let fingerprint = + cohort_fingerprint(&fingerprint, cohort_video_contract(&members).as_deref()); let admitted = admitted_ids(&members); if !matches!( &old_status, @@ -597,16 +646,21 @@ impl ModelDiscoveryController { continue; }; let fingerprint = fingerprint.clone(); - let Some(member_key) = group - .cohorts - .get(&fingerprint) - .and_then(|members| members.first()) - else { + // The status carries the cohort fingerprint, which folds in the + // agreed video contract, so it is not a key into `cohorts`. A + // queued group has exactly one cohort — more than one is a + // conflict — so take that one. + let Some((_, member_keys)) = group.sole_cohort() else { continue; }; - let Some(representative) = self.desired.get(member_key).cloned() else { + let members = member_keys + .iter() + .filter_map(|member_key| self.desired.get(member_key).cloned()) + .collect::>(); + let Some(representative) = members.first().cloned() else { continue; }; + let video_contract = cohort_video_contract(&members); let cancellation = CancellationToken::new(); let generation = self.next_build_generation; @@ -616,6 +670,7 @@ impl ModelDiscoveryController { fingerprint: fingerprint.clone(), generation, representative, + video_contract, }; group.status = GroupStatus::Building { fingerprint, @@ -661,6 +716,21 @@ impl ModelDiscoveryController { } return; }; + // `cohorts` is keyed by the per-card fingerprint while the spec carries + // the cohort fingerprint, which also folds in the agreed video + // contract. Derive the group's current cohort fingerprint rather than + // looking the spec's up, so a build stays current only while the cohort + // it was started for still describes the group. + let cohort_members = group + .sole_cohort() + .map(|(fingerprint, member_keys)| (fingerprint.clone(), member_keys.clone())) + .filter(|(fingerprint, member_keys)| { + cohort_fingerprint( + fingerprint, + cohort_video_contract(&self.members(member_keys)).as_deref(), + ) == result.spec.fingerprint + }) + .map(|(_, member_keys)| member_keys); let is_current = matches!( &group.status, GroupStatus::Building { @@ -668,8 +738,7 @@ impl ModelDiscoveryController { generation, .. } if fingerprint == &result.spec.fingerprint && *generation == result.spec.generation - ) && group.cohorts.len() == 1 - && group.cohorts.contains_key(&result.spec.fingerprint); + ) && cohort_members.is_some(); if !is_current { if let BuildOutcome::Prepared(prepared) = result.outcome { self.host.discard_prepared(prepared); @@ -680,11 +749,7 @@ impl ModelDiscoveryController { match result.outcome { BuildOutcome::Prepared(prepared) => { - let member_keys = group - .cohorts - .get(&result.spec.fingerprint) - .cloned() - .unwrap_or_default(); + let member_keys = cohort_members.unwrap_or_default(); let members = self.members(&member_keys); let adapters = self.adapters_for_members(&member_keys); group.admission_tx.send_replace(admitted_ids(&members)); @@ -1045,6 +1110,7 @@ mod tests { card, fingerprint: "spec".to_string(), projection_fingerprint: "projection".to_string(), + video_contract: None, })) } @@ -1161,6 +1227,14 @@ mod tests { group_key: group_key(), fingerprint: fingerprint.to_string(), projection_fingerprint: fingerprint.to_string(), + video_contract: None, + } + } + + fn instance_with_contract(id: u64, fingerprint: &str, contract: &str) -> DesiredInstance { + DesiredInstance { + video_contract: Some(contract.to_string()), + ..instance(id, fingerprint) } } @@ -1258,6 +1332,107 @@ mod tests { assert_eq!(host.members(&group_key()), BTreeSet::from([compatible.key])); } + /// A worker that predates the video contract publishes none, so a rolling + /// upgrade routinely mixes it with workers that do. The card checksum is + /// the same either way, so the two share a cohort and text serving + /// survives; only exact video routing is withdrawn, and it comes back once + /// the group agrees again. + #[tokio::test] + async fn a_worker_without_the_video_contract_joins_instead_of_conflicting() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + let current = instance_with_contract(1, "spec", "contract-a"); + controller.apply_added(current.clone()); + controller.start_queued_builds(); + assert_eq!( + starts.recv().await.unwrap().video_contract.as_deref(), + Some("contract-a") + ); + host.release.add_permits(1); + finish_build(&mut controller).await; + + let legacy = instance(2, "spec"); + controller.apply_added(legacy.clone()); + assert!( + !matches!( + controller.groups[&group_key()].status, + GroupStatus::Conflict + ), + "a member that publishes no contract must not conflict with one that does" + ); + controller.start_queued_builds(); + assert_eq!( + starts.recv().await.unwrap().video_contract, + None, + "the group must rebuild without a contract it cannot serve to every member" + ); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!( + host.members(&group_key()), + BTreeSet::from([current.key.clone(), legacy.key.clone()]) + ); + + controller.apply_removed(&legacy.key); + controller.start_queued_builds(); + assert_eq!( + starts.recv().await.unwrap().video_contract.as_deref(), + Some("contract-a"), + "draining the older worker restores exact video routing" + ); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!(host.members(&group_key()), BTreeSet::from([current.key])); + } + + /// Two workers of the same version can still publish different contracts, + /// because engine-level `--mm-processor-kwargs` changes the video token + /// layout. They serve as one group, without exact video routing, rather + /// than one being served with the other's contract. + #[tokio::test] + async fn differing_video_contracts_serve_together_without_exact_video_routing() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + let first = instance_with_contract(1, "spec", "contract-a"); + controller.apply_added(first.clone()); + controller.start_queued_builds(); + starts.recv().await.unwrap(); + host.release.add_permits(1); + finish_build(&mut controller).await; + + let second = instance_with_contract(2, "spec", "contract-b"); + controller.apply_added(second.clone()); + controller.start_queued_builds(); + assert_eq!(starts.recv().await.unwrap().video_contract, None); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!( + host.members(&group_key()), + BTreeSet::from([first.key, second.key]) + ); + } + + /// The contract only forces a rebuild when the agreement itself changes. + #[tokio::test] + async fn an_agreeing_member_joins_without_rebuilding_the_group() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + let first = instance_with_contract(1, "spec", "contract-a"); + controller.apply_added(first.clone()); + controller.start_queued_builds(); + starts.recv().await.unwrap(); + host.release.add_permits(1); + finish_build(&mut controller).await; + + let second = instance_with_contract(2, "spec", "contract-a"); + controller.apply_added(second.clone()); + assert_eq!( + host.members(&group_key()), + BTreeSet::from([first.key, second.key]) + ); + assert_eq!(host.starts.load(Ordering::SeqCst), 1); + } + #[tokio::test] async fn conflict_during_build_cancels_without_publishing_either_cohort() { let (host, mut starts) = FakeHost::new(); diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index 6190950a225a..fbe48c853b84 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -36,6 +36,7 @@ use crate::{ }, local_model::runtime_config::{ ModelRuntimeConfig, TokenizerBackend, VLLM_INFERENCE_V1_GENERATE_CAPABILITY, + VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY, }, model_card::ModelDeploymentCard, model_type::{ModelInput, ModelType}, @@ -59,7 +60,6 @@ use crate::{ tensor::{NvCreateTensorRequest, NvCreateTensorResponse}, }, types::generic::realtime::{RealtimeClientEvent, RealtimeServerEvent}, - utils::canonicalize_json, worker_type::WorkerType, }; @@ -434,6 +434,28 @@ where validate_selector_worker_role(card, self.require_typed_worker_role)?; + // One video-routing processor serves the whole WorkerSet, so it may + // only use a contract every member published. Where they disagree, + // withhold the contract from the card the pipeline is built from: the + // preprocessor then takes the same path as a worker that published no + // contract and leaves exact video routing off. Text serving, and every + // other member of the group, are untouched. + if spec.video_contract.is_none() + && card + .runtime_config + .runtime_data + .remove(VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY) + .is_some() + { + tracing::warn!( + target: "mm_routing", + model_name = card.name(), + group = %spec.key.id(), + "WorkerSet members publish different Qwen video prompt-expansion contracts; \ + exact video routing disabled for this group" + ); + } + // Use per-worker-set router config if the worker provided one in its MDC, // otherwise fall back to the frontend-level global config. Policy selections // are process-local, so preserve them when the MDC supplies the base config. @@ -1042,6 +1064,7 @@ where }; let fingerprint = materialization_fingerprint(&card, &self.router_config)?; let projection_fingerprint = lora_projection_fingerprint(&card)?; + let video_contract = qwen_video_contract_digest(&card); Ok(Some(DesiredInstance { key: mcid.to_path(), mcid, @@ -1050,6 +1073,7 @@ where group_key, fingerprint, projection_fingerprint, + video_contract, })) } @@ -1329,6 +1353,48 @@ fn lora_projection_fingerprint(card: &ModelDeploymentCard) -> anyhow::Result Option { + let mut contract = card + .runtime_config + .runtime_data + .get(VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY)? + .clone(); + canonicalize_json(&mut contract); + Some(blake3::hash(contract.to_string().as_bytes()).to_string()) +} + +fn canonicalize_json(value: &mut serde_json::Value) { + match value { + serde_json::Value::Object(object) => { + let mut entries = std::mem::take(object).into_iter().collect::>(); + entries.sort_by(|left, right| left.0.cmp(&right.0)); + for (key, mut value) in entries { + canonicalize_json(&mut value); + object.insert(key, value); + } + } + serde_json::Value::Array(values) => { + for value in values { + canonicalize_json(value); + } + } + _ => {} + } +} + #[cfg(test)] mod tests { use super::*; @@ -1457,6 +1523,7 @@ mod tests { endpoint_id, fingerprint: materialization_fingerprint(&card, &router_config).unwrap(), projection_fingerprint: lora_projection_fingerprint(&card).unwrap(), + video_contract: qwen_video_contract_digest(&card), card, group_key: key.clone(), }; @@ -1464,6 +1531,7 @@ mod tests { key, fingerprint: desired.fingerprint.clone(), generation: 1, + video_contract: desired.video_contract.clone(), representative: desired.clone(), }; let (_admission_tx, admission_rx) = tokio::sync::watch::channel(vec![1]); @@ -1553,6 +1621,7 @@ mod tests { endpoint_id, fingerprint: materialization_fingerprint(&card, &router_config).unwrap(), projection_fingerprint: lora_projection_fingerprint(&card).unwrap(), + video_contract: qwen_video_contract_digest(&card), card, group_key: key.clone(), }; @@ -1560,6 +1629,7 @@ mod tests { key, fingerprint: desired.fingerprint.clone(), generation: 1, + video_contract: desired.video_contract.clone(), representative: desired, }; let (_admission_tx, admission_rx) = tokio::sync::watch::channel(vec![1]); @@ -2183,6 +2253,49 @@ mod tests { ); } + #[test] + fn qwen_video_contract_digest_identifies_the_contract_not_the_worker() { + use crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY; + + fn card_with_contract(contract: serde_json::Value) -> ModelDeploymentCard { + let mut card = ModelDeploymentCard::with_name_only("model"); + card.runtime_config.runtime_data.insert( + VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY.to_string(), + contract, + ); + card + } + + let legacy_ceil = serde_json::json!({ + "placeholder_target": "bare_video_token", + "resize_mode": "legacy_ceil", + }); + let round_ties_even = serde_json::json!({ + "placeholder_target": "bare_video_token", + "resize_mode": "round_ties_even", + }); + + assert_eq!( + qwen_video_contract_digest(&ModelDeploymentCard::with_name_only("model")), + None, + "a worker that predates the contract publishes none" + ); + assert_ne!( + qwen_video_contract_digest(&card_with_contract(legacy_ceil.clone())), + qwen_video_contract_digest(&card_with_contract(round_ties_even)), + ); + + // `serde_json` preserves key order in this workspace, so the same + // contract written in a different key order must still agree. + assert_eq!( + qwen_video_contract_digest(&card_with_contract(legacy_ceil)), + qwen_video_contract_digest(&card_with_contract(serde_json::json!({ + "resize_mode": "legacy_ceil", + "placeholder_target": "bare_video_token", + }))), + ); + } + #[tokio::test] async fn discovery_normalization_joins_v12_and_current_prefill() { use dynamo_runtime::{Runtime, distributed::DistributedConfig}; diff --git a/lib/llm/src/model_card.rs b/lib/llm/src/model_card.rs index 2b3015949511..111133033060 100644 --- a/lib/llm/src/model_card.rs +++ b/lib/llm/src/model_card.rs @@ -1222,16 +1222,9 @@ impl ModelDeploymentCard { bytes_to_hash.extend_from_slice(b"\0vllm_enable_tower_connector_lora\0true"); } - // Workers derive this contract from their own packages, so hash - // its content. Canonicalize: serde_json preserves key order. - if let Some(contract) = self.runtime_config.runtime_data.get( - crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY, - ) { - let mut contract = contract.clone(); - crate::utils::canonicalize_json(&mut contract); - bytes_to_hash.extend_from_slice(b"\0vllm_qwen_video_processor_contract\0"); - bytes_to_hash.extend(blake3::hash(contract.to_string().as_bytes()).as_bytes()); - } + // The Qwen video prompt-expansion contract is deliberately absent + // here. It is a per-WorkerSet routing input, resolved across the + // whole cohort in `discovery::controller`, not a reason to split one. // TODO: Do we want any other user_data or runtime_config? @@ -3212,19 +3205,16 @@ mod ownership_tests { assert_ne!(missing.mdcsum(), enabled.mdcsum()); } - /// The group builds one video-routing processor from a single - /// representative card, so every member of a WorkerSet is served with the - /// representative's Qwen video contract. Cards whose contract differs - - /// including a card that withholds it because engine-level - /// `--mm-processor-kwargs` may change the video token layout - must - /// therefore hash apart. - /// - /// Note: `mdcsum()` caches its result on first call via `OnceLock`, so each - /// case builds a fresh card rather than mutating one and re-hashing. #[test] - fn qwen_video_processor_contract_isolates_worker_sets() { + fn qwen_video_processor_contract_stays_out_of_the_checksum() { use crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY; + // A checksum split becomes a cohort conflict, which withdraws the whole + // group. Disagreement over the contract disables exact video routing + // instead; see `discovery::controller::cohort_video_contract`. + // `mdcsum()` caches on first call via `OnceLock`, so each case builds a + // fresh card rather than mutating one and re-hashing. + fn card_with_contract(contract: serde_json::Value) -> ModelDeploymentCard { let mut card = ModelDeploymentCard::with_name_only("model"); card.runtime_config.runtime_data.insert( @@ -3242,32 +3232,20 @@ mod ownership_tests { } let withheld = ModelDeploymentCard::with_name_only("model"); - assert_ne!( + assert_eq!( withheld.mdcsum(), card_with_contract(legacy_ceil()).mdcsum(), - "a worker that never published the contract must not join the WorkerSet of one that did" + "a worker that predates the contract must still join the WorkerSet of one that publishes it" ); - assert_ne!( - card_with_contract(legacy_ceil()).mdcsum(), - card_with_contract(serde_json::json!({ - "placeholder_target": "bare_video_token", - "resize_mode": "smart_resize", - })) - .mdcsum(), - "two workers publishing different contracts must not share a WorkerSet" - ); - - // serde_json preserves key order here, so the same contract written in - // a different key order must still hash the same. assert_eq!( card_with_contract(legacy_ceil()).mdcsum(), card_with_contract(serde_json::json!({ - "resize_mode": "legacy_ceil", "placeholder_target": "bare_video_token", + "resize_mode": "round_ties_even", })) .mdcsum(), - "key order within the contract must not change the checksum" + "two workers publishing different contracts must still serve as one group" ); // Negative control: a deployment that never carries this key is diff --git a/lib/llm/src/utils/mod.rs b/lib/llm/src/utils/mod.rs index e1153f0b26c8..e5e3170fb9f0 100644 --- a/lib/llm/src/utils/mod.rs +++ b/lib/llm/src/utils/mod.rs @@ -5,28 +5,3 @@ pub mod lora; pub mod zmq; pub use lora::lora_name_to_id; - -/// Sort every JSON object in `value` by key, recursively. -/// -/// `serde_json` runs with `preserve_order` in this workspace, so an object -/// keeps the key order it was parsed in. Anything that hashes JSON as an -/// identity has to canonicalize first, or two logically identical values -/// hash apart. -pub(crate) fn canonicalize_json(value: &mut serde_json::Value) { - match value { - serde_json::Value::Object(object) => { - let mut entries = std::mem::take(object).into_iter().collect::>(); - entries.sort_by(|left, right| left.0.cmp(&right.0)); - for (key, mut value) in entries { - canonicalize_json(&mut value); - object.insert(key, value); - } - } - serde_json::Value::Array(values) => { - for value in values { - canonicalize_json(value); - } - } - _ => {} - } -} From 29461b4667fab75e1de1daeb2259d73e745f9ea5 Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 10 Sep 2026 22:49:46 +0000 Subject: [PATCH 04/25] fix(discovery): treat a contract-only card update as a change A worker republishing its card with a different Qwen video contract keeps the same materialization fingerprint, because the contract is not part of it. Comparing the contract in `apply_added` keeps the group from holding an agreement one of its members no longer publishes. Signed-off-by: svc-glamr@nvidia.com --- lib/llm/src/discovery/controller.rs | 34 +++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 3b84b474d051..5aa4577fb265 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -357,6 +357,7 @@ impl ModelDiscoveryController { if let Some(existing) = self.desired.get(&instance.key) { if existing.fingerprint == instance.fingerprint && existing.projection_fingerprint == instance.projection_fingerprint + && existing.video_contract == instance.video_contract { return false; } @@ -1412,6 +1413,39 @@ mod tests { ); } + /// The contract sits outside the materialization fingerprint, so an + /// instance that republishes its card with a different contract looks + /// unchanged by fingerprint alone. It is still an update: the group's + /// agreement moves with it. + #[tokio::test] + async fn republishing_only_the_video_contract_updates_the_group() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + let first = instance_with_contract(1, "spec", "contract-a"); + controller.apply_added(first.clone()); + controller.start_queued_builds(); + assert_eq!( + starts.recv().await.unwrap().video_contract.as_deref(), + Some("contract-a") + ); + host.release.add_permits(1); + finish_build(&mut controller).await; + + assert!( + controller.apply_added(instance_with_contract(1, "spec", "contract-b")), + "a contract-only change must not be discarded as a duplicate" + ); + controller.start_queued_builds(); + assert_eq!( + starts.recv().await.unwrap().video_contract.as_deref(), + Some("contract-b"), + "the group must rebuild with the contract the worker now publishes" + ); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!(host.members(&group_key()), BTreeSet::from([first.key])); + } + /// The contract only forces a rebuild when the agreement itself changes. #[tokio::test] async fn an_agreeing_member_joins_without_rebuilding_the_group() { From 0613b8dbf5976350884e5590090cfb9f23f83e9a Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 20:13:18 +0000 Subject: [PATCH 05/25] docs(discovery): shorten video contract rationale Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 28 +++++----------------------- lib/llm/src/discovery/watcher.rs | 14 +------------- lib/llm/src/model_card.rs | 4 +--- 3 files changed, 7 insertions(+), 39 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 5aa4577fb265..4f0e9e5be370 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -75,15 +75,7 @@ fn cohort_video_contract(members: &[DesiredInstance]) -> Option { .then_some(agreed) } -/// Fold the cohort's video contract into the fingerprint the status machine -/// compares against. -/// -/// The contract is outside the card checksum, so a member joining or leaving -/// can change what the cohort agrees on without changing any member's -/// fingerprint. Folding it in here rebuilds the group on that transition -/// instead of leaving it serving with a contract one of its members never -/// published. A cohort with no agreement keeps the bare fingerprint, so -/// deployments that never carry the contract are unaffected. +/// Add the agreed contract to the status-machine fingerprint. fn cohort_fingerprint(fingerprint: &str, video_contract: Option<&str>) -> String { match video_contract { Some(contract) => format!("{fingerprint}\0video_contract\0{contract}"), @@ -1333,11 +1325,7 @@ mod tests { assert_eq!(host.members(&group_key()), BTreeSet::from([compatible.key])); } - /// A worker that predates the video contract publishes none, so a rolling - /// upgrade routinely mixes it with workers that do. The card checksum is - /// the same either way, so the two share a cohort and text serving - /// survives; only exact video routing is withdrawn, and it comes back once - /// the group agrees again. + // Legacy workers join the cohort without exact video routing. #[tokio::test] async fn a_worker_without_the_video_contract_joins_instead_of_conflicting() { let (host, mut starts) = FakeHost::new(); @@ -1386,10 +1374,7 @@ mod tests { assert_eq!(host.members(&group_key()), BTreeSet::from([current.key])); } - /// Two workers of the same version can still publish different contracts, - /// because engine-level `--mm-processor-kwargs` changes the video token - /// layout. They serve as one group, without exact video routing, rather - /// than one being served with the other's contract. + // Contract disagreement disables exact video routing for the cohort. #[tokio::test] async fn differing_video_contracts_serve_together_without_exact_video_routing() { let (host, mut starts) = FakeHost::new(); @@ -1413,10 +1398,7 @@ mod tests { ); } - /// The contract sits outside the materialization fingerprint, so an - /// instance that republishes its card with a different contract looks - /// unchanged by fingerprint alone. It is still an update: the group's - /// agreement moves with it. + // Contract-only card updates must refresh the cohort agreement. #[tokio::test] async fn republishing_only_the_video_contract_updates_the_group() { let (host, mut starts) = FakeHost::new(); @@ -1446,7 +1428,7 @@ mod tests { assert_eq!(host.members(&group_key()), BTreeSet::from([first.key])); } - /// The contract only forces a rebuild when the agreement itself changes. + // An agreeing member does not rebuild the cohort. #[tokio::test] async fn an_agreeing_member_joins_without_rebuilding_the_group() { let (host, mut starts) = FakeHost::new(); diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index fbe48c853b84..6c2c35b286a8 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -1353,19 +1353,7 @@ fn lora_projection_fingerprint(card: &ModelDeploymentCard) -> anyhow::Result Option { let mut contract = card .runtime_config diff --git a/lib/llm/src/model_card.rs b/lib/llm/src/model_card.rs index 111133033060..a089cfcb7119 100644 --- a/lib/llm/src/model_card.rs +++ b/lib/llm/src/model_card.rs @@ -1222,9 +1222,7 @@ impl ModelDeploymentCard { bytes_to_hash.extend_from_slice(b"\0vllm_enable_tower_connector_lora\0true"); } - // The Qwen video prompt-expansion contract is deliberately absent - // here. It is a per-WorkerSet routing input, resolved across the - // whole cohort in `discovery::controller`, not a reason to split one. + // The Qwen video contract is resolved per cohort, not per card. // TODO: Do we want any other user_data or runtime_config? From 0fbe58ced25f1ea7a81b9a84525a12d951f6aa1c Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 20:14:58 +0000 Subject: [PATCH 06/25] docs(model-card): shorten contract test note Signed-off-by: GLAMR --- lib/llm/src/model_card.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/lib/llm/src/model_card.rs b/lib/llm/src/model_card.rs index a089cfcb7119..61e01cebcfe1 100644 --- a/lib/llm/src/model_card.rs +++ b/lib/llm/src/model_card.rs @@ -3207,11 +3207,7 @@ mod ownership_tests { fn qwen_video_processor_contract_stays_out_of_the_checksum() { use crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY; - // A checksum split becomes a cohort conflict, which withdraws the whole - // group. Disagreement over the contract disables exact video routing - // instead; see `discovery::controller::cohort_video_contract`. - // `mdcsum()` caches on first call via `OnceLock`, so each case builds a - // fresh card rather than mutating one and re-hashing. + // `mdcsum()` caches via `OnceLock`, so each case uses a fresh card. fn card_with_contract(contract: serde_json::Value) -> ModelDeploymentCard { let mut card = ModelDeploymentCard::with_name_only("model"); From 05dd4a6037abcc2f47196ebd6fa3c8742884108c Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 21:10:10 +0000 Subject: [PATCH 07/25] style(discovery): deduplicate video contract comments Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 18 +++--------------- lib/llm/src/discovery/watcher.rs | 7 +------ 2 files changed, 4 insertions(+), 21 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 4f0e9e5be370..de4782c651b5 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -89,8 +89,7 @@ pub(crate) struct GroupSpec { pub(crate) fingerprint: String, pub(crate) generation: u64, pub(crate) representative: DesiredInstance, - /// The cohort-wide contract from [`cohort_video_contract`]. `None` means - /// the group must build with exact video routing disabled. + /// The cohort-wide contract, if any. pub(crate) video_contract: Option, } @@ -639,10 +638,7 @@ impl ModelDiscoveryController { continue; }; let fingerprint = fingerprint.clone(); - // The status carries the cohort fingerprint, which folds in the - // agreed video contract, so it is not a key into `cohorts`. A - // queued group has exactly one cohort — more than one is a - // conflict — so take that one. + // The status fingerprint includes the contract; take the sole cohort. let Some((_, member_keys)) = group.sole_cohort() else { continue; }; @@ -709,11 +705,7 @@ impl ModelDiscoveryController { } return; }; - // `cohorts` is keyed by the per-card fingerprint while the spec carries - // the cohort fingerprint, which also folds in the agreed video - // contract. Derive the group's current cohort fingerprint rather than - // looking the spec's up, so a build stays current only while the cohort - // it was started for still describes the group. + // The spec fingerprint includes the contract, unlike the `cohorts` key. let cohort_members = group .sole_cohort() .map(|(fingerprint, member_keys)| (fingerprint.clone(), member_keys.clone())) @@ -1325,7 +1317,6 @@ mod tests { assert_eq!(host.members(&group_key()), BTreeSet::from([compatible.key])); } - // Legacy workers join the cohort without exact video routing. #[tokio::test] async fn a_worker_without_the_video_contract_joins_instead_of_conflicting() { let (host, mut starts) = FakeHost::new(); @@ -1374,7 +1365,6 @@ mod tests { assert_eq!(host.members(&group_key()), BTreeSet::from([current.key])); } - // Contract disagreement disables exact video routing for the cohort. #[tokio::test] async fn differing_video_contracts_serve_together_without_exact_video_routing() { let (host, mut starts) = FakeHost::new(); @@ -1398,7 +1388,6 @@ mod tests { ); } - // Contract-only card updates must refresh the cohort agreement. #[tokio::test] async fn republishing_only_the_video_contract_updates_the_group() { let (host, mut starts) = FakeHost::new(); @@ -1428,7 +1417,6 @@ mod tests { assert_eq!(host.members(&group_key()), BTreeSet::from([first.key])); } - // An agreeing member does not rebuild the cohort. #[tokio::test] async fn an_agreeing_member_joins_without_rebuilding_the_group() { let (host, mut starts) = FakeHost::new(); diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index 6c2c35b286a8..6ad0aa3ac852 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -434,12 +434,7 @@ where validate_selector_worker_role(card, self.require_typed_worker_role)?; - // One video-routing processor serves the whole WorkerSet, so it may - // only use a contract every member published. Where they disagree, - // withhold the contract from the card the pipeline is built from: the - // preprocessor then takes the same path as a worker that published no - // contract and leaves exact video routing off. Text serving, and every - // other member of the group, are untouched. + // Prepare without exact video routing unless the cohort agreed on a contract. if spec.video_contract.is_none() && card .runtime_config From b8997674a85589d74847134b15a5258a7ce2fbf8 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 21:21:36 +0000 Subject: [PATCH 08/25] style(discovery): shorten video contract rationale Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 25 +++++++------------------ lib/llm/src/discovery/watcher.rs | 6 ++---- 2 files changed, 9 insertions(+), 22 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index de4782c651b5..37ea5e89b2af 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -47,9 +47,7 @@ pub(crate) struct DesiredInstance { pub(crate) group_key: GroupKey, pub(crate) fingerprint: String, pub(crate) projection_fingerprint: String, - /// Digest of the Qwen video prompt-expansion contract this worker - /// published, or `None` when it published none. Deliberately not part of - /// `fingerprint`: it is resolved across the cohort rather than per card. + /// Digest of this worker's Qwen video prompt-expansion contract. pub(crate) video_contract: Option, } @@ -59,14 +57,10 @@ impl DesiredInstance { } } -/// The Qwen video prompt-expansion contract every member of the cohort -/// published, or `None` when they do not all publish the same one. +/// Returns the cohort's shared contract, if every member publishes one. /// -/// A WorkerSet is served by a single video-routing processor, so it may only -/// use a contract the whole cohort agrees on. Members can legitimately -/// disagree: workers older than the contract publish nothing, and engine-level -/// `--mm-processor-kwargs` can change the video token layout between two -/// workers of the same version. +/// A WorkerSet has one video-routing processor, so exact video routing is +/// enabled only when all members agree. fn cohort_video_contract(members: &[DesiredInstance]) -> Option { let mut members = members.iter(); let agreed = members.next()?.video_contract.clone()?; @@ -75,7 +69,7 @@ fn cohort_video_contract(members: &[DesiredInstance]) -> Option { .then_some(agreed) } -/// Add the agreed contract to the status-machine fingerprint. +/// Identifies a cohort's active video contract. fn cohort_fingerprint(fingerprint: &str, video_contract: Option<&str>) -> String { match video_contract { Some(contract) => format!("{fingerprint}\0video_contract\0{contract}"), @@ -89,7 +83,7 @@ pub(crate) struct GroupSpec { pub(crate) fingerprint: String, pub(crate) generation: u64, pub(crate) representative: DesiredInstance, - /// The cohort-wide contract, if any. + /// Contract shared by the cohort, if any. pub(crate) video_contract: Option, } @@ -638,7 +632,7 @@ impl ModelDiscoveryController { continue; }; let fingerprint = fingerprint.clone(); - // The status fingerprint includes the contract; take the sole cohort. + // The status fingerprint identifies one contract cohort. let Some((_, member_keys)) = group.sole_cohort() else { continue; }; @@ -1338,13 +1332,11 @@ mod tests { controller.groups[&group_key()].status, GroupStatus::Conflict ), - "a member that publishes no contract must not conflict with one that does" ); controller.start_queued_builds(); assert_eq!( starts.recv().await.unwrap().video_contract, None, - "the group must rebuild without a contract it cannot serve to every member" ); host.release.add_permits(1); finish_build(&mut controller).await; @@ -1358,7 +1350,6 @@ mod tests { assert_eq!( starts.recv().await.unwrap().video_contract.as_deref(), Some("contract-a"), - "draining the older worker restores exact video routing" ); host.release.add_permits(1); finish_build(&mut controller).await; @@ -1404,13 +1395,11 @@ mod tests { assert!( controller.apply_added(instance_with_contract(1, "spec", "contract-b")), - "a contract-only change must not be discarded as a duplicate" ); controller.start_queued_builds(); assert_eq!( starts.recv().await.unwrap().video_contract.as_deref(), Some("contract-b"), - "the group must rebuild with the contract the worker now publishes" ); host.release.add_permits(1); finish_build(&mut controller).await; diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index 6ad0aa3ac852..fa2dc342b5e6 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -1348,7 +1348,7 @@ fn lora_projection_fingerprint(card: &ModelDeploymentCard) -> anyhow::Result Option { let mut contract = card .runtime_config @@ -2261,15 +2261,13 @@ mod tests { assert_eq!( qwen_video_contract_digest(&ModelDeploymentCard::with_name_only("model")), None, - "a worker that predates the contract publishes none" ); assert_ne!( qwen_video_contract_digest(&card_with_contract(legacy_ceil.clone())), qwen_video_contract_digest(&card_with_contract(round_ties_even)), ); - // `serde_json` preserves key order in this workspace, so the same - // contract written in a different key order must still agree. + // Canonicalization makes key order irrelevant. assert_eq!( qwen_video_contract_digest(&card_with_contract(legacy_ceil)), qwen_video_contract_digest(&card_with_contract(serde_json::json!({ From ea1ef0c015b53443b5e7bf49292c40269af8d902 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 21:52:33 +0000 Subject: [PATCH 09/25] fix(discovery): retain groups during contract rebuilds Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 107 ++++++++++++++++++++++++---- 1 file changed, 92 insertions(+), 15 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 978d5ae4b82e..e35bee671076 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -132,11 +132,13 @@ enum GroupStatus { Idle, Queued { mdc_checksum: String, + committed_members: Option>, }, Building { mdc_checksum: String, generation: u64, cancellation: CancellationToken, + committed_members: Option>, }, Ready { mdc_checksum: String, @@ -145,10 +147,12 @@ enum GroupStatus { Retrying { mdc_checksum: String, deadline: Instant, + committed_members: Option>, }, Blocked { mdc_checksum: String, deadline: Instant, + committed_members: Option>, }, BlockedReady { mdc_checksum: String, @@ -467,15 +471,32 @@ impl ModelDiscoveryController { &mdc_checksum, cohort_video_contract(&members).as_deref(), ); - if status_checksum(&old_status).is_some_and(|previous| previous != fingerprint) { - group.admission_tx.send_replace(Vec::new()); + let fingerprint_changed = + status_checksum(&old_status).is_some_and(|previous| previous != fingerprint); + let mut retained_commit = status_committed_members(&old_status).cloned(); + if fingerprint_changed { + retained_commit = retained_commit.filter(|committed_members| { + committed_members != &member_keys + && (status_checksum(&old_status) == Some(mdc_checksum.as_str()) + || status_checksum(&old_status).is_some_and(|previous| { + previous + .strip_prefix(mdc_checksum.as_str()) + .is_some_and(|suffix| suffix.starts_with("\0video_contract\0")) + })) + }); cancel_build(&old_status); - if status_has_commit(&old_status) { - self.host.remove_group(key); + if retained_commit.is_some() { + let (admission_tx, _) = watch::channel(Vec::new()); + group.admission_tx = admission_tx; + } else { + group.admission_tx.send_replace(Vec::new()); + if status_has_commit(&old_status) { + self.host.remove_group(key); + } + // Retained pipeline clients must never observe the successor's IDs. + let (admission_tx, _) = watch::channel(Vec::new()); + group.admission_tx = admission_tx; } - // Retained pipeline clients must never observe the successor's IDs. - let (admission_tx, _) = watch::channel(Vec::new()); - group.admission_tx = admission_tx; group.reported_rejections.clear(); group.retry_attempt = 0; } @@ -484,7 +505,8 @@ impl ModelDiscoveryController { if !matches!( &old_status, GroupStatus::Ready { .. } | GroupStatus::BlockedReady { .. } - ) { + ) || (fingerprint_changed && retained_commit.is_some()) + { group.admission_tx.send_replace(admitted); } @@ -555,29 +577,37 @@ impl ModelDiscoveryController { mdc_checksum: building_checksum, generation, cancellation, + committed_members, } if building_checksum == fingerprint => GroupStatus::Building { mdc_checksum: fingerprint, generation, cancellation, + committed_members, }, GroupStatus::Queued { mdc_checksum: queued_checksum, + committed_members, } if queued_checksum == fingerprint => GroupStatus::Queued { mdc_checksum: fingerprint, + committed_members, }, GroupStatus::Retrying { mdc_checksum: retry_checksum, deadline, + committed_members, } if retry_checksum == fingerprint && !desired_changed => GroupStatus::Retrying { mdc_checksum: fingerprint, deadline, + committed_members, }, GroupStatus::Blocked { mdc_checksum: blocked_checksum, deadline, + committed_members, } if blocked_checksum == fingerprint && !desired_changed => GroupStatus::Blocked { mdc_checksum: fingerprint, deadline, + committed_members, }, previous => { cancel_build(&previous); @@ -585,6 +615,7 @@ impl ModelDiscoveryController { group.retry_attempt = 0; GroupStatus::Queued { mdc_checksum: fingerprint, + committed_members: retained_commit, } } }; @@ -667,6 +698,13 @@ impl ModelDiscoveryController { let GroupStatus::Queued { .. } = &group.status else { continue; }; + let committed_members = match &group.status { + GroupStatus::Queued { + committed_members, + .. + } => committed_members.clone(), + _ => unreachable!("queued status was checked above"), + }; let Some(mdc_checksum) = group.selected_checksum().map(str::to_string) else { continue; }; @@ -698,6 +736,7 @@ impl ModelDiscoveryController { mdc_checksum: fingerprint, generation, cancellation: cancellation.clone(), + committed_members, }; let host = self.host.clone(); @@ -756,6 +795,13 @@ impl ModelDiscoveryController { cohort_video_contract(&self.members(member_keys)).as_deref(), ) == result.spec.fingerprint }); + let committed_members = match &group.status { + GroupStatus::Building { + committed_members, + .. + } => committed_members.clone(), + _ => None, + }; if !is_current { if let BuildOutcome::Prepared(prepared) = result.outcome { self.host.discard_prepared(prepared); @@ -796,6 +842,7 @@ impl ModelDiscoveryController { group.status = GroupStatus::Blocked { mdc_checksum: result.spec.fingerprint, deadline: Instant::now() + retry_delay(group.retry_attempt), + committed_members, }; } } @@ -813,11 +860,13 @@ impl ModelDiscoveryController { group.status = GroupStatus::Retrying { mdc_checksum: result.spec.fingerprint, deadline: Instant::now() + delay, + committed_members, }; } BuildOutcome::Cancelled => { group.status = GroupStatus::Queued { mdc_checksum: result.spec.fingerprint, + committed_members, }; } } @@ -840,15 +889,17 @@ impl ModelDiscoveryController { let now = Instant::now(); let mut retained_retries = Vec::new(); for (key, group) in &mut self.groups { - let (mdc_checksum, deadline) = match &group.status { + let (mdc_checksum, deadline, committed_members) = match &group.status { GroupStatus::Retrying { mdc_checksum, deadline, + committed_members, } | GroupStatus::Blocked { mdc_checksum, deadline, - } => (mdc_checksum, deadline), + committed_members, + } => (mdc_checksum, deadline, committed_members.clone()), GroupStatus::BlockedReady { mdc_checksum, committed_members, @@ -866,6 +917,7 @@ impl ModelDiscoveryController { if *deadline <= now { group.status = GroupStatus::Queued { mdc_checksum: mdc_checksum.clone(), + committed_members, }; } } @@ -977,7 +1029,7 @@ fn cancel_build(status: &GroupStatus) { fn status_checksum(status: &GroupStatus) -> Option<&str> { match status { GroupStatus::Idle => None, - GroupStatus::Queued { mdc_checksum } + GroupStatus::Queued { mdc_checksum, .. } | GroupStatus::Building { mdc_checksum, .. } | GroupStatus::Ready { mdc_checksum, .. } | GroupStatus::Retrying { mdc_checksum, .. } @@ -987,10 +1039,35 @@ fn status_checksum(status: &GroupStatus) -> Option<&str> { } fn status_has_commit(status: &GroupStatus) -> bool { - matches!( - status, - GroupStatus::Ready { .. } | GroupStatus::BlockedReady { .. } - ) + status_committed_members(status).is_some() +} + +fn status_committed_members(status: &GroupStatus) -> Option<&BTreeSet> { + match status { + GroupStatus::Ready { + committed_members, .. + } + | GroupStatus::BlockedReady { + committed_members, .. + } => Some(committed_members), + GroupStatus::Queued { + committed_members, + .. + } + | GroupStatus::Building { + committed_members, + .. + } + | GroupStatus::Retrying { + committed_members, + .. + } + | GroupStatus::Blocked { + committed_members, + .. + } => committed_members.as_ref(), + GroupStatus::Idle => None, + } } async fn wait_for_deadline(deadline: Option) { From ef3a1063cb7efffbf5d0fa231e163c98ed1a93e6 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 22:04:28 +0000 Subject: [PATCH 10/25] fix(llm): replace retained video contract worker sets Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 31 +++++++++-- lib/llm/src/discovery/model_manager.rs | 27 +++++++++- lib/llm/src/discovery/watcher.rs | 74 ++++++++++++++++++++++++++ 3 files changed, 127 insertions(+), 5 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index e35bee671076..345a140cfa6e 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -120,6 +120,14 @@ pub(crate) trait ControllerHost: Send + Sync + 'static { adapters: &[DesiredInstance], ) -> anyhow::Result<()>; + fn replace_prepared_group( + &self, + spec: &GroupSpec, + prepared: Self::Prepared, + members: &[DesiredInstance], + adapters: &[DesiredInstance], + ) -> anyhow::Result<()>; + fn remove_group(&self, key: &GroupKey); fn discard_prepared(&self, prepared: Self::Prepared); @@ -820,10 +828,14 @@ impl ModelDiscoveryController { let members = self.members(&member_keys); let adapters = self.adapters_for_members(&member_keys); group.admission_tx.send_replace(admitted_ids(&members)); - match self - .host - .commit_group(&result.spec, prepared, &members, &adapters) - { + let commit_result = if committed_members.is_some() { + self.host + .replace_prepared_group(&result.spec, prepared, &members, &adapters) + } else { + self.host + .commit_group(&result.spec, prepared, &members, &adapters) + }; + match commit_result { Ok(()) => { group.retry_attempt = 0; group.status = GroupStatus::Ready { @@ -1296,6 +1308,17 @@ mod tests { Ok(()) } + fn replace_prepared_group( + &self, + spec: &GroupSpec, + prepared: Self::Prepared, + members: &[DesiredInstance], + adapters: &[DesiredInstance], + ) -> anyhow::Result<()> { + let Prepared(_build) = prepared; + self.replace_group(&spec.key, members, adapters) + } + fn remove_group(&self, key: &GroupKey) { self.committed.lock().unwrap().remove(&key.id()); self.adapters.lock().unwrap().remove(&key.id()); diff --git a/lib/llm/src/discovery/model_manager.rs b/lib/llm/src/discovery/model_manager.rs index b31d22129ce5..66520276f941 100644 --- a/lib/llm/src/discovery/model_manager.rs +++ b/lib/llm/src/discovery/model_manager.rs @@ -799,6 +799,7 @@ impl ModelManager { pub(crate) fn replace_discovery_group( &self, group_id: &str, + replacement_worker_set: Option, members: Vec<(String, ModelDeploymentCard)>, adapters: Vec<(String, ModelDeploymentCard)>, ) -> anyhow::Result<()> { @@ -808,8 +809,13 @@ impl ModelManager { .get(group_id) .ok_or_else(|| anyhow::anyhow!("committed discovery group {group_id:?} not found"))?; let primary = group.primary.clone(); + let namespace = group.namespace.clone(); let worker_set_key = group.worker_set_key.clone(); - let worker_set = group.worker_set.clone(); + let aliases = group.aliases.clone(); + let replacing_worker_set = replacement_worker_set.is_some(); + let worker_set = replacement_worker_set + .map(Arc::new) + .unwrap_or_else(|| group.worker_set.clone()); let previous_member_keys = group.cards.keys().cloned().collect::>(); let previous_adapter_keys = group.adapters.keys().cloned().collect::>(); let previous_adapter_names = group @@ -823,6 +829,10 @@ impl ModelManager { !members.is_empty(), "cannot replace with an empty discovery group" ); + anyhow::ensure!( + worker_set.namespace() == namespace, + "replacement WorkerSet namespace does not match committed discovery group" + ); self.validate_adapter_claims(&primary, adapters.iter().map(|(_, card)| card))?; let members = members.into_iter().collect::>(); let adapters = adapters.into_iter().collect::>(); @@ -843,6 +853,15 @@ impl ModelManager { .collect::>(); let lora_before = self.lora_projection_locked(); + if replacing_worker_set { + self.get_or_create_model(&primary) + .add_worker_set(worker_set_key.clone(), worker_set.clone()); + for alias in &aliases { + self.get_or_create_model(alias) + .add_worker_set(worker_set_key.clone(), worker_set.clone()); + } + } + for key in previous_member_keys.difference(&desired_member_keys) { self.cards.remove(key); } @@ -864,6 +883,9 @@ impl ModelManager { .expect("non-empty members checked above"); group.cards = members; group.adapters = adapters; + if replacing_worker_set { + group.worker_set = worker_set; + } drop(group); for (name, adapter_view) in adapter_views { @@ -878,6 +900,9 @@ impl ModelManager { } let lora_after = self.lora_projection_locked(); self.publish_lora_projection_locked(Self::union_lora_projection(&lora_before, &lora_after)); + if replacing_worker_set { + self.reconcile_discovery_topology(&primary, &namespace); + } self.publish_catalog_locked(); self.publish_lora_projection_locked(lora_after); Ok(()) diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index 5b59e6cca282..d2e1e4fe0cec 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -1181,6 +1181,7 @@ where .collect::>(); self.manager.replace_discovery_group( &group_id, + None, members .iter() .map(|member| (member.key.clone(), member.card.clone())) @@ -1207,6 +1208,79 @@ where Ok(()) } + fn replace_prepared_group( + &self, + spec: &GroupSpec, + mut prepared: Self::Prepared, + members: &[DesiredInstance], + adapters: &[DesiredInstance], + ) -> anyhow::Result<()> { + let adapter_was_available = adapters + .iter() + .map(|adapter| { + ( + adapter.card.name().to_string(), + self.manager + .get_committed_model(adapter.card.name()) + .is_some(), + ) + }) + .collect::>(); + let worker_set = prepared + .worker_set + .take() + .ok_or_else(|| anyhow::anyhow!("prepared WorkerSet was already consumed"))?; + let mut committed_members = members + .iter() + .map(|member| (member.key.clone(), member.card.clone())) + .collect::>(); + if let Some((_, card)) = committed_members + .iter_mut() + .find(|(key, _)| key == &spec.representative.key) + { + *card = prepared.card.clone(); + } + let group_id = spec.key.id(); + let previous = self + .manager + .discovery_group_adapter_cards(&group_id) + .into_iter() + .map(|card| (card.name().to_string(), card)) + .collect::>(); + let desired = adapters + .iter() + .map(|adapter| (adapter.card.name().to_string(), adapter.card.clone())) + .collect::>(); + self.manager.replace_discovery_group( + &group_id, + Some(worker_set), + committed_members, + adapters + .iter() + .map(|adapter| (adapter.key.clone(), adapter.card.clone())) + .collect(), + )?; + self.emit_update(ModelUpdate::Added(prepared.card.clone())); + for (name, card) in &desired { + if !adapter_was_available.get(name).copied().unwrap_or(false) + && self.manager.get_committed_model(name).is_some() + { + self.emit_update(ModelUpdate::Added(card.clone())); + } + } + for (name, card) in previous { + if adapter_was_available.get(&name).copied().unwrap_or(false) + && self.manager.get_committed_model(&name).is_none() + { + self.emit_update(ModelUpdate::Removed(card)); + } + } + if prepared.card.model_type.supports_chat() { + self.notify_on_model.notify_waiters(); + } + Ok(()) + } + fn remove_group(&self, key: &GroupKey) { let Some(removed) = self.manager.remove_discovery_group(&key.id()) else { return; From 74c971d67ed0e3fad6156ce38e699d112662a726 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 22:10:15 +0000 Subject: [PATCH 11/25] style(llm): format retained group replacement Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 345a140cfa6e..8789e2c74e88 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -829,8 +829,12 @@ impl ModelDiscoveryController { let adapters = self.adapters_for_members(&member_keys); group.admission_tx.send_replace(admitted_ids(&members)); let commit_result = if committed_members.is_some() { - self.host - .replace_prepared_group(&result.spec, prepared, &members, &adapters) + self.host.replace_prepared_group( + &result.spec, + prepared, + &members, + &adapters, + ) } else { self.host .commit_group(&result.spec, prepared, &members, &adapters) From 12eaa0bce4f5cdb470996419726b068d66e78daa Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 22:35:42 +0000 Subject: [PATCH 12/25] fix(discovery): retain contract-only replacements Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 8789e2c74e88..4cf83ab2a4e1 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -483,14 +483,13 @@ impl ModelDiscoveryController { status_checksum(&old_status).is_some_and(|previous| previous != fingerprint); let mut retained_commit = status_committed_members(&old_status).cloned(); if fingerprint_changed { - retained_commit = retained_commit.filter(|committed_members| { - committed_members != &member_keys - && (status_checksum(&old_status) == Some(mdc_checksum.as_str()) - || status_checksum(&old_status).is_some_and(|previous| { - previous - .strip_prefix(mdc_checksum.as_str()) - .is_some_and(|suffix| suffix.starts_with("\0video_contract\0")) - })) + retained_commit = retained_commit.filter(|_| { + status_checksum(&old_status) == Some(mdc_checksum.as_str()) + || status_checksum(&old_status).is_some_and(|previous| { + previous + .strip_prefix(mdc_checksum.as_str()) + .is_some_and(|suffix| suffix.starts_with("\0video_contract\0")) + }) }); cancel_build(&old_status); if retained_commit.is_some() { From a30adb143de703af463b78628e68082f056391e0 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 22:58:35 +0000 Subject: [PATCH 13/25] fix(discovery): preserve routing updates during replacement Signed-off-by: GLAMR --- lib/llm/src/discovery/model_manager.rs | 15 +++++++++---- lib/llm/src/discovery/watcher.rs | 30 +++++++++++++++----------- 2 files changed, 29 insertions(+), 16 deletions(-) diff --git a/lib/llm/src/discovery/model_manager.rs b/lib/llm/src/discovery/model_manager.rs index 66520276f941..e65f18350d8a 100644 --- a/lib/llm/src/discovery/model_manager.rs +++ b/lib/llm/src/discovery/model_manager.rs @@ -854,11 +854,18 @@ impl ModelManager { let lora_before = self.lora_projection_locked(); if replacing_worker_set { - self.get_or_create_model(&primary) - .add_worker_set(worker_set_key.clone(), worker_set.clone()); + let primary_model = self.get_or_create_model(&primary); + if let Some(displaced_worker_set) = primary_model.get_worker_set(&worker_set_key) + { + Self::clear_worker_set_targets(&displaced_worker_set); + } + primary_model.add_worker_set(worker_set_key.clone(), worker_set.clone()); for alias in &aliases { - self.get_or_create_model(alias) - .add_worker_set(worker_set_key.clone(), worker_set.clone()); + let alias_model = self.get_or_create_model(alias); + if let Some(displaced_worker_set) = alias_model.get_worker_set(&worker_set_key) { + Self::clear_worker_set_targets(&displaced_worker_set); + } + alias_model.add_worker_set(worker_set_key.clone(), worker_set.clone()); } } diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index d2e1e4fe0cec..46c0e2205a36 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -1215,13 +1215,26 @@ where members: &[DesiredInstance], adapters: &[DesiredInstance], ) -> anyhow::Result<()> { - let adapter_was_available = adapters - .iter() - .map(|adapter| { + let group_id = spec.key.id(); + let previous = self + .manager + .discovery_group_adapter_cards(&group_id) + .into_iter() + .map(|card| (card.name().to_string(), card)) + .collect::>(); + let adapter_was_available = previous + .keys() + .cloned() + .chain( + adapters + .iter() + .map(|adapter| adapter.card.name().to_string()), + ) + .map(|name| { ( - adapter.card.name().to_string(), + name.clone(), self.manager - .get_committed_model(adapter.card.name()) + .get_committed_model(&name) .is_some(), ) }) @@ -1240,13 +1253,6 @@ where { *card = prepared.card.clone(); } - let group_id = spec.key.id(); - let previous = self - .manager - .discovery_group_adapter_cards(&group_id) - .into_iter() - .map(|card| (card.name().to_string(), card)) - .collect::>(); let desired = adapters .iter() .map(|adapter| (adapter.card.name().to_string(), adapter.card.clone())) From 39a67d2c7cb43241bb594375717e1d9fdef6d1f9 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 23:04:39 +0000 Subject: [PATCH 14/25] style(discovery): format replacement target cleanup Signed-off-by: GLAMR --- lib/llm/src/discovery/model_manager.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/lib/llm/src/discovery/model_manager.rs b/lib/llm/src/discovery/model_manager.rs index e65f18350d8a..d06397da5ec7 100644 --- a/lib/llm/src/discovery/model_manager.rs +++ b/lib/llm/src/discovery/model_manager.rs @@ -855,8 +855,7 @@ impl ModelManager { if replacing_worker_set { let primary_model = self.get_or_create_model(&primary); - if let Some(displaced_worker_set) = primary_model.get_worker_set(&worker_set_key) - { + if let Some(displaced_worker_set) = primary_model.get_worker_set(&worker_set_key) { Self::clear_worker_set_targets(&displaced_worker_set); } primary_model.add_worker_set(worker_set_key.clone(), worker_set.clone()); From 877c7be0ad467839f17217f06362988e314e6c6b Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 23:34:32 +0000 Subject: [PATCH 15/25] style(discovery): format video contract reconciliation Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 4cf83ab2a4e1..d1e8ef83faba 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -707,8 +707,7 @@ impl ModelDiscoveryController { }; let committed_members = match &group.status { GroupStatus::Queued { - committed_members, - .. + committed_members, .. } => committed_members.clone(), _ => unreachable!("queued status was checked above"), }; @@ -764,9 +763,7 @@ impl ModelDiscoveryController { }); let outcome = match future.catch_unwind().await { Ok(outcome) => outcome, - Err(_) => BuildOutcome::Failed(anyhow::anyhow!( - "model materialization panicked" - )), + Err(_) => BuildOutcome::Failed(anyhow::anyhow!("model materialization panicked")), }; BuildResult { spec: task_spec, @@ -804,8 +801,7 @@ impl ModelDiscoveryController { }); let committed_members = match &group.status { GroupStatus::Building { - committed_members, - .. + committed_members, .. } => committed_members.clone(), _ => None, }; @@ -828,12 +824,8 @@ impl ModelDiscoveryController { let adapters = self.adapters_for_members(&member_keys); group.admission_tx.send_replace(admitted_ids(&members)); let commit_result = if committed_members.is_some() { - self.host.replace_prepared_group( - &result.spec, - prepared, - &members, - &adapters, - ) + self.host + .replace_prepared_group(&result.spec, prepared, &members, &adapters) } else { self.host .commit_group(&result.spec, prepared, &members, &adapters) From a5c8d87ae41cd60d40a1f99c18e15e6e90ae1d31 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 23:43:18 +0000 Subject: [PATCH 16/25] style(discovery): wrap reconciliation formatting Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index d1e8ef83faba..610b5b02c71a 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -753,7 +753,11 @@ impl ModelDiscoveryController { tokio::select! { biased; _ = cancellation.cancelled() => BuildOutcome::Cancelled, - result = host.prepare(task_spec.clone(), admitted_ids, cancellation.clone()) => { + result = host.prepare( + task_spec.clone(), + admitted_ids, + cancellation.clone(), + ) => { match result { Ok(prepared) => BuildOutcome::Prepared(prepared), Err(error) => BuildOutcome::Failed(error), @@ -763,7 +767,9 @@ impl ModelDiscoveryController { }); let outcome = match future.catch_unwind().await { Ok(outcome) => outcome, - Err(_) => BuildOutcome::Failed(anyhow::anyhow!("model materialization panicked")), + Err(_) => BuildOutcome::Failed(anyhow::anyhow!( + "model materialization panicked" + )), }; BuildResult { spec: task_spec, From cd725a32ebb6fbe929e00f45bb211486235416d1 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 23:55:01 +0000 Subject: [PATCH 17/25] fix(discovery): retain blocked status metadata Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 610b5b02c71a..eef164ec5cd9 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -576,6 +576,7 @@ impl ModelDiscoveryController { GroupStatus::Blocked { mdc_checksum: fingerprint, deadline: Instant::now() + retry_delay(group.retry_attempt), + committed_members: None, } } } From bf9bc4900f40a14f2c35e7dc995a8ecdc7fe2640 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 23:58:43 +0000 Subject: [PATCH 18/25] style(discovery): normalize status match patterns Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 36 +++++++---------------------- 1 file changed, 8 insertions(+), 28 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index eef164ec5cd9..e8aaddfd8975 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -707,9 +707,7 @@ impl ModelDiscoveryController { continue; }; let committed_members = match &group.status { - GroupStatus::Queued { - committed_members, .. - } => committed_members.clone(), + GroupStatus::Queued { committed_members, .. } => committed_members.clone(), _ => unreachable!("queued status was checked above"), }; let Some(mdc_checksum) = group.selected_checksum().map(str::to_string) else { @@ -807,9 +805,7 @@ impl ModelDiscoveryController { ) == result.spec.fingerprint }); let committed_members = match &group.status { - GroupStatus::Building { - committed_members, .. - } => committed_members.clone(), + GroupStatus::Building { committed_members, .. } => committed_members.clone(), _ => None, }; if !is_current { @@ -1058,28 +1054,12 @@ fn status_has_commit(status: &GroupStatus) -> bool { fn status_committed_members(status: &GroupStatus) -> Option<&BTreeSet> { match status { - GroupStatus::Ready { - committed_members, .. - } - | GroupStatus::BlockedReady { - committed_members, .. - } => Some(committed_members), - GroupStatus::Queued { - committed_members, - .. - } - | GroupStatus::Building { - committed_members, - .. - } - | GroupStatus::Retrying { - committed_members, - .. - } - | GroupStatus::Blocked { - committed_members, - .. - } => committed_members.as_ref(), + GroupStatus::Ready { committed_members, .. } + | GroupStatus::BlockedReady { committed_members, .. } => Some(committed_members), + GroupStatus::Queued { committed_members, .. } + | GroupStatus::Building { committed_members, .. } + | GroupStatus::Retrying { committed_members, .. } + | GroupStatus::Blocked { committed_members, .. } => committed_members.as_ref(), GroupStatus::Idle => None, } } From 3f055fbf319bd91c17f67e04bf2a54819ec7e60b Mon Sep 17 00:00:00 2001 From: GLAMR Date: Sat, 12 Sep 2026 00:06:52 +0000 Subject: [PATCH 19/25] style(discovery): compact group commit call Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index e8aaddfd8975..9d8cb346b38a 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -830,8 +830,7 @@ impl ModelDiscoveryController { self.host .replace_prepared_group(&result.spec, prepared, &members, &adapters) } else { - self.host - .commit_group(&result.spec, prepared, &members, &adapters) + self.host.commit_group(&result.spec, prepared, &members, &adapters) }; match commit_result { Ok(()) => { From 1f0b0cc636e4e5f7c8cdab68764a0e302873e02d Mon Sep 17 00:00:00 2001 From: GLAMR Date: Sat, 12 Sep 2026 00:40:53 +0000 Subject: [PATCH 20/25] fix(discovery): retain commits only across contract rebuilds Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 9d8cb346b38a..e33e0276b617 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -483,13 +483,14 @@ impl ModelDiscoveryController { status_checksum(&old_status).is_some_and(|previous| previous != fingerprint); let mut retained_commit = status_committed_members(&old_status).cloned(); if fingerprint_changed { - retained_commit = retained_commit.filter(|_| { - status_checksum(&old_status) == Some(mdc_checksum.as_str()) - || status_checksum(&old_status).is_some_and(|previous| { - previous - .strip_prefix(mdc_checksum.as_str()) - .is_some_and(|suffix| suffix.starts_with("\0video_contract\0")) - }) + retained_commit = retained_commit.filter(|committed_members| { + committed_members == &member_keys + && (status_checksum(&old_status) == Some(mdc_checksum.as_str()) + || status_checksum(&old_status).is_some_and(|previous| { + previous + .strip_prefix(mdc_checksum.as_str()) + .is_some_and(|suffix| suffix.starts_with("\0video_contract\0")) + })) }); cancel_build(&old_status); if retained_commit.is_some() { From 37231059ec34ff6750f7865560849874a63f932d Mon Sep 17 00:00:00 2001 From: GLAMR Date: Sat, 12 Sep 2026 00:43:02 +0000 Subject: [PATCH 21/25] style(discovery): format rebuild admission guard Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index e33e0276b617..242dc5bf71a2 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -513,8 +513,7 @@ impl ModelDiscoveryController { if !matches!( &old_status, GroupStatus::Ready { .. } | GroupStatus::BlockedReady { .. } - ) || (fingerprint_changed && retained_commit.is_some()) - { + ) || (fingerprint_changed && retained_commit.is_some()) { group.admission_tx.send_replace(admitted); } From d6789bb74741397405f1c72d00beef64c5e3c89d Mon Sep 17 00:00:00 2001 From: GLAMR Date: Sat, 12 Sep 2026 00:46:25 +0000 Subject: [PATCH 22/25] style(discovery): apply rustfmt Signed-off-by: GLAMR --- lib/llm/src/discovery/controller.rs | 50 +++++++++++++++++++---------- 1 file changed, 33 insertions(+), 17 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 242dc5bf71a2..00fd605341d3 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -475,10 +475,8 @@ impl ModelDiscoveryController { .to_string(); let member_keys = group.cohorts[&mdc_checksum].clone(); let members = self.members(&member_keys); - let fingerprint = cohort_fingerprint( - &mdc_checksum, - cohort_video_contract(&members).as_deref(), - ); + let fingerprint = + cohort_fingerprint(&mdc_checksum, cohort_video_contract(&members).as_deref()); let fingerprint_changed = status_checksum(&old_status).is_some_and(|previous| previous != fingerprint); let mut retained_commit = status_committed_members(&old_status).cloned(); @@ -513,7 +511,8 @@ impl ModelDiscoveryController { if !matches!( &old_status, GroupStatus::Ready { .. } | GroupStatus::BlockedReady { .. } - ) || (fingerprint_changed && retained_commit.is_some()) { + ) || (fingerprint_changed && retained_commit.is_some()) + { group.admission_tx.send_replace(admitted); } @@ -707,7 +706,9 @@ impl ModelDiscoveryController { continue; }; let committed_members = match &group.status { - GroupStatus::Queued { committed_members, .. } => committed_members.clone(), + GroupStatus::Queued { + committed_members, .. + } => committed_members.clone(), _ => unreachable!("queued status was checked above"), }; let Some(mdc_checksum) = group.selected_checksum().map(str::to_string) else { @@ -766,9 +767,9 @@ impl ModelDiscoveryController { }); let outcome = match future.catch_unwind().await { Ok(outcome) => outcome, - Err(_) => BuildOutcome::Failed(anyhow::anyhow!( - "model materialization panicked" - )), + Err(_) => { + BuildOutcome::Failed(anyhow::anyhow!("model materialization panicked")) + } }; BuildResult { spec: task_spec, @@ -805,7 +806,9 @@ impl ModelDiscoveryController { ) == result.spec.fingerprint }); let committed_members = match &group.status { - GroupStatus::Building { committed_members, .. } => committed_members.clone(), + GroupStatus::Building { + committed_members, .. + } => committed_members.clone(), _ => None, }; if !is_current { @@ -830,7 +833,8 @@ impl ModelDiscoveryController { self.host .replace_prepared_group(&result.spec, prepared, &members, &adapters) } else { - self.host.commit_group(&result.spec, prepared, &members, &adapters) + self.host + .commit_group(&result.spec, prepared, &members, &adapters) }; match commit_result { Ok(()) => { @@ -1053,12 +1057,24 @@ fn status_has_commit(status: &GroupStatus) -> bool { fn status_committed_members(status: &GroupStatus) -> Option<&BTreeSet> { match status { - GroupStatus::Ready { committed_members, .. } - | GroupStatus::BlockedReady { committed_members, .. } => Some(committed_members), - GroupStatus::Queued { committed_members, .. } - | GroupStatus::Building { committed_members, .. } - | GroupStatus::Retrying { committed_members, .. } - | GroupStatus::Blocked { committed_members, .. } => committed_members.as_ref(), + GroupStatus::Ready { + committed_members, .. + } + | GroupStatus::BlockedReady { + committed_members, .. + } => Some(committed_members), + GroupStatus::Queued { + committed_members, .. + } + | GroupStatus::Building { + committed_members, .. + } + | GroupStatus::Retrying { + committed_members, .. + } + | GroupStatus::Blocked { + committed_members, .. + } => committed_members.as_ref(), GroupStatus::Idle => None, } } From 99baf2717a306ae9fbf4759601772d7ed362622d Mon Sep 17 00:00:00 2001 From: GLAMR Date: Sat, 12 Sep 2026 00:48:12 +0000 Subject: [PATCH 23/25] style(discovery): format watcher replacement Signed-off-by: GLAMR --- lib/llm/src/discovery/watcher.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index 46c0e2205a36..00ab73528dfe 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -1233,9 +1233,7 @@ where .map(|name| { ( name.clone(), - self.manager - .get_committed_model(&name) - .is_some(), + self.manager.get_committed_model(&name).is_some(), ) }) .collect::>(); From 5f2819e820cf523f01e03d314043490ef04a5e3a Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Mon, 14 Sep 2026 21:45:29 +0000 Subject: [PATCH 24/25] fix(discovery): keep serving while a video-contract rebuild is prepared Retention of a committed WorkerSet required the desired membership to equal the committed membership. Removing the last legacy worker of a rolling upgrade changes membership and contract agreement at once, so retention was refused and the healthy group was withdrawn before its replacement had started building. Retain the committed workers that are still present instead, and require that set to be non-empty. Retention also tested the shape of the previous fingerprint rather than the workers themselves, so a worker that republished a different contract under the same instance ID kept its pipeline and kept expanding video prompts with the contract that pipeline was built from. Check the members instead: every worker the retained pipeline still serves must publish the contract it was built with. The retained pipeline also lost its admission sender, because the retained branch overwrote it with the successor's channel. Its client saw the channel close, left its update loop, and froze on the worker list it last observed. Keep the retained sender on the group, update it with the committed workers that still exist, and retire it when the replacement commits or the commit is withdrawn. The successor still builds against a fresh channel, so retained clients never observe the successor's worker IDs. Give the controller tests a contract-bearing fixture and cover the three behaviours, and make the fake host enforce the ModelManager rules that committing rejects an already committed group while replacing requires one. Signed-off-by: svc-glamr@nvidia.com --- lib/llm/src/discovery/controller.rs | 297 ++++++++++++++++++++++++++-- lib/llm/src/discovery/watcher.rs | 2 +- 2 files changed, 285 insertions(+), 14 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index 00fd605341d3..a6fb0dce86a2 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -77,6 +77,14 @@ fn cohort_fingerprint(mdc_checksum: &str, video_contract: Option<&str>) -> Strin } } +/// Splits a fingerprint back into the MDC checksum and video contract it carries. +fn fingerprint_parts(fingerprint: &str) -> (&str, Option<&str>) { + match fingerprint.split_once("\0video_contract\0") { + Some((mdc_checksum, contract)) => (mdc_checksum, Some(contract)), + None => (fingerprint, None), + } +} + #[derive(Clone, Debug)] pub(crate) struct GroupSpec { pub(crate) key: GroupKey, @@ -175,6 +183,9 @@ struct DesiredGroup { cohort_order: VecDeque, reported_rejections: HashSet, admission_tx: watch::Sender>, + /// Sender of the committed pipeline that keeps serving while its replacement + /// is prepared. Retired once the replacement commits or the commit is withdrawn. + retained_admission_tx: Option>>, status: GroupStatus, } @@ -187,6 +198,7 @@ impl DesiredGroup { cohort_order: VecDeque::new(), reported_rejections: HashSet::new(), admission_tx, + retained_admission_tx: None, status: GroupStatus::Idle, } } @@ -462,6 +474,9 @@ impl ModelDiscoveryController { if group.cohorts.is_empty() { group.admission_tx.send_replace(Vec::new()); + if let Some(retained_tx) = group.retained_admission_tx.take() { + retained_tx.send_replace(Vec::new()); + } cancel_build(&old_status); if status_has_commit(&old_status) { self.host.remove_group(key); @@ -481,21 +496,54 @@ impl ModelDiscoveryController { status_checksum(&old_status).is_some_and(|previous| previous != fingerprint); let mut retained_commit = status_committed_members(&old_status).cloned(); if fingerprint_changed { - retained_commit = retained_commit.filter(|committed_members| { - committed_members == &member_keys - && (status_checksum(&old_status) == Some(mdc_checksum.as_str()) - || status_checksum(&old_status).is_some_and(|previous| { - previous - .strip_prefix(mdc_checksum.as_str()) - .is_some_and(|suffix| suffix.starts_with("\0video_contract\0")) - })) + retained_commit = retained_commit.and_then(|committed_members| { + let (previous_checksum, retained_contract) = + fingerprint_parts(status_checksum(&old_status)?); + if previous_checksum != mdc_checksum { + return None; + } + // Keep serving the committed workers that are still here rather than + // demanding the whole membership, so dropping the last worker of a + // rolling upgrade does not withdraw a healthy group. + let surviving = members + .iter() + .filter(|member| committed_members.contains(&member.key)) + .collect::>(); + if surviving.is_empty() { + return None; + } + // The committed pipeline expands video prompts with the contract it + // was built from, so every worker it still serves must publish that + // same contract; one that republished a different contract under the + // same ID would otherwise keep receiving requests against the old one. + if let Some(contract) = retained_contract + && !surviving + .iter() + .all(|member| member.video_contract.as_deref() == Some(contract)) + { + return None; + } + Some( + surviving + .into_iter() + .map(|member| member.key.clone()) + .collect::>(), + ) }); cancel_build(&old_status); if retained_commit.is_some() { - let (admission_tx, _) = watch::channel(Vec::new()); - group.admission_tx = admission_tx; + // The retained pipeline keeps its own sender, or its client leaves the + // update loop and freezes on the worker list it last saw. Retained + // pipeline clients must never observe the successor's IDs, so the + // successor builds against a fresh channel. + let (successor_tx, _) = watch::channel(Vec::new()); + let previous_tx = std::mem::replace(&mut group.admission_tx, successor_tx); + group.retained_admission_tx.get_or_insert(previous_tx); } else { group.admission_tx.send_replace(Vec::new()); + if let Some(retained_tx) = group.retained_admission_tx.take() { + retained_tx.send_replace(Vec::new()); + } if status_has_commit(&old_status) { self.host.remove_group(key); } @@ -506,6 +554,20 @@ impl ModelDiscoveryController { group.reported_rejections.clear(); group.retry_attempt = 0; } + // Follow the retained pipeline's membership until it is retired, so a worker + // that leaves during a slow or failing rebuild stops receiving requests. + if let Some(retained_members) = &retained_commit { + if let Some(retained_tx) = &group.retained_admission_tx { + let still_serving = members + .iter() + .filter(|member| retained_members.contains(&member.key)) + .cloned() + .collect::>(); + retained_tx.send_replace(admitted_ids(&still_serving)); + } + } else if let Some(retained_tx) = group.retained_admission_tx.take() { + retained_tx.send_replace(Vec::new()); + } group.report_rejections(key); let admitted = admitted_ids(&members); if !matches!( @@ -838,6 +900,9 @@ impl ModelDiscoveryController { }; match commit_result { Ok(()) => { + // The replacement has taken over, so retire the sender the + // superseded pipeline was watching. + group.retained_admission_tx = None; group.retry_attempt = 0; group.status = GroupStatus::Ready { mdc_checksum: result.spec.fingerprint, @@ -942,6 +1007,9 @@ impl ModelDiscoveryController { async fn shutdown_builds(&mut self) { for group in self.groups.values_mut() { group.admission_tx.send_replace(Vec::new()); + if let Some(retained_tx) = group.retained_admission_tx.take() { + retained_tx.send_replace(Vec::new()); + } cancel_build(&group.status); } self.builds.abort_all(); @@ -1107,6 +1175,8 @@ mod tests { }; use tokio::sync::{Semaphore, mpsc}; + use crate::local_model::runtime_config::VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY; + struct Prepared(u64); struct FakeHost { @@ -1122,6 +1192,7 @@ mod tests { admissions: Mutex>>>, removed_groups: AtomicUsize, discarded: AtomicUsize, + prepared_replacements: AtomicUsize, } impl FakeHost { @@ -1141,6 +1212,7 @@ mod tests { admissions: Mutex::new(Vec::new()), removed_groups: AtomicUsize::new(0), discarded: AtomicUsize::new(0), + prepared_replacements: AtomicUsize::new(0), }), start_rx, ) @@ -1230,7 +1302,7 @@ mod tests { group_key: group_key(), mdc_checksum: card.mdcsum().to_string(), projection_fingerprint: card.source_path.clone().unwrap(), - video_contract: None, + video_contract: crate::discovery::watcher::qwen_video_contract_digest(&card), card, })) } @@ -1274,6 +1346,12 @@ mod tests { { anyhow::bail!("injected commit conflict"); } + // `ModelManager::commit_discovery_group` refuses a group ID it already holds. + anyhow::ensure!( + !self.committed.lock().unwrap().contains_key(&spec.key.id()), + "discovery group {:?} is already committed", + spec.key.id() + ); self.committed.lock().unwrap().insert( spec.key.id(), members.iter().map(|member| member.key.clone()).collect(), @@ -1297,6 +1375,12 @@ mod tests { { anyhow::bail!("injected replacement conflict"); } + // `ModelManager::replace_discovery_group` requires a committed group ID. + anyhow::ensure!( + self.committed.lock().unwrap().contains_key(&key.id()), + "committed discovery group {:?} not found", + key.id() + ); self.committed.lock().unwrap().insert( key.id(), members.iter().map(|member| member.key.clone()).collect(), @@ -1313,6 +1397,7 @@ mod tests { adapters: &[DesiredInstance], ) -> anyhow::Result<()> { let Prepared(_build) = prepared; + self.prepared_replacements.fetch_add(1, Ordering::SeqCst); self.replace_group(&spec.key, members, adapters) } @@ -1341,8 +1426,22 @@ mod tests { } fn instance(id: u64, mdc_checksum: &str) -> DesiredInstance { + build_instance(id, mdc_checksum, None) + } + + fn build_instance( + id: u64, + mdc_checksum: &str, + video_contract: Option, + ) -> DesiredInstance { let mut card = ModelDeploymentCard::with_name_only("model"); card.source_path = Some(mdc_checksum.to_string()); + if let Some(contract) = video_contract { + card.runtime_config.runtime_data.insert( + VLLM_QWEN_VIDEO_PROCESSOR_CONTRACT_RUNTIME_KEY.to_string(), + contract, + ); + } let mcid = ModelCardInstanceId { namespace: "namespace".to_string(), component: "worker".to_string(), @@ -1360,12 +1459,39 @@ mod tests { }, group_key: group_key(), mdc_checksum: card.mdcsum().to_string(), - card, projection_fingerprint: mdc_checksum.to_string(), - video_contract: None, + video_contract: crate::discovery::watcher::qwen_video_contract_digest(&card), + card, } } + /// Builds a worker that publishes a Qwen video prompt-expansion contract. + /// + /// The contract stays out of `mdcsum()`, so workers that differ only in + /// `resize_mode` share a cohort and disagree only on the contract. + fn instance_with_contract(id: u64, mdc_checksum: &str, resize_mode: &str) -> DesiredInstance { + let instance = build_instance( + id, + mdc_checksum, + Some(serde_json::json!({ + "placeholder_target": "bare_video_token", + "resize_mode": resize_mode, + })), + ); + assert!(instance.video_contract.is_some()); + assert_eq!( + instance.mdc_checksum, + self::instance(id, mdc_checksum).mdc_checksum + ); + instance + } + + fn contract_digest(resize_mode: &str) -> String { + instance_with_contract(0, "spec", resize_mode) + .video_contract + .unwrap() + } + fn discovery_instance(instance: &DesiredInstance) -> DiscoveryInstance { DiscoveryInstance::Model { namespace: instance.mcid.namespace.clone(), @@ -1383,6 +1509,151 @@ mod tests { controller.apply_build_result(result); } + #[tokio::test] + async fn contract_agreement_drives_the_fingerprint_and_survives_a_rolling_upgrade() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + // A legacy worker that predates the contract, alongside one that publishes it. + let legacy = instance(1, "same"); + let upgraded = instance_with_contract(2, "same", "round_ties_even"); + assert_eq!(legacy.mdc_checksum, upgraded.mdc_checksum); + + controller.apply_added(legacy.clone()); + controller.apply_added(upgraded.clone()); + controller.start_queued_builds(); + let mixed_spec = starts.recv().await.unwrap(); + // The cohort disagrees, so exact video routing stays off. + assert_eq!(mixed_spec.video_contract, None); + assert_eq!(mixed_spec.fingerprint, mixed_spec.mdc_checksum); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!( + host.members(&group_key()), + BTreeSet::from([legacy.key.clone(), upgraded.key.clone()]) + ); + let committed_admissions = host.admissions.lock().unwrap()[0].clone(); + assert_eq!(*committed_admissions.borrow(), vec![1, 2]); + + // Removing the last legacy worker changes membership and contract agreement + // at once. Text service must keep running while the replacement is prepared. + controller.apply_removed(&legacy.key); + assert_eq!( + host.members(&group_key()), + BTreeSet::from([legacy.key.clone(), upgraded.key.clone()]), + "the committed group must survive the upgrade's last legacy worker leaving" + ); + assert_eq!(host.removed_groups.load(Ordering::SeqCst), 0); + // The retained pipeline keeps its channel and stops admitting the worker + // that left, so it never routes to a worker that is gone. + assert_eq!(*committed_admissions.borrow(), vec![2]); + assert!(committed_admissions.has_changed().is_ok()); + + controller.start_queued_builds(); + let upgraded_spec = starts.recv().await.unwrap(); + assert_eq!( + upgraded_spec.video_contract, + Some(contract_digest("round_ties_even")) + ); + assert_eq!( + upgraded_spec.fingerprint, + cohort_fingerprint( + &upgraded_spec.mdc_checksum, + Some(&contract_digest("round_ties_even")) + ) + ); + host.release.add_permits(1); + finish_build(&mut controller).await; + + // A retained commit is replaced in place, never committed a second time. + assert_eq!(host.prepared_replacements.load(Ordering::SeqCst), 1); + assert_eq!(host.members(&group_key()), BTreeSet::from([upgraded.key])); + assert_eq!(host.removed_groups.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn republished_contract_withdraws_the_commit_it_no_longer_matches() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + let worker = instance_with_contract(1, "same", "round_ties_even"); + let peer = instance_with_contract(2, "same", "round_ties_even"); + + controller.apply_added(worker.clone()); + controller.apply_added(peer.clone()); + controller.start_queued_builds(); + let spec = starts.recv().await.unwrap(); + assert_eq!( + spec.video_contract, + Some(contract_digest("round_ties_even")) + ); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!(host.members(&group_key()).len(), 2); + + // The same instance ID republishes a different contract. The committed + // pipeline expands video prompts with the old one, so it cannot be kept. + let republished = instance_with_contract(1, "same", "legacy_ceil"); + assert_eq!(republished.key, worker.key); + assert_ne!(republished.video_contract, worker.video_contract); + assert!(controller.apply_added(republished)); + + assert!( + host.members(&group_key()).is_empty(), + "a worker serving a contract it no longer publishes must lose its commit" + ); + assert_eq!(host.removed_groups.load(Ordering::SeqCst), 1); + + // The cohort now disagrees, so the rebuild turns exact video routing off. + controller.start_queued_builds(); + let rebuilt = starts.recv().await.unwrap(); + assert_eq!(rebuilt.video_contract, None); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!(host.prepared_replacements.load(Ordering::SeqCst), 0); + assert_eq!(host.members(&group_key()).len(), 2); + } + + #[tokio::test(start_paused = true)] + async fn a_retained_pipeline_keeps_receiving_worker_updates_while_rebuilding() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + let first = instance_with_contract(1, "same", "round_ties_even"); + let second = instance_with_contract(2, "same", "round_ties_even"); + let legacy = instance(3, "same"); + + controller.apply_added(first.clone()); + controller.apply_added(second.clone()); + controller.apply_added(legacy.clone()); + controller.start_queued_builds(); + starts.recv().await.unwrap(); + host.release.add_permits(1); + finish_build(&mut controller).await; + let committed_admissions = host.admissions.lock().unwrap()[0].clone(); + assert_eq!(*committed_admissions.borrow(), vec![1, 2, 3]); + + // Dropping the legacy worker retains the commit and starts a rebuild that fails. + host.failures.store(1, Ordering::SeqCst); + controller.apply_removed(&legacy.key); + assert_eq!(*committed_admissions.borrow(), vec![1, 2]); + controller.start_queued_builds(); + starts.recv().await.unwrap(); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!(host.members(&group_key()).len(), 3); + + // A worker leaving during the failed rebuild must still reach the retained + // pipeline, or it keeps routing to a worker that is gone. + controller.apply_removed(&second.key); + assert_eq!(*committed_admissions.borrow(), vec![1]); + assert!(committed_admissions.has_changed().is_ok()); + + controller.start_queued_builds(); + starts.recv().await.unwrap(); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!(host.members(&group_key()), BTreeSet::from([first.key])); + assert_eq!(host.prepared_replacements.load(Ordering::SeqCst), 1); + } + #[tokio::test] async fn membership_churn_keeps_one_build_and_commits_latest_members() { let (host, mut starts) = FakeHost::new(); diff --git a/lib/llm/src/discovery/watcher.rs b/lib/llm/src/discovery/watcher.rs index 00ab73528dfe..724428e393c3 100644 --- a/lib/llm/src/discovery/watcher.rs +++ b/lib/llm/src/discovery/watcher.rs @@ -1402,7 +1402,7 @@ fn lora_projection_fingerprint(card: &ModelDeploymentCard) -> anyhow::Result Option { +pub(super) fn qwen_video_contract_digest(card: &ModelDeploymentCard) -> Option { let mut contract = card .runtime_config .runtime_data From 25770fcfa902a0e86215860b5681dbdc501f963d Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Wed, 16 Sep 2026 02:38:22 +0000 Subject: [PATCH 25/25] fix(discovery): recheck the retained contract during pending rebuilds A retained pipeline keeps expanding video prompts with the contract it was built from, but the group status carries the replacement's fingerprint while the rebuild is pending. A worker that republished a different contract without moving that fingerprint therefore kept serving against the contract it dropped. Bind the contract to the retained pipeline and revalidate it on every worker update, not only when the fingerprint changes. Signed-off-by: svc-glamr@nvidia.com --- lib/llm/src/discovery/controller.rs | 237 +++++++++++++++++++++------- 1 file changed, 184 insertions(+), 53 deletions(-) diff --git a/lib/llm/src/discovery/controller.rs b/lib/llm/src/discovery/controller.rs index a6fb0dce86a2..550fca7427ca 100644 --- a/lib/llm/src/discovery/controller.rs +++ b/lib/llm/src/discovery/controller.rs @@ -177,15 +177,24 @@ enum GroupStatus { }, } +/// The committed pipeline that keeps serving while its replacement is prepared. +/// Retired once the replacement commits or the commit is withdrawn. +struct RetainedPipeline { + admission_tx: watch::Sender>, + /// Contract this pipeline expands video prompts with. Held here because the + /// group's status carries the *replacement's* fingerprint while a rebuild is + /// pending, so it is the only record of what the retained workers must publish. + /// `None` means it was built without exact video routing. + contract: Option, +} + struct DesiredGroup { retry_attempt: u32, cohorts: HashMap>, cohort_order: VecDeque, reported_rejections: HashSet, admission_tx: watch::Sender>, - /// Sender of the committed pipeline that keeps serving while its replacement - /// is prepared. Retired once the replacement commits or the commit is withdrawn. - retained_admission_tx: Option>>, + retained: Option, status: GroupStatus, } @@ -198,7 +207,7 @@ impl DesiredGroup { cohort_order: VecDeque::new(), reported_rejections: HashSet::new(), admission_tx, - retained_admission_tx: None, + retained: None, status: GroupStatus::Idle, } } @@ -474,8 +483,8 @@ impl ModelDiscoveryController { if group.cohorts.is_empty() { group.admission_tx.send_replace(Vec::new()); - if let Some(retained_tx) = group.retained_admission_tx.take() { - retained_tx.send_replace(Vec::new()); + if let Some(retained) = group.retained.take() { + retained.admission_tx.send_replace(Vec::new()); } cancel_build(&old_status); if status_has_commit(&old_status) { @@ -495,40 +504,22 @@ impl ModelDiscoveryController { let fingerprint_changed = status_checksum(&old_status).is_some_and(|previous| previous != fingerprint); let mut retained_commit = status_committed_members(&old_status).cloned(); + // The contract the currently committed pipeline serves with. While a rebuild + // is pending, `old_status` names the replacement, so the retained pipeline is + // the authority on what its own workers must publish. + let committed_contract = match &group.retained { + Some(retained) => retained.contract.clone(), + None => fingerprint_parts(status_checksum(&old_status).unwrap_or_default()) + .1 + .map(str::to_string), + }; if fingerprint_changed { retained_commit = retained_commit.and_then(|committed_members| { - let (previous_checksum, retained_contract) = - fingerprint_parts(status_checksum(&old_status)?); + let (previous_checksum, _) = fingerprint_parts(status_checksum(&old_status)?); if previous_checksum != mdc_checksum { return None; } - // Keep serving the committed workers that are still here rather than - // demanding the whole membership, so dropping the last worker of a - // rolling upgrade does not withdraw a healthy group. - let surviving = members - .iter() - .filter(|member| committed_members.contains(&member.key)) - .collect::>(); - if surviving.is_empty() { - return None; - } - // The committed pipeline expands video prompts with the contract it - // was built from, so every worker it still serves must publish that - // same contract; one that republished a different contract under the - // same ID would otherwise keep receiving requests against the old one. - if let Some(contract) = retained_contract - && !surviving - .iter() - .all(|member| member.video_contract.as_deref() == Some(contract)) - { - return None; - } - Some( - surviving - .into_iter() - .map(|member| member.key.clone()) - .collect::>(), - ) + surviving_commit(&members, &committed_members, committed_contract.as_deref()) }); cancel_build(&old_status); if retained_commit.is_some() { @@ -538,11 +529,14 @@ impl ModelDiscoveryController { // successor builds against a fresh channel. let (successor_tx, _) = watch::channel(Vec::new()); let previous_tx = std::mem::replace(&mut group.admission_tx, successor_tx); - group.retained_admission_tx.get_or_insert(previous_tx); + group.retained.get_or_insert_with(|| RetainedPipeline { + admission_tx: previous_tx, + contract: committed_contract, + }); } else { group.admission_tx.send_replace(Vec::new()); - if let Some(retained_tx) = group.retained_admission_tx.take() { - retained_tx.send_replace(Vec::new()); + if let Some(retained) = group.retained.take() { + retained.admission_tx.send_replace(Vec::new()); } if status_has_commit(&old_status) { self.host.remove_group(key); @@ -553,20 +547,34 @@ impl ModelDiscoveryController { } group.reported_rejections.clear(); group.retry_attempt = 0; + } else if group.retained.is_some() { + // A rebuild is already pending, so the fingerprint tracks the replacement + // and can stay put while the retained workers change underneath it. Its + // contract is checked again here, or a worker that republished a different + // one keeps receiving requests expanded with the contract it dropped. + let had_commit = retained_commit.is_some(); + retained_commit = retained_commit.and_then(|committed_members| { + surviving_commit(&members, &committed_members, committed_contract.as_deref()) + }); + if had_commit && retained_commit.is_none() { + self.host.remove_group(key); + } } // Follow the retained pipeline's membership until it is retired, so a worker // that leaves during a slow or failing rebuild stops receiving requests. if let Some(retained_members) = &retained_commit { - if let Some(retained_tx) = &group.retained_admission_tx { + if let Some(retained) = &group.retained { let still_serving = members .iter() .filter(|member| retained_members.contains(&member.key)) .cloned() .collect::>(); - retained_tx.send_replace(admitted_ids(&still_serving)); + retained + .admission_tx + .send_replace(admitted_ids(&still_serving)); } - } else if let Some(retained_tx) = group.retained_admission_tx.take() { - retained_tx.send_replace(Vec::new()); + } else if let Some(retained) = group.retained.take() { + retained.admission_tx.send_replace(Vec::new()); } group.report_rejections(key); let admitted = admitted_ids(&members); @@ -642,41 +650,44 @@ impl ModelDiscoveryController { } } } + // These arms hold a replacement whose fingerprint has not moved. They carry + // `retained_commit` rather than the commit the status arrived with, so a + // commit withdrawn above is not claimed again by the pending rebuild. GroupStatus::Building { mdc_checksum: building_checksum, generation, cancellation, - committed_members, + .. } if building_checksum == fingerprint => GroupStatus::Building { mdc_checksum: fingerprint, generation, cancellation, - committed_members, + committed_members: retained_commit, }, GroupStatus::Queued { mdc_checksum: queued_checksum, - committed_members, + .. } if queued_checksum == fingerprint => GroupStatus::Queued { mdc_checksum: fingerprint, - committed_members, + committed_members: retained_commit, }, GroupStatus::Retrying { mdc_checksum: retry_checksum, deadline, - committed_members, + .. } if retry_checksum == fingerprint && !desired_changed => GroupStatus::Retrying { mdc_checksum: fingerprint, deadline, - committed_members, + committed_members: retained_commit, }, GroupStatus::Blocked { mdc_checksum: blocked_checksum, deadline, - committed_members, + .. } if blocked_checksum == fingerprint && !desired_changed => GroupStatus::Blocked { mdc_checksum: fingerprint, deadline, - committed_members, + committed_members: retained_commit, }, previous => { cancel_build(&previous); @@ -902,7 +913,7 @@ impl ModelDiscoveryController { Ok(()) => { // The replacement has taken over, so retire the sender the // superseded pipeline was watching. - group.retained_admission_tx = None; + group.retained = None; group.retry_attempt = 0; group.status = GroupStatus::Ready { mdc_checksum: result.spec.fingerprint, @@ -1007,8 +1018,8 @@ impl ModelDiscoveryController { async fn shutdown_builds(&mut self) { for group in self.groups.values_mut() { group.admission_tx.send_replace(Vec::new()); - if let Some(retained_tx) = group.retained_admission_tx.take() { - retained_tx.send_replace(Vec::new()); + if let Some(retained) = group.retained.take() { + retained.admission_tx.send_replace(Vec::new()); } cancel_build(&group.status); } @@ -1107,6 +1118,42 @@ fn cancel_build(status: &GroupStatus) { } } +/// The committed workers that may keep serving: those still present, provided they +/// all still publish `contract`. `None` withdraws the commit. +fn surviving_commit( + members: &[DesiredInstance], + committed_members: &BTreeSet, + contract: Option<&str>, +) -> Option> { + // Keep serving the committed workers that are still here rather than demanding + // the whole membership, so dropping the last worker of a rolling upgrade does + // not withdraw a healthy group. + let surviving = members + .iter() + .filter(|member| committed_members.contains(&member.key)) + .collect::>(); + if surviving.is_empty() { + return None; + } + // The committed pipeline expands video prompts with the contract it was built + // from, so every worker it still serves must publish that same contract; one + // that republished a different contract under the same ID would otherwise keep + // receiving requests against the old one. + if let Some(contract) = contract + && !surviving + .iter() + .all(|member| member.video_contract.as_deref() == Some(contract)) + { + return None; + } + Some( + surviving + .into_iter() + .map(|member| member.key.clone()) + .collect(), + ) +} + fn status_checksum(status: &GroupStatus) -> Option<&str> { match status { GroupStatus::Idle => None, @@ -1612,6 +1659,90 @@ mod tests { assert_eq!(host.members(&group_key()).len(), 2); } + #[tokio::test] + async fn a_republished_contract_during_a_pending_rebuild_withdraws_the_retained_commit() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + let first = instance_with_contract(1, "same", "round_ties_even"); + let second = instance_with_contract(2, "same", "round_ties_even"); + + controller.apply_added(first.clone()); + controller.apply_added(second.clone()); + controller.start_queued_builds(); + starts.recv().await.unwrap(); + host.release.add_permits(1); + finish_build(&mut controller).await; + let committed_admissions = host.admissions.lock().unwrap()[0].clone(); + assert_eq!(*committed_admissions.borrow(), vec![1, 2]); + + // A worker without a contract joins, so the cohort stops agreeing and a + // replacement with exact video routing off is queued. The commit built from + // `round_ties_even` keeps serving while that replacement is prepared. + let legacy = instance(3, "same"); + controller.apply_added(legacy.clone()); + assert_eq!(host.members(&group_key()).len(), 2); + assert_eq!(*committed_admissions.borrow(), vec![1, 2]); + assert_eq!(host.removed_groups.load(Ordering::SeqCst), 0); + + // Before that rebuild completes, a retained worker republishes a different + // contract. The cohort still has no common contract, so the replacement's + // fingerprint does not move -- but the retained pipeline still expands video + // prompts with `round_ties_even`, which worker 1 no longer publishes. + let republished = instance_with_contract(1, "same", "legacy_ceil"); + assert_eq!(republished.key, first.key); + assert!(controller.apply_added(republished)); + + assert!( + host.members(&group_key()).is_empty(), + "a retained pipeline must not keep serving a worker that republished a different contract" + ); + assert_eq!(host.removed_groups.load(Ordering::SeqCst), 1); + assert!(committed_admissions.borrow().is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_republished_contract_during_a_failed_rebuild_withdraws_the_retained_commit() { + let (host, mut starts) = FakeHost::new(); + let mut controller = ModelDiscoveryController::new(host.clone()); + let first = instance_with_contract(1, "same", "round_ties_even"); + let second = instance_with_contract(2, "same", "round_ties_even"); + + controller.apply_added(first.clone()); + controller.apply_added(second.clone()); + controller.start_queued_builds(); + starts.recv().await.unwrap(); + host.release.add_permits(1); + finish_build(&mut controller).await; + let committed_admissions = host.admissions.lock().unwrap()[0].clone(); + assert_eq!(*committed_admissions.borrow(), vec![1, 2]); + + // A worker without a contract joins and the replacement build fails, so the + // commit built from `round_ties_even` keeps serving across the retry. + host.failures.store(1, Ordering::SeqCst); + controller.apply_added(instance(3, "same")); + controller.start_queued_builds(); + starts.recv().await.unwrap(); + host.release.add_permits(1); + finish_build(&mut controller).await; + assert_eq!(host.members(&group_key()).len(), 2); + assert_eq!(host.removed_groups.load(Ordering::SeqCst), 0); + assert_eq!(*committed_admissions.borrow(), vec![1, 2]); + + // A retained worker republishes a different contract while the failed rebuild + // waits to retry. The retained pipeline still expands video prompts with + // `round_ties_even`, so it cannot keep serving that worker. + let republished = instance_with_contract(1, "same", "legacy_ceil"); + assert_eq!(republished.key, first.key); + assert!(controller.apply_added(republished)); + + assert!( + host.members(&group_key()).is_empty(), + "a failed rebuild must not keep a commit whose worker republished a different contract" + ); + assert_eq!(host.removed_groups.load(Ordering::SeqCst), 1); + assert!(committed_admissions.borrow().is_empty()); + } + #[tokio::test(start_paused = true)] async fn a_retained_pipeline_keeps_receiving_worker_updates_while_rebuilding() { let (host, mut starts) = FakeHost::new();