Runs SendWalHeartbeat in parallel (#675)

SendWalHeartbeat now runs every 10 seconds in parallel during the
PeerFlow phase of CDC (after setup and snapshot)
cancelActivity is called where previously the .Get() on the activity was
called in `cdc_flow.go`

flow/activities/flowable.go

@@ -671,13 +671,27 @@ func (a *FlowableActivity) SendWALHeartbeat(ctx context.Context, config *protos.
 		return fmt.Errorf("failed to get destination connector: %w", err)
 	}
 	defer connectors.CloseConnector(srcConn)
-
-	err = srcConn.SendWALHeartbeat()
-	if err != nil {
-		return fmt.Errorf("failed to send WAL heartbeat: %w", err)
+	log.WithFields(log.Fields{"flowName": config.FlowJobName}).Info("sending walheartbeat every 10 minutes")
+	ticker := time.NewTicker(10 * time.Minute)
+	for {
+		select {
+		case <-ctx.Done():
+			log.WithFields(
+				log.Fields{
+					"flowName": config.FlowJobName,
+				}).Info("context is done, exiting wal heartbeat send loop")
+			return nil
+		case <-ticker.C:
+			err = srcConn.SendWALHeartbeat()
+			if err != nil {
+				return fmt.Errorf("failed to send WAL heartbeat: %w", err)
+			}
+			log.WithFields(
+				log.Fields{
+					"flowName": config.FlowJobName,
+				}).Info("sent wal heartbeat")
+		}
 	}
-
-	return nil
 }
 
 func (a *FlowableActivity) QRepWaitUntilNewRows(ctx context.Context,

flow/workflows/cdc_flow.go

@@ -100,18 +100,6 @@ func (s *CDCFlowWorkflowState) TruncateProgress() {
 	}
 }
 
-func (s *CDCFlowWorkflowState) SendWALHeartbeat(ctx workflow.Context, cfg *protos.FlowConnectionConfigs) error {
-	walHeartbeatCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
-		StartToCloseTimeout: 5 * time.Minute,
-	})
-
-	if err := workflow.ExecuteActivity(walHeartbeatCtx, flowable.SendWALHeartbeat, cfg).Get(ctx, nil); err != nil {
-		return fmt.Errorf("failed to send WAL heartbeat: %w", err)
-	}
-
-	return nil
-}
-
 // CDCFlowWorkflowExecution represents the state for execution of a peer flow.
 type CDCFlowWorkflowExecution struct {
 	flowExecutionID string
@@ -277,6 +265,12 @@ func CDCFlowWorkflowWithConfig(
 		state.Progress = append(state.Progress, "executed setup flow and snapshot flow")
 	}
 
+	heartbeatCancelCtx, cancelHeartbeat := workflow.WithCancel(ctx)
+	walHeartbeatCtx := workflow.WithActivityOptions(heartbeatCancelCtx, workflow.ActivityOptions{
+		StartToCloseTimeout: 7 * 24 * time.Hour,
+	})
+	workflow.ExecuteActivity(walHeartbeatCtx, flowable.SendWALHeartbeat, cfg)
+
 	syncFlowOptions := &protos.SyncFlowOptions{
 		BatchSize: int32(limits.MaxBatchSize),
 	}
@@ -419,10 +413,8 @@ func CDCFlowWorkflowWithConfig(
 		selector.Select(ctx)
 	}
 
-	// send WAL heartbeat
-	if err := state.SendWALHeartbeat(ctx, cfg); err != nil {
-		return state, err
-	}
+	// cancel the SendWalHeartbeat activity
+	defer cancelHeartbeat()
 
 	state.TruncateProgress()
 	return nil, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflowWithConfig, cfg, limits, state)