build(deps): bump predicates from 2.1.4 to 3.0.2

src/environmentd/Cargo.toml

@@ -104,7 +104,7 @@ postgres = { git = "https://github.com/MaterializeInc/rust-postgres", features =
 postgres-openssl = { git = "https://github.com/MaterializeInc/rust-postgres" }
 postgres-protocol = { git = "https://github.com/MaterializeInc/rust-postgres" }
 postgres_array = { git = "https://github.com/MaterializeInc/rust-postgres-array" }
-predicates = "2.1.4"
+predicates = "3.0.2"
 regex = "1.7.0"
 reqwest = { version = "0.11.13", features = ["blocking"] }
 serde_json = "1.0.89"

src/storage/src/source/postgres.rs

@@ -1036,6 +1036,7 @@ async fn produce_replication<'a>(
             let mut stream = Box::pin(LogicalReplicationStream::new(copy_stream));
 
             let mut last_data_message = Instant::now();
+            let mut err_counter = 0;
 
             // The inner loop
             loop {
@@ -1051,6 +1052,10 @@ async fn produce_replication<'a>(
                 )
                 .await;
 
+                if res.is_ok() {
+                    err_counter = 0;
+                }
+
                 // The upstream will periodically request status updates by setting the keepalive's
                 // reply field to 1. However, we cannot rely on these messages arriving on time. For
                 // example, when the upstream is sending a big transaction its keepalive messages are
@@ -1281,6 +1286,15 @@ async fn produce_replication<'a>(
                             "if our request timed out, it must have been at least long enough to require \
                              a status update"
                         );
+
+                        // If we've timed out multiple times, try reconnecting. This should have
+                        // given us plenty of time to receive a KeepAlive message from the server if
+                        // we're going to.
+                        if err_counter > 1 {
+                            break;
+                        }
+
+                        err_counter += 1;
                     }
                 }
                 if needs_status_update {