diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index 78399fbcb9e..f0235b8df6e 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -5902,6 +5902,14 @@ impl BackpressureThread { == ConcurrentBootstrapPhase::Synchronizing; for (epid, ep) in controller.status.input_status().iter() { + // `add_input_endpoint` registers an endpoint before `open` + // hands back its reader, so an endpoint can show up here with + // no reader yet. Skip it: recording the id in + // `running_endpoints` now would swallow the endpoint's one and + // only `extend()` and leave it paused forever. + let Some(reader) = ep.reader.as_ref() else { + continue; + }; let should_run = globally_running && !bootstrap_in_progress && !concurrent_synchronize @@ -5909,16 +5917,12 @@ impl BackpressureThread { && !ep.is_full(); match should_run { true => { - if running_endpoints.insert(*epid) - && let Some(reader) = ep.reader.as_ref() - { + if running_endpoints.insert(*epid) { reader.extend() } } false => { - if running_endpoints.remove(epid) - && let Some(reader) = ep.reader.as_ref() - { + if running_endpoints.remove(epid) { reader.pause() } } @@ -7422,6 +7426,11 @@ impl ControllerInner { }; if fault_tolerance < self.fault_tolerance { + // The endpoint is registered and open by now, so undo both. A + // rejected endpoint left in the map can never run, yet it counts + // toward the pipeline's statistics and reserves its own name + // against a later attempt to add the connector. + self.disconnect_input(&endpoint_id); return Err(ControllerError::input_transport_error( endpoint_name, true, diff --git a/crates/adapters/src/controller/test.rs b/crates/adapters/src/controller/test.rs index 24745d020a4..e839a41b260 100644 --- a/crates/adapters/src/controller/test.rs +++ b/crates/adapters/src/controller/test.rs @@ -6,7 +6,7 @@ use crate::{ test::{ DEFAULT_TIMEOUT_MS, TestStruct, generate_test_batch, init_test_logger, test_circuit, wait, }, - transport::set_barrier, + transport::{input_transport_config_to_endpoint, set_barrier}, }; use anyhow::anyhow; use crossbeam::sync::Parker; @@ -959,6 +959,92 @@ fn test_connector_init_error() { assert!(result.is_err()); } +/// A connector rejected by the fault-tolerance check must leave nothing behind. +/// +/// `add_input_endpoint` registers and opens the endpoint before it compares the +/// endpoint's fault tolerance with the pipeline's, so the rejection path has to +/// undo both. It used to return the error while leaving the endpoint in the +/// status map, where the endpoint could never run yet still reserved its own +/// name against a retry. +#[test] +fn fault_tolerance_mismatch_unregisters_the_endpoint() { + init_test_logger(); + + let tempdir = TempDir::new().unwrap(); + let storage_dir = tempdir.path().join("storage"); + create_dir(&storage_dir).unwrap(); + let input_file = tempdir.path().join("input.csv"); + File::create(&input_file).unwrap(); + + // `file_input` reports at-least-once for a path with a barrier, weaker than + // this pipeline's exactly-once requirement. + set_barrier(input_file.to_str().unwrap(), 0); + + let config: PipelineConfig = serde_json::from_value(json!({ + "name": "test", + "workers": 4, + "storage_config": { + "path": storage_dir, + }, + "storage": true, + "fault_tolerance": {}, + "clock_resolution_usecs": null, + })) + .unwrap(); + + let controller = Controller::with_test_config( + |circuit_config| { + Ok(test_circuit::( + circuit_config, + &TestStruct::schema(), + &[None], + )) + }, + &config, + Box::new(|e, _| panic!("error: {e}")), + ) + .unwrap(); + + // `add_input_endpoint` refuses to run while the pipeline is still restoring, + // which on a fault-tolerant pipeline is the state it starts in. + controller.start(); + wait(|| !controller.is_replaying(), DEFAULT_TIMEOUT_MS).unwrap(); + + let connector = json!({ + "name": "file_input", + "config": { + "path": input_file, + "follow": true + } + }); + let endpoint = input_transport_config_to_endpoint( + &serde_json::from_value(connector.clone()).unwrap(), + "weak_ft", + tempdir.path(), + ) + .unwrap() + .unwrap(); + let endpoint_config: InputEndpointConfig = serde_json::from_value(json!({ + "stream": "test_input1", + "transport": connector, + "format": { "name": "csv" } + })) + .unwrap(); + + let error = controller + .add_input_endpoint("weak_ft", endpoint_config, endpoint, None) + .expect_err("an at-least-once connector must be rejected by an exactly-once pipeline"); + assert!(error.to_string().contains("fault tolerance"), "{error}"); + + assert!( + controller.input_endpoint_id_by_name("weak_ft").is_err(), + "the rejected endpoint stayed registered" + ); + assert!(controller.status().input_status().is_empty()); + + controller.stop().unwrap(); +} + #[test] fn ft_with_checkpoints() { test_ft(&[