feldera · ryzhyk · Aug 1, 2026 · Aug 1, 2026 · Aug 1, 2026
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -5902,23 +5902,27 @@ 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
                     && !ep.is_paused_by_user()
                     && !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
@@ -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::<TestStruct>(
+                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(&[