storage-client: don't panic if reading Kafka/Postgres secret fails

The surrounding code paths have been sufficiently refactored such that
it is now possible to gracefully return an error here.

Fix #18438.

src/storage-client/src/types/connections.rs

@@ -349,7 +349,7 @@ impl KafkaConnection {
                 k,
                 v.get_string(&*connection_context.secrets_reader)
                     .await
-                    .expect("reading kafka secret unexpectedly failed"),
+                    .context("reading kafka secret")?,
             );
         }
         for (k, v) in extra_options {

src/storage/src/source/postgres.rs

@@ -294,11 +294,28 @@ impl SourceRender for PostgresSourceConnection {
 
             let resume_lsn = Arc::new(AtomicU64::new(start_offset.offset));
 
-            let connection_config = self
+            let connection_config = match self
                 .connection
                 .config(&*connection_context.secrets_reader)
                 .await
-                .expect("Postgres connection unexpectedly missing secrets");
+            {
+                Ok(config) => config,
+                Err(e) => {
+                    let update = HealthStatusUpdate {
+                        update: HealthStatus::StalledWithError {
+                            error: format!("failed creating postgres config: {e:#}"),
+                            hint: None,
+                        },
+                        should_halt: true,
+                    };
+                    health_output.give(&health_capability, update).await;
+                    // IMPORTANT: wedge forever until the `SuspendAndRestart` is processed.
+                    // Returning would incorrectly present to the remap operator as progress to the
+                    // empty frontier which would be incorrectly recorded to the remap shard.
+                    std::future::pending::<()>().await;
+                    unreachable!("pending future never returns");
+                }
+            };
 
             let mut source_tables = BTreeMap::new();
             let tables_iter = self.publication_details.tables.iter();