storage: configure pg connections the same as storage during purifica…

…tion
MaterializeInc · Dec 12, 2023 · cbcbad5 · cbcbad5
1 parent f9a51d3
commit cbcbad5
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 5 deletions.
diff --git a/src/postgres-util/src/tunnel.rs b/src/postgres-util/src/tunnel.rs
@@ -90,8 +90,12 @@ pub struct Config {
 }
 
 impl Config {
-    pub fn new(inner: tokio_postgres::Config, tunnel: TunnelConfig) -> Result<Self, PostgresError> {
-        let config = Self { inner, tunnel }.tcp_timeouts(TcpTimeoutConfig::default());
+    pub fn new(
+        inner: tokio_postgres::Config,
+        tunnel: TunnelConfig,
+        tcp_timeouts: TcpTimeoutConfig,
+    ) -> Result<Self, PostgresError> {
+        let config = Self { inner, tunnel }.tcp_timeouts(tcp_timeouts);
 
         // Early validate that the configuration contains only a single TCP
         // server.
@@ -100,7 +104,7 @@ impl Config {
         Ok(config)
     }
 
-    pub fn tcp_timeouts(mut self, tcp_timeouts: TcpTimeoutConfig) -> Config {
+    fn tcp_timeouts(mut self, tcp_timeouts: TcpTimeoutConfig) -> Config {
         if let Some(connect_timeout) = tcp_timeouts.connect_timeout {
             self.inner.connect_timeout(connect_timeout);
         }

diff --git a/src/storage-types/src/connections.rs b/src/storage-types/src/connections.rs
@@ -1052,7 +1052,7 @@ impl PostgresConnection<InlinedConnection> {
     pub async fn config(
         &self,
         secrets_reader: &dyn mz_secrets::SecretsReader,
-        _storage_configuration: &StorageConfiguration,
+        storage_configuration: &StorageConfiguration,
     ) -> Result<mz_postgres_util::Config, anyhow::Error> {
         let mut config = tokio_postgres::Config::new();
         config
@@ -1100,7 +1100,14 @@ impl PostgresConnection<InlinedConnection> {
             }
         };
 
-        Ok(mz_postgres_util::Config::new(config, tunnel)?)
+        Ok(mz_postgres_util::Config::new(
+            config,
+            tunnel,
+            storage_configuration
+                .parameters
+                .pg_source_tcp_timeouts
+                .clone(),
+        )?)
     }
 
     async fn validate(