From 7397d16b47229e5e0a68eb0a6bddeaecc3f08c79 Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Sun, 1 Nov 2026 00:26:35 -0500 Subject: [PATCH 1/5] fix: don't forward session cookies on intra-cluster syncs Without P_CLUSTER_SECRET, create_intracluster_auth_headermap forwarded the caller's browser cookie to peers when the request had no Authorization header. Sessions exist only on the node that issued them, so ingestors rejected every user and role sync made from the UI ("Your session has expired..."), and the querier only logged it. Users created in the UI could not ingest, roles created in the UI were missing, and roles removed in the UI stayed in effect on ingestors until restart. Fall back to the node token instead. The querier has already authorized the caller, and the ingestor sync handlers reconcile from storage, so authenticating as the node grants nothing beyond what storage says. The Prism field-stats path called the helper with an empty node token; it now forwards only the caller's Authorization header and otherwise lets send_query_request use the querier's token. Co-Authored-By: Claude Opus 5.5 --- src/storage/field_stats.rs | 26 +++++------- src/utils/mod.rs | 87 ++++++++++++++++++++++++++++++++++---- 2 files changed, 88 insertions(+), 25 deletions(-) diff --git a/src/storage/field_stats.rs b/src/storage/field_stats.rs index 3f81d2dbe..a4b6ba85e 100644 --- a/src/storage/field_stats.rs +++ b/src/storage/field_stats.rs @@ -36,7 +36,6 @@ use crate::parseable::PARSEABLE; use crate::storage::ObjectStorageError; use crate::storage::StreamType; use crate::tenants::TENANT_METADATA; -use crate::utils::create_intracluster_auth_headermap; use crate::utils::get_tenant_id_from_request; use crate::utils::get_user_from_request; use crate::utils::json::apply_generic_flattening_for_partition; @@ -916,21 +915,16 @@ pub async fn get_dataset_stats( ); Some(map) } else { - let auth = create_intracluster_auth_headermap( - req.headers(), - "", - &get_user_from_request(&req).unwrap(), - ); - let mut map = HeaderMap::new(); - - for (key, value) in auth.iter() { - if let Ok(name) = HeaderName::from_bytes(key.as_str().as_bytes()) - && let Ok(val) = HeaderValue::from_bytes(value.as_bytes()) - { - map.insert(name, val); - } - } - Some(map) + // Forward the caller's credentials if they sent any. A session + // cookie is only valid on this node, so without credentials + // `send_query_request` falls back to the querier's own token. + req.headers() + .get(actix_web::http::header::AUTHORIZATION) + .map(|auth| { + let mut map = HeaderMap::new(); + map.insert(actix_web::http::header::AUTHORIZATION, auth.clone()); + map + }) }; let response = match send_query_request(auth, &query_request, &tenant_id).await { diff --git a/src/utils/mod.rs b/src/utils/mod.rs index bbf2de6e2..18e9c17d5 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -253,16 +253,37 @@ pub fn is_admin_for_session(session_key: &SessionKey) -> bool { false } +/// Auth headers for a request this node sends to a peer on behalf of `req`. +/// +/// With a cluster secret the peer authenticates the caller by `userid`. +/// Without one, the caller's `Authorization` header is forwarded if present; +/// otherwise the request authenticates as this node with `node_token`. A +/// session cookie is never forwarded, since sessions live only on the node +/// that issued them and the peer would reject it. pub fn create_intracluster_auth_headermap( req: &HeaderMap, node_token: &str, userid: &str, +) -> reqwest::header::HeaderMap { + intracluster_auth_headermap( + req, + node_token, + userid, + CLUSTER_SECRET.get().map(|(_, hash)| hash.as_str()), + ) +} + +fn intracluster_auth_headermap( + req: &HeaderMap, + node_token: &str, + userid: &str, + cluster_secret_hash: Option<&str>, ) -> reqwest::header::HeaderMap { let mut map = reqwest::header::HeaderMap::new(); let tenant_id = req .get(TENANT_ID) .map(|tenant_value| tenant_value.to_str().unwrap().to_owned()); - if let Some((_, hash)) = CLUSTER_SECRET.get() { + if let Some(hash) = cluster_secret_hash { // also insert server secret along with other required headers for query forwarding map.insert( reqwest::header::HeaderName::from_static(CLUSTER_SECRET_HEADER), @@ -282,14 +303,6 @@ pub fn create_intracluster_auth_headermap( reqwest::header::AUTHORIZATION, reqwest::header::HeaderValue::from_bytes(auth.as_bytes()).unwrap(), ); - } else if req.contains_key(actix_web::http::header::COOKIE) { - // multiple cookies - for cookie in req.get_all(actix_web::http::header::COOKIE) { - map.insert( - reqwest::header::COOKIE, - reqwest::header::HeaderValue::from_bytes(cookie.as_bytes()).unwrap(), - ); - } } else { map.insert( reqwest::header::AUTHORIZATION, @@ -298,3 +311,59 @@ pub fn create_intracluster_auth_headermap( } map } + +#[cfg(test)] +mod tests { + use actix_web::http::header::{AUTHORIZATION, COOKIE, HeaderValue}; + + use super::*; + + const NODE_TOKEN: &str = "Basic bm9kZTpzZWNyZXQ="; + + fn headers(pairs: &[(actix_web::http::header::HeaderName, &'static str)]) -> HeaderMap { + let mut map = HeaderMap::new(); + for (name, value) in pairs { + map.append(name.clone(), HeaderValue::from_static(value)); + } + map + } + + #[test] + fn session_cookie_is_replaced_by_node_token() { + let req = headers(&[(COOKIE, "session=01M6ARA6TCK200TXKYC6CWRN5D; username=admin")]); + let map = intracluster_auth_headermap(&req, NODE_TOKEN, "admin", None); + assert!(map.get(reqwest::header::COOKIE).is_none()); + assert_eq!(map.get(reqwest::header::AUTHORIZATION).unwrap(), NODE_TOKEN); + } + + #[test] + fn caller_authorization_is_forwarded() { + let req = headers(&[ + (AUTHORIZATION, "Basic dXNlcjpwYXNz"), + (COOKIE, "session=01M6ARA6TCK200TXKYC6CWRN5D"), + ]); + let map = intracluster_auth_headermap(&req, NODE_TOKEN, "user", None); + assert!(map.get(reqwest::header::COOKIE).is_none()); + assert_eq!( + map.get(reqwest::header::AUTHORIZATION).unwrap(), + "Basic dXNlcjpwYXNz" + ); + } + + #[test] + fn no_credentials_use_node_token() { + let map = intracluster_auth_headermap(&HeaderMap::new(), NODE_TOKEN, "admin", None); + assert_eq!(map.get(reqwest::header::AUTHORIZATION).unwrap(), NODE_TOKEN); + } + + #[test] + fn cluster_secret_identifies_caller_without_credentials() { + let req = headers(&[(COOKIE, "session=01M6ARA6TCK200TXKYC6CWRN5D")]); + let map = intracluster_auth_headermap(&req, NODE_TOKEN, "admin", Some("hash")); + assert_eq!(map.get(CLUSTER_SECRET_HEADER).unwrap(), "hash"); + assert_eq!(map.get("intra-cluster-userid").unwrap(), "admin"); + assert_eq!(map.get("intra-cluster-tenant").unwrap(), DEFAULT_TENANT); + assert!(map.get(reqwest::header::COOKIE).is_none()); + assert!(map.get(reqwest::header::AUTHORIZATION).is_none()); + } +} From 021a107e8e66cf15b4da130c05bb43a6f9225029 Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Sun, 1 Nov 2026 00:39:16 -0500 Subject: [PATCH 2/5] fix: drop session cookies from stream sync; fail closed in Prism stats sync_streams_with_ingestors copied every caller header, including the session cookie, so a stream created or updated from the UI was rejected by ingestors ("session ... no longer valid"). Drop the cookie and, when no other credential (Authorization, X-API-KEY, cluster secret) is left, authenticate with the ingestor's node token. The Prism dataset-stats path no longer passes no auth to send_query_request when there is no cluster secret and no Authorization header. That fell back to the querier's root token, which skips the per-dataset checks the querier runs for the caller. It now returns an error asking for P_CLUSTER_SECRET instead. Co-Authored-By: Claude Opus 5.5 --- src/handlers/http/cluster/mod.rs | 85 +++++++++++++++++++++++++++----- src/storage/field_stats.rs | 60 ++++++++++++++++++---- 2 files changed, 124 insertions(+), 21 deletions(-) diff --git a/src/handlers/http/cluster/mod.rs b/src/handlers/http/cluster/mod.rs index 1130557d5..1de139b2f 100644 --- a/src/handlers/http/cluster/mod.rs +++ b/src/handlers/http/cluster/mod.rs @@ -43,6 +43,7 @@ use url::Url; use utils::{IngestionStats, QueriedStats, StorageStats, check_liveness, to_url_string}; use crate::INTRA_CLUSTER_CLIENT; +use crate::handlers::http::middleware::CLUSTER_SECRET_HEADER; use crate::handlers::http::modal::ingest::SyncRole; use crate::handlers::http::query::{Query, QueryError, TIME_ELAPSED_HEADER}; use crate::metrics::prom_utils::Metrics; @@ -426,23 +427,41 @@ pub fn fallback_intracluster_userid( generated_text_reqwest_header_value(username, "configured username").map(Some) } -// forward the create/update stream request to all ingestors to keep them in sync -pub async fn sync_streams_with_ingestors( - headers: HeaderMap, - body: Bytes, - stream_name: &str, - tenant_id: &Option, -) -> Result<(), StreamError> { - let mut reqwest_headers = reqwest::header::HeaderMap::new(); - +/// Copies the caller's headers for a stream sync, minus the session cookie: +/// sessions live only on the node that issued them, so a peer would reject +/// it. The flag is true when the copy carries no other credential, in which +/// case the sync authenticates with the target node's token. +fn stream_sync_headers(headers: &HeaderMap) -> (reqwest::header::HeaderMap, bool) { + let mut forwarded = reqwest::header::HeaderMap::new(); for (key, value) in headers.iter() { + if key == actix_web::http::header::COOKIE { + continue; + } // Convert actix header name/value to reqwest header name/value if let Ok(name) = reqwest::header::HeaderName::from_bytes(key.as_str().as_bytes()) && let Ok(val) = reqwest::header::HeaderValue::from_bytes(value.as_bytes()) { - reqwest_headers.insert(name, val); + forwarded.insert(name, val); } } + let has_credentials = [ + CLUSTER_SECRET_HEADER, + reqwest::header::AUTHORIZATION.as_str(), + "x-api-key", + ] + .into_iter() + .any(|name| forwarded.contains_key(name)); + (forwarded, !has_credentials) +} + +// forward the create/update stream request to all ingestors to keep them in sync +pub async fn sync_streams_with_ingestors( + headers: HeaderMap, + body: Bytes, + stream_name: &str, + tenant_id: &Option, +) -> Result<(), StreamError> { + let (mut reqwest_headers, needs_node_token) = stream_sync_headers(&headers); if let Some(userid) = fallback_intracluster_userid( reqwest_headers.contains_key(INTRA_CLUSTER_USERID), @@ -466,9 +485,15 @@ pub async fn sync_streams_with_ingestors( base_path_without_preceding_slash(), stream_name ); - let headers = reqwest_headers_clone.clone(); + let mut headers = reqwest_headers_clone.clone(); let body = body_clone.clone(); async move { + if needs_node_token { + headers.insert( + reqwest::header::AUTHORIZATION, + generated_text_reqwest_header_value(&ingestor.token, "node token")?, + ); + } let res = INTRA_CLUSTER_CLIENT .put(url) .headers(headers) @@ -2092,6 +2117,44 @@ pub async fn send_query_request( #[cfg(test)] mod header_build_tests { use super::*; + use actix_web::http::header::{HeaderName, HeaderValue}; + + fn actix_headers(pairs: &[(&'static str, &'static str)]) -> HeaderMap { + let mut map = HeaderMap::new(); + for (name, value) in pairs { + map.append( + HeaderName::from_static(name), + HeaderValue::from_static(value), + ); + } + map + } + + #[test] + fn stream_sync_drops_session_cookie_and_needs_node_token() { + let (forwarded, needs_node_token) = stream_sync_headers(&actix_headers(&[ + ("cookie", "session=abc"), + ("x-p-log-source", "json"), + ])); + assert!(forwarded.get(reqwest::header::COOKIE).is_none()); + assert_eq!(forwarded.get("x-p-log-source").unwrap(), "json"); + assert!(needs_node_token); + } + + #[test] + fn stream_sync_keeps_caller_credentials() { + for credential in [ + ("authorization", "Basic dXNlcjpwYXNz"), + ("x-api-key", "key"), + (CLUSTER_SECRET_HEADER, "hash"), + ] { + let (forwarded, needs_node_token) = + stream_sync_headers(&actix_headers(&[("cookie", "session=abc"), credential])); + assert!(forwarded.get(reqwest::header::COOKIE).is_none()); + assert_eq!(forwarded.get(credential.0).unwrap(), credential.1); + assert!(!needs_node_token, "{} is a credential", credential.0); + } + } #[test] fn node_header_failure_wins_over_earlier_network_failure() { diff --git a/src/storage/field_stats.rs b/src/storage/field_stats.rs index d6e642463..b78ca9b17 100644 --- a/src/storage/field_stats.rs +++ b/src/storage/field_stats.rs @@ -999,16 +999,7 @@ pub async fn get_dataset_stats( TENANT_METADATA.get_global_query_auth(tenant).as_deref(), )?) } else { - // Forward the caller's credentials if they sent any. A session - // cookie is only valid on this node, so without credentials - // `send_query_request` falls back to the querier's own token. - req.headers() - .get(actix_web::http::header::AUTHORIZATION) - .map(|auth| { - let mut map = HeaderMap::new(); - map.insert(actix_web::http::header::AUTHORIZATION, auth.clone()); - map - }) + Some(prism_auth_without_secret(req.headers())?) }; let response = match send_query_request(auth, &query_request, &tenant_id).await { @@ -1033,6 +1024,24 @@ pub async fn get_dataset_stats( Ok(response) } +/// Auth for forwarding a Prism dataset-stats query to a querier without the +/// cluster secret: the caller's `Authorization` header, so the querier checks +/// the caller's own access. Anything else is rejected. A session cookie is +/// only valid on this node, and leaving the auth empty would make +/// `send_query_request` use the querier's own token and skip those checks. +fn prism_auth_without_secret(headers: &HeaderMap) -> Result { + let auth = headers + .get(actix_web::http::header::AUTHORIZATION) + .ok_or_else(|| { + QueryError::CustomError( + "P_CLUSTER_SECRET is required to forward this request to a querier".to_string(), + ) + })?; + let mut map = HeaderMap::new(); + map.insert(actix_web::http::header::AUTHORIZATION, auth.clone()); + Ok(map) +} + /// Headers for forwarding a Prism dataset-stats query to a querier with the /// cluster secret. All values are server-side (stored global query auth, /// cluster secret hash, the authenticated user id) apart from the tenant, @@ -1857,6 +1866,37 @@ mod tests { mod forwarding_header_tests { use super::*; + #[test] + fn prism_auth_without_secret_forwards_only_authorization() { + let mut headers = HeaderMap::new(); + headers.insert( + actix_web::http::header::AUTHORIZATION, + HeaderValue::from_static("Basic dXNlcjpwYXNz"), + ); + headers.insert( + actix_web::http::header::COOKIE, + HeaderValue::from_static("session=abc"), + ); + let map = prism_auth_without_secret(&headers).unwrap(); + assert_eq!(map.get("authorization").unwrap(), "Basic dXNlcjpwYXNz"); + assert_eq!(map.len(), 1); + } + + #[test] + fn prism_auth_without_secret_rejects_cookie_and_api_key_callers() { + for (name, value) in [("cookie", "session=abc"), ("x-api-key", "key")] { + let mut headers = HeaderMap::new(); + headers.insert( + HeaderName::from_static(name), + HeaderValue::from_static(value), + ); + assert!(matches!( + prism_auth_without_secret(&headers), + Err(QueryError::CustomError(_)) + )); + } + } + #[test] fn prism_forwarding_headers_include_auth_secret_tenant_and_user() { let map = prism_forwarding_headers("hash", "acme", "alice", Some("Basic abc")) From b16b4c57178854d528bc04fe84030d5e631b6a02 Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Sun, 1 Nov 2026 00:52:48 -0500 Subject: [PATCH 3/5] fix: keep node-token stream syncs to the default tenant The node token authenticates as root in the default tenant, so an ingestor given a tenant's stream sync with it would create or update the stream in the default tenant. Without a caller credential, a tenant's stream sync now fails (and is logged) and asks for P_CLUSTER_SECRET, as the other multi-tenant cluster paths do. A node token that can't be encoded now fails only that node's sync as a logged sync error. The stream already exists locally, so it is not returned as a header-build error. Co-Authored-By: Claude Opus 5.5 --- src/handlers/http/cluster/mod.rs | 84 +++++++++++++++++++++++++++++--- 1 file changed, 77 insertions(+), 7 deletions(-) diff --git a/src/handlers/http/cluster/mod.rs b/src/handlers/http/cluster/mod.rs index 1de139b2f..3f8055bd3 100644 --- a/src/handlers/http/cluster/mod.rs +++ b/src/handlers/http/cluster/mod.rs @@ -454,6 +454,43 @@ fn stream_sync_headers(headers: &HeaderMap) -> (reqwest::header::HeaderMap, bool (forwarded, !has_credentials) } +/// Without a credential from the caller, a stream sync authenticates with the +/// node token, i.e. as root in the default tenant. The ingestor would create a +/// tenant's stream in the default tenant, so that needs the cluster secret. +fn check_stream_sync_tenant( + needs_node_token: bool, + tenant_id: &Option, +) -> Result<(), StreamError> { + if needs_node_token && tenant_id.is_some() { + return Err(StreamError::Custom { + msg: "P_CLUSTER_SECRET is required to sync a tenant's stream to ingestors".to_string(), + status: StatusCode::INTERNAL_SERVER_ERROR, + }); + } + Ok(()) +} + +/// The headers for one node's stream sync. A node token that can't be encoded +/// fails only this node's sync: the stream already exists locally, so it is a +/// sync failure rather than a header-build error. +fn stream_sync_node_headers( + headers: &reqwest::header::HeaderMap, + needs_node_token: bool, + node_token: &str, +) -> Result { + let mut headers = headers.clone(); + if needs_node_token { + let token = generated_text_reqwest_header_value(node_token, "node token").map_err(|e| { + StreamError::Custom { + msg: e.to_string(), + status: StatusCode::INTERNAL_SERVER_ERROR, + } + })?; + headers.insert(reqwest::header::AUTHORIZATION, token); + } + Ok(headers) +} + // forward the create/update stream request to all ingestors to keep them in sync pub async fn sync_streams_with_ingestors( headers: HeaderMap, @@ -462,6 +499,7 @@ pub async fn sync_streams_with_ingestors( tenant_id: &Option, ) -> Result<(), StreamError> { let (mut reqwest_headers, needs_node_token) = stream_sync_headers(&headers); + check_stream_sync_tenant(needs_node_token, tenant_id)?; if let Some(userid) = fallback_intracluster_userid( reqwest_headers.contains_key(INTRA_CLUSTER_USERID), @@ -485,15 +523,11 @@ pub async fn sync_streams_with_ingestors( base_path_without_preceding_slash(), stream_name ); - let mut headers = reqwest_headers_clone.clone(); + let headers = + stream_sync_node_headers(&reqwest_headers_clone, needs_node_token, &ingestor.token); let body = body_clone.clone(); async move { - if needs_node_token { - headers.insert( - reqwest::header::AUTHORIZATION, - generated_text_reqwest_header_value(&ingestor.token, "node token")?, - ); - } + let headers = headers?; let res = INTRA_CLUSTER_CLIENT .put(url) .headers(headers) @@ -2141,6 +2175,42 @@ mod header_build_tests { assert!(needs_node_token); } + #[test] + fn stream_sync_node_headers_add_token_only_without_credentials() { + let (base, needs_node_token) = stream_sync_headers(&actix_headers(&[ + ("cookie", "session=abc"), + (INTRA_CLUSTER_USERID, "alice"), + ])); + let headers = stream_sync_node_headers(&base, needs_node_token, "Basic node").unwrap(); + assert_eq!( + headers.get(reqwest::header::AUTHORIZATION).unwrap(), + "Basic node" + ); + assert_eq!(headers.get(INTRA_CLUSTER_USERID).unwrap(), "alice"); + + let headers = stream_sync_node_headers(&base, false, "Basic node").unwrap(); + assert!(headers.get(reqwest::header::AUTHORIZATION).is_none()); + } + + #[test] + fn stream_sync_bad_node_token_is_a_sync_failure() { + let err = stream_sync_node_headers(&reqwest::header::HeaderMap::new(), true, "Basic josé") + .unwrap_err(); + assert!(matches!(err, StreamError::Custom { .. })); + assert!(!err.to_string().contains("josé")); + } + + #[test] + fn stream_sync_node_token_is_only_for_the_default_tenant() { + let tenant = Some("acme".to_string()); + assert!(check_stream_sync_tenant(true, &None).is_ok()); + assert!(check_stream_sync_tenant(false, &tenant).is_ok()); + assert!(matches!( + check_stream_sync_tenant(true, &tenant), + Err(StreamError::Custom { .. }) + )); + } + #[test] fn stream_sync_keeps_caller_credentials() { for credential in [ From 267ae8d3145847f900c6b34067e235dfab0c2a8e Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Sun, 1 Nov 2026 01:26:05 -0500 Subject: [PATCH 4/5] fix: name the stream, tenant and ingestor in stream sync errors Co-Authored-By: Claude Opus 5.5 --- src/handlers/http/cluster/mod.rs | 44 +++++++++++++++++++++++--------- 1 file changed, 32 insertions(+), 12 deletions(-) diff --git a/src/handlers/http/cluster/mod.rs b/src/handlers/http/cluster/mod.rs index 3f8055bd3..c7250c7f2 100644 --- a/src/handlers/http/cluster/mod.rs +++ b/src/handlers/http/cluster/mod.rs @@ -459,11 +459,14 @@ fn stream_sync_headers(headers: &HeaderMap) -> (reqwest::header::HeaderMap, bool /// tenant's stream in the default tenant, so that needs the cluster secret. fn check_stream_sync_tenant( needs_node_token: bool, + stream_name: &str, tenant_id: &Option, ) -> Result<(), StreamError> { - if needs_node_token && tenant_id.is_some() { + if needs_node_token && let Some(tenant) = tenant_id { return Err(StreamError::Custom { - msg: "P_CLUSTER_SECRET is required to sync a tenant's stream to ingestors".to_string(), + msg: format!( + "failed to sync stream {stream_name} of tenant {tenant} to ingestors: P_CLUSTER_SECRET is required" + ), status: StatusCode::INTERNAL_SERVER_ERROR, }); } @@ -476,13 +479,14 @@ fn check_stream_sync_tenant( fn stream_sync_node_headers( headers: &reqwest::header::HeaderMap, needs_node_token: bool, + node: &str, node_token: &str, ) -> Result { let mut headers = headers.clone(); if needs_node_token { let token = generated_text_reqwest_header_value(node_token, "node token").map_err(|e| { StreamError::Custom { - msg: e.to_string(), + msg: format!("failed to sync stream to ingestor {node}: {e}"), status: StatusCode::INTERNAL_SERVER_ERROR, } })?; @@ -499,7 +503,7 @@ pub async fn sync_streams_with_ingestors( tenant_id: &Option, ) -> Result<(), StreamError> { let (mut reqwest_headers, needs_node_token) = stream_sync_headers(&headers); - check_stream_sync_tenant(needs_node_token, tenant_id)?; + check_stream_sync_tenant(needs_node_token, &stream_name, tenant_id)?; if let Some(userid) = fallback_intracluster_userid( reqwest_headers.contains_key(INTRA_CLUSTER_USERID), @@ -524,7 +528,12 @@ pub async fn sync_streams_with_ingestors( stream_name ); let headers = - stream_sync_node_headers(&reqwest_headers_clone, needs_node_token, &ingestor.token); + stream_sync_node_headers( + &reqwest_headers_clone, + needs_node_token, + &ingestor.domain_name, + &ingestor.token, + ); let body = body_clone.clone(); async move { let headers = headers?; @@ -2181,34 +2190,45 @@ mod header_build_tests { ("cookie", "session=abc"), (INTRA_CLUSTER_USERID, "alice"), ])); - let headers = stream_sync_node_headers(&base, needs_node_token, "Basic node").unwrap(); + let headers = + stream_sync_node_headers(&base, needs_node_token, "node", "Basic node").unwrap(); assert_eq!( headers.get(reqwest::header::AUTHORIZATION).unwrap(), "Basic node" ); assert_eq!(headers.get(INTRA_CLUSTER_USERID).unwrap(), "alice"); - let headers = stream_sync_node_headers(&base, false, "Basic node").unwrap(); + let headers = stream_sync_node_headers(&base, false, "node", "Basic node").unwrap(); assert!(headers.get(reqwest::header::AUTHORIZATION).is_none()); } #[test] fn stream_sync_bad_node_token_is_a_sync_failure() { - let err = stream_sync_node_headers(&reqwest::header::HeaderMap::new(), true, "Basic josé") - .unwrap_err(); + let err = stream_sync_node_headers( + &reqwest::header::HeaderMap::new(), + true, + "node", + "Basic josé", + ) + .unwrap_err(); assert!(matches!(err, StreamError::Custom { .. })); assert!(!err.to_string().contains("josé")); + assert!(err.to_string().contains("ingestor node")); } #[test] fn stream_sync_node_token_is_only_for_the_default_tenant() { let tenant = Some("acme".to_string()); - assert!(check_stream_sync_tenant(true, &None).is_ok()); - assert!(check_stream_sync_tenant(false, &tenant).is_ok()); + assert!(check_stream_sync_tenant(true, "logs", &None).is_ok()); + assert!(check_stream_sync_tenant(false, "logs", &tenant).is_ok()); assert!(matches!( - check_stream_sync_tenant(true, &tenant), + check_stream_sync_tenant(true, "logs", &tenant), Err(StreamError::Custom { .. }) )); + let msg = check_stream_sync_tenant(true, "logs", &tenant) + .unwrap_err() + .to_string(); + assert!(msg.contains("logs") && msg.contains("acme"), "{msg}"); } #[test] From 209bf3442e6289644a61617a857e37736d1ea1f8 Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Sun, 1 Nov 2026 01:33:39 -0500 Subject: [PATCH 5/5] fix: drop needless borrow flagged by clippy Co-Authored-By: Claude Opus 5.5 --- src/handlers/http/cluster/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/handlers/http/cluster/mod.rs b/src/handlers/http/cluster/mod.rs index c7250c7f2..ef91339db 100644 --- a/src/handlers/http/cluster/mod.rs +++ b/src/handlers/http/cluster/mod.rs @@ -503,7 +503,7 @@ pub async fn sync_streams_with_ingestors( tenant_id: &Option, ) -> Result<(), StreamError> { let (mut reqwest_headers, needs_node_token) = stream_sync_headers(&headers); - check_stream_sync_tenant(needs_node_token, &stream_name, tenant_id)?; + check_stream_sync_tenant(needs_node_token, stream_name, tenant_id)?; if let Some(userid) = fallback_intracluster_userid( reqwest_headers.contains_key(INTRA_CLUSTER_USERID),