diff --git a/model_gateway/src/routers/common/worker_selection.rs b/model_gateway/src/routers/common/worker_selection.rs index ee3af291c3..f1dab86fbb 100644 --- a/model_gateway/src/routers/common/worker_selection.rs +++ b/model_gateway/src/routers/common/worker_selection.rs @@ -16,7 +16,10 @@ use crate::{ common::header_utils::{apply_provider_headers, extract_auth_header}, error, }, - worker::{ConnectionMode, ProviderType, RuntimeType, Worker, WorkerRegistry, WorkerType}, + worker::{ + ConnectionMode, ProviderType, RuntimeType, Worker, WorkerRegistry, WorkerType, + UNKNOWN_MODEL_ID, + }, }; /// Holds references to shared infrastructure needed for worker selection. @@ -102,16 +105,53 @@ impl<'a> WorkerSelector<'a> { }) } + /// Available workers passing every request filter, ready for load-based + /// selection. + /// + /// When the model is known, this uses the registry's bounded, + /// wildcard-safe per-model lookup (`get_candidates_for_model`) instead + /// of scanning the whole fleet — the O(total fleet) scan was the hot-path + /// cost this addresses. If that bounded set turns up no available + /// candidates, it falls back to the full scan so behavior is never worse + /// (the bounded index can briefly lag a concurrent registration, and we + /// must never regress a servable model into "no worker found"). fn get_candidates(&self, req: &SelectWorkerRequest<'_>) -> Vec> { - let workers = self.registry.get_workers_filtered( - None, // model_id index lookup not used — we filter via supports_model - req.worker_type, - req.connection_mode, - req.runtime_type, - false, // we filter availability ourselves for consistent behavior - ); + if req.model_id != UNKNOWN_MODEL_ID { + let bounded = + Self::filter_candidates(self.registry.get_candidates_for_model(req.model_id), req); + if !bounded.is_empty() { + return bounded; + } + } + + // Unknown model (caller wants any worker) or empty bounded set: + // fall back to the full scan, preserving the original behavior. + Self::filter_candidates( + self.registry.get_workers_filtered( + None, // full scan — wildcard-safe via supports_model below + req.worker_type, + req.connection_mode, + req.runtime_type, + false, // we filter availability ourselves for consistent behavior + ), + req, + ) + } - let candidates: Vec<_> = workers.into_iter().filter(|w| w.is_available()).collect(); + /// Apply the request's worker_type/connection_mode/runtime_type, + /// availability, and provider filters to a candidate set. + /// + /// `get_workers_filtered` already applies the type/mode/runtime filters, + /// so re-applying them to that path is a cheap no-op; the bounded + /// per-model path relies on them here. + fn filter_candidates( + workers: Vec>, + req: &SelectWorkerRequest<'_>, + ) -> Vec> { + let candidates: Vec<_> = workers + .into_iter() + .filter(|w| Self::matches_filters(w, req) && w.is_available()) + .collect(); match &req.provider { Some(provider) => filter_by_provider(candidates, provider), @@ -119,6 +159,28 @@ impl<'a> WorkerSelector<'a> { } } + /// Per-worker worker_type/connection_mode/runtime_type filter, mirroring + /// `WorkerRegistry::get_workers_filtered`. Applied to the bounded + /// per-model candidate set (which is not pre-filtered). + fn matches_filters(worker: &Arc, req: &SelectWorkerRequest<'_>) -> bool { + if let Some(ref wtype) = req.worker_type { + if *worker.worker_type() != *wtype { + return false; + } + } + if let Some(ref conn) = req.connection_mode { + if worker.connection_mode() != conn { + return false; + } + } + if let Some(ref rt) = req.runtime_type { + if worker.metadata().spec.runtime_type != *rt { + return false; + } + } + true + } + fn find_best_worker(&self, req: &SelectWorkerRequest<'_>) -> Option> { self.get_candidates(req) .into_iter() @@ -129,6 +191,19 @@ impl<'a> WorkerSelector<'a> { /// Check if any healthy worker supports the model (regardless of circuit breaker). /// Used to distinguish "model not found" from "all workers circuit-broken". fn any_worker_supports_model(&self, req: &SelectWorkerRequest<'_>) -> bool { + if req.model_id != UNKNOWN_MODEL_ID { + let bounded = Self::healthy_supporting_candidates( + self.registry.get_candidates_for_model(req.model_id), + req, + ); + if bounded.iter().any(|w| w.supports_model(req.model_id)) { + return true; + } + // Bounded set yielded no supporting worker — fall through to the + // full scan so we never falsely report "model not found" if the + // per-model index briefly lags a registration. + } + let workers = self.registry.get_workers_filtered( None, req.worker_type, @@ -136,11 +211,26 @@ impl<'a> WorkerSelector<'a> { req.runtime_type, true, // healthy only — model exists even if circuit-broken ); - let candidates = match &req.provider { - Some(p) => filter_by_provider(workers, p), - None => workers, - }; - candidates.iter().any(|w| w.supports_model(req.model_id)) + Self::healthy_supporting_candidates(workers, req) + .iter() + .any(|w| w.supports_model(req.model_id)) + } + + /// Filter a candidate set to healthy workers passing the request's + /// type/mode/runtime and provider filters (no circuit-breaker check — + /// the model exists even if every worker is circuit-broken). + fn healthy_supporting_candidates( + workers: Vec>, + req: &SelectWorkerRequest<'_>, + ) -> Vec> { + let candidates: Vec<_> = workers + .into_iter() + .filter(|w| Self::matches_filters(w, req) && w.is_healthy()) + .collect(); + match &req.provider { + Some(p) => filter_by_provider(candidates, p), + None => candidates, + } } /// Refresh model lists for healthy external workers in parallel. @@ -274,3 +364,196 @@ async fn refresh_worker_models( } } } + +#[cfg(test)] +mod tests { + use openai_protocol::{model_card::ModelCard, worker::HealthCheckConfig}; + + use super::*; + use crate::worker::BasicWorkerBuilder; + + fn ready_worker(builder: BasicWorkerBuilder) -> Arc { + // disable_health_check makes the worker start Ready (and thus + // is_available, since a fresh circuit breaker permits execution). + let worker: Arc = Arc::new( + builder + .health_config(HealthCheckConfig { + disable_health_check: true, + ..Default::default() + }) + .build(), + ); + assert!(worker.is_available(), "test worker must be routable"); + worker + } + + fn selector_request(model_id: &str) -> SelectWorkerRequest<'_> { + SelectWorkerRequest { + model_id, + ..Default::default() + } + } + + fn client() -> reqwest::Client { + reqwest::Client::new() + } + + /// The core regression: a wildcard worker (no models declared) must be + /// selectable for an arbitrary model via the bounded candidate path, + /// even though it is not in `get_by_model(arbitrary_model)`. + #[tokio::test] + async fn wildcard_worker_selected_for_arbitrary_model_via_bounded_path() { + let registry = WorkerRegistry::new(); + registry + .register(ready_worker(BasicWorkerBuilder::new( + "http://wildcard:8080", + ))) + .unwrap(); + + // Sanity: not in the per-model index for this arbitrary model. + assert!(registry.get_by_model("totally-made-up-model").is_empty()); + + let client = client(); + let selector = WorkerSelector::new(®istry, &client); + let chosen = selector + .select_worker(&selector_request("totally-made-up-model")) + .await + .expect("wildcard worker should serve any model"); + assert_eq!(chosen.url(), "http://wildcard:8080"); + } + + /// Per-model bounding returns the right worker for a normal model and + /// excludes workers serving other models. + #[tokio::test] + async fn bounded_path_selects_correct_model_and_excludes_others() { + let registry = WorkerRegistry::new(); + registry + .register(ready_worker( + BasicWorkerBuilder::new("http://a:8080").model(ModelCard::new("model-a")), + )) + .unwrap(); + registry + .register(ready_worker( + BasicWorkerBuilder::new("http://b:8080").model(ModelCard::new("model-b")), + )) + .unwrap(); + + let client = client(); + let selector = WorkerSelector::new(®istry, &client); + + let chosen = selector + .select_worker(&selector_request("model-a")) + .await + .expect("model-a worker exists"); + assert_eq!( + chosen.url(), + "http://a:8080", + "must not pick model-b worker" + ); + } + + /// Empty-bounded-set fallback: a worker reachable only by an *alias* + /// (not its indexed id, and not a wildcard) is found via the full-scan + /// fallback so we never regress a servable model into model_not_found. + #[tokio::test] + async fn empty_bounded_set_falls_back_to_full_scan() { + let registry = WorkerRegistry::new(); + // Indexed under id "gpt-4"; supports "gpt-4-latest" only via alias. + registry + .register(ready_worker( + BasicWorkerBuilder::new("http://aliased:8080") + .model(ModelCard::new("gpt-4").with_alias("gpt-4-latest")), + )) + .unwrap(); + + // The bounded lookup keys on the literal id and finds nothing for the + // alias (no wildcard workers either) — proving the fallback is what + // surfaces the worker. + assert!( + registry.get_candidates_for_model("gpt-4-latest").is_empty(), + "alias is not an index key, so the bounded set is empty" + ); + + let client = client(); + let selector = WorkerSelector::new(®istry, &client); + let chosen = selector + .select_worker(&selector_request("gpt-4-latest")) + .await + .expect("alias-only worker must be found via fallback"); + assert_eq!(chosen.url(), "http://aliased:8080"); + } + + /// A genuinely unknown model with no wildcard workers yields + /// model_not_found (the fallback does not invent workers). + #[tokio::test] + async fn unknown_model_without_wildcard_is_not_found() { + let registry = WorkerRegistry::new(); + registry + .register(ready_worker( + BasicWorkerBuilder::new("http://a:8080").model(ModelCard::new("model-a")), + )) + .unwrap(); + + let client = client(); + let selector = WorkerSelector::new(®istry, &client); + let result = selector + .select_worker(&selector_request("nonexistent")) + .await; + assert!(result.is_err(), "no worker and no wildcard → error"); + } + + /// With UNKNOWN_MODEL_ID the common/HTTP path takes the full-scan branch + /// (no per-model index lookup). A wildcard worker — which `supports_model` + /// accepts for the sentinel — is selectable, matching the pre-bounding + /// behavior. (A *specific* worker does not `supports_model` the sentinel, + /// so this path only resolves to wildcards, unchanged by this PR.) + #[tokio::test] + async fn unknown_model_id_selects_wildcard_via_full_scan() { + let registry = WorkerRegistry::new(); + registry + .register(ready_worker(BasicWorkerBuilder::new( + "http://wildcard:8080", + ))) + .unwrap(); + + let client = client(); + let selector = WorkerSelector::new(®istry, &client); + let chosen = selector + .select_worker(&selector_request(UNKNOWN_MODEL_ID)) + .await + .expect("wildcard worker for unknown model id"); + assert_eq!(chosen.url(), "http://wildcard:8080"); + } + + /// The bounded path still honors the worker_type filter. + #[tokio::test] + async fn bounded_path_honors_worker_type_filter() { + let registry = WorkerRegistry::new(); + registry + .register(ready_worker( + BasicWorkerBuilder::new("http://regular:8080") + .model(ModelCard::new("m")) + .worker_type(WorkerType::Regular), + )) + .unwrap(); + registry + .register(ready_worker( + BasicWorkerBuilder::new("http://prefill:8080") + .model(ModelCard::new("m")) + .worker_type(WorkerType::Prefill), + )) + .unwrap(); + + let client = client(); + let selector = WorkerSelector::new(®istry, &client); + let chosen = selector + .select_worker(&SelectWorkerRequest { + model_id: "m", + worker_type: Some(WorkerType::Prefill), + ..Default::default() + }) + .await + .expect("prefill worker exists for model m"); + assert_eq!(chosen.url(), "http://prefill:8080"); + } +} diff --git a/model_gateway/src/worker/registry.rs b/model_gateway/src/worker/registry.rs index a1d8431fa0..17dfeeadd5 100644 --- a/model_gateway/src/worker/registry.rs +++ b/model_gateway/src/worker/registry.rs @@ -17,7 +17,7 @@ use std::{ sync::Arc, }; -use dashmap::{mapref::entry::Entry, DashMap}; +use dashmap::{mapref::entry::Entry, DashMap, DashSet}; use openai_protocol::worker::WorkerStatus; use tokio::sync::broadcast; use uuid::Uuid; @@ -94,6 +94,19 @@ pub struct WorkerRegistry { /// Uses Arc<[T]> instead of Arc>> for lock-free reads. model_index: ModelIndex, + /// IDs of wildcard workers (those advertising no specific models, which + /// `supports_model` accepts for *any* model_id). Wildcards are not + /// reliably indexed under a single key in `model_index` — `model_id()` + /// falls back to a `model_id` label or [`crate::worker::UNKNOWN_MODEL_ID`], + /// so a per-model lookup cannot find them. This set lets the bounded + /// candidate lookup ([`Self::get_candidates_for_model`]) union them in + /// without an O(total fleet) scan. Maintained under the per-worker + /// mutation lock in register / replace / remove. Membership is captured + /// from `has_models_discovered()` at mutation time; a worker that later + /// discovers models via lazy refresh stays in the set until its next + /// mutation, which is harmless — `supports_model` re-filters the union. + wildcard_workers: Arc>, + /// Consistent hash rings per model for O(log n) routing. /// Rebuilt on worker add/remove (copy-on-write). hash_rings: Arc>>, @@ -139,6 +152,7 @@ impl WorkerRegistry { Self { workers: Arc::new(DashMap::new()), model_index: Arc::new(DashMap::new()), + wildcard_workers: Arc::new(DashSet::new()), hash_rings: Arc::new(DashMap::new()), type_workers: Arc::new(DashMap::new()), connection_workers: Arc::new(DashMap::new()), @@ -292,6 +306,48 @@ impl WorkerRegistry { .unwrap_or_else(|| Arc::from(Self::EMPTY_WORKERS)) } + /// Return every worker that could serve `model_id`: the per-model index + /// slice unioned with all wildcard workers, deduplicated by URL. + /// + /// This is the bounded, wildcard-safe candidate lookup. `get_by_model` + /// alone misses wildcard workers (those advertising no specific models), + /// because they are indexed under a `model_id` label or + /// [`crate::worker::UNKNOWN_MODEL_ID`] rather than under `model_id`. + /// `WorkerSelector` previously scanned the entire fleet to catch them; + /// this restores that correctness at O(per-model + wildcards) cost + /// instead of O(total fleet). Wildcards are few, so the union is cheap. + /// + /// Callers still apply `supports_model` (and any other filters) to the + /// result — a wildcard worker that has since discovered specific models + /// is correctly rejected there. + pub fn get_candidates_for_model(&self, model_id: &str) -> Vec> { + let indexed = self.get_by_model(model_id); + + // Fast path: no wildcard workers, so the per-model slice is the full + // candidate set. Avoids the dedup HashSet allocation entirely. + if self.wildcard_workers.is_empty() { + return indexed.to_vec(); + } + + let mut seen: HashSet = HashSet::with_capacity(indexed.len()); + let mut candidates: Vec> = Vec::with_capacity(indexed.len()); + for worker in indexed.iter() { + if seen.insert(worker.url().to_string()) { + candidates.push(Arc::clone(worker)); + } + } + for entry in self.wildcard_workers.iter() { + if let Some(worker) = self.get(entry.key()) { + // A wildcard worker may also be model-indexed (e.g. under a + // `model_id` label that equals `model_id`); dedup by URL. + if seen.insert(worker.url().to_string()) { + candidates.push(worker); + } + } + } + candidates + } + /// Return all workers of a given type as an immutable shared slice. /// /// Unified with [`Self::get_by_model`] on `Arc<[_]>` so callers can @@ -751,6 +807,11 @@ impl WorkerRegistry { self.rebuild_hash_ring(kept_model); } + // Re-evaluate wildcard membership for the replacement: the new + // worker may declare specific models where the old was wildcard, + // or vice versa. + self.update_wildcard_membership(worker_id, &new_worker); + self.warn_on_sampling_defaults_divergence_for_worker(&new_worker); if old_worker.worker_type() != new_worker.worker_type() { @@ -971,6 +1032,7 @@ impl WorkerRegistry { // We hold _guard; drop the DashMap entry but the Mutex stays alive via Arc. self.worker_mutation_locks.remove(worker_id); self.worker_origins.remove(worker_id); + self.wildcard_workers.remove(worker_id); for model_id in Self::worker_model_ids(&worker) { self.remove_worker_from_model_index(&model_id, worker.url()); @@ -1085,6 +1147,21 @@ impl WorkerRegistry { model_ids } + /// Add or remove `worker_id` from the wildcard set based on whether + /// `worker` currently advertises no specific models. Called under the + /// per-worker mutation lock from register / replace so the set stays in + /// sync with the worker object installed in `workers`. + fn update_wildcard_membership(&self, worker_id: &WorkerId, worker: &Arc) { + // A worker is a routing wildcard when it has not (yet) had specific + // models discovered/declared — exactly when `supports_model` accepts + // any model_id. `has_models_discovered()` is the trait-level predicate. + if worker.has_models_discovered() { + self.wildcard_workers.remove(worker_id); + } else { + self.wildcard_workers.insert(worker_id.clone()); + } + } + fn sampling_defaults_label(worker: &Arc) -> Option<&str> { worker .metadata() @@ -1201,6 +1278,9 @@ impl WorkerRegistry { self.add_worker_to_model_index(&model_id, worker.clone()); self.rebuild_hash_ring(&model_id); } + // Track wildcard workers separately: they are not reliably indexed + // under a per-model key, so the bounded candidate lookup unions them. + self.update_wildcard_membership(&worker_id, &worker); self.warn_on_sampling_defaults_divergence_for_worker(&worker); // Update type index (clone needed for DashMap key ownership) @@ -1873,6 +1953,135 @@ mod tests { assert_eq!(llama_workers_after[0].url(), "http://worker2:8080"); } + #[test] + fn test_get_candidates_for_model_unions_wildcard_workers() { + let registry = WorkerRegistry::new(); + + // Specific worker for "llama-3". + let specific: Arc = Arc::new( + BasicWorkerBuilder::new("http://specific:8080") + .model(ModelCard::new("llama-3")) + .build(), + ); + // Wildcard worker (no models declared) — indexed under UNKNOWN_MODEL_ID, + // NOT under "llama-3", so get_by_model("llama-3") would miss it. + let wildcard: Arc = + Arc::new(BasicWorkerBuilder::new("http://wildcard:8080").build()); + + registry.register(specific).unwrap(); + let wildcard_id = registry.register(wildcard).unwrap(); + + // get_by_model alone misses the wildcard. + assert_eq!(registry.get_by_model("llama-3").len(), 1); + + // The bounded candidate lookup unions the wildcard in. + let candidates = registry.get_candidates_for_model("llama-3"); + let urls: HashSet = candidates.iter().map(|w| w.url().to_string()).collect(); + assert_eq!(candidates.len(), 2, "specific + wildcard"); + assert!(urls.contains("http://specific:8080")); + assert!(urls.contains("http://wildcard:8080")); + + // An arbitrary, never-registered model still surfaces the wildcard + // (the regression this guards against). + let arbitrary = registry.get_candidates_for_model("some-random-model"); + let arb_urls: HashSet = arbitrary.iter().map(|w| w.url().to_string()).collect(); + assert_eq!(arbitrary.len(), 1); + assert!(arb_urls.contains("http://wildcard:8080")); + + // Every surfaced wildcard actually supports the arbitrary model. + assert!(arbitrary + .iter() + .all(|w| w.supports_model("some-random-model"))); + + // Removing the wildcard drops it from the bounded set. + registry.remove(&wildcard_id); + assert!(registry + .get_candidates_for_model("some-random-model") + .is_empty()); + assert_eq!(registry.get_candidates_for_model("llama-3").len(), 1); + } + + #[test] + fn test_get_candidates_for_model_excludes_other_models() { + let registry = WorkerRegistry::new(); + + let a: Arc = Arc::new( + BasicWorkerBuilder::new("http://a:8080") + .model(ModelCard::new("model-a")) + .build(), + ); + let b: Arc = Arc::new( + BasicWorkerBuilder::new("http://b:8080") + .model(ModelCard::new("model-b")) + .build(), + ); + registry.register(a).unwrap(); + registry.register(b).unwrap(); + + // No wildcards registered: the bounded set is exactly the per-model + // slice and excludes other models' workers. + let candidates = registry.get_candidates_for_model("model-a"); + assert_eq!(candidates.len(), 1); + assert_eq!(candidates[0].url(), "http://a:8080"); + + assert!(registry.get_candidates_for_model("model-c").is_empty()); + } + + #[test] + fn test_get_candidates_for_model_dedups_wildcard_indexed_under_model_key() { + // A wildcard worker carrying a `model_id` label is indexed under that + // label in the model index AND tracked in the wildcard set. The + // union must not return it twice. + let registry = WorkerRegistry::new(); + let mut labels = HashMap::new(); + labels.insert("model_id".to_string(), "labelled".to_string()); + let wildcard: Arc = Arc::new( + BasicWorkerBuilder::new("http://wildcard:8080") + .labels(labels) + .build(), + ); + registry.register(wildcard).unwrap(); + + // Indexed under its label key... + assert_eq!(registry.get_by_model("labelled").len(), 1); + // ...and the union dedups by URL rather than returning it twice. + let candidates = registry.get_candidates_for_model("labelled"); + assert_eq!( + candidates.len(), + 1, + "no duplicate from index + wildcard set" + ); + } + + #[test] + fn test_wildcard_membership_updates_on_replace() { + let registry = WorkerRegistry::new(); + + // Register a wildcard worker. + let wildcard: Arc = Arc::new(BasicWorkerBuilder::new("http://w:8080").build()); + let id = registry.register(wildcard).unwrap(); + assert_eq!(registry.get_candidates_for_model("anything").len(), 1); + + // Replace it with a worker declaring specific models — it must drop + // out of the wildcard set. + let specific: Arc = Arc::new( + BasicWorkerBuilder::new("http://w:8080") + .model(ModelCard::new("specific-model")) + .build(), + ); + assert!(registry.replace(&id, specific)); + assert!( + registry.get_candidates_for_model("anything").is_empty(), + "replacement declares specific models, no longer wildcard" + ); + assert_eq!(registry.get_candidates_for_model("specific-model").len(), 1); + + // Replace back to wildcard — it must re-enter the wildcard set. + let wildcard2: Arc = Arc::new(BasicWorkerBuilder::new("http://w:8080").build()); + assert!(registry.replace(&id, wildcard2)); + assert_eq!(registry.get_candidates_for_model("anything").len(), 1); + } + // Health-checker integration tests moved to worker/manager.rs along with // the loop itself. The registry is now a pure collection — see // `worker::manager::WorkerManager` tests. diff --git a/model_gateway/src/worker/service.rs b/model_gateway/src/worker/service.rs index 24fcd8be31..1925b76791 100644 --- a/model_gateway/src/worker/service.rs +++ b/model_gateway/src/worker/service.rs @@ -17,7 +17,7 @@ use tracing::warn; use crate::{ config::RouterConfig, - worker::{registry::WorkerId, worker::worker_to_info, WorkerRegistry, WorkerType}, + worker::{registry::WorkerId, worker::worker_to_info, Worker, WorkerRegistry, WorkerType}, workflow::{Job, JobQueue}, }; @@ -343,27 +343,53 @@ impl WorkerService { /// /// The counts reflect the returned (filtered) list, not the whole /// registry. + /// + /// The `?model` path uses the registry's bounded, wildcard-safe per-model + /// candidate lookup (per-model index unioned with wildcard workers) + /// instead of scanning every worker, so it stays O(per-model + wildcards). + /// The unfiltered path must serialize all workers, so it remains O(workers) + /// — acceptable for this admin endpoint. pub fn list_workers(&self, model: Option<&str>) -> ListWorkersResult { - let mut worker_infos = Vec::new(); let mut prefill_count = 0; let mut decode_count = 0; let mut regular_count = 0; - for (worker_id, worker) in self.worker_registry.get_all_with_ids() { - if let Some(model) = model { - if !worker.supports_model(model) { - continue; - } - } + let mut count_and_build = |worker_id: String, worker: &Arc| { match worker.worker_type() { WorkerType::Prefill => prefill_count += 1, WorkerType::Decode => decode_count += 1, WorkerType::Regular => regular_count += 1, } - let mut info = worker_to_info(&worker); - info.id = worker_id.as_str().to_string(); - worker_infos.push(info); - } + let mut info = worker_to_info(worker); + info.id = worker_id; + info + }; + + let worker_infos: Vec = match model { + Some(model) => self + .worker_registry + .get_candidates_for_model(model) + .into_iter() + // The bounded set may include wildcard workers that have since + // discovered specific models; re-filter via `supports_model` + // so the listing matches the unfiltered semantics exactly. + .filter(|worker| worker.supports_model(model)) + .map(|worker| { + let id = self + .worker_registry + .get_id_by_url(worker.url()) + .map(|id| id.as_str().to_string()) + .unwrap_or_default(); + count_and_build(id, &worker) + }) + .collect(), + None => self + .worker_registry + .get_all_with_ids() + .into_iter() + .map(|(worker_id, worker)| count_and_build(worker_id.as_str().to_string(), &worker)) + .collect(), + }; ListWorkersResult { total: worker_infos.len(),