Add max_attempts for close slot activity, and specify query mode for slot check (#4148)

- Noticed that for a mirror which fails due to snapshot invalid
identifier the final closeSlot activity can hang as it has infinite
retries and is run on defer when the session worker is no longer
available. Adding max_attempts of 2 here
- Noticed this error:
```
error checking for replication slot - <slot_name> : expected 0 arguments, got 1
```
even though we clearly set $1 for that query. Turns out it can happen
with prepared statements to poolers. Specifying query mode at query
level like we do in other places

Author: Amogh-Bharadwaj

flow/connectors/postgres/client.go

@@ -249,7 +249,7 @@ func (c *PostgresConnector) checkSlotAndPublication(ctx context.Context, slot st
 	var slotName pgtype.Text
 	err := c.conn.QueryRow(ctx,
 		"SELECT slot_name FROM pg_replication_slots WHERE slot_name = $1",
-		slot).Scan(&slotName)
+		pgx.QueryExecModeSimpleProtocol, slot).Scan(&slotName)
 	if err != nil {
 		// check if the error is a "no rows" error
 		if !errors.Is(err, pgx.ErrNoRows) {
@@ -262,7 +262,7 @@ func (c *PostgresConnector) checkSlotAndPublication(ctx context.Context, slot st
 	// Check if the publication exists
 	var pubName pgtype.Text
 	if err := c.conn.QueryRow(ctx,
-		"SELECT pubname FROM pg_publication WHERE pubname = $1", publication,
+		"SELECT pubname FROM pg_publication WHERE pubname = $1", pgx.QueryExecModeSimpleProtocol, publication,
 	).Scan(&pubName); err != nil {
 		// check if the error is a "no rows" error
 		if !errors.Is(err, pgx.ErrNoRows) {

flow/workflows/snapshot_flow.go

@@ -83,9 +83,10 @@ func (s *SnapshotFlowExecution) closeSlotKeepAlive(
 	s.logger.Info("closing slot keep alive for peer flow")
 
 	ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
-		StartToCloseTimeout: 10 * time.Minute,
+		StartToCloseTimeout: 3 * time.Minute,
 		RetryPolicy: &temporal.RetryPolicy{
 			InitialInterval: 1 * time.Minute,
+			MaximumAttempts: 3,
 		},
 	})
 
@@ -94,7 +95,6 @@ func (s *SnapshotFlowExecution) closeSlotKeepAlive(
 	}
 
 	s.logger.Info("closed slot keep alive for peer flow")
-
 	return nil
 }
 
@@ -357,7 +357,6 @@ func SnapshotFlowWorkflow(
 		if err := se.closeSlotKeepAlive(sessionCtx); err != nil {
 			return fmt.Errorf("failed to close slot keep alive: %w", err)
 		}
-
 		return nil
 	}