try see if conn closed earlier

PeerDB-io · Feb 15, 2024 · 3bc529b · 3bc529b
1 parent 1183f84
commit 3bc529b
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go
@@ -171,6 +171,11 @@ func (p *PostgresCDCSource) PullRecords(ctx context.Context, req *model.PullReco
 	waitingForCommit := false
 
 	for {
+		_, err := p.getCurrentLSN(ctx)
+		if err != nil {
+			return fmt.Errorf("!!! current LSN: %w", err)
+		}
+
 		if pkmRequiresResponse {
 			err := pglogrepl.SendStandbyStatusUpdate(ctx, conn,
 				pglogrepl.StandbyStatusUpdate{WALWritePosition: consumedXLogPos})

diff --git a/flow/connectors/postgres/postgres.go b/flow/connectors/postgres/postgres.go
@@ -366,6 +366,12 @@ func (c *PostgresConnector) PullRecords(ctx context.Context, catalogPool *pgxpoo
 	if err != nil {
 		return err
 	}
+
+	_, err = c.getCurrentLSN(ctx)
+	if err != nil {
+		return fmt.Errorf("??? current LSN: %w", err)
+	}
+
 	req.RecordStream.Close()
 	c.replState.Offset = req.RecordStream.GetLastCheckpoint()