restore async replication connection keepalive when PullRecords is not active (#4405)

Prior to #4324, we had an async
`maintainReplConn` goroutine that was running on a ticker every 15
seconds. When we moved the keepalive inside of PullRecords, we
introduced an edge case where there can be _a large time gap_ between
PullRecords where we are no longer keeping the connection alive(e.g.
when normalize backpressure kicks in). This led to a new failure mode
when PullRecords resumes and tries to read from PG, it hit
wal_sender_timeout error.

This PR brings back the async goroutine ~but only run the keepalive when
PullRecords is not active; otherwise, it let the PullRecords handle the
keepalive logic in the loop.~

The keepalive is hard-coded to 15 seconds. When we merge
#4363 later, we can use the
computed keepalive instead. ~Also note it's still safe to remove
`replLock` because this ReplPing only runs when PullRecords is inactive,
so there should be no lock contention (but doesn't hurt if we wish to
keep it either for safety).~ There can be edge cases (see discussion
below) where a race condition can happen, so we will want to keep the
replLock around when.

Also note that #4404 addresses
slightly separate edge case, with the goal of ensuring ReplPing has full
coverage _inside_ a PullRecord cycle.

Author: jgao54

flow/connectors/postgres/cdc.go

@@ -737,8 +737,8 @@ func PullCdcRecords[Items model.Items](
 
 		if err != nil && pgconn.Timeout(err) {
 			// On timeout, always send a ping before moving on to read more messages
-			if err := p.ReplPing(ctx); err != nil {
-				return fmt.Errorf("ReplPing failed: %w", err)
+			if err := sendStandbyAfterReplLock("receive-timeout"); err != nil {
+				return err
 			}
 			// After that, we let the condition checks at the beginging of the loop,
 			// specifically "if we are past the next standby deadline (?)" to

flow/connectors/postgres/postgres_source.go

@@ -80,11 +80,38 @@ func (c *PostgresConnector) SetupReplConn(ctx context.Context, env map[string]st
 		return err
 	}
 	c.replConn = conn
+	go c.runReplConnKeepalive(ctx, 15*time.Second, c.ReplPing)
 	return nil
 }
 
-// To keep connection alive between sync batches.
-// By default postgres drops connection after 1 minute of inactivity.
+// runReplConnKeepalive pings the replication connection on a ticker so the
+// walsender doesn't hit wal_sender_timeout. This happens whenever keepalive
+// from the PullRecords stop: (1) between PullRecords calls (2) within
+// PullRecords when AddRecord blocks on a full record stream.
+//
+// While PullRecords holds replLock inside ReceiveMessage, ReplPing's TryLock
+// makes it a no-op. This is safe because ReceiveMessage will always run its
+// own keepalive on timeout.
+func (c *PostgresConnector) runReplConnKeepalive(
+	ctx context.Context,
+	interval time.Duration,
+	keepalive func(context.Context) error,
+) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := keepalive(ctx); err != nil {
+				c.logger.Warn("replication keepalive failed", slog.Any("error", err))
+			}
+		}
+	}
+}
+
+// ReplPing sends a standby status update so Postgres doesn't terminate the walsender for inactivity.
 func (c *PostgresConnector) ReplPing(ctx context.Context) error {
 	if c.replLock.TryLock() {
 		defer c.replLock.Unlock()