pr review

PeerDB-io · Jan 24, 2024 · bcea38b · bcea38b
1 parent 5075598
commit bcea38b
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 12 deletions.
diff --git a/flow/shared/constants.go b/flow/shared/constants.go
@@ -14,6 +14,8 @@ const (
 	// Signals
 	FlowSignalName                 = "peer-flow-signal"
 	CDCDynamicPropertiesSignalName = "cdc-dynamic-properties"
+	NormalizeSyncSignalName        = "normalize-sync"
+	NormalizeSyncDoneSignalName    = "normalize-sync-done"
 
 	// Queries
 	CDCFlowStateQuery  = "q-cdc-flow-status"

diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go
@@ -423,11 +423,11 @@ func CDCFlowWorkflowWithConfig(
 
 	var normWaitChan workflow.ReceiveChannel
 	if !peerdbenv.PeerDBEnableParallelSyncNormalize() {
-		normWaitChan = workflow.GetSignalChannel(ctx, "SyncDone")
+		normWaitChan = workflow.GetSignalChannel(ctx, shared.NormalizeSyncDoneSignalName)
 	}
 
 	finishNormalize := func() {
-		childNormalizeFlowFuture.SignalChildWorkflow(ctx, "Sync", model.NormalizeSignal{
+		childNormalizeFlowFuture.SignalChildWorkflow(ctx, shared.NormalizeSyncSignalName, model.NormalizeSignal{
 			Done:        true,
 			SyncBatchID: -1,
 		})
@@ -597,7 +597,7 @@ func CDCFlowWorkflowWithConfig(
 					}
 				}
 
-				signalFuture := childNormalizeFlowFuture.SignalChildWorkflow(ctx, "Sync", model.NormalizeSignal{
+				signalFuture := childNormalizeFlowFuture.SignalChildWorkflow(ctx, shared.NormalizeSyncSignalName, model.NormalizeSignal{
 					Done:                   false,
 					SyncBatchID:            childSyncFlowRes.CurrentSyncBatchID,
 					TableNameSchemaMapping: normalizeTableNameSchemaMapping,

diff --git a/flow/workflows/normalize_flow.go b/flow/workflows/normalize_flow.go
@@ -9,6 +9,7 @@ import (
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/model"
 	"github.com/PeerDB-io/peer-flow/peerdbenv"
+	"github.com/PeerDB-io/peer-flow/shared"
 )
 
 func NormalizeFlowWorkflow(ctx workflow.Context,
@@ -25,7 +26,7 @@ func NormalizeFlowWorkflow(ctx workflow.Context,
 
 	results := make([]model.NormalizeResponse, 0, 4)
 	errors := make([]string, 0)
-	syncChan := workflow.GetSignalChannel(ctx, "Sync")
+	syncChan := workflow.GetSignalChannel(ctx, shared.NormalizeSyncSignalName)
 
 	var stopLoop, canceled bool
 	var lastSyncBatchID, syncBatchID int64
@@ -54,6 +55,11 @@ func NormalizeFlowWorkflow(ctx workflow.Context,
 			selector.Select(ctx)
 		}
 		if canceled || (stopLoop && lastSyncBatchID == syncBatchID) {
+			if canceled {
+				logger.Info("normalize canceled - ", config.FlowJobName)
+			} else {
+				logger.Info("normalize finished - ", config.FlowJobName)
+			}
 			break
 		}
 		if lastSyncBatchID != syncBatchID {
@@ -81,7 +87,7 @@ func NormalizeFlowWorkflow(ctx workflow.Context,
 				ctx,
 				parent.ID,
 				parent.RunID,
-				"SyncDone",
+				shared.NormalizeSyncDoneSignalName,
 				struct{}{},
 			)
 		}

diff --git a/flow/workflows/qrep_flow.go b/flow/workflows/qrep_flow.go
@@ -366,9 +366,7 @@ func (q *QRepFlowExecution) handleTableRenameForResync(ctx workflow.Context, sta
 	return nil
 }
 
-func (q *QRepFlowExecution) receiveAndHandleSignalAsync(ctx workflow.Context) {
-	signalChan := workflow.GetSignalChannel(ctx, shared.FlowSignalName)
-
+func (q *QRepFlowExecution) receiveAndHandleSignalAsync(signalChan workflow.ReceiveChannel) {
 	var signalVal shared.CDCFlowSignal
 	ok := signalChan.ReceiveAsync(&signalVal)
 	if ok {
@@ -513,11 +511,11 @@ func QRepFlowWorkflow(
 
 	// here, we handle signals after the end of the flow because a new workflow does not inherit the signals
 	// and the chance of missing a signal is much higher if the check is before the time consuming parts run
-	q.receiveAndHandleSignalAsync(ctx)
+	signalChan := workflow.GetSignalChannel(ctx, shared.FlowSignalName)
+	q.receiveAndHandleSignalAsync(signalChan)
 	if q.activeSignal == shared.PauseSignal {
 		startTime := time.Now()
 		state.CurrentFlowStatus = protos.FlowStatus_STATUS_PAUSED
-		signalChan := workflow.GetSignalChannel(ctx, shared.FlowSignalName)
 		var signalVal shared.CDCFlowSignal
 
 		for q.activeSignal == shared.PauseSignal {

diff --git a/flow/workflows/xmin_flow.go b/flow/workflows/xmin_flow.go
@@ -117,11 +117,11 @@ func XminFlowWorkflow(
 
 	// here, we handle signals after the end of the flow because a new workflow does not inherit the signals
 	// and the chance of missing a signal is much higher if the check is before the time consuming parts run
-	q.receiveAndHandleSignalAsync(ctx)
+	signalChan := workflow.GetSignalChannel(ctx, shared.FlowSignalName)
+	q.receiveAndHandleSignalAsync(signalChan)
 	if x.activeSignal == shared.PauseSignal {
 		startTime := time.Now()
 		state.CurrentFlowStatus = protos.FlowStatus_STATUS_PAUSED
-		signalChan := workflow.GetSignalChannel(ctx, shared.FlowSignalName)
 		var signalVal shared.CDCFlowSignal
 
 		for x.activeSignal == shared.PauseSignal {