added timeout for long CDC pulls (#574)

fixes #562

flow/connectors/postgres/cdc.go

@@ -156,6 +156,16 @@ func (p *PostgresCDCSource) consumeStream(
 
 	var standByLastLogged time.Time
 
+	shutdown := utils.HeartbeatRoutine(p.ctx, 10*time.Second, func() string {
+		jobName := req.FlowJobName
+		currRecords := len(records.Records)
+		return fmt.Sprintf("pulling records for job - %s, currently have %d records", jobName, currRecords)
+	})
+
+	defer func() {
+		shutdown <- true
+	}()
+
 	for {
 		if time.Now().After(nextStandbyMessageDeadline) ||
 			(len(records.Records) >= int(req.MaxBatchSize)) {

flow/connectors/utils/heartbeat.go

@@ -19,7 +19,7 @@ func HeartbeatRoutine(
 	go func() {
 		for {
 			msg := fmt.Sprintf("heartbeat #%d: %s", counter, message())
-			activity.RecordHeartbeat(ctx, msg)
+			RecordHeartbeatWithRecover(ctx, msg)
 			counter += 1
 			to := time.After(interval)
 			select {

flow/workflows/sync_flow.go

@@ -83,9 +83,7 @@ func (s *SyncFlowExecution) executeSyncFlow(
 
 	startFlowCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 		StartToCloseTimeout: 72 * time.Hour,
-		// TODO: activity needs to call heartbeat.
-		// see https://github.com/PeerDB-io/nexus/issues/216
-		HeartbeatTimeout: 24 * time.Hour,
+		HeartbeatTimeout:    5 * time.Minute,
 	})
 
 	// execute StartFlow on the peers to start the flow