diff --git a/src/config/mod.rs b/src/config/mod.rs index 11df7567..3c5289f3 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -1,6 +1,6 @@ #![allow(clippy::result_large_err)] use std::borrow::Cow; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::time::Duration; use anyhow::{Result, anyhow}; @@ -49,6 +49,27 @@ pub enum DeliveryMode { Push, } +/// The applications a pool admits. An empty set admits every application. +#[derive(Clone, Debug, Default, PartialEq, Deserialize, Serialize)] +#[serde(transparent)] +pub struct Applications(BTreeSet); + +impl Applications { + pub fn admits(&self, application: &str) -> bool { + self.0.is_empty() || self.0.contains(application) + } + + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } +} + +impl FromIterator for Applications { + fn from_iter>(iter: T) -> Self { + Self(iter.into_iter().collect()) + } +} + #[derive(Clone, PartialEq, Debug, Deserialize, Serialize, Validate)] pub struct Config { /// Deprecated configuration options. Not meant to be used. @@ -216,6 +237,12 @@ pub struct Config { #[validate(length(min = 1))] pub worker_map: BTreeMap, + /// The applications this pool can execute. Activations for any other + /// application are discarded by the consumer. Normalization derives this + /// from `worker_map` in push mode; left empty, the check is disabled. + #[serde(default)] + pub applications: Applications, + /// The namespace to assign to raw mode activations. pub raw_namespace: Option, @@ -291,6 +318,7 @@ impl Default for Config { status_update_batch_size: 1, status_update_interval_ms: 100, worker_map: [].into(), + applications: Applications::default(), raw_namespace: None, raw_application: None, raw_taskname: None, @@ -810,6 +838,42 @@ impl Config { } } + self.normalize_applications(); + self.validate_applications()?; + + Ok(()) + } + + /// Push pools derive the set from `worker_map`. Pull pools have no such list and + /// are left empty, which admits everything. + fn normalize_applications(&mut self) { + if !self.applications.is_empty() { + return; + } + + if self.delivery_mode == DeliveryMode::Push { + self.applications = self.worker_map.keys().cloned().collect(); + } else { + warn!("applications is unset, activations will not be checked against it"); + } + } + + /// A `worker_map` entry outside `applications` is unreachable in push mode. Pull + /// pools never deliver through `worker_map` and are given a default entry from + /// `from_args` that they never read, so the check does not apply to them. + fn validate_applications(&self) -> Result<()> { + if self.applications.is_empty() || self.delivery_mode != DeliveryMode::Push { + return Ok(()); + } + + for application in self.worker_map.keys() { + if !self.applications.admits(application) { + return Err(anyhow!( + "worker_map application '{application}' is not in applications" + )); + } + } + Ok(()) } @@ -1253,6 +1317,112 @@ mod tests { }); } + #[test] + fn test_applications_derived_from_worker_map_in_push_mode() { + Jail::expect_with(|jail| { + jail.set_env("TASKBROKER_DELIVERY_MODE", "push"); + jail.set_env( + "TASKBROKER_WORKER_MAP", + "{launchpad=http://127.0.0.1:50052}", + ); + + let args = Args { + run: Run::Broker, + config: None, + }; + let config = Config::from_args(&args).unwrap(); + + assert!(config.applications.admits("launchpad")); + assert!(!config.applications.admits("sentry")); + + Ok(()) + }); + } + + /// Pull pools are not given a `worker_map`, so there is nothing to derive from + /// and everything stays admitted until the pool opts in. + #[test] + fn test_applications_unset_in_pull_mode_admits_everything() { + Jail::expect_with(|_jail| { + let args = Args { + run: Run::Broker, + config: None, + }; + let config = Config::from_args(&args).unwrap(); + + assert_eq!(config.delivery_mode, DeliveryMode::Pull); + assert!(config.applications.is_empty()); + assert!(config.applications.admits("launchpad")); + + Ok(()) + }); + } + + /// `applications` uses the same env list encoding as other sequences. + #[test] + fn test_applications_from_env() { + Jail::expect_with(|jail| { + jail.set_env("TASKBROKER_APPLICATIONS", "[sentry,launchpad]"); + + let args = Args { + run: Run::Broker, + config: None, + }; + let config = Config::from_args(&args).unwrap(); + + assert!(config.applications.admits("sentry")); + assert!(config.applications.admits("launchpad")); + assert!(!config.applications.admits("seer")); + + Ok(()) + }); + } + + /// `from_args` gives pull pools a default `worker_map` they never deliver through, + /// which must not block them from opting in to an unrelated application. + #[test] + fn test_pull_applications_ignore_default_worker_map() { + Jail::expect_with(|jail| { + jail.create_file("config.yaml", "applications:\n - launchpad\n")?; + + let args = Args { + run: Run::Broker, + config: Some("config.yaml".to_owned()), + }; + let config = Config::from_args(&args).unwrap(); + + assert!(config.applications.admits("launchpad")); + assert!(!config.applications.admits("sentry")); + + Ok(()) + }); + } + + #[test] + fn test_applications_rejects_unreachable_worker_map_entry() { + Jail::expect_with(|jail| { + jail.create_file( + "config.yaml", + r#" + delivery_mode: push + applications: + - launchpad + worker_map: + sentry: http://sentry:50052 + "#, + )?; + + let args = Args { + run: Run::Broker, + config: Some("config.yaml".to_owned()), + }; + + assert!(Config::from_args(&args).is_err()); + + Ok(()) + }); + } + #[test] fn test_kafka_consumer_config() { let args = Args { diff --git a/src/grpc/server.rs b/src/grpc/server.rs index ef39c507..1b39025f 100644 --- a/src/grpc/server.rs +++ b/src/grpc/server.rs @@ -43,6 +43,25 @@ impl ConsumerService for TaskbrokerServer { let application = &request.get_ref().application; let namespace = &request.get_ref().namespace; + // Worker pointed at the wrong pool fails + if let Some(application) = application + && !self.config.applications.admits(application) + { + metrics::counter!( + "grpc_server.unserved_application", + "method" => "get_task", + ) + .increment(1); + error!( + application, + "Requested application is not served by this pool" + ); + + return Err(Status::failed_precondition(format!( + "Broker does not serve application '{application}'" + ))); + } + let inflight = self .store .claim_activation_for_pull(application.as_deref(), namespace.as_deref()) @@ -177,6 +196,24 @@ impl ConsumerService for TaskbrokerServer { return Ok(Response::new(SetTaskStatusResponse { task: None })); }; + // Same check as get_task. Returning an error here would make the worker + // re-send the status update, so an unserved application just gets no task. + if let Some(application) = application + && !self.config.applications.admits(application) + { + metrics::counter!( + "grpc_server.unserved_application", + "method" => "fetch_next_task", + ) + .increment(1); + error!( + application, + "Requested application is not served by this pool" + ); + + return Ok(Response::new(SetTaskStatusResponse { task: None })); + } + let start_time = Instant::now(); let res = match self diff --git a/src/grpc/server_tests.rs b/src/grpc/server_tests.rs index 720c6a7a..3a6bafda 100644 --- a/src/grpc/server_tests.rs +++ b/src/grpc/server_tests.rs @@ -202,6 +202,73 @@ async fn test_get_task_with_application_success(#[case] adapter: &str) { assert_eq!(task.application, Some("hammers".into())); } +#[tokio::test] +async fn test_get_task_unserved_application_returns_failed_precondition() { + let store = create_test_store("sqlite").await; + let mut config = Config { + applications: ["sentry".to_owned()].into_iter().collect(), + ..Config::default() + }; + config.normalize_and_validate().unwrap(); + + let service = TaskbrokerServer { + store, + config: Arc::new(config), + update_tx: None, + }; + + let request = GetTaskRequest { + namespace: None, + application: Some("hammers".into()), + }; + + let response = service.get_task(Request::new(request)).await; + + assert!(response.is_err()); + assert_eq!(response.unwrap_err().code(), Code::FailedPrecondition); +} + +/// A leftover activation for an unserved application must not be handed out on the +/// fetch-next path either. +#[tokio::test] +#[allow(deprecated)] +async fn test_fetch_next_task_skips_unserved_application() { + let store = create_test_store("sqlite").await; + let mut config = Config { + applications: ["sentry".to_owned()].into_iter().collect(), + ..Config::default() + }; + config.normalize_and_validate().unwrap(); + + let mut activations = make_activations(2); + let mut payload = TaskActivation::decode(&activations[1].activation as &[u8]).unwrap(); + payload.application = Some("launchpad".into()); + activations[1].activation = payload.encode_to_vec(); + activations[1].application = "launchpad".into(); + store.store(&activations).await.unwrap(); + + let service = TaskbrokerServer { + store, + config: Arc::new(config), + update_tx: None, + }; + + let request = SetTaskStatusRequest { + id: "id_0".to_string(), + status: 5, // Complete + fetch_next_task: Some(FetchNextTask { + namespace: None, + application: Some("launchpad".into()), + }), + max_attempts: None, + delay_on_retry: None, + }; + + let response = service.set_task_status(Request::new(request)).await; + assert!(response.is_ok()); + assert!(response.unwrap().get_ref().task.is_none()); +} + #[tokio::test] #[rstest] #[case::sqlite("sqlite")] diff --git a/src/kafka/activation_batcher.rs b/src/kafka/activation_batcher.rs index 2cc127ad..dddae044 100644 --- a/src/kafka/activation_batcher.rs +++ b/src/kafka/activation_batcher.rs @@ -1,3 +1,4 @@ +use std::collections::BTreeMap; use std::mem::replace; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -5,8 +6,9 @@ use std::time::{Duration, Instant}; use chrono::Utc; use futures::future::join_all; use rdkafka::config::ClientConfig; +use tracing::error; -use crate::config::Config; +use crate::config::{Applications, Config}; use crate::kafka::producer::KafkaProducer; use crate::runtime_config::RuntimeConfigManager; use crate::store::activation::Activation; @@ -20,6 +22,8 @@ pub struct ActivationBatcherConfig { pub producer_config: ClientConfig, pub kafka_topic: String, pub kafka_long_topic: String, + pub kafka_deadletter_topic: String, + pub applications: Applications, pub send_timeout_ms: u64, pub max_batch_time_ms: u64, pub max_batch_len: usize, @@ -35,6 +39,8 @@ impl ActivationBatcherConfig { producer_config: config.kafka_producer_config(), kafka_topic: topic_name.to_owned(), kafka_long_topic: config.kafka_long_topic.clone(), + kafka_deadletter_topic: config.kafka_deadletter_topic.clone(), + applications: config.applications.clone(), send_timeout_ms: config.kafka_send_timeout_ms, max_batch_time_ms: config.store.insert_batch_max_time_ms, max_batch_len: config.store.insert_batch_max_length, @@ -46,7 +52,8 @@ impl ActivationBatcherConfig { pub struct ActivationBatcher { batch: Vec, batch_size: usize, - forward_batch: Vec>, // payload + forward_batch: Vec>, // payload + deadletter_batch: Vec<(String, Vec)>, // application, payload config: ActivationBatcherConfig, runtime_config_manager: Arc, producer: Arc, @@ -74,12 +81,53 @@ impl ActivationBatcher { batch: Vec::with_capacity(config.max_batch_len), batch_size: 0, forward_batch: Vec::with_capacity(config.max_batch_len), + deadletter_batch: Vec::new(), config, runtime_config_manager, producer, producer_cluster, } } + + /// The cluster the deadletter topic lives on. The forwarding producer + /// authenticates against it, so it is also the forwarding default. + fn deadletter_cluster(&self) -> String { + self.config + .producer_config + .get("bootstrap.servers") + .expect("producer config always sets bootstrap.servers") + .to_string() + } + + /// The producer, recreated when the target cluster changed. + fn producer_for(&mut self, cluster: String) -> Arc { + if self.producer_cluster != cluster { + let mut config = self.config.producer_config.clone(); + config.set("bootstrap.servers", &cluster); + self.producer = Arc::new( + KafkaProducer::new(config, Duration::from_millis(self.config.send_timeout_ms)) + .expect("Could not create kafka producer in activation batcher"), + ); + self.producer_cluster = cluster; + } + + self.producer.clone() + } + + /// Produce every payload to `topic`, returning (attempts, successes). + async fn send_all<'a>( + &self, + producer: &KafkaProducer, + topic: &str, + payloads: impl Iterator>, + ) -> (usize, usize) { + let sends = payloads.map(|payload| producer.send(topic, payload)); + + let results = join_all(sends).await; + let successes = results.iter().filter(|r| r.is_ok()).count(); + + (results.len(), successes) + } } impl Reducer for ActivationBatcher { @@ -137,6 +185,17 @@ impl Reducer for ActivationBatcher { return Ok(()); } + // Discard tasks for applications this pool's workers do not serve. + if !self.config.applications.admits(&t.application) { + metrics::counter!( + "filter.unknown_application", + "topic" => self.config.kafka_topic.clone(), + ) + .increment(1); + self.deadletter_batch.push((t.application, t.activation)); + return Ok(()); + } + if runtime_config.demoted_namespaces.contains(namespace) { if forward_topic == self.config.kafka_topic { metrics::counter!( @@ -166,7 +225,10 @@ impl Reducer for ActivationBatcher { } async fn flush(&mut self) -> Result, anyhow::Error> { - if self.batch.is_empty() && self.forward_batch.is_empty() { + if self.batch.is_empty() + && self.forward_batch.is_empty() + && self.deadletter_batch.is_empty() + { return Ok(None); } @@ -175,58 +237,71 @@ impl Reducer for ActivationBatcher { metrics::histogram!("consumer.batch_bytes", "topic" => self.config.kafka_topic.clone()) .record(self.batch_size as f64); + let runtime_config = self.runtime_config_manager.read().await; + // Send all forward batch in parallel if !self.forward_batch.is_empty() { - let runtime_config = self.runtime_config_manager.read().await; // The forwarding producer authenticates against the deadletter // cluster, so default demoted forwarding there too (and consistently // with upkeep) when no demoted_topic_cluster is configured. - let forward_cluster = - runtime_config - .demoted_topic_cluster - .clone() - .unwrap_or_else(|| { - self.config - .producer_config - .get("bootstrap.servers") - .expect("producer config always sets bootstrap.servers") - .to_string() - }); - if self.producer_cluster != forward_cluster { - let mut new_config = self.config.producer_config.clone(); - new_config.set("bootstrap.servers", &forward_cluster); - self.producer = Arc::new( - KafkaProducer::new( - new_config, - Duration::from_millis(self.config.send_timeout_ms), - ) - .expect("Could not create kafka producer in activation batcher"), - ); - self.producer_cluster = forward_cluster; - } + let forward_cluster = runtime_config + .demoted_topic_cluster + .clone() + .unwrap_or_else(|| self.deadletter_cluster()); + let producer = self.producer_for(forward_cluster); let forward_topic = runtime_config .demoted_topic .clone() .unwrap_or(self.config.kafka_long_topic.clone()); - let sends = self - .forward_batch - .iter() - .map(|payload| self.producer.send(&forward_topic, payload)); - let results = join_all(sends).await; - let success_count = results.iter().filter(|r| r.is_ok()).count(); + let (attempts, successes) = self + .send_all(&producer, &forward_topic, self.forward_batch.iter()) + .await; let topic = self.config.kafka_topic.clone(); metrics::histogram!("consumer.forward_attempts", "topic" => topic.clone()) - .record(results.len() as f64); + .record(attempts as f64); metrics::histogram!("consumer.forward_successes", "topic" => topic.clone()) - .record(success_count as f64); + .record(successes as f64); metrics::histogram!("consumer.forward_failures", "topic" => topic) - .record((results.len() - success_count) as f64); + .record((attempts - successes) as f64); self.forward_batch.clear(); } + if !self.deadletter_batch.is_empty() { + let mut counts: BTreeMap<&str, usize> = BTreeMap::new(); + for (application, _) in &self.deadletter_batch { + *counts.entry(application.as_str()).or_default() += 1; + } + error!( + topic = self.config.kafka_topic.as_str(), + ?counts, + "Discarding activations for applications this pool does not serve" + ); + + let deadletter_cluster = self.deadletter_cluster(); + let producer = self.producer_for(deadletter_cluster); + let deadletter_topic = self.config.kafka_deadletter_topic.clone(); + let (attempts, successes) = self + .send_all( + &producer, + &deadletter_topic, + self.deadletter_batch.iter().map(|(_, payload)| payload), + ) + .await; + + let topic = self.config.kafka_topic.clone(); + metrics::histogram!("consumer.unknown_application_attempts", "topic" => topic.clone()) + .record(attempts as f64); + metrics::histogram!("consumer.unknown_application_successes", "topic" => topic.clone()) + .record(successes as f64); + metrics::histogram!("consumer.unknown_application_failures", "topic" => topic) + .record((attempts - successes) as f64); + + self.deadletter_batch.clear(); + } + self.batch_size = 0; Ok(Some(replace( @@ -238,11 +313,14 @@ impl Reducer for ActivationBatcher { fn reset(&mut self) { self.batch_size = 0; self.forward_batch.clear(); + self.deadletter_batch.clear(); self.batch.clear(); } async fn is_full(&self) -> bool { self.batch.len() >= self.config.max_batch_len + || self.forward_batch.len() >= self.config.max_batch_len + || self.deadletter_batch.len() >= self.config.max_batch_len || self.batch_size >= self.config.max_batch_size } @@ -262,11 +340,17 @@ mod tests { use std::sync::Arc; use chrono::Utc; + use prost::Message; + use sentry_protos::taskbroker::v1::TaskActivation; use tempfile::NamedTempFile; + use crate::config::deprecated::DeprecatedConfig; use crate::config::store::StoreConfig; use crate::store::activation::ActivationBuilder; - use crate::test_utils::{TaskActivationBuilder, generate_unique_namespace}; + use crate::test_utils::{ + TaskActivationBuilder, consume_topic, create_integration_config_from_base, + generate_unique_namespace, reset_topic, + }; use super::{ ActivationBatcher, ActivationBatcherConfig, Config, Reducer, RuntimeConfigManager, @@ -387,6 +471,153 @@ demoted_namespaces: []"#; assert_eq!(batcher.batch.len(), 0); } + #[tokio::test] + async fn test_discard_task_from_unserved_application() { + let runtime_config = Arc::new(RuntimeConfigManager::new(None).await); + let mut config = Config { + applications: ["sentry".to_owned()].into_iter().collect(), + ..Default::default() + }; + config.normalize_and_validate().unwrap(); + let config = Arc::new(config); + let mut batcher = ActivationBatcher::new( + ActivationBatcherConfig::from_topic(&config, config.consumable_topics().unwrap()[0].0), + runtime_config, + ); + + let namespace = generate_unique_namespace(); + + let unserved = ActivationBuilder::new() + .id("0") + .taskname("taskname") + .namespace(&namespace) + .application("launchpad") + .build(TaskActivationBuilder::new()); + + batcher.reduce(unserved).await.unwrap(); + assert_eq!(batcher.batch.len(), 0); + assert_eq!(batcher.deadletter_batch.len(), 1); + assert_eq!(batcher.deadletter_batch[0].0, "launchpad"); + + let served = ActivationBuilder::new() + .id("1") + .taskname("taskname") + .namespace(&namespace) + .application("sentry") + .build(TaskActivationBuilder::new()); + + batcher.reduce(served).await.unwrap(); + assert_eq!(batcher.batch.len(), 1); + assert_eq!(batcher.deadletter_batch.len(), 1); + } + + #[tokio::test] + async fn test_discarded_task_is_published_to_deadletter_topic() { + let topic = "taskbroker-test-unknown-application"; + let config = Arc::new(create_integration_config_from_base(Config { + deprecated: DeprecatedConfig { + kafka_topic: Some(topic.to_owned()), + ..DeprecatedConfig::default() + }, + kafka_deadletter_topic: format!("{topic}-dlq"), + applications: ["sentry".to_owned()].into_iter().collect(), + ..Default::default() + })); + reset_topic(config.clone()).await; + + let runtime_config = Arc::new(RuntimeConfigManager::new(None).await); + + let mut batcher = ActivationBatcher::new( + ActivationBatcherConfig::from_topic(&config, config.consumable_topics().unwrap()[0].0), + runtime_config, + ); + + let unserved = ActivationBuilder::new() + .id("0") + .taskname("taskname") + .namespace(generate_unique_namespace()) + .application("launchpad") + .build(TaskActivationBuilder::new()); + let expected = TaskActivation::decode(&unserved.activation as &[u8]).unwrap(); + + batcher.reduce(unserved).await.unwrap(); + assert_eq!(batcher.deadletter_batch.len(), 1); + + let flushed = batcher + .flush() + .await + .unwrap() + .expect("flush produced a batch"); + assert!(flushed.is_empty()); + assert_eq!(batcher.deadletter_batch.len(), 0); + + let messages = + consume_topic(config.clone(), config.kafka_deadletter_topic.as_ref(), 1).await; + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].id, expected.id); + assert_eq!(messages[0].application, Some("launchpad".to_owned())); + assert_eq!(messages[0].parameters_bytes, expected.parameters_bytes); + } + + /// Discards bypass `batch`, so without them `is_full` never trips and a misrouted + /// topic only flushes on the timer. + #[tokio::test] + async fn test_deadletter_batch_counts_towards_full() { + let runtime_config = Arc::new(RuntimeConfigManager::new(None).await); + let mut config = Config { + applications: ["sentry".to_owned()].into_iter().collect(), + store: StoreConfig { + insert_batch_max_length: 2, + ..StoreConfig::default() + }, + ..Default::default() + }; + config.normalize_and_validate().unwrap(); + let config = Arc::new(config); + let mut batcher = ActivationBatcher::new( + ActivationBatcherConfig::from_topic(&config, config.consumable_topics().unwrap()[0].0), + runtime_config, + ); + + let namespace = generate_unique_namespace(); + for id in 0..2 { + let unserved = ActivationBuilder::new() + .id(id.to_string()) + .taskname("taskname") + .namespace(&namespace) + .application("launchpad") + .build(TaskActivationBuilder::new()); + batcher.reduce(unserved).await.unwrap(); + } + + assert!(batcher.batch.is_empty()); + assert!(batcher.is_full().await); + } + + /// A pool that has not opted in keeps running every application it is sent. + #[tokio::test] + async fn test_unset_applications_keeps_every_task() { + let runtime_config = Arc::new(RuntimeConfigManager::new(None).await); + let mut config = Config::default(); + config.normalize_and_validate().unwrap(); + let config = Arc::new(config); + let mut batcher = ActivationBatcher::new( + ActivationBatcherConfig::from_topic(&config, config.consumable_topics().unwrap()[0].0), + runtime_config, + ); + + let activation = ActivationBuilder::new() + .id("0") + .taskname("taskname") + .namespace(generate_unique_namespace()) + .application("launchpad") + .build(TaskActivationBuilder::new()); + + batcher.reduce(activation).await.unwrap(); + assert_eq!(batcher.batch.len(), 1); + assert_eq!(batcher.deadletter_batch.len(), 0); + } + #[tokio::test] async fn test_close_by_bytes_limit() { let runtime_config = Arc::new(RuntimeConfigManager::new(None).await); diff --git a/src/push/tests.rs b/src/push/tests.rs index 7bb3dea6..b908c011 100644 --- a/src/push/tests.rs +++ b/src/push/tests.rs @@ -14,7 +14,7 @@ use crate::store::activation::{Activation, ActivationStatus}; use crate::store::traits::ActivationStore; use crate::store::types::{Claim, FailedTasksForwarder, TopicPartition}; use crate::test_utils::make_activations; -use crate::worker::test_worker_map; +use crate::worker::{WorkerMap, test_worker_map}; use super::*; @@ -22,12 +22,17 @@ use super::*; #[derive(Default, Clone)] struct MockStore { marked_processing: Arc>>, + status_updates: Arc>>, } impl MockStore { fn marked_ids(&self) -> Vec { self.marked_processing.lock().unwrap().clone() } + + fn recorded_statuses(&self) -> Vec<(String, ActivationStatus)> { + self.status_updates.lock().unwrap().clone() + } } #[async_trait] @@ -72,11 +77,16 @@ impl ActivationStore for MockStore { async fn set_status( &self, - _id: &str, - _status: ActivationStatus, + id: &str, + status: ActivationStatus, _max_attempts: Option, _delay_on_retry: Option, ) -> Result> { + self.status_updates + .lock() + .unwrap() + .push((id.to_string(), status)); + Ok(None) } @@ -339,3 +349,50 @@ async fn push_pool_start_does_not_mark_activation_processing_on_push_failure() { "mark_activation_processing should not be called when push fails" ); } + +/// An activation the pool has no worker for must reach a terminal status, otherwise +/// the claim expires and it is fetched again forever. +#[tokio::test] +async fn push_pool_start_fails_activation_without_worker_mapping() { + let config = Arc::new(Config { + worker_map: [("sentry".into(), "unused".into())].into(), + push: PushConfig { + threads: 1, + queue: PushQueueConfig { + size: 10, + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }); + + let store = Arc::new(MockStore::default()); + let (sender, receiver) = flume::bounded(config.push.queue.size); + let pool = Arc::new(PushPool::new(receiver, config)); + + let updater = test_eager_updater(store.clone()); + + tokio::spawn({ + let store = store.clone(); + async move { + pool.start(vec![WorkerMap::new()], updater, store.clone()) + .await + } + }); + + let activation = make_activations(1).remove(0); + let id = activation.id.clone(); + + sender + .send_async((activation, Instant::now())) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + + assert_eq!( + store.recorded_statuses(), + vec![(id, ActivationStatus::Failure)] + ); + assert!(store.marked_ids().is_empty()); +} diff --git a/src/push/thread.rs b/src/push/thread.rs index fd3b828a..e5411289 100644 --- a/src/push/thread.rs +++ b/src/push/thread.rs @@ -9,7 +9,7 @@ use flume::Receiver; use tracing::{debug, error}; use crate::push::updater::Updater; -use crate::store::activation::Activation; +use crate::store::activation::{Activation, ActivationStatus}; use crate::store::traits::ActivationStore; use crate::timed; use crate::worker::WorkerMap; @@ -68,6 +68,19 @@ impl PushThread { "Application has no worker mapping" ); + // Fail the task so upkeep discards it instead of the claim expiring. + if let Err(e) = self + .store + .set_status(id, ActivationStatus::Failure, None, None) + .await + { + error!( + task_id = %id, + error = ?e, + "Failed to fail activation without a worker mapping" + ); + } + return; };