From 294473a8e030f7d9c4de2d0f3973bc7c03db1167 Mon Sep 17 00:00:00 2001 From: Evan Hicks Date: Fri, 6 Nov 2026 10:17:38 -0500 Subject: [PATCH 1/3] fix(liveness): Remove the upkeep interval based health check This check does not make sense when AlloyDB is the storage, since then the upkeep check is not strictly testing just the one pod. Even with SQLite pools, this check has caused unnecessary restarts during incidents, since a pod that is overloaded can also cause a restart, but that is the opposite of the desired behaviour when the pool is already behind. For now, make this check a no-op. That will help with the current incidents. In the future it can be expanded to do deadlock checks for the different tokio threads. --- src/upkeep.rs | 33 ++++++++------------------------- 1 file changed, 8 insertions(+), 25 deletions(-) diff --git a/src/upkeep.rs b/src/upkeep.rs index d26c4606..d512146b 100644 --- a/src/upkeep.rs +++ b/src/upkeep.rs @@ -563,36 +563,19 @@ fn create_retry_activation(mut new_activation: TaskActivation) -> TaskActivation /// Update health based on upkeep intervals /// -/// Because SQLite is a shared component for upkeep, grpc, and consumer loops, when one gets slow -/// they all do. We see taskbroker getting slow in production when the underlying cloud disk -/// degrades. Restarting the application typically solves this problem. -/// Track metrics and update status so that we can measure health failures, and apply probes -/// incrementally. +/// For an AlloyDB backed taskbroker, this check doesn't validate a specific pod, but AlloyDB +/// in general. For now it's a no-op, and it will get more sophisticated as the AlloyDB +/// taskbroker gets used more. pub async fn check_health( last_run: Instant, config: &Config, health_reporter: HealthReporter, ) -> Instant { - let now = Instant::now(); - if config.health_check_killswitched { - metrics::counter!("upkeep.health", "status" => "killswitched").increment(1); - health_reporter - .set_service_status(SERVICE_NAME, ServingStatus::Serving) - .await; - return now; - } - if now - last_run > Duration::from_millis(config.upkeep_unhealthy_interval_ms) { - metrics::counter!("upkeep.health", "status" => "unhealthy").increment(1); - health_reporter - .set_service_status(SERVICE_NAME, ServingStatus::NotServing) - .await; - } else { - metrics::counter!("upkeep.health", "status" => "healthy").increment(1); - health_reporter - .set_service_status(SERVICE_NAME, ServingStatus::Serving) - .await; - } - now + metrics::counter!("upkeep.health", "status" => "healthy").increment(1); + health_reporter + .set_service_status(SERVICE_NAME, ServingStatus::Serving) + .await; + Instant::now() } #[cfg(test)] From bb949488a4cdbc4ce8a611c1765f861a5a5da825 Mon Sep 17 00:00:00 2001 From: Evan Hicks Date: Fri, 6 Nov 2026 10:39:51 -0500 Subject: [PATCH 2/3] fix(liveness): Drop unused check_health args --- src/upkeep.rs | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/src/upkeep.rs b/src/upkeep.rs index d512146b..e78639c4 100644 --- a/src/upkeep.rs +++ b/src/upkeep.rs @@ -45,7 +45,6 @@ pub async fn upkeep( let guard = elegant_departure::get_shutdown_guard().shutdown_on_drop(); let mut timer = time::interval(Duration::from_millis(config.upkeep_task_interval_ms)); timer.set_missed_tick_behavior(time::MissedTickBehavior::Delay); - let mut last_run = Instant::now(); let mut last_vacuum = Instant::now(); let mut last_backtrace_log = Instant::now(); let mut emitted_partitions: HashSet = HashSet::new(); @@ -61,7 +60,7 @@ pub async fn upkeep( &mut last_vacuum, &mut emitted_partitions, ).await; - last_run = check_health(last_run, &config, health_reporter.clone()).await; + check_health(health_reporter.clone()).await; if config.log_async_backtrace && last_backtrace_log.elapsed() >= ASYNC_BACKTRACE_LOG_INTERVAL @@ -566,16 +565,11 @@ fn create_retry_activation(mut new_activation: TaskActivation) -> TaskActivation /// For an AlloyDB backed taskbroker, this check doesn't validate a specific pod, but AlloyDB /// in general. For now it's a no-op, and it will get more sophisticated as the AlloyDB /// taskbroker gets used more. -pub async fn check_health( - last_run: Instant, - config: &Config, - health_reporter: HealthReporter, -) -> Instant { +pub async fn check_health(health_reporter: HealthReporter) { metrics::counter!("upkeep.health", "status" => "healthy").increment(1); health_reporter .set_service_status(SERVICE_NAME, ServingStatus::Serving) .await; - Instant::now() } #[cfg(test)] From d9432e05b06db4b8df62222a4719f3721d57b988 Mon Sep 17 00:00:00 2001 From: Evan Hicks Date: Fri, 6 Nov 2026 12:03:14 -0500 Subject: [PATCH 3/3] ref(upkeep): Remove check_health and unused health config --- src/config/mod.rs | 9 --------- src/main.rs | 1 - src/upkeep.rs | 17 ----------------- 3 files changed, 27 deletions(-) diff --git a/src/config/mod.rs b/src/config/mod.rs index bbbb208b..501759b3 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -159,13 +159,6 @@ pub struct Config { /// (discarding, retrying activations, etc.) are executed. pub upkeep_task_interval_ms: u64, - /// The number of milliseconds between upkeep runs that indicates unhealthy - /// performance that should trigger a restart. - pub upkeep_unhealthy_interval_ms: u64, - - /// Whether to skip the health check if the pods are in a bad state. - pub health_check_killswitched: bool, - /// The number of seconds that deadline resets /// are skipped after startup. This delay allows workers /// time to publish results after a broker restart. @@ -303,8 +296,6 @@ impl Default for Config { store: StoreConfig::default(), runtime_config_path: None, upkeep_task_interval_ms: 1000, - upkeep_unhealthy_interval_ms: 5000, - health_check_killswitched: false, upkeep_deadline_reset_skip_after_startup_sec: 60, maintenance_task_interval_ms: 6000, runtime_metrics_interval: Duration::from_secs(1), diff --git a/src/main.rs b/src/main.rs index 30436b4c..65c92dfc 100644 --- a/src/main.rs +++ b/src/main.rs @@ -134,7 +134,6 @@ async fn main() -> Result<(), Error> { upkeep_store, startup_time, runtime_config_manager.clone(), - health_reporter.clone(), ) .await?; Ok(()) diff --git a/src/upkeep.rs b/src/upkeep.rs index e78639c4..5677f606 100644 --- a/src/upkeep.rs +++ b/src/upkeep.rs @@ -10,12 +10,9 @@ use prost_types::Timestamp; use sentry_arroyo::backends::ProducerError; use sentry_protos::taskbroker::v1::TaskActivation; use tokio::{fs, join, select, time}; -use tonic_health::ServingStatus; -use tonic_health::server::HealthReporter; use tracing::{debug, error, info, instrument, warn}; use uuid::Uuid; -use crate::SERVICE_NAME; use crate::config::Config; use crate::kafka::producer::KafkaProducer; use crate::runtime_config::RuntimeConfigManager; @@ -29,7 +26,6 @@ pub async fn upkeep( store: Arc, startup_time: DateTime, runtime_config_manager: Arc, - health_reporter: HealthReporter, ) -> Result<(), anyhow::Error> { const ASYNC_BACKTRACE_LOG_INTERVAL: Duration = Duration::from_secs(30); @@ -60,7 +56,6 @@ pub async fn upkeep( &mut last_vacuum, &mut emitted_partitions, ).await; - check_health(health_reporter.clone()).await; if config.log_async_backtrace && last_backtrace_log.elapsed() >= ASYNC_BACKTRACE_LOG_INTERVAL @@ -560,18 +555,6 @@ fn create_retry_activation(mut new_activation: TaskActivation) -> TaskActivation new_activation } -/// Update health based on upkeep intervals -/// -/// For an AlloyDB backed taskbroker, this check doesn't validate a specific pod, but AlloyDB -/// in general. For now it's a no-op, and it will get more sophisticated as the AlloyDB -/// taskbroker gets used more. -pub async fn check_health(health_reporter: HealthReporter) { - metrics::counter!("upkeep.health", "status" => "healthy").increment(1); - health_reporter - .set_service_status(SERVICE_NAME, ServingStatus::Serving) - .await; -} - #[cfg(test)] mod tests { use std::collections::HashSet;