Fix normalize-done race causing sync/normalize to execute in parallel…

… after pause/unpause

Fixes #1448

flow/model/signals.go

@@ -84,6 +84,11 @@ func (self TypedReceiveChannel[T]) ReceiveAsyncWithMoreFlag() (T, bool, bool) {
 	return result, ok, more
 }
 
+func (self TypedReceiveChannel[T]) Drain() {
+	for self.Chan.ReceiveAsync(nil) {
+	}
+}
+
 func (self TypedReceiveChannel[T]) AddToSelector(selector workflow.Selector, f func(T, bool)) workflow.Selector {
 	return selector.AddReceive(self.Chan, func(c workflow.ReceiveChannel, more bool) {
 		var result T

flow/workflows/cdc_flow.go

@@ -479,6 +479,7 @@ func CDCFlowWorkflow(
 	})
 	if !parallel {
 		normDoneChan := model.NormalizeDoneSignal.GetSignalChannel(ctx)
+		normDoneChan.Drain()
 		normDoneChan.AddToSelector(mainLoopSelector, func(x struct{}, _ bool) {
 			if syncFlowFuture != nil {
 				_ = model.NormalizeDoneSignal.SignalChildWorkflow(ctx, syncFlowFuture, x).Get(ctx, nil)

flow/workflows/normalize_flow.go

@@ -104,7 +104,7 @@ func NormalizeFlowWorkflow(
 		}
 	}
 
-	if !state.Stop {
+	if ctx.Err() == nil && !state.Stop {
 		parallel := GetSideEffect(ctx, func(_ workflow.Context) bool {
 			return peerdbenv.PeerDBEnableParallelSyncNormalize()
 		})

flow/workflows/sync_flow.go

@@ -79,6 +79,7 @@ func SyncFlowWorkflow(
 		waitSelector = workflow.NewNamedSelector(ctx, "NormalizeWait")
 		waitSelector.AddReceive(ctx.Done(), func(_ workflow.ReceiveChannel, _ bool) {})
 		waitChan := model.NormalizeDoneSignal.GetSignalChannel(ctx)
+		waitChan.Drain()
 		waitChan.AddToSelector(waitSelector, func(_ struct{}, _ bool) {})
 		stopChan.AddToSelector(waitSelector, func(_ struct{}, _ bool) {
 			stop = true