From 39fa9a61194db1df9011e1cde0b3315ec82e67b3 Mon Sep 17 00:00:00 2001 From: Rafa Moreno Date: Mon, 19 Oct 2026 16:38:08 +0200 Subject: [PATCH 1/2] Discover regional clusters and route CLI requests to deployment hosts --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/client.rs | 32 +++-- src/client/discovery.rs | 140 +++++++++++++++++++++ src/client/routing.rs | 48 ++++++-- src/client/routing_tests.rs | 239 +++++++++++++++++++++++++++++++++--- tests/browser_login.rs | 33 ++++- 7 files changed, 450 insertions(+), 46 deletions(-) create mode 100644 src/client/discovery.rs diff --git a/Cargo.lock b/Cargo.lock index e25e84c..2a686c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1502,7 +1502,7 @@ dependencies = [ [[package]] name = "rawtree-cli" -version = "0.7.1" +version = "0.7.2" dependencies = [ "anyhow", "axoupdater", diff --git a/Cargo.toml b/Cargo.toml index c411f5b..6f103af 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rawtree-cli" -version = "0.7.1" +version = "0.7.2" edition = "2021" repository = "https://github.com/rawtreedb/rawtree-cli" diff --git a/src/client.rs b/src/client.rs index 9e29b91..7422419 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,3 +1,4 @@ +mod discovery; mod regions; mod routing; #[cfg(test)] @@ -39,11 +40,18 @@ impl ApiClient { } fn unrouted_request(&self, method: reqwest::Method, path: &str) -> RequestBuilder { - let request = with_client_header( - self.client - .request(method, format!("{}{}", self.base_url, path)), - ); - let request = match &self.deployment { + self.deployment_request(method, path, self.deployment.as_ref()) + } + + fn deployment_request( + &self, + method: reqwest::Method, + path: &str, + deployment: Option<&(String, String)>, + ) -> RequestBuilder { + let url = routing::deployment_url(&format!("{}{}", self.base_url, path), deployment); + let request = with_client_header(self.client.request(method, url)); + let request = match deployment { Some((deployment, region)) => request .header("X-RawTree-Deployment-Id", deployment) .header("X-RawTree-Region", region), @@ -135,9 +143,17 @@ impl ApiClient { } pub fn get(&self, path: &str) -> Result { - let req = self.routed_request(reqwest::Method::GET, path)?; - let resp = req.send().context("failed to connect to server")?; - let value: Value = handle_response(resp)?; + let url = reqwest::Url::parse(&format!("{}{}", self.base_url, path))?; + let organization = url.query_pairs().find(|(key, _)| key == "organization"); + let value: Value = if url.path() == "/v1/clusters" + && self.has_user_credential() + && self.deployment.is_none() + { + self.discover_clusters(organization.as_ref().map(|(_, name)| name.as_ref()), None)? + } else { + let req = self.routed_request(reqwest::Method::GET, path)?; + handle_response(req.send().context("failed to connect to server")?)? + }; self.remember_routing_metadata(path, &value); serde_json::from_value(value).context("failed to parse server response") } diff --git a/src/client/discovery.rs b/src/client/discovery.rs new file mode 100644 index 0000000..3d7b007 --- /dev/null +++ b/src/client/discovery.rs @@ -0,0 +1,140 @@ +use super::{ + handle_response, + regions::Regions, + routing::{before_deadline, valid_routing}, + ApiClient, +}; +use anyhow::{Context, Result}; +use reqwest::Method; +use serde::Deserialize; +use serde_json::{json, Value}; +use std::{collections::HashMap, time::Instant}; + +#[derive(Deserialize)] +struct Organization { + organization_id: String, + name: String, +} + +#[derive(Deserialize)] +struct Organizations { + organizations: Vec, +} + +impl ApiClient { + pub(super) fn has_user_credential(&self) -> bool { + self.token + .as_deref() + .is_some_and(|token| !token.starts_with("rt_")) + } + + /// Discover each enabled deployment before publishing or caching a complete list. + pub(super) fn discover_clusters( + &self, + organization: Option<&str>, + deadline: Option, + ) -> Result { + let organizations: Organizations = handle_response( + before_deadline( + self.unrouted_request(Method::GET, "/v1/organizations"), + deadline, + )? + .send()?, + )?; + let selected: Vec<_> = organizations + .organizations + .into_iter() + .filter(|item| organization.is_none_or(|name| item.name == name)) + .collect(); + anyhow::ensure!( + organization.is_none() || !selected.is_empty(), + "Organization not found." + ); + let mut result = json!({"clusters": []}); + let mut clusters = Vec::new(); + let mut positions = HashMap::new(); + for owner in selected { + let regions: Regions = handle_response( + before_deadline( + self.unrouted_request( + Method::GET, + &format!( + "/v1/organizations/{}/regions", + urlencoding::encode(&owner.organization_id) + ), + ), + deadline, + )? + .send()?, + )?; + let mut deployments = Vec::new(); + for region in regions.regions.iter().filter(|region| region.enabled) { + let pair = region + .deployment_id + .as_ref() + .filter(|id| valid_routing(id, ®ion.cloud_region)) + .map(|id| (id.clone(), region.cloud_region.clone())); + if !deployments.contains(&pair) { + deployments.push(pair); + } + } + let regions: HashMap<_, _> = regions + .regions + .iter() + .map(|region| (region.id.as_str(), region)) + .collect(); + for deployment in deployments { + let response: Value = handle_response( + before_deadline( + self.deployment_request( + Method::GET, + &format!( + "/v1/clusters?organization={}", + urlencoding::encode(&owner.name) + ), + deployment.as_ref(), + ), + deadline, + )? + .send()?, + )?; + let items = response + .get("clusters") + .and_then(Value::as_array) + .context("Invalid cluster catalog response")?; + if organization.is_some() { + result = response.clone(); + } + for item in items { + let mut cluster = item.clone(); + let id = item + .get("id") + .and_then(Value::as_str) + .context("Cluster is missing its ID")?; + let region_id = item + .get("region_id") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()); + if let Some(region_id) = region_id { + let region = regions.get(region_id).with_context(|| { + format!("Region '{region_id}' for cluster '{id}' was not found.") + })?; + cluster["deployment_id"] = json!(region.deployment_id); + cluster["cloud_region"] = json!(region.cloud_region); + } + // A gateway fallback must never replace an authoritative regional record. + match positions.get(id) { + Some(&index) if region_id.is_some() => clusters[index] = cluster, + Some(_) => {} + None => { + positions.insert(id.to_owned(), clusters.len()); + clusters.push(cluster); + } + } + } + } + } + result["clusters"] = json!(clusters); + Ok(result) + } +} diff --git a/src/client/routing.rs b/src/client/routing.rs index 067b8e2..3db375a 100644 --- a/src/client/routing.rs +++ b/src/client/routing.rs @@ -126,12 +126,8 @@ impl ApiClient { self.routing.clusters.borrow_mut().clear(); *self.routing.organizations.borrow_mut() = None; } - let mut request = self.unrouted_request(method, path); - if let Some((deployment, region)) = pair { - request = request - .header("X-RawTree-Deployment-Id", deployment) - .header("X-RawTree-Region", region); - } + let request = + self.deployment_request(method, path, pair.as_ref().or(self.deployment.as_ref())); before_deadline(request, deadline) } @@ -166,10 +162,8 @@ impl ApiClient { let clusters = match cached { Some(clusters) => clusters, None => { - let response: Clusters = handle_response( - before_deadline(self.unrouted_request(Method::GET, &collection), deadline)? - .send()?, - )?; + let response: Clusters = + serde_json::from_value(self.discover_clusters(Some(&organization), deadline)?)?; self.routing .clusters .borrow_mut() @@ -192,7 +186,10 @@ impl ApiClient { } } -fn before_deadline(request: RequestBuilder, deadline: Option) -> Result { +pub(super) fn before_deadline( + request: RequestBuilder, + deadline: Option, +) -> Result { match deadline { Some(deadline) => { let remaining = deadline @@ -251,7 +248,7 @@ pub(super) fn valid_routing(deployment: &str, region: &str) -> bool { let mut parts = region.rsplit('-'); let number = parts.next().unwrap_or_default(); let names: Vec<_> = parts.collect(); - deployment.len() == 8 + matches!(deployment.len(), 8 | 32) && deployment .bytes() .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) @@ -264,3 +261,30 @@ pub(super) fn valid_routing(deployment: &str, region: &str) -> bool { .iter() .all(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_lowercase())) } + +/// Only public RawTree API origins use deployment hosts; custom endpoints keep their URL. +pub(super) fn deployment_url(url: &str, pair: Option<&(String, String)>) -> String { + let Some((deployment, region)) = + pair.filter(|(deployment, region)| valid_routing(deployment, region)) + else { + return url.to_owned(); + }; + let Ok(mut target) = Url::parse(url) else { + return url.to_owned(); + }; + if !matches!( + target.origin().ascii_serialization().as_str(), + "https://api.rawtree.com" | "https://api.rawtree.dev" + ) { + return url.to_owned(); + } + let host = target.host_str().unwrap_or_default().replacen( + "api.", + &format!("api.{deployment}.{region}.aws."), + 1, + ); + if target.set_host(Some(&host)).is_err() { + return url.to_owned(); + } + target.to_string() +} diff --git a/src/client/routing_tests.rs b/src/client/routing_tests.rs index 27dba86..34256da 100644 --- a/src/client/routing_tests.rs +++ b/src/client/routing_tests.rs @@ -67,11 +67,26 @@ fn server(responses: Vec<(&'static str, Value)>) -> (String, thread::JoinHandle< (base, handle) } +fn discovery_catalog() -> Vec<(&'static str, Value)> { + vec![ + ( + "200 OK", + json!({"organizations":[{"organization_id":ORG,"name":"acme"}]}), + ), + ( + "200 OK", + json!({"regions":[{"id":"legacy", "name":"Virginia", "cloud_region":"us-east-1", "deployment_id":null,"enabled":true,"organization_id":null}]}), + ), + ] +} + fn metadata(pair: Value) -> Vec<(&'static str, Value)> { - vec![( + let mut responses = discovery_catalog(); + responses.push(( "200 OK", json!({"clusters": [{"id": "cluster-id", "name": "production", "deployment_id": pair["deployment_id"], "cloud_region": pair["cloud_region"]}]}), - )] + )); + responses } #[test] @@ -93,19 +108,19 @@ fn session_requests_resolve_scope_and_reuse_metadata_for_queries_and_gzip() { .unwrap(); let requests = server.join().unwrap(); assert_eq!( - requests[0].path, + requests[2].path, "GET /v1/clusters?organization=acme HTTP/1.1" ); - assert!(!requests[0].headers.contains_key("x-rawtree-region")); - for request in &requests[1..] { + assert!(!requests[2].headers.contains_key("x-rawtree-region")); + for request in &requests[3..] { assert_eq!(request.headers["x-rawtree-deployment-id"], "12abcdef"); assert_eq!(request.headers["x-rawtree-region"], "eu-west-1"); assert_eq!(request.headers["authorization"], "Bearer session.fixture"); assert_eq!(request.headers["x-rawtree-client"], "cli"); } - assert_eq!(requests[2].headers["content-encoding"], "gzip"); + assert_eq!(requests[4].headers["content-encoding"], "gzip"); let mut decoded = String::new(); - flate2::read::GzDecoder::new(requests[2].body.as_slice()) + flate2::read::GzDecoder::new(requests[4].body.as_slice()) .read_to_string(&mut decoded) .unwrap(); assert_eq!(decoded, "{\"value\":1}"); @@ -127,8 +142,8 @@ fn absent_partial_and_invalid_metadata_keep_default_routing() { .get("/v1/databases?organization=acme&cluster=production") .unwrap(); let requests = server.join().unwrap(); - assert!(!requests[1].headers.contains_key("x-rawtree-region")); - assert!(!requests[1].headers.contains_key("x-rawtree-deployment-id")); + assert!(!requests[3].headers.contains_key("x-rawtree-region")); + assert!(!requests[3].headers.contains_key("x-rawtree-deployment-id")); } } @@ -159,15 +174,16 @@ fn cluster_id_actions_preserve_organization_query_and_serving_headers() { .unwrap(); let requests = server.join().unwrap(); assert_eq!( - requests[1].path, + requests[3].path, "POST /v1/clusters/cluster-id/stop?organization=acme HTTP/1.1" ); - assert_eq!(requests[1].headers["x-rawtree-deployment-id"], "12abcdef"); + assert_eq!(requests[3].headers["x-rawtree-deployment-id"], "12abcdef"); } #[test] fn discovery_failure_does_not_execute_the_operation() { let mut responses = metadata(json!({})); + responses.truncate(1); responses[0] = ( "503 Service Unavailable", json!({"message":"discovery unavailable"}), @@ -182,7 +198,8 @@ fn discovery_failure_does_not_execute_the_operation() { #[test] fn cluster_creation_invalidates_previously_discovered_empty_collection() { - let responses = vec![ + let mut responses = discovery_catalog(); + responses.extend([ ("200 OK", json!({"clusters": []})), ("200 OK", json!({"id": "new-id"})), ( @@ -190,7 +207,8 @@ fn cluster_creation_invalidates_previously_discovered_empty_collection() { json!({"clusters": [{"id":"new-id", "name":"new", "deployment_id":"12abcdef", "cloud_region":"eu-west-1"}]}), ), ("200 OK", json!({"data":[]})), - ]; + ]); + responses.splice(4..4, discovery_catalog()); let (base, server) = server(responses); let client = ApiClient::new(base, Some("session.fixture".to_string())); let collection = "/v1/clusters?organization=acme"; @@ -200,8 +218,8 @@ fn cluster_creation_invalidates_previously_discovered_empty_collection() { .get("/v1/databases?organization=acme&cluster_id=new-id") .unwrap(); let requests = server.join().unwrap(); - assert_eq!(requests[2].path, format!("GET {collection} HTTP/1.1")); - assert_eq!(requests[3].headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(requests[6].path, format!("GET {collection} HTTP/1.1")); + assert_eq!(requests[7].headers["x-rawtree-region"], "eu-west-1"); } #[test] @@ -216,7 +234,7 @@ fn explicit_timeout_includes_routing_discovery() { let mut reader = BufReader::new(socket); let mut line = String::new(); reader.read_line(&mut line).unwrap(); - assert_eq!(line.trim(), "GET /v1/clusters?organization=acme HTTP/1.1"); + assert_eq!(line.trim(), "GET /v1/organizations HTTP/1.1"); thread::sleep(Duration::from_millis(400)); }); let client = ApiClient::new(base, Some("session.fixture".to_string())); @@ -233,7 +251,8 @@ fn explicit_timeout_includes_routing_discovery() { #[test] fn explicit_id_and_name_selectors_cannot_collide() { - let (base, server) = server(vec![ + let mut responses = discovery_catalog(); + responses.extend([ ( "200 OK", json!({"clusters":[ @@ -249,6 +268,7 @@ fn explicit_id_and_name_selectors_cannot_collide() { ("200 OK", json!({})), ("200 OK", json!({})), ]); + let (base, server) = server(responses); let client = ApiClient::new(base, Some("session.fixture".to_string())); let _: Value = client .get("/v1/clusters/target-id?organization=acme") @@ -264,9 +284,9 @@ fn explicit_id_and_name_selectors_cannot_collide() { )) .unwrap(); let requests = server.join().unwrap(); - assert_eq!(requests[1].headers["x-rawtree-region"], "eu-west-1"); assert_eq!(requests[3].headers["x-rawtree-region"], "eu-west-1"); - assert_eq!(requests[4].headers["x-rawtree-region"], "ap-south-1"); + assert_eq!(requests[5].headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(requests[6].headers["x-rawtree-region"], "ap-south-1"); } fn region_catalog(overrides: Value) -> Vec<(&'static str, Value)> { @@ -302,7 +322,7 @@ fn creation_and_sizes_use_selected_region_without_mutating_default_client() { &json!({"region_id":"region-id"}), ) .unwrap(); - let _: Value = client.get("/v1/clusters?organization=acme").unwrap(); + let _: Value = client.get("/v1/clusters/sizes").unwrap(); let requests = server.join().unwrap(); assert_eq!( requests[1].path, @@ -361,3 +381,182 @@ fn api_keys_cannot_override_their_embedded_deployment() { .to_string() .contains("user credential")); } + +fn multi_region_catalog(deployment: &str) -> Vec<(&'static str, Value)> { + vec![ + ( + "200 OK", + json!({"organizations":[{"organization_id":ORG,"name":"acme"}]}), + ), + ( + "200 OK", + json!({"regions":[ + {"id":"legacy","name":"Virginia","cloud_region":"us-east-1","deployment_id":null,"organization_id":null,"enabled":true}, + {"id":"eu","name":"Ireland","cloud_region":"eu-west-1","deployment_id":deployment,"organization_id":null,"enabled":true}, + {"id":"duplicate","name":"Same deployment","cloud_region":"eu-west-1","deployment_id":deployment,"organization_id":null,"enabled":true}, + {"id":"pending","name":"Pending","cloud_region":"bad.region","deployment_id":"bad","organization_id":null,"enabled":true}, + {"id":"disabled","name":"Disabled","cloud_region":"ap-south-1","deployment_id":"abcd1234","organization_id":null,"enabled":false} + ]}), + ), + ( + "200 OK", + json!({"organization":{"name":"acme"},"clusters":[{"id":"us","name":"america","region_id":"legacy"}]}), + ), + ( + "200 OK", + json!({"organization":{"name":"acme"},"clusters":[ + {"id":"eu","name":"europe","region_id":"eu","deployment_id":"wrong","cloud_region":"us-east-1","s3_storage":{"data":{"bucket":"data"}}}, + {"id":"eu","name":"europe","region_id":"eu","s3_storage":{"data":{"bucket":"data"}}} + ]}), + ), + ] +} + +#[test] +fn discovery_merges_regions_and_uses_authoritative_metadata_for_queries() { + for deployment in ["1234abcd", "0123456789abcdef0123456789abcdef"] { + let mut responses = multi_region_catalog(deployment); + responses.push(("200 OK", json!({"data":[]}))); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".into())); + let result: Value = client.get("/v1/clusters?organization=acme").unwrap(); + assert_eq!(result["clusters"].as_array().unwrap().len(), 2); + assert_eq!(result["organization"]["name"], "acme"); + assert_eq!(result["clusters"][1]["deployment_id"], deployment); + assert_eq!(result["clusters"][1]["cloud_region"], "eu-west-1"); + assert_eq!( + result["clusters"][1]["s3_storage"]["data"]["bucket"], + "data" + ); + let _: Value = client + .get("/v1/databases?organization=acme&cluster=europe") + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests.len(), 5); + assert!(!requests[2].headers.contains_key("x-rawtree-region")); + for index in [3, 4] { + assert_eq!( + requests[index].headers["x-rawtree-deployment-id"], + deployment + ); + assert_eq!(requests[index].headers["x-rawtree-region"], "eu-west-1"); + assert_eq!( + requests[index].headers["authorization"], + "Bearer session.fixture" + ); + } + } +} + +#[test] +fn discovery_runs_before_a_direct_cluster_request_and_retries_after_regional_failure() { + let mut responses = multi_region_catalog("1234abcd"); + responses[3] = ( + "503 Service Unavailable", + json!({"message":"Europe unavailable"}), + ); + responses.extend(multi_region_catalog("1234abcd")); + responses.push(("200 OK", json!({"data":[]}))); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".into())); + assert!(client + .get::("/v1/clusters?organization=acme") + .unwrap_err() + .to_string() + .contains("Europe unavailable")); + let _: Value = client + .get("/v1/databases?organization=acme&cluster=europe") + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests.len(), 9); + assert_eq!(requests[8].headers["x-rawtree-region"], "eu-west-1"); +} + +#[test] +fn discovery_rejects_unknown_region_references() { + let mut responses = multi_region_catalog("1234abcd"); + responses[3].1["clusters"][0]["region_id"] = json!("missing"); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".into())); + assert!(client + .get::("/v1/databases?organization=acme&cluster=europe") + .unwrap_err() + .to_string() + .contains("Region 'missing'")); + assert_eq!(server.join().unwrap().len(), 4); +} + +#[test] +fn public_origins_use_deployment_hosts_and_keep_paths_queries_and_headers() { + for deployment in ["1234abcd", "0123456789abcdef0123456789abcdef"] { + for env in ["com", "dev"] { + let client = ApiClient::new( + format!("https://api.rawtree.{env}"), + Some("session.fixture".into()), + ); + client.remember_routing_metadata("/v1/clusters?organization=acme", &json!({"clusters":[{ + "id":"eu", "name":"europe", "deployment_id":deployment,"cloud_region":"eu-west-1" + }]})); + let request = client + .routed_request( + reqwest::Method::GET, + "/v1/databases?organization=acme&cluster=europe", + ) + .unwrap() + .build() + .unwrap(); + assert_eq!(request.url().as_str(), format!("https://api.{deployment}.eu-west-1.aws.rawtree.{env}/v1/databases?organization=acme&cluster=europe")); + assert_eq!(request.headers()["x-rawtree-deployment-id"], deployment); + } + } + for url in [ + "http://localhost:8080/v1/query", + "https://custom.example.com/v1/query", + "https://api.rawtree.com:8443/v1/query", + ] { + assert_eq!( + super::routing::deployment_url(url, Some(&("1234abcd".into(), "eu-west-1".into()))), + url + ); + } +} + +#[test] +fn complete_region_metadata_wins_over_gateway_duplicates_in_either_order() { + for gateway_first in [true, false] { + let mut responses = multi_region_catalog("1234abcd"); + responses[2].1["clusters"] + .as_array_mut() + .unwrap() + .push(json!({"id":"eu","name":"europe"})); + if !gateway_first { + responses[1].1["regions"].as_array_mut().unwrap().swap(0, 1); + responses.swap(2, 3); + } + responses.push(("200 OK", json!({}))); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".into())); + let _: Value = client + .get("/v1/databases?organization=acme&cluster=europe") + .unwrap(); + assert_eq!( + server.join().unwrap().last().unwrap().headers["x-rawtree-region"], + "eu-west-1" + ); + } +} + +#[test] +fn api_keys_preserve_the_public_gateway_url() { + let client = ApiClient::new("https://api.rawtree.com".into(), Some("rt_fixture".into())); + let request = client + .routed_request( + reqwest::Method::GET, + "/v1/databases?organization=acme&cluster=europe", + ) + .unwrap() + .build() + .unwrap(); + assert_eq!(request.url().host_str(), Some("api.rawtree.com")); + assert!(!request.headers().contains_key("x-rawtree-region")); +} diff --git a/tests/browser_login.rs b/tests/browser_login.rs index 7696ac8..7a43cff 100644 --- a/tests/browser_login.rs +++ b/tests/browser_login.rs @@ -45,7 +45,15 @@ fn organizations(names: &[&str]) -> Response { ( "GET /v1/organizations".into(), "200 OK", - json!({"organizations": names.iter().map(|name| json!({"name": name, "role": "owner"})).collect::>()}), + json!({"organizations": names.iter().map(|name| json!({"name": name, "organization_id": format!("{name}-id"), "role": "owner"})).collect::>()}), + ) +} + +fn regions() -> Response { + ( + "GET /v1/organizations/team-id/regions".into(), + "200 OK", + json!({"regions":[{"id":"legacy", "name":"Virginia", "cloud_region":"us-east-1", "deployment_id":null, "organization_id":null, "enabled":true}]}), ) } @@ -122,7 +130,12 @@ fn missing_organization_returns_available_choices_without_saving_authentication( fn missing_cluster_returns_choices_in_the_selected_organization() { for names in [vec![], vec!["production", "staging"]] { let mut responses = approved_login(); - responses.extend([organizations(&["team", "other"]), clusters(&names)]); + responses.extend([ + organizations(&["team", "other"]), + organizations(&["team"]), + regions(), + clusters(&names), + ]); assert_selection( &responses, &["login", "--org", "team"], @@ -137,6 +150,8 @@ fn missing_database_returns_choices_in_the_selected_cluster() { let mut responses = approved_login(); responses.extend([ organizations(&["team"]), + organizations(&["team"]), + regions(), clusters(&["staging", "production"]), databases(&names), ]); @@ -161,6 +176,8 @@ fn approval_is_reported_before_polling_completes_and_only_once() { ); responses.extend([ organizations(&["team"]), + organizations(&["team"]), + regions(), clusters(&["production"]), databases(&["analytics"]), ]); @@ -203,6 +220,8 @@ fn explicit_json_and_selectors_complete_login_with_no_browser() { let mut responses = approved_login(); responses.extend([ organizations(&["other", "team"]), + organizations(&["team"]), + regions(), clusters(&["staging", "production"]), databases(&["billing", "analytics"]), ]); @@ -239,7 +258,11 @@ fn invalid_selectors_return_json_errors_and_preserve_config() { let mut responses = approved_login(); responses.push(organizations(&["team"])); if flag != "--org" { - responses.push(clusters(&["production"])); + responses.extend([ + organizations(&["team"]), + regions(), + clusters(&["production"]), + ]); } if flag == "--database" { responses.push(databases(&["analytics"])); @@ -259,10 +282,12 @@ fn invalid_selectors_return_json_errors_and_preserve_config() { #[test] fn discovery_errors_are_not_treated_as_empty_choices_or_success() { - for stage in 0..3 { + for stage in 0..5 { let mut responses = approved_login(); let discovery = [ organizations(&["team"]), + organizations(&["team"]), + regions(), clusters(&["production"]), databases(&["analytics"]), ]; From d86eb220d3b473b9fcf43872ae8fd6fac266da56 Mon Sep 17 00:00:00 2001 From: Rafa Moreno Date: Tue, 20 Oct 2026 09:30:20 +0200 Subject: [PATCH 2/2] Preserve owning deployment cluster metadata and bump to 0.7.4 --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/client/discovery.rs | 31 ++++++++++++++++--- src/client/routing_tests.rs | 61 +++++++++++++++++++++++++------------ tests/storage_contracts.rs | 12 ++++++++ 5 files changed, 82 insertions(+), 26 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e6365e1..39f509e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1502,7 +1502,7 @@ dependencies = [ [[package]] name = "rawtree-cli" -version = "0.7.3" +version = "0.7.4" dependencies = [ "anyhow", "axoupdater", diff --git a/Cargo.toml b/Cargo.toml index 8600cdd..2db37ca 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rawtree-cli" -version = "0.7.3" +version = "0.7.4" edition = "2021" repository = "https://github.com/rawtreedb/rawtree-cli" diff --git a/src/client/discovery.rs b/src/client/discovery.rs index 3d7b007..2bcde8f 100644 --- a/src/client/discovery.rs +++ b/src/client/discovery.rs @@ -10,6 +10,14 @@ use serde::Deserialize; use serde_json::{json, Value}; use std::{collections::HashMap, time::Instant}; +// Metadata identifies placement; only the owning catalog supplies authoritative fields. +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +enum CatalogAuthority { + Fallback, + RegionMetadata, + OwningDeployment, +} + #[derive(Deserialize)] struct Organization { organization_id: String, @@ -115,19 +123,34 @@ impl ApiClient { .get("region_id") .and_then(Value::as_str) .filter(|id| !id.is_empty()); + let mut authority = CatalogAuthority::Fallback; if let Some(region_id) = region_id { let region = regions.get(region_id).with_context(|| { format!("Region '{region_id}' for cluster '{id}' was not found.") })?; + let owns_cluster = match deployment.as_ref() { + Some((id, cloud_region)) => { + region.deployment_id.as_ref() == Some(id) + && region.cloud_region == *cloud_region + } + None => region.deployment_id.is_none(), + }; + authority = if owns_cluster { + CatalogAuthority::OwningDeployment + } else { + CatalogAuthority::RegionMetadata + }; cluster["deployment_id"] = json!(region.deployment_id); cluster["cloud_region"] = json!(region.cloud_region); } - // A gateway fallback must never replace an authoritative regional record. - match positions.get(id) { - Some(&index) if region_id.is_some() => clusters[index] = cluster, + match positions.get_mut(id) { + Some((index, previous)) if authority > *previous => { + clusters[*index] = cluster; + *previous = authority; + } Some(_) => {} None => { - positions.insert(id.to_owned(), clusters.len()); + positions.insert(id.to_owned(), (clusters.len(), authority)); clusters.push(cluster); } } diff --git a/src/client/routing_tests.rs b/src/client/routing_tests.rs index 34256da..e4751a3 100644 --- a/src/client/routing_tests.rs +++ b/src/client/routing_tests.rs @@ -522,27 +522,48 @@ fn public_origins_use_deployment_hosts_and_keep_paths_queries_and_headers() { } #[test] -fn complete_region_metadata_wins_over_gateway_duplicates_in_either_order() { - for gateway_first in [true, false] { - let mut responses = multi_region_catalog("1234abcd"); - responses[2].1["clusters"] - .as_array_mut() - .unwrap() - .push(json!({"id":"eu","name":"europe"})); - if !gateway_first { - responses[1].1["regions"].as_array_mut().unwrap().swap(0, 1); - responses.swap(2, 3); +fn owning_catalog_wins_over_duplicates_in_either_order() { + for source in ["gateway", "same-region-deployment"] { + for region_id in [Value::Null, json!("eu")] { + for owner_first in [true, false] { + let mut responses = multi_region_catalog("1234abcd"); + if source == "same-region-deployment" { + responses[1].1["regions"][3]["enabled"] = json!(false); + responses[1].1["regions"][0] = json!({ + "id":"other", "name":"Other Ireland", "cloud_region":"eu-west-1", + "deployment_id":"abcd1234", "organization_id":ORG, "enabled":true + }); + } + responses[2].1["clusters"] = json!([{ + "id":"eu", "name":"europe", "region_id":region_id, + "deployment_id":"stale", "cloud_region":"us-east-1", "status":"stale" + }]); + for cluster in responses[3].1["clusters"].as_array_mut().unwrap() { + cluster["status"] = json!("running"); + } + if owner_first { + responses[1].1["regions"].as_array_mut().unwrap().swap(0, 1); + responses.swap(2, 3); + } + responses.push(("200 OK", json!({}))); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".into())); + let catalog: Value = client.get("/v1/clusters?organization=acme").unwrap(); + assert_eq!(catalog["clusters"].as_array().unwrap().len(), 1); + assert_eq!( + catalog["clusters"][0]["s3_storage"], + json!({"data":{"bucket":"data"}}) + ); + assert_eq!(catalog["clusters"][0]["status"], "running"); + let _: Value = client + .get("/v1/databases?organization=acme&cluster=europe") + .unwrap(); + let requests = server.join().unwrap(); + let headers = &requests.last().unwrap().headers; + assert_eq!(headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(headers["x-rawtree-deployment-id"], "1234abcd"); + } } - responses.push(("200 OK", json!({}))); - let (base, server) = server(responses); - let client = ApiClient::new(base, Some("session.fixture".into())); - let _: Value = client - .get("/v1/databases?organization=acme&cluster=europe") - .unwrap(); - assert_eq!( - server.join().unwrap().last().unwrap().headers["x-rawtree-region"], - "eu-west-1" - ); } } diff --git a/tests/storage_contracts.rs b/tests/storage_contracts.rs index 295e374..2177feb 100644 --- a/tests/storage_contracts.rs +++ b/tests/storage_contracts.rs @@ -61,6 +61,16 @@ fn database_storage_create_uses_current_response_and_persists_only_on_success() ]); let (output, saved, bodies) = common::run_cli( &[ + ( + "GET /v1/organizations".into(), + "200 OK", + json!({"organizations":[{"organization_id":"team-id","name":"team alpha"}]}), + ), + ( + "GET /v1/organizations/team-id/regions".into(), + "200 OK", + json!({"regions":[{"id":"legacy","name":"Virginia","cloud_region":"us-east-1","deployment_id":null,"organization_id":null,"enabled":true}]}), + ), ( "GET /v1/clusters?organization=team%20alpha".into(), "200 OK", @@ -87,6 +97,8 @@ fn database_storage_create_uses_current_response_and_persists_only_on_success() assert_eq!( bodies, vec![ + Value::Null, + Value::Null, Value::Null, json!({"name": "analytics", "s3_storage": expected}) ]