sync flow: check stop channel at end to avoid error ignoring pause (#…

…1895)
PeerDB-io · Jul 1, 2024 · 21ec626 · 21ec626
1 parent e493e42
commit 21ec626
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 26 deletions.
diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go
@@ -438,16 +438,15 @@ func CDCFlowWorkflow(
 	mainLoopSelector := workflow.NewNamedSelector(ctx, "MainLoop")
 	mainLoopSelector.AddReceive(ctx.Done(), func(_ workflow.ReceiveChannel, _ bool) {})
 	mainLoopSelector.AddFuture(syncFlowFuture, func(f workflow.Future) {
-		err := f.Get(ctx, nil)
-		if err != nil {
+		if err := f.Get(ctx, nil); err != nil {
 			handleError("sync", err)
 		}
 
 		logger.Info("sync finished, finishing normalize")
 		syncFlowFuture = nil
 		restart = true
 		if normFlowFuture != nil {
-			err = model.NormalizeSignal.SignalChildWorkflow(ctx, normFlowFuture, model.NormalizePayload{
+			err := model.NormalizeSignal.SignalChildWorkflow(ctx, normFlowFuture, model.NormalizePayload{
 				Done:        true,
 				SyncBatchID: -1,
 			}).Get(ctx, nil)

diff --git a/flow/workflows/sync_flow.go b/flow/workflows/sync_flow.go
@@ -60,15 +60,12 @@ func SyncFlowWorkflow(
 
 	selector := workflow.NewNamedSelector(ctx, "SyncLoop")
 	selector.AddReceive(ctx.Done(), func(_ workflow.ReceiveChannel, _ bool) {})
-	if fMaintain != nil {
-		selector.AddFuture(fMaintain, func(f workflow.Future) {
-			err := f.Get(ctx, nil)
-			if err != nil {
-				logger.Error("MaintainPull failed", slog.Any("error", err))
-				syncErr = true
-			}
-		})
-	}
+	selector.AddFuture(fMaintain, func(f workflow.Future) {
+		if err := f.Get(ctx, nil); err != nil {
+			logger.Error("MaintainPull failed", slog.Any("error", err))
+			syncErr = true
+		}
+	})
 
 	stopChan := model.SyncStopSignal.GetSignalChannel(ctx)
 	stopChan.AddToSelector(selector, func(_ struct{}, _ bool) {
@@ -218,24 +215,25 @@ func SyncFlowWorkflow(
 		return err
 	}
 
-	if fMaintain != nil {
-		unmaintainCtx := workflow.WithActivityOptions(syncSessionCtx, workflow.ActivityOptions{
-			RetryPolicy:         &temporal.RetryPolicy{MaximumAttempts: 1},
-			StartToCloseTimeout: time.Minute,
-			HeartbeatTimeout:    time.Minute,
-			WaitForCancellation: true,
-		})
-		if err := workflow.ExecuteActivity(
-			unmaintainCtx,
-			flowable.UnmaintainPull,
-			sessionID,
-		).Get(unmaintainCtx, nil); err != nil {
-			logger.Warn("UnmaintainPull failed", slog.Any("error", err))
-		}
+	unmaintainCtx := workflow.WithActivityOptions(syncSessionCtx, workflow.ActivityOptions{
+		RetryPolicy:         &temporal.RetryPolicy{MaximumAttempts: 1},
+		StartToCloseTimeout: time.Minute,
+		HeartbeatTimeout:    time.Minute,
+		WaitForCancellation: true,
+	})
+	if err := workflow.ExecuteActivity(
+		unmaintainCtx,
+		flowable.UnmaintainPull,
+		sessionID,
+	).Get(unmaintainCtx, nil); err != nil {
+		logger.Warn("UnmaintainPull failed", slog.Any("error", err))
 	}
 
 	if stop {
 		return nil
+	} else if _, stop := stopChan.ReceiveAsync(); stop {
+		// if sync flow erroring may outrace receiving stop
+		return nil
 	}
 	return workflow.NewContinueAsNewError(ctx, SyncFlowWorkflow, config, options)
 }