rust connectors mega commit

Cargo.toml

@@ -159,6 +159,7 @@ tracing-subscriber = { version = "0.3", features = [
     "fmt",
 ] }
 shared_child = "1.0.0"
+zeroize = "1.6"
 
 unicode-bom = "1.1"
 unicode-normalization = "0.1"

Makefile

@@ -300,6 +300,7 @@ install-tools: ${PKGDIR}/bin/deno ${PKGDIR}/bin/etcd ${PKGDIR}/bin/sops
 
 .PHONY: rust-gnu-test
 rust-gnu-test:
+	PATH=${PKGDIR}/bin:$$PATH ;\
 	cargo test --release --locked --workspace --exclude parser --exclude network-tunnel --exclude schemalate --exclude connector-init
 
 .PHONY: rust-musl-test

crates/agent/Cargo.toml

@@ -13,10 +13,13 @@ license.workspace = true
 [dependencies]
 agent-sql = { path = "../agent-sql" }
 async-process = { path = "../async-process" }
+build = { path = "../build" }
 doc = { path = "../doc" }
 models = { path = "../models" }
 proto-flow = { path = "../proto-flow" }
+runtime = { path = "../runtime" }
 tables = { path = "../tables", features = ["persist"] }
+validation = { path = "../validation" }
 
 anyhow = { workspace = true }
 async-trait = { workspace = true }
@@ -29,7 +32,6 @@ itertools = { workspace = true }
 lazy_static = { workspace = true }
 regex = { workspace = true }
 reqwest = { workspace = true }
-rusqlite = { workspace = true }
 schemars = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

crates/agent/src/logs.rs

@@ -16,6 +16,43 @@ pub struct Line {
 // Tx is the channel sender of log Lines.
 pub type Tx = tokio::sync::mpsc::Sender<Line>;
 
+impl Line {
+    pub(crate) fn new(token: Uuid, stream: String, line: String) -> Self {
+        Line {
+            token,
+            stream,
+            line,
+        }
+    }
+}
+
+/// ops_handler is a ops::Log handler that dispatches to `tx`
+/// using the given `stream` and `token`.
+pub fn ops_handler(
+    tx: Tx,
+    stream: String,
+    token: Uuid,
+) -> impl Fn(&proto_flow::ops::Log) + Send + Sync + Clone + 'static {
+    let log_handler = move |log: &proto_flow::ops::Log| {
+        // TODO(johnny): format into a user-facing representation for our UI.
+        let line = serde_json::to_string(log).unwrap();
+
+        let Err(tokio::sync::mpsc::error::TrySendError::Full(line)) = tx.try_send(Line {
+            token: token.clone(),
+            stream: stream.clone(),
+            line,
+        }) else { return };
+
+        // Perform an expensive "move" of all other tasks scheduled on the
+        // current async executor thread, so that we can block until there's capacity.
+        let tx_clone = tx.clone();
+        _ = tokio::task::block_in_place(move || {
+            tokio::runtime::Handle::current().block_on(tx_clone.send(line))
+        });
+    };
+    log_handler
+}
+
 // capture_job_logs consumes newline-delimited lines from the AsyncRead and
 // streams each as a Line to the channel Sender.
 #[tracing::instrument(err, skip(tx, reader))]
@@ -35,13 +72,9 @@ where
         let line = String::from_utf8(line)
             .unwrap_or_else(|err| String::from_utf8_lossy(err.as_bytes()).into_owned());
 
-        tx.send(Line {
-            token,
-            stream: stream.clone(),
-            line,
-        })
-        .await
-        .unwrap();
+        tx.send(Line::new(token, stream.clone(), line))
+            .await
+            .unwrap();
     }
     Ok(())
 }

crates/agent/src/publications.rs

@@ -301,7 +301,6 @@ impl PublishHandler {
             &self.builds_root,
             &draft_catalog,
             &self.connector_network,
-            &self.bindir,
             row.logs_token,
             &self.logs_tx,
             row.pub_id,