[adapters] Add ability to support connector orchestration for multihost.

blp · blp · commit 1d9c024abe52 · 2026-01-22T00:19:22.000Z
Signed-off-by: Ben Pfaff &lt;blp@feldera.com&gt;
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -125,6 +125,7 @@ use std::{
     thread::JoinHandle,
     time::{Duration, Instant},
 };
+use tokio::sync::Notify;
 use tokio::{
     fs::File,
     io::{AsyncReadExt, BufReader},
@@ -1722,6 +1723,20 @@ impl Controller {
             .map(|(_step, snapshot)| snapshot.clone())
     }
 
+    pub fn incomplete_labels(&self) -> HashSet<String> {
+        let mut incomplete_labels = HashSet::new();
+        for status in self.inner.status.inputs.read().values() {
+            if !status.metrics.end_of_input.load(Ordering::Relaxed) {
+                incomplete_labels.extend(status.config.connector_config.labels.iter().cloned());
+            }
+        }
+        incomplete_labels
+    }
+
+    pub fn input_completion_notify(&self) -> Arc<Notify> {
+        self.inner.input_completion_notify.clone()
+    }
+
     pub fn adhoc_catalog(&self) -> AdHocCatalog {
         let mut tables = Vec::new();
         for (name, clh) in self.inner.catalog.output_iter() {
@@ -4724,6 +4739,7 @@ pub struct ControllerInner {
     transaction_sender: tokio::sync::watch::Sender<TransactionCoordination>,
     coordination_request: Mutex<Option<StepRequest>>,
     coordination_prepare_checkpoint: AtomicBool,
+    input_completion_notify: Arc<Notify>,
     // The mutex is acquired from async context by actix and
     // from the sync context by the circuit thread.
     transaction_info: Mutex<TransactionInfo>,
@@ -4807,6 +4823,7 @@ impl ControllerInner {
                 transaction_receiver,
                 coordination_request: Mutex::new(is_multihost.then_some(StepRequest::new_idle(0))),
                 coordination_prepare_checkpoint: AtomicBool::new(false),
+                input_completion_notify: Arc::new(Notify::new()),
             }
         });
 
@@ -5773,7 +5790,8 @@ impl ControllerInner {
             endpoint_id,
             &self.circuit_thread_unparker,
             &self.backpressure_thread_unparker,
-        )
+        );
+        self.input_completion_notify.notify_waiters();
     }
 
     fn output_buffers_full(&self) -> bool {
diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs
@@ -60,7 +60,7 @@ use feldera_types::completion_token::{
     CompletionStatusArgs, CompletionStatusResponse, CompletionTokenResponse,
 };
 use feldera_types::constants::STATUS_FILE;
-use feldera_types::coordination::{AdHocScan, CoordinationActivate, Step, StepRequest};
+use feldera_types::coordination::{AdHocScan, CoordinationActivate, Labels, Step, StepRequest};
 use feldera_types::pipeline_diff::PipelineDiff;
 use feldera_types::query_params::{
     ActivateParams, MetricsFormat, MetricsParameters, SamplyProfileParams,
@@ -1203,6 +1203,7 @@ where
         .service(coordination_adhoc_catalog)
         .service(coordination_adhoc_lease)
         .service(coordination_adhoc_scan)
+        .service(coordination_labels_incomplete)
 }
 
 /// Implements `/start`, `/pause`, `/activate`:
@@ -2442,6 +2443,30 @@ async fn coordination_adhoc_scan(
     ))
 }
 
+/// Stream the set of incomplete labels.
+#[get("/coordination/labels/incomplete")]
+async fn coordination_labels_incomplete(
+    state: WebData<ServerState>,
+) -> Result<HttpResponse, PipelineError> {
+    let controller = state.controller()?;
+    let notify = controller.input_completion_notify();
+
+    let response_stream = async_stream::stream! {
+        loop {
+            let notify = notify.notified();
+            let labels = format!("{}\n", serde_json::to_string(&Labels {
+                incomplete: controller.incomplete_labels()
+            }).unwrap());
+            yield Ok::<_, actix_web::Error>(web::Bytes::from(labels));
+            notify.await;
+        }
+    };
+
+    Ok(HttpResponse::Ok()
+        .content_type("application/x-ndjson")
+        .streaming(response_stream))
+}
+
 #[derive(Copy, Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
 struct StoredStatus {
     /// Desired status.
diff --git a/crates/feldera-types/src/coordination.rs b/crates/feldera-types/src/coordination.rs
@@ -53,7 +53,7 @@
 
 use std::{
     borrow::Cow,
-    collections::{BTreeMap, HashMap},
+    collections::{BTreeMap, HashMap, HashSet},
     net::SocketAddr,
 };
 
@@ -268,3 +268,10 @@ pub struct AdHocScan {
     /// Columnar projection.
     pub projection: Option<Vec<usize>>,
 }
+
+/// `/coordination/labels/incomplete` reply, streamed from pipeline to coordinator.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Labels {
+    /// All the labels on incomplete connectors.
+    pub incomplete: HashSet<String>,
+}
