sundry cdc fixes

PeerDB-io · Oct 22, 2023 · 1c24d3e · 1c24d3e
1 parent d5e6d36
commit 1c24d3e
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go
@@ -154,6 +154,8 @@ func (p *PostgresCDCSource) consumeStream(
 		consumedXLogPos = clientXLogPos - 1
 	}
 
+	var standByLastLogged time.Time
+
 	for {
 		if time.Now().After(nextStandbyMessageDeadline) ||
 			(len(records.Records) >= int(req.MaxBatchSize)) {
@@ -167,7 +169,12 @@ func (p *PostgresCDCSource) consumeStream(
 
 			numRowsProcessedMessage := fmt.Sprintf("processed %d rows", len(records.Records))
 			utils.RecordHeartbeatWithRecover(p.ctx, numRowsProcessedMessage)
-			log.Infof("Sent Standby status message. %s", numRowsProcessedMessage)
+
+			if time.Since(standByLastLogged) > 10*time.Second {
+				log.Infof("Sent Standby status message. %s", numRowsProcessedMessage)
+				standByLastLogged = time.Now()
+			}
+
 			nextStandbyMessageDeadline = time.Now().Add(standbyMessageTimeout)
 
 			if !p.commitLock && (len(records.Records) >= int(req.MaxBatchSize)) {

diff --git a/flow/workflows/sync_flow.go b/flow/workflows/sync_flow.go
@@ -82,10 +82,10 @@ func (s *SyncFlowExecution) executeSyncFlow(
 	}
 
 	startFlowCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
-		StartToCloseTimeout: 24 * time.Hour,
+		StartToCloseTimeout: 72 * time.Hour,
 		// TODO: activity needs to call heartbeat.
 		// see https://github.com/PeerDB-io/nexus/issues/216
-		HeartbeatTimeout: 30 * time.Second,
+		HeartbeatTimeout: 24 * time.Hour,
 	})
 
 	// execute StartFlow on the peers to start the flow