Skip to content

fix: make service pool selection and routing reliable #1737

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ctkm-aelf merged 4 commits into from
Oct 4, 2026
Merged

fix: make service pool selection and routing reliable #1737

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a24d72b
fix: group pool connections and clarify routing controls
ctkm-aelf Oct 2, 2026
38d2f23
docs: clarify pool candidate search metadata scope
ctkm-aelf Oct 2, 2026
ae4a695
fix: preserve pool configuration and refresh connection eligibility
ctkm-aelf Oct 4, 2026
6fc7448
fix: clarify pool picker loading and failure states
ctkm-aelf Oct 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
400 changes: 400 additions & 0 deletions backend/src/handlers/service_pool_inspection_search_tests.rs
View file
Open in desktop

Large diffs are not rendered by default.

136 changes: 2 additions & 134 deletions backend/src/handlers/service_pool_inspection_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -632,137 +632,5 @@ fn pool_credential_unavailability_is_narrowly_classified() {
)));
}

#[tokio::test]
async fn pool_inspection_name_search_precedes_paging_and_preserves_scope() {
let mut fixture = fixture("pool_name_search", StatusCode::OK, "priority", false).await;
let db = &fixture.state.db;
let services: Vec<Document> = db
.collection::<Document>("user_services")
.find(doc! {})
.sort(doc! {"_id":1})
.await
.unwrap()
.try_collect()
.await
.unwrap();
let first_id = services[0].get_str("_id").unwrap();
let last = &services[1];
let last_id = last.get_str("_id").unwrap();
let label = "Preferred member [West]";
db.collection::<Document>("user_endpoints")
.update_one(
doc! {"_id":last.get_str("endpoint_id").unwrap()},
doc! {"$set":{"label":label}},
)
.await
.unwrap();
let decrypts = fixture.state.encryption_keys.decrypt_stats();
let page = inspect(&fixture, json!({"check_operation":false,"limit":1}), false).await;
assert_eq!(page.candidates[0].user_service_id, first_id);
for search in ["preferred MEMBER [west]", last.get_str("slug").unwrap()] {
let found = inspect(
&fixture,
json!({"check_operation":false,"search":search,"limit":1}),
false,
)
.await;
assert_eq!(found.candidates.len(), 1);
assert_eq!(found.candidates[0].user_service_id, last_id);
assert_eq!(found.candidates[0].name, label);
assert!(!found.has_more);
}
assert!(
inspect(
&fixture,
json!({"check_operation":false,"search":".*"}),
false
)
.await
.candidates
.is_empty()
);
// Matching foreign-owned instances must never enter the candidate result.
let mut foreign = last.clone();
foreign.insert("_id", Uuid::new_v4().to_string());
foreign.insert("user_id", Uuid::new_v4().to_string());
db.collection::<Document>("user_services")
.insert_one(foreign)
.await
.unwrap();
assert_eq!(
inspect(
&fixture,
json!({"check_operation":false,"search":label}),
false
)
.await
.candidates
.len(),
1
);
fixture.auth.allow_all_services = false;
fixture.auth.allowed_service_ids = vec![first_id.to_owned()];
assert!(
inspect(
&fixture,
json!({"check_operation":false,"search":label}),
false
)
.await
.candidates
.is_empty()
);
let selected = inspect(&fixture, json!({"check_operation":false,"search":"no match","selected_only":true,"peer_ids":format!("{first_id},{last_id}"),"limit":1,"after":"999"}), false).await;
assert_eq!(selected.candidates.len(), 1);
assert_eq!(selected.candidates[0].user_service_id, first_id);
fixture.auth.allow_all_services = true;
// Platform resolution also displays the endpoint label, with no credential materialization.
db.collection::<Document>("downstream_services").update_one(
doc! {"_id":last.get_str("catalog_service_id").unwrap()},
doc! {"$set":{"auth_method":"bearer","service_category":"internal","requires_user_credential":false,"visibility":"public","credential_encrypted":mongodb::bson::Binary{subtype:mongodb::bson::spec::BinarySubtype::Generic,bytes:vec![1]}}},
).await.unwrap();
db.collection::<Document>("user_services")
.update_one(
doc! {"_id":last_id},
doc! {"$set":{"credential_binding":"platform","auth_method":"bearer"}},
)
.await
.unwrap();
let platform = inspect(
&fixture,
json!({"check_operation":false,"search":label}),
false,
)
.await;
assert_eq!(platform.candidates.len(), 1);
assert_eq!(platform.candidates[0].name, label);
assert_eq!(platform.candidates[0].credential_binding, "platform");
assert!(platform.candidates[0].eligible);
// Matching-name pages retain stable pagination, without consuming unrelated rows.
db.collection::<Document>("user_endpoints")
.update_one(
doc! {"_id":services[0].get_str("endpoint_id").unwrap()},
doc! {"$set":{"label":"Another member [West]"}},
)
.await
.unwrap();
let page = inspect(
&fixture,
json!({"check_operation":false,"search":"member [West]","limit":1}),
false,
)
.await;
assert_eq!(page.candidates.len(), 1);
assert!(page.has_more);
let next = inspect(&fixture, json!({"check_operation":false,"search":"member [West]","limit":1,"after":page.next_cursor}), false).await;
assert_eq!(next.candidates.len(), 1);
assert!(!next.has_more);
assert_ne!(
page.candidates[0].user_service_id,
next.candidates[0].user_service_id
);
assert_eq!(decrypts, fixture.state.encryption_keys.decrypt_stats());
assert!(fixture.first.requests.lock().await.is_empty());
assert!(fixture.second.requests.lock().await.is_empty());
db.drop().await.unwrap();
}
#[path = "service_pool_inspection_search_tests.rs"]
mod search;
6 changes: 6 additions & 0 deletions backend/src/handlers/service_pools_handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -599,6 +599,10 @@ pub struct PoolCandidateResponse {
pub credential_binding: String,
pub protocol: Option<crate::models::downstream_service::InferenceWireProtocol>,
pub catalog_service_id: Option<String>,
/// Authoritative original catalog grouping metadata. Custom connections
/// use the stable "Custom connections" group and a null catalog ID.
pub group_name: String,
pub group_slug: Option<String>,
pub requires_compatibility_declaration: bool,
pub cooldown_until: Option<String>,
pub consecutive_failures: i64,
Expand Down Expand Up @@ -722,6 +726,8 @@ async fn inspect_pool_candidates(
credential_binding: row.credential_binding,
protocol: row.protocol,
catalog_service_id: row.catalog_service_id,
group_name: row.group_name,
group_slug: row.group_slug,
requires_compatibility_declaration: row.requires_compatibility_declaration,
cooldown_until: row.cooldown_until.map(|t| t.to_rfc3339()),
consecutive_failures: row.consecutive_failures,
Expand Down
139 changes: 131 additions & 8 deletions backend/src/services/service_pool_inspection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,31 @@ use crate::{
crypto::aes::EncryptionKeys,
errors::{AppError, AppResult},
models::{
downstream_service::COLLECTION_NAME as DOWNSTREAM_SERVICES,
service_pool::{PoolMemberContract, PoolStrategy, ServicePool},
user_service::UserService,
},
};
use futures::TryStreamExt;
use mongodb::bson::doc;
use serde::Deserialize;
use std::collections::HashSet;

#[derive(Debug, Deserialize)]
struct CatalogDisplayMetadata {
#[serde(rename = "_id")]
id: String,
name: String,
slug: String,
}

#[derive(Debug, Deserialize)]
struct ConnectionDisplayMetadata {
#[serde(rename = "_id")]
id: String,
label: String,
}

pub struct CandidateInspection {
pub user_service_id: String,
pub name: String,
Expand All @@ -22,6 +39,8 @@ pub struct CandidateInspection {
pub credential_binding: String,
pub protocol: Option<crate::models::downstream_service::InferenceWireProtocol>,
pub catalog_service_id: Option<String>,
pub group_name: String,
pub group_slug: Option<String>,
pub requires_compatibility_declaration: bool,
pub cooldown_until: Option<chrono::DateTime<chrono::Utc>>,
pub consecutive_failures: i64,
Expand Down Expand Up @@ -122,7 +141,7 @@ pub async fn inspect(
.search
.filter(|search| !query.members_only && !query.selected_only && !search.is_empty());
let mut services: Vec<UserService> = if let Some(search) = search {
search_services(db, filter, search, offset, limit + 1).await?
search_services(db, filter, owner, search, offset, limit + 1).await?
} else {
crate::services::service_history::collection(db, "user_services")
.find(filter)
Expand All @@ -136,6 +155,65 @@ pub async fn inspect(
let has_more = services.len() > limit;
services.truncate(limit);
let next_cursor = has_more.then(|| (offset + limit as u64).to_string());

// Resolve only the bounded page's safe display metadata. This keeps a
// failed credential from hiding its connection label and lets the picker
// group by the authoritative catalog identity without materializing any
// credential or broadening the inventory ACL.
let catalog_ids: Vec<String> = services
.iter()
.filter_map(|service| service.catalog_service_id.clone())
.collect();
let catalog_metadata: std::collections::HashMap<String, (String, String)> =
if catalog_ids.is_empty() {
std::collections::HashMap::new()
} else {
db.collection::<CatalogDisplayMetadata>(DOWNSTREAM_SERVICES)
.find(doc! {"_id": {"$in": &catalog_ids}})
.projection(doc! {"_id": 1, "name": 1, "slug": 1})
.await?
.try_collect::<Vec<_>>()
.await?
.into_iter()
.map(|service| (service.id, (service.name, service.slug)))
.collect()
};
let endpoint_ids: Vec<String> = services
.iter()
.map(|service| service.endpoint_id.clone())
.filter(|id| !id.is_empty())
.collect();
let endpoint_labels: std::collections::HashMap<String, String> = if endpoint_ids.is_empty() {
std::collections::HashMap::new()
} else {
db.collection::<ConnectionDisplayMetadata>("user_endpoints")
.find(doc! {"_id": {"$in": &endpoint_ids}, "user_id": owner})
.projection(doc! {"_id": 1, "label": 1})
.await?
.try_collect::<Vec<_>>()
.await?
.into_iter()
.map(|endpoint| (endpoint.id, endpoint.label))
.collect()
};
let key_ids: Vec<String> = services
.iter()
.filter(|service| super::platform_key_service::binding(service) != "platform")
.filter_map(|service| service.api_key_id.clone())
.collect();
let key_labels: std::collections::HashMap<String, String> = if key_ids.is_empty() {
std::collections::HashMap::new()
} else {
db.collection::<ConnectionDisplayMetadata>("user_api_keys")
.find(doc! {"_id": {"$in": &key_ids}, "user_id": owner})
.projection(doc! {"_id": 1, "label": 1})
.await?
.try_collect::<Vec<_>>()
.await?
.into_iter()
.map(|key| (key.id, key.label))
.collect()
};
let selected = if let Some(peers) = query.peer_ids.filter(|_| !query.members_only) {
let members: Vec<_> = peers
.iter()
Expand Down Expand Up @@ -169,7 +247,15 @@ pub async fn inspect(
});
let mut row = CandidateInspection {
user_service_id: service.id.clone(),
name: service.slug.clone(),
name: service
.api_key_id
.as_ref()
.filter(|_| super::platform_key_service::binding(&service) != "platform")
.and_then(|id| key_labels.get(id))
.or_else(|| endpoint_labels.get(&service.endpoint_id))
.filter(|label| !label.is_empty())
.cloned()
.unwrap_or_else(|| service.slug.clone()),
slug: service.slug.clone(),
is_active: service.is_active,
eligible: true,
Expand All @@ -180,6 +266,17 @@ pub async fn inspect(
.unwrap_or_else(|| "user".into()),
protocol: None,
catalog_service_id: service.catalog_service_id.clone(),
group_name: match service.catalog_service_id.as_ref() {
Some(id) => catalog_metadata
.get(id)
.map(|metadata| metadata.0.clone())
.unwrap_or_else(|| "Unavailable catalog service".into()),
None => "Custom connections".into(),
},
group_slug: service
.catalog_service_id
.as_ref()
.and_then(|id| catalog_metadata.get(id).map(|metadata| metadata.1.clone())),
requires_compatibility_declaration: service.catalog_service_id.is_none(),
cooldown_until: None,
consecutive_failures: 0,
Expand All @@ -200,7 +297,6 @@ pub async fn inspect(
.await;
match resolution {
Ok(Some(resolution)) => {
row.name = resolution.target.service.name.clone();
row.credential_binding = if resolution.master_credential {
"platform"
} else if resolution.target.auth_method == "none" {
Expand Down Expand Up @@ -383,6 +479,8 @@ pub async fn inspect(
credential_binding: "unavailable".into(),
protocol: None,
catalog_service_id: None,
group_name: "Custom connections".into(),
group_slug: None,
requires_compatibility_declaration: false,
cooldown_until: None,
consecutive_failures: 0,
Expand Down Expand Up @@ -438,11 +536,12 @@ pub async fn inspect(
})
}

/// Search before pagination, projecting only the referenced endpoint label.
/// Both personal and platform resolution use this label as the displayed name.
/// Search connection and original catalog labels before pagination.
/// Lookups project display metadata only, after owner and caller scope filtering.
async fn search_services(
db: &mongodb::Database,
filter: mongodb::bson::Document,
owner: &str,
search: &str,
offset: u64,
limit: usize,
Expand All @@ -455,12 +554,36 @@ async fn search_services(
doc! {"$sort":{"_id":1}},
doc! {"$lookup":{
"from":"user_endpoints","localField":"endpoint_id","foreignField":"_id",
"pipeline":[{"$project":{"_id":0,"label":1}}],"as":"search_endpoint",
"pipeline":[
{"$match":{"user_id":owner}},
{"$project":{"_id":0,"label":1}},
],"as":"search_endpoint",
}},
doc! {"$lookup":{
"from":DOWNSTREAM_SERVICES,"localField":"catalog_service_id","foreignField":"_id",
"pipeline":[{"$project":{"_id":0,"name":1,"slug":1}}],"as":"search_catalog",
}},
doc! {"$lookup":{
"from":"user_api_keys","localField":"api_key_id","foreignField":"_id",
"let":{"binding":"$credential_binding"},
"pipeline":[
{"$match":{"user_id":owner,"$expr":{"$ne":[{"$ifNull":["$$binding","user"]},"platform"]}}},
{"$project":{"_id":0,"label":1}},
],"as":"search_key",
}},
doc! {"$match":{"$or":[{"slug":&pattern},{"search_endpoint.label":pattern}]}},
doc! {"$set":{"search_label":{"$ifNull":[
{"$arrayElemAt":["$search_key.label",0]},
{"$arrayElemAt":["$search_endpoint.label",0]},
]}}},
doc! {"$match":{"$or":[
{"slug":&pattern},
{"search_label":&pattern},
{"search_catalog.name":&pattern},
{"search_catalog.slug":&pattern},
]}},
doc! {"$skip":offset as i64},
doc! {"$limit":limit as i64},
doc! {"$unset":"search_endpoint"},
doc! {"$unset":["search_endpoint","search_catalog","search_key","search_label"]},
])
.max_time(std::time::Duration::from_secs(5))
.await?
Expand Down
Loading
Loading