From 213c0ffb42ecccdda416d0249b70cea47e15d0e9 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:22:09 +0200 Subject: [PATCH 1/9] fix(security): redact Hetzner token and quoted prefixes from diagnostics --- src/error.rs | 51 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/src/error.rs b/src/error.rs index 3dfe619..66b086e 100644 --- a/src/error.rs +++ b/src/error.rs @@ -13,6 +13,23 @@ fn is_rate_limit(err: &hcloud::apis::Error) -> bool { matches!(err, hcloud::apis::Error::ResponseError(response) if response.status.as_u16() == 429) } +/// Remove the Hetzner API token (including a quoted prefix) from diagnostics. +/// +/// Hetzner error bodies can contain a token prefix, not only the complete token. +/// Eight bytes is the shortest prefix we treat as secret so short unrelated +/// text does not get unnecessarily replaced. +#[must_use] +pub fn redact(message: &str, token: &str) -> String { + const MIN_PREFIX: usize = 8; + let mut safe = message.to_string(); + for len in (MIN_PREFIX..=token.len()).rev() { + if let Some(prefix) = token.get(..len) { + safe = safe.replace(prefix, "[REDACTED]"); + } + } + safe +} + /// All possible errors that can occur in the robotlb operator. #[derive(Debug, Error)] pub enum RobotLBError { @@ -181,3 +198,37 @@ impl RobotLBError { .is_some_and(|code| code == "target_already_defined") } } + +#[cfg(test)] +mod redaction_tests { + use super::redact; + + const TOKEN: &str = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ01"; + + #[test] + fn redacts_complete_token_and_quoted_prefix() { + assert_eq!( + redact(&format!("bad token {TOKEN}"), TOKEN), + "bad token [REDACTED]" + ); + assert_eq!( + redact(&format!("invalid token {}", &TOKEN[..24]), TOKEN), + "invalid token [REDACTED]" + ); + } + + #[test] + fn leaves_unrelated_messages_and_short_prefixes_intact() { + assert_eq!(redact("Hetzner 500", TOKEN), "Hetzner 500"); + assert_eq!(redact("short abcdefg", TOKEN), "short abcdefg"); + assert_eq!(redact("empty token", ""), "empty token"); + } + + #[test] + fn redacts_non_ascii_surrounding_text() { + assert_eq!( + redact(&format!("échec {} 🦀", &TOKEN[..8]), TOKEN), + "échec [REDACTED] 🦀" + ); + } +} From 4a9b331de5532ddaa8f409a95b88e19a8c3f4c62 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:22:41 +0200 Subject: [PATCH 2/9] fix(controller): report and redact all reconciliation failures --- src/controller/mod.rs | 144 +++++++++++++++++++++++++++++------------- 1 file changed, 99 insertions(+), 45 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 8192583..5856f26 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -92,15 +92,25 @@ pub async fn reconcile_service( context: Arc, ) -> RobotLBResult { let mut timer = ReconcileTimer::new(context.metrics.clone()); + let result = reconcile_service_inner(svc.clone(), context.clone()).await; + if let Err(error) = &result { + if !matches!(error, RobotLBError::SkipService) { + timer.set_failed(); + publish_reconciliation_warning(&svc, &context, error).await; + } + } + result +} +async fn reconcile_service_inner( + svc: Arc, + context: Arc, +) -> RobotLBResult { if !context.is_leader.load(Ordering::Relaxed) { return Err(RobotLBError::SkipService); } if let Some(delay) = context.rate_limit.remaining(std::time::Instant::now()) { - tracing::debug!( - ?delay, - "Hetzner API rate limit pause; skipping reconciliation" - ); + tracing::debug!(?delay, "Hetzner API rate limit pause; skipping reconciliation"); return Ok(Action::requeue(delay)); } @@ -116,39 +126,20 @@ pub async fn reconcile_service( ensure_service_is_supported(&svc)?; - tracing::info!("Starting service reconcilation"); + tracing::info!("Starting service reconciliation"); - let lb = match LoadBalancer::try_from_svc(&svc, &context) { - Ok(lb) => lb, - Err(e) => { - timer.set_failed(); - return Err(e); - } - }; + let lb = LoadBalancer::try_from_svc(&svc, &context)?; - // Add finalizer if it's not there yet. - if !finalizers::check(&svc) - && let Err(e) = finalizers::add(context.client.clone(), &svc).await - { - timer.set_failed(); - return Err(e); + if !finalizers::check(&svc) { + finalizers::add(context.client.clone(), &svc).await?; } - // Based on the service type, we will reconcile the load balancer. - match reconcile_load_balancer(lb, svc.clone(), context.clone()).await { - Ok(action) => Ok(action), - Err(e) => { - timer.set_failed(); - if matches!(e, RobotLBError::InvalidLoadBalancerConfiguration(_)) { - publish_invalid_configuration_warning(&svc, &context, &e).await; - } - Err(e) - } - } + reconcile_load_balancer(lb, svc, context).await } -/// Emit a Warning Event while keeping reconciliation errors independent of Event RBAC. -async fn publish_invalid_configuration_warning( +/// Display the last error on the Service without leaking Hetzner API credentials. +/// Event publishing errors must not hide the original reconciliation failure. +async fn publish_reconciliation_warning( svc: &Service, context: &CurrentContext, error: &RobotLBError, @@ -163,18 +154,40 @@ async fn publish_invalid_configuration_warning( let event = Event { type_: EventType::Warning, action: "Reconcile".to_string(), - reason: "InvalidLoadBalancerConfiguration".to_string(), - note: Some(error.to_string()), + reason: warning_reason(error).to_string(), + note: Some(event_note(error, &context.config.hcloud_token)), secondary: None, }; if let Err(publish_error) = recorder.publish(&event, &svc.object_ref(&())).await { tracing::warn!( error = %publish_error, - "Cannot publish invalid configuration Warning Event" + "Cannot publish reconciliation Warning Event" ); } } +const fn warning_reason(error: &RobotLBError) -> &'static str { + if matches!(error, RobotLBError::InvalidLoadBalancerConfiguration(_)) { + "InvalidLoadBalancerConfiguration" + } else { + "SyncLoadBalancerFailed" + } +} + +/// Kubernetes Events cap notes at 1024 bytes, not characters. +fn event_note(error: &RobotLBError, token: &str) -> String { + const MAX_NOTE_BYTES: usize = 1024; + let mut note = crate::error::redact(&error.to_string(), token); + if note.len() > MAX_NOTE_BYTES { + let mut end = MAX_NOTE_BYTES; + while !note.is_char_boundary(end) { + end -= 1; + } + note.truncate(end); + } + note +} + /// Reconcile the `LoadBalancer` type of service. /// /// This function will find the nodes based on the node selector @@ -264,6 +277,7 @@ pub async fn run( shutdown: tokio_util::sync::CancellationToken, ) { let controller_shutdown = shutdown.clone(); + let hcloud_token = context.config.hcloud_token.clone(); Controller::new( kube::Api::::all(kube_client), watcher::Config::default(), @@ -275,20 +289,31 @@ pub async fn run( ) .run(reconcile_service, on_error, context) .take_until(controller_shutdown.cancelled()) - .for_each(|reconcilation_result| async move { - match reconcilation_result { - Ok((service, _action)) => { - tracing::info!("Reconcilation of a service {} was successful", service.name); - } - Err(err) => match err { - kube::runtime::controller::Error::ReconcilerFailed( + .for_each(|reconciliation_result| { + let token = hcloud_token.clone(); + async move { + match reconciliation_result { + Ok((service, _action)) => { + tracing::info!("Reconciliation of a service {} was successful", service.name); + } + Err(kube::runtime::controller::Error::ReconcilerFailed( RobotLBError::SkipService, _, - ) => {} - _ => { - tracing::error!("Error reconciling service: {:#?}", err); + )) => {} + Err(kube::runtime::controller::Error::ReconcilerFailed(error, service)) => { + tracing::error!( + service = %service, + error = %crate::error::redact(&error.to_string(), &token), + "Service reconciliation failed" + ); + } + Err(error) => { + tracing::error!( + error = %crate::error::redact(&error.to_string(), &token), + "Controller reconciliation failed" + ); } - }, + } } }) .await; @@ -323,6 +348,35 @@ mod tests { assert!(ensure_service_is_supported(&svc).is_ok()); } + #[test] + fn all_errors_are_reported_except_skipped_services() { + assert_eq!( + warning_reason(&RobotLBError::InvalidLoadBalancerConfiguration( + "too many targets".to_string() + )), + "InvalidLoadBalancerConfiguration" + ); + assert_eq!( + warning_reason(&RobotLBError::HCloudError("error".to_string())), + "SyncLoadBalancerFailed" + ); + } + + #[test] + fn warning_notes_redact_hetzner_token_and_fit_event_limit() { + let token = "0123456789abcdef"; + let error = RobotLBError::HCloudError(format!( + "invalid token {} {}", + &token[..12], + "é".repeat(600) + )); + let note = event_note(&error, token); + assert!(note.contains("[REDACTED]")); + assert!(!note.contains(&token[..12])); + assert!(note.len() <= 1024); + assert!(note.is_char_boundary(note.len())); + } + #[test] fn on_error_awaits_change_for_skipped_service() { let action = action_for_error(&RobotLBError::SkipService); From 26abec512bf66a60c891dd98bf7a710606ec3302 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:23:15 +0200 Subject: [PATCH 3/9] fix(lb): retry transient Hetzner deletion errors without retrying 429 --- src/lb/mod.rs | 46 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 44 insertions(+), 2 deletions(-) diff --git a/src/lb/mod.rs b/src/lb/mod.rs index 9414447..68926d8 100644 --- a/src/lb/mod.rs +++ b/src/lb/mod.rs @@ -500,7 +500,7 @@ impl LoadBalancer { else { return Ok(()); }; - api::delete_load_balancer(&self.hcloud_config, lb.id).await + retry_temporary(|| api::delete_load_balancer(&self.hcloud_config, lb.id)).await } /// Release without parsing annotations: malformed config cannot block deletion. @@ -516,7 +516,7 @@ impl LoadBalancer { let Some(lb) = api::get_owned_load_balancer(&context.hcloud_config, uid).await? else { return Ok(()); }; - api::delete_load_balancer(&context.hcloud_config, lb.id).await + retry_temporary(|| api::delete_load_balancer(&context.hcloud_config, lb.id)).await } /// Existing resources must have the matching Service UID label. @@ -683,6 +683,48 @@ mod tests { use hcloud::apis::configuration::Configuration as HcloudConfig; use std::sync::Mutex; + fn delete_error(status: u16) -> RobotLBError { + RobotLBError::HcloudLBDeleteError(hcloud::apis::Error::ResponseError( + hcloud::apis::ResponseContent { + status: status.try_into().expect("valid HTTP status"), + content: String::new(), + entity: None, + }, + )) + } + + #[tokio::test] + async fn delete_retries_locked_then_succeeds() { + let attempts = std::sync::atomic::AtomicUsize::new(0); + retry_temporary(|| { + let attempt = attempts.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + async move { + if attempt == 0 { + Err(delete_error(423)) + } else { + Ok(()) + } + } + }) + .await + .expect("second deletion succeeds"); + assert_eq!(attempts.load(std::sync::atomic::Ordering::Relaxed), 2); + } + + #[tokio::test] + async fn delete_does_not_retry_permanent_or_rate_limit_errors() { + for status in [403, 429] { + let attempts = std::sync::atomic::AtomicUsize::new(0); + let result: RobotLBResult<()> = retry_temporary(|| { + attempts.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + async move { Err(delete_error(status)) } + }) + .await; + assert!(result.is_err()); + assert_eq!(attempts.load(std::sync::atomic::Ordering::Relaxed), 1); + } + } + #[test] fn capacity_validation_rejects_oversized_configuration() { let mut lb = test_load_balancer(); From 5ef1b7f9cb4e03c5a646d184227fab97e84ee47c Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:25:53 +0200 Subject: [PATCH 4/9] style: satisfy rustfmt and clippy in reconciliation reporting --- src/controller/mod.rs | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 5856f26..56d070d 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -93,11 +93,11 @@ pub async fn reconcile_service( ) -> RobotLBResult { let mut timer = ReconcileTimer::new(context.metrics.clone()); let result = reconcile_service_inner(svc.clone(), context.clone()).await; - if let Err(error) = &result { - if !matches!(error, RobotLBError::SkipService) { - timer.set_failed(); - publish_reconciliation_warning(&svc, &context, error).await; - } + if let Err(error) = &result + && !matches!(error, RobotLBError::SkipService) + { + timer.set_failed(); + publish_reconciliation_warning(&svc, &context, error).await; } result } @@ -110,7 +110,10 @@ async fn reconcile_service_inner( return Err(RobotLBError::SkipService); } if let Some(delay) = context.rate_limit.remaining(std::time::Instant::now()) { - tracing::debug!(?delay, "Hetzner API rate limit pause; skipping reconciliation"); + tracing::debug!( + ?delay, + "Hetzner API rate limit pause; skipping reconciliation" + ); return Ok(Action::requeue(delay)); } @@ -294,7 +297,10 @@ pub async fn run( async move { match reconciliation_result { Ok((service, _action)) => { - tracing::info!("Reconciliation of a service {} was successful", service.name); + tracing::info!( + "Reconciliation of a service {} was successful", + service.name + ); } Err(kube::runtime::controller::Error::ReconcilerFailed( RobotLBError::SkipService, From c7e32ccc14a1d3ac7468d789ef1a3421ea8e9934 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:38:59 +0200 Subject: [PATCH 5/9] fix(controller): throttle identical warning event publications --- src/controller/event_throttle.rs | 121 +++++++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 src/controller/event_throttle.rs diff --git a/src/controller/event_throttle.rs b/src/controller/event_throttle.rs new file mode 100644 index 0000000..241c69b --- /dev/null +++ b/src/controller/event_throttle.rs @@ -0,0 +1,121 @@ +//! Bound repeated Kubernetes Warning Events during persistent reconciliation failures. +//! +//! Reconciliation and error logging continue on every attempt. Only identical +//! Event publications are throttled, so a changed error remains visible. + +use std::{ + collections::HashMap, + sync::{Mutex, PoisonError}, + time::{Duration, Instant}, +}; + +const REPEAT_INTERVAL: Duration = Duration::from_secs(300); + +/// Identifies the exact error shown on a particular Service. +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +pub(crate) struct EventKey { + pub service_uid: String, + pub reason: String, + pub note: String, +} + +/// Records recent Event publication attempts for one controller instance. +#[derive(Debug, Default)] +pub(crate) struct EventThrottle { + recent: Mutex>, +} + +impl EventThrottle { + /// Reserve an Event. Concurrent reconciliations cannot reserve the same key. + /// Expired entries are removed to avoid retaining deleted Services indefinitely. + pub fn reserve(&self, key: &EventKey, now: Instant) -> bool { + let mut recent = self.recent.lock().unwrap_or_else(PoisonError::into_inner); + recent.retain(|_, last| now.saturating_duration_since(*last) < REPEAT_INTERVAL); + if recent.contains_key(key) { + return false; + } + recent.insert(key.clone(), now); + true + } + + /// Permit another attempt if publishing failed. Do not clear a newer reservation. + pub fn release(&self, key: &EventKey, reserved_at: Instant) { + let mut recent = self.recent.lock().unwrap_or_else(PoisonError::into_inner); + if recent.get(key) == Some(&reserved_at) { + recent.remove(key); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(uid: &str, reason: &str, note: &str) -> EventKey { + EventKey { + service_uid: uid.to_string(), + reason: reason.to_string(), + note: note.to_string(), + } + } + + #[test] + fn identical_event_is_suppressed_for_five_minutes_then_resent() { + let throttle = EventThrottle::default(); + let first = Instant::now(); + let event = key("service-uid", "SyncLoadBalancerFailed", "Hetzner 503"); + assert!(throttle.reserve(&event, first)); + assert!(!throttle.reserve(&event, first + Duration::from_secs(299))); + assert!(throttle.reserve(&event, first + REPEAT_INTERVAL)); + assert!(!throttle.reserve(&event, first + REPEAT_INTERVAL)); + } + + #[test] + fn different_service_reason_or_message_is_not_suppressed() { + let throttle = EventThrottle::default(); + let now = Instant::now(); + assert!(throttle.reserve(&key("uid-a", "SyncLoadBalancerFailed", "503"), now)); + assert!(throttle.reserve(&key("uid-b", "SyncLoadBalancerFailed", "503"), now)); + assert!(throttle.reserve(&key("uid-a", "InvalidLoadBalancerConfiguration", "503"), now)); + assert!(throttle.reserve(&key("uid-a", "SyncLoadBalancerFailed", "502"), now)); + } + + #[test] + fn failed_event_publication_does_not_suppress_retry() { + let throttle = EventThrottle::default(); + let now = Instant::now(); + let event = key("uid", "SyncLoadBalancerFailed", "503"); + assert!(throttle.reserve(&event, now)); + throttle.release(&event, now); + assert!(throttle.reserve(&event, now + Duration::from_secs(30))); + } + + #[test] + fn stale_failure_cannot_cancel_a_new_reservation() { + let throttle = EventThrottle::default(); + let now = Instant::now(); + let event = key("uid", "SyncLoadBalancerFailed", "503"); + assert!(throttle.reserve(&event, now)); + assert!(throttle.reserve(&event, now + REPEAT_INTERVAL)); + throttle.release(&event, now); + assert!(!throttle.reserve(&event, now + REPEAT_INTERVAL)); + } + + #[test] + fn only_one_concurrent_reconcile_may_publish_an_event() { + let throttle = std::sync::Arc::new(EventThrottle::default()); + let now = Instant::now(); + let event = key("uid", "SyncLoadBalancerFailed", "503"); + let reservations: usize = std::thread::scope(|scope| { + let tasks: Vec<_> = (0..8) + .map(|_| { + let throttle = throttle.clone(); + let event = event.clone(); + scope.spawn(move || usize::from(throttle.reserve(&event, now))) + }) + .collect(); + tasks.into_iter().map(|task| task.join().unwrap()).sum() + }); + assert_eq!(reservations, 1); + } +} From 83d22ef6335cf58e626440b837d54d832b5e77fd Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:39:16 +0200 Subject: [PATCH 6/9] fix(controller): suppress duplicate warnings per service, reason and note --- src/controller/mod.rs | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 56d070d..6474d81 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -3,9 +3,13 @@ //! This module contains the reconciliation logic for Kubernetes services //! with the `robotlb` load balancer class. +mod event_throttle; mod nodes; mod status; +pub(crate) use event_throttle::EventThrottle; +use event_throttle::EventKey; + use std::{ sync::{Arc, atomic::Ordering}, time::Duration, @@ -147,6 +151,22 @@ async fn publish_reconciliation_warning( context: &CurrentContext, error: &RobotLBError, ) { + let reason = warning_reason(error); + let note = event_note(error, &context.config.hcloud_token); + let event_key = EventKey { + service_uid: svc + .metadata + .uid + .clone() + .unwrap_or_else(|| format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any())), + reason: reason.to_string(), + note: note.clone(), + }; + let reserved_at = std::time::Instant::now(); + if !context.event_throttle.reserve(&event_key, reserved_at) { + return; + } + let recorder = Recorder::new( context.client.clone(), Reporter { @@ -157,11 +177,12 @@ async fn publish_reconciliation_warning( let event = Event { type_: EventType::Warning, action: "Reconcile".to_string(), - reason: warning_reason(error).to_string(), - note: Some(event_note(error, &context.config.hcloud_token)), + reason: reason.to_string(), + note: Some(note), secondary: None, }; if let Err(publish_error) = recorder.publish(&event, &svc.object_ref(&())).await { + context.event_throttle.release(&event_key, reserved_at); tracing::warn!( error = %publish_error, "Cannot publish reconciliation Warning Event" From 97d3ea6b0570ebf7d1775e5c20a6e21ae3c2c523 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:39:24 +0200 Subject: [PATCH 7/9] fix(controller): share warning event throttle in reconciliation context --- src/main.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/main.rs b/src/main.rs index 85f272d..e0ea12e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -72,6 +72,8 @@ pub struct CurrentContext { pub metrics: Arc, /// Hetzner project-wide API throttle shared across all Services. pub rate_limit: Arc, + /// Suppress repeated identical Kubernetes Warning Events. + pub(crate) event_throttle: Arc, } impl CurrentContext { @@ -91,6 +93,7 @@ impl CurrentContext { is_leader, metrics, rate_limit: Arc::default(), + event_throttle: Arc::default(), } } } From 6590f40fbc4e0d10ff66d9f3ce8818368e3f6456 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:41:55 +0200 Subject: [PATCH 8/9] test(controller): satisfy clippy while preserving true concurrent event test --- src/controller/event_throttle.rs | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/controller/event_throttle.rs b/src/controller/event_throttle.rs index 241c69b..5d4137d 100644 --- a/src/controller/event_throttle.rs +++ b/src/controller/event_throttle.rs @@ -13,7 +13,7 @@ const REPEAT_INTERVAL: Duration = Duration::from_secs(300); /// Identifies the exact error shown on a particular Service. #[derive(Clone, Debug, Eq, Hash, PartialEq)] -pub(crate) struct EventKey { +pub struct EventKey { pub service_uid: String, pub reason: String, pub note: String, @@ -21,7 +21,7 @@ pub(crate) struct EventKey { /// Records recent Event publication attempts for one controller instance. #[derive(Debug, Default)] -pub(crate) struct EventThrottle { +pub struct EventThrottle { recent: Mutex>, } @@ -76,7 +76,10 @@ mod tests { let now = Instant::now(); assert!(throttle.reserve(&key("uid-a", "SyncLoadBalancerFailed", "503"), now)); assert!(throttle.reserve(&key("uid-b", "SyncLoadBalancerFailed", "503"), now)); - assert!(throttle.reserve(&key("uid-a", "InvalidLoadBalancerConfiguration", "503"), now)); + assert!(throttle.reserve( + &key("uid-a", "InvalidLoadBalancerConfiguration", "503"), + now + )); assert!(throttle.reserve(&key("uid-a", "SyncLoadBalancerFailed", "502"), now)); } @@ -107,13 +110,13 @@ mod tests { let now = Instant::now(); let event = key("uid", "SyncLoadBalancerFailed", "503"); let reservations: usize = std::thread::scope(|scope| { - let tasks: Vec<_> = (0..8) - .map(|_| { - let throttle = throttle.clone(); - let event = event.clone(); - scope.spawn(move || usize::from(throttle.reserve(&event, now))) - }) - .collect(); + // Spawn all workers before joining to exercise concurrent reservations. + let mut tasks = Vec::new(); + for _ in 0..8 { + let throttle = throttle.clone(); + let event = event.clone(); + tasks.push(scope.spawn(move || usize::from(throttle.reserve(&event, now)))); + } tasks.into_iter().map(|task| task.join().unwrap()).sum() }); assert_eq!(reservations, 1); From afe49ec8ecd4b2800c6813b71db2afb1a6154199 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:42:05 +0200 Subject: [PATCH 9/9] style(controller): apply rustfmt layout to event throttling --- src/controller/mod.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 6474d81..317203c 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -7,8 +7,8 @@ mod event_throttle; mod nodes; mod status; -pub(crate) use event_throttle::EventThrottle; use event_throttle::EventKey; +pub(crate) use event_throttle::EventThrottle; use std::{ sync::{Arc, atomic::Ordering}, @@ -154,11 +154,9 @@ async fn publish_reconciliation_warning( let reason = warning_reason(error); let note = event_note(error, &context.config.hcloud_token); let event_key = EventKey { - service_uid: svc - .metadata - .uid - .clone() - .unwrap_or_else(|| format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any())), + service_uid: svc.metadata.uid.clone().unwrap_or_else(|| { + format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any()) + }), reason: reason.to_string(), note: note.clone(), };