various unrelated small fixes and tweaks

crates/agent/src/proxy_connectors.rs

@@ -201,7 +201,7 @@ impl<L: runtime::LogHandler> ProxyConnectors<L> {
 
         let mut proxy_client =
             proto_grpc::runtime::connector_proxy_client::ConnectorProxyClient::with_interceptor(
-                gazette::dial_channel(reactor_address).await?,
+                gazette::dial_channel(reactor_address)?,
                 metadata.clone(),
             );
         let mut proxy_responses = proxy_client
@@ -238,7 +238,7 @@ impl<L: runtime::LogHandler> ProxyConnectors<L> {
         };
 
         Ok((
-            gazette::dial_channel(&address).await?,
+            gazette::dial_channel(&address)?,
             metadata,
             (cancel_tx, log_loop),
         ))

crates/data-plane-controller/src/controller.rs

@@ -1,4 +1,4 @@
-use super::{run_cmd, stack};
+use super::{logs, run_cmd, stack};
 use crate::repo;
 use anyhow::Context;
 use itertools::{EitherOrBoth, Itertools};
@@ -518,6 +518,15 @@ impl Controller {
         )
         .await?;
 
+        self.logs_tx
+            .send(logs::Line {
+                token: state.logs_token,
+                stream: "controller".to_string(),
+                line: format!("Waiting {DNS_TTL:?} for DNS propagation before continuing."),
+            })
+            .await
+            .context("failed to send to logs sink")?;
+
         state.status = Status::AwaitDNS1;
         state.last_pulumi_up = chrono::Utc::now();
 
@@ -672,6 +681,15 @@ impl Controller {
         )
         .await?;
 
+        self.logs_tx
+            .send(logs::Line {
+                token: state.logs_token,
+                stream: "controller".to_string(),
+                line: format!("Waiting {DNS_TTL:?} for DNS propagation before continuing."),
+            })
+            .await
+            .context("failed to send to logs sink")?;
+
         state.status = Status::AwaitDNS2;
         state.last_pulumi_up = chrono::Utc::now();
 

crates/data-plane-controller/src/lib.rs

@@ -23,7 +23,7 @@ pub struct Args {
     #[clap(long = "database-ca", env = "DPC_DATABASE_CA")]
     database_ca: Option<String>,
     /// Number of tasks which may be polled concurrently.
-    #[clap(long = "concurrency", env = "DPC_CONCURRENCY", default_value = "1")]
+    #[clap(long = "concurrency", env = "DPC_CONCURRENCY", default_value = "2")]
     concurrency: u32,
     /// Interval between polls for dequeue-able tasks when otherwise idle.
     #[clap(
@@ -39,7 +39,7 @@ pub struct Args {
     #[clap(
         long = "heartbeat-timeout",
         env = "DPC_HEARTBEAT_TIMEOUT",
-        default_value = "30s"
+        default_value = "60s"
     )]
     #[serde(with = "humantime_serde")]
     #[arg(value_parser = humantime::parse_duration)]

crates/derive-sqlite/src/snapshots/derive_sqlite__validate__test__stub_generation.snap

@@ -0,0 +1,15 @@
+---
+source: crates/derive-sqlite/src/validate.rs
+expression: validated
+---
+{
+  "transforms": [
+    {
+      "readOnly": true
+    }
+  ],
+  "generatedFiles": {
+    "file://path/to/migration.stub": "\n-- Use migrations to create or alter tables that your derivation will use.\n-- Each migration is run only once, and new migrations will be applied as needed.\n--\n-- For example, create the join table below, and then use it across multiple lambdas:\n--\n-- A first lambda that updates indexed state:\n--\n--   INSERT INTO my_join_table (id, joined_value) VALUES ($id, $my::value)\n--     ON CONFLICT REPLACE;\n--\n-- A second lambda that reads out and joins over the indexed state:\n--\n--    SELECT $id, $other$value, j.joined_value FROM my_join_table WHERE id = $id;\n\nCREATE TABLE my_join_table (\n    -- A common ID that's joined over.\n    id           INTEGER PRIMARY KEY NOT NULL,\n    -- A value that's updated by one lambda, and read by another.\n    joined_value TEXT NOT NULL\n);\n\n",
+    "file://path/to/transform.stub.sql": "\n-- Example statement which passes-through source acmeCo/foo/bar documents without modification.\n-- Use a WHERE clause to filter, for example: WHERE $my$column = 1234\nSELECT JSON($flow_document);\n\n-- Example statement demonstrating how to SELECT specific locations from documents of acmeCo/foo/bar.\n-- This statement is effectively disabled by its WHERE FALSE clause and does not emit any documents.\n--\n-- You can rename a location by using the SQL \"AS\" syntax, for example:\n--   SELECT $some$column AS \"my_new_column_name;\n--\n-- You can also filter by using locations in a WHERE clause, for example:\n--   SELECT $some$column WHERE $other$column = 1234;\nSELECT\n    -- Key id at /id\n    $id,\n    -- Partitioned field part at /part\n    $part,\n    -- Field nested/int at /nested/int\n    $nested$int,\n    -- Field value at /value\n    $value\n-- Disable this statement, so that it emits no documents.\nWHERE FALSE;\n"
+  }
+}

crates/derive-sqlite/src/validate.rs

@@ -107,20 +107,20 @@ const MIGRATION_STUB: &str = r#"
 --
 -- A second lambda that reads out and joins over the indexed state:
 --
---    SELECT $id, $other::value, j.joined_value FROM my_join_table WHERE id = $id;
+--    SELECT $id, $other$value, j.joined_value FROM my_join_table WHERE id = $id;
 
-create table my_join_table (
+CREATE TABLE my_join_table (
     -- A common ID that's joined over.
-    id           integer primary key not null,
+    id           INTEGER PRIMARY KEY NOT NULL,
     -- A value that's updated by one lambda, and read by another.
-    joined_value text not null
+    joined_value TEXT NOT NULL
 );
 
 "#;
 
 fn lambda_stub(
     Transform {
-        name,
+        name: _,
         source,
         block: _,
         params,
@@ -129,11 +129,32 @@ fn lambda_stub(
     use std::fmt::Write;
     let mut w = String::with_capacity(4096);
 
+    let root = params.iter().find(|param| param.projection.ptr.is_empty());
+
+    if let Some(root) = root {
+        _ = write!(
+            w,
+            r#"
+-- Example statement which passes-through source {source} documents without modification.
+-- Use a WHERE clause to filter, for example: WHERE $my$column = 1234
+SELECT JSON({});
+"#,
+            root.canonical_encoding
+        );
+    }
+
     _ = write!(
         w,
         r#"
--- Example select statement for transform {name} of source collection {source}.
-select
+-- Example statement demonstrating how to SELECT specific locations from documents of {source}.
+-- This statement is effectively disabled by its WHERE FALSE clause and does not emit any documents.
+--
+-- You can rename a location by using the SQL "AS" syntax, for example:
+--   SELECT $some$column AS "my_new_column_name;
+--
+-- You can also filter by using locations in a WHERE clause, for example:
+--   SELECT $some$column WHERE $other$column = 1234;
+SELECT
 "#,
     );
 
@@ -167,11 +188,62 @@ select
 
     let params: Vec<String> = params
         .iter()
-        .map(|p| format!("    -- {}\n    {}", comment(p), p.canonical_encoding))
+        .filter_map(|p| {
+            if p.projection.ptr.is_empty() {
+                None // Skip projection of the document root.
+            } else {
+                Some(format!(
+                    "    -- {}\n    {}",
+                    comment(p),
+                    p.canonical_encoding
+                ))
+            }
+        })
         .collect();
 
     w.push_str(&params.join(",\n"));
-    w.push_str("\n;");
+    w.push_str(
+        r#"
+-- Disable this statement, so that it emits no documents.
+WHERE FALSE;
+"#,
+    );
 
     w
 }
+
+#[cfg(test)]
+mod test {
+    use super::super::test_param;
+    use super::do_validate;
+    use crate::Transform;
+
+    #[test]
+    fn test_stub_generation() {
+        let mut params = vec![
+            test_param("id", "/id", false, false, false),
+            test_param("part", "/part", false, false, false),
+            test_param("flow_document", "", false, false, false),
+            test_param("nested/int", "/nested/int", false, true, false),
+            test_param("value", "/value", false, false, false),
+        ];
+        params[0].projection.is_primary_key = true;
+        params[1].projection.is_partition_key = true;
+
+        let migrations = vec![
+            "file://path/to/migration.stub".to_string(),
+            "CREATE TABLE foo (one TEXT, two INTEGER);".to_string(),
+        ];
+
+        let transforms = vec![Transform {
+            block: "file://path/to/transform.stub.sql".to_string(),
+            name: "fromFoobar".to_string(),
+            source: "acmeCo/foo/bar".to_string(),
+            params,
+        }];
+
+        let validated = do_validate(&migrations, &transforms).unwrap();
+
+        insta::assert_json_snapshot!(validated);
+    }
+}

crates/flowctl/src/poll.rs

@@ -12,7 +12,7 @@ pub async fn poll_while_queued(
 
     tokio::select! {
         outcome = poll_table(client, table, id) => return outcome,
-        result = stream_logs(client, logs_token) => return Err(result.unwrap_err()),
+        result = stream_logs(client, logs_token, None) => return Err(result.unwrap_err()),
     };
 }
 
@@ -47,9 +47,11 @@ pub async fn poll_table(
     Ok(outcome)
 }
 
-pub async fn stream_logs(client: &crate::Client, logs_token: &str) -> anyhow::Result<()> {
-    let mut last_logged_at = None;
-
+pub async fn stream_logs(
+    client: &crate::Client,
+    logs_token: &str,
+    mut last_logged_at: Option<crate::Timestamp>,
+) -> anyhow::Result<()> {
     loop {
         tokio::time::sleep(std::time::Duration::from_millis(500)).await;
 

crates/flowctl/src/raw/mod.rs

@@ -180,11 +180,27 @@ pub struct BearerLogs {
     /// Bearer logs token.
     #[clap(long)]
     pub token: uuid::Uuid,
+    /// Start reading from this far in the past.
+    #[clap(long, default_value = "1h")]
+    pub since: Option<humantime::Duration>,
 }
 
 impl BearerLogs {
     pub async fn run(&self, ctx: &mut crate::CliContext) -> anyhow::Result<()> {
-        crate::poll::stream_logs(&ctx.client, &self.token.to_string()).await
+        let bound = match self.since {
+            None => None,
+            Some(since) => {
+                let since: std::time::Duration = since.into();
+                Some(crate::Timestamp::from_unix_timestamp(
+                    (std::time::SystemTime::now() - since)
+                        .duration_since(std::time::SystemTime::UNIX_EPOCH)
+                        .unwrap()
+                        .as_secs() as i64,
+                )?)
+            }
+        };
+
+        crate::poll::stream_logs(&ctx.client, &self.token.to_string(), bound).await
     }
 }
 

crates/gazette/Cargo.toml

@@ -20,6 +20,7 @@ futures = { workspace = true }
 futures-core = { workspace = true }
 hyper-util = { workspace = true }
 jsonwebtoken = { workspace = true }
+rand = { workspace = true }
 reqwest = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }

crates/gazette/src/journal/list.rs

@@ -51,7 +51,7 @@ impl Client {
         router: &crate::Router,
         req: &broker::ListRequest,
     ) -> crate::Result<tonic::Streaming<broker::ListResponse>> {
-        let mut client = self.into_sub(router.route(None, false, &self.default).await?);
+        let mut client = self.into_sub(router.route(None, false, &self.default)?);
         Ok(client.list(req.clone()).await?.into_inner())
     }
 }

crates/gazette/src/journal/mod.rs

@@ -50,7 +50,7 @@ impl Client {
 
     /// Invoke the Gazette journal Apply API.
     pub async fn apply(&self, req: broker::ApplyRequest) -> crate::Result<broker::ApplyResponse> {
-        let mut client = self.into_sub(self.router.route(None, false, &self.default).await?);
+        let mut client = self.into_sub(self.router.route(None, false, &self.default)?);
 
         let resp = client
             .apply(req)
@@ -66,7 +66,7 @@ impl Client {
         &self,
         req: broker::FragmentsRequest,
     ) -> crate::Result<broker::FragmentsResponse> {
-        let mut client = self.into_sub(self.router.route(None, false, &self.default).await?);
+        let mut client = self.into_sub(self.router.route(None, false, &self.default)?);
 
         let resp = client
             .list_fragments(req)
@@ -77,11 +77,13 @@ impl Client {
         check_ok(resp.status(), resp)
     }
 
-    fn into_sub(&self, channel: Channel) -> SubClient {
+    fn into_sub(&self, (channel, _local): (Channel, bool)) -> SubClient {
         proto_grpc::broker::journal_client::JournalClient::with_interceptor(
             channel,
             self.metadata.clone(),
         )
+        // TODO(johnny): Use `_local` to selectively enable LZ4 compression
+        // when traversing a non-local zone.
     }
 }
 

crates/gazette/src/journal/read.rs

@@ -36,6 +36,8 @@ impl Client {
                         // Surface error to the caller, which can either drop us
                         // or poll us again to retry.
                         () = co.yield_(Err(err)).await;
+                        // Restart route discovery.
+                        req.header = None;
                     }
                 }
             }
@@ -49,7 +51,7 @@ impl Client {
         write_head: &mut i64,
     ) -> crate::Result<()> {
         let route = req.header.as_ref().and_then(|hdr| hdr.route.as_ref());
-        let mut client = self.into_sub(self.router.route(route, false, &self.default).await?);
+        let mut client = self.into_sub(self.router.route(route, false, &self.default)?);
 
         // Fetch metadata first before we start the actual read.
         req.metadata_only = true;
@@ -89,7 +91,9 @@ impl Client {
             }
             match (resp.status(), &resp.fragment, resp.content.is_empty()) {
                 // Metadata response telling us of a new fragment being read.
-                (broker::Status::Ok, Some(_fragment), true) => {
+                (broker::Status::Ok, Some(fragment), true) => {
+                    tracing::trace!(fragment=?ops::DebugJson(fragment), "read fragment metadata");
+
                     // Offset jumps happen if content is removed from the middle of a journal,
                     // or when reading from the journal head (offset -1).
                     if req.offset != resp.offset {
@@ -138,6 +142,8 @@ async fn read_fragment_url(
         .and_then(reqwest::Response::error_for_status)
         .map_err(Error::FetchFragment)?;
 
+    tracing::trace!(fragment=?ops::DebugJson(&fragment), "started direct fragment read");
+
     let raw_reader = response
         // Map into a Stream<Item = Result<Bytes, _>>.
         .bytes_stream()