Remove the need for metadata db in eventhub group peer (#983)

We already use catalog in cases where this isn't passed, but validation
failed for eventhub group peer.

nexus/analyzer/src/lib.rs

@@ -663,11 +663,8 @@ fn parse_db_options(
             Some(config)
         }
         DbType::Eventhub => {
-            let conn_str: String = opts
-                .get("metadata_db")
-                .map(|s| s.to_string())
-                .unwrap_or_default();
-            let metadata_db = parse_metadata_db_info(&conn_str)?;
+            let conn_str = opts.get("metadata_db");
+            let metadata_db = parse_metadata_db_info(conn_str.copied())?;
             let subscription_id = opts
                 .get("subscription_id")
                 .map(|s| s.to_string())
@@ -711,11 +708,8 @@ fn parse_db_options(
             Some(config)
         }
         DbType::S3 => {
-            let s3_conn_str: String = opts
-                .get("metadata_db")
-                .map(|s| s.to_string())
-                .unwrap_or_default();
-            let metadata_db = parse_metadata_db_info(&s3_conn_str)?;
+            let s3_conn_str = opts.get("metadata_db");
+            let metadata_db = parse_metadata_db_info(s3_conn_str.copied())?;
             let s3_config = S3Config {
                 url: opts
                     .get("url")
@@ -754,10 +748,8 @@ fn parse_db_options(
             Some(config)
         }
         DbType::EventhubGroup => {
-            let conn_str = opts
-                .get("metadata_db")
-                .context("no metadata db specified")?;
-            let metadata_db = parse_metadata_db_info(conn_str)?;
+            let conn_str = opts.get("metadata_db");
+            let metadata_db = parse_metadata_db_info(conn_str.copied())?;
 
             // metadata_db is required for eventhub group
             if metadata_db.is_none() {
@@ -808,7 +800,12 @@ fn parse_db_options(
     Ok(config)
 }
 
-fn parse_metadata_db_info(conn_str: &str) -> anyhow::Result<Option<PostgresConfig>> {
+fn parse_metadata_db_info(conn_str: Option<&str>) -> anyhow::Result<Option<PostgresConfig>> {
+    let conn_str = match conn_str {
+        Some(conn_str) => conn_str,
+        None => return Ok(None),
+    };
+
     if conn_str.is_empty() {
         return Ok(None);
     }

nexus/flow-rs/src/grpc.rs

@@ -113,7 +113,7 @@ impl FlowGrpcClient {
             requested_flow_state: state.into(),
             source_peer: Some(workflow_details.source_peer),
             destination_peer: Some(workflow_details.destination_peer),
-            flow_state_update: None
+            flow_state_update: None,
         };
         let response = self.client.flow_state_change(state_change_req).await?;
         let state_change_response = response.into_inner();