From 40ea053137822c8ea94a7b54104481d6828ea3a2 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 31 Jul 2026 23:38:50 -0700 Subject: [PATCH 1/2] [adapters] Start input endpoints whose reader arrives late The backpressure thread recorded an endpoint id in `running_endpoints` before checking for that endpoint's reader, so an endpoint that was registered but not yet opened lost its only `extend()`: `insert` returned false on every later scan, leaving the endpoint paused forever. Skip a reader-less endpoint instead, and the scan that follows `add_input_endpoint`'s unpark starts it. `add_input_endpoint` registers an endpoint with `reader: None` and installs the reader only after `open` returns, a window open since d24ba4024. Only endpoints added to a running pipeline can land in that window, because the connectors named in the pipeline configuration are all created before the backpressure thread starts. Every HTTP `/ingress` push creates such an endpoint, and the endpoint blocks in `complete_request` until `extend` arrives, so a lost command hangs the request until the client gives up. That is how `server::test_http::test_concurrent_bootstrap` failed in CI, where the client timed out after 120 seconds, and the same window can hang an ingress request in production. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/controller.rs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index 78399fbcb9e..22064321cb2 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() } } From 426cbdfbf70bd85434ec068ce2cfd1d77bedad7a Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Sat, 1 Aug 2026 00:05:04 -0700 Subject: [PATCH 2/2] [adapters] Unregister an input endpoint rejected for weak fault tolerance `add_input_endpoint` compares the endpoint's fault tolerance with the pipeline's only after it has registered and opened the endpoint, and the rejection path returned the error without undoing either step. The endpoint stayed in the status map, where it could never run because the error path also skipped the unpark that starts it, yet it still counted toward the pipeline's statistics and reserved its own name against a later attempt to add the connector. Disconnect it instead, which both removes the entry and shuts the reader down. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/controller.rs | 5 ++ crates/adapters/src/controller/test.rs | 88 +++++++++++++++++++++++++- 2 files changed, 92 insertions(+), 1 deletion(-) diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index 22064321cb2..f0235b8df6e 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -7426,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(&[