CustomSync

PeerDB-io · Dec 10, 2024 · 328a81b · 328a81b
1 parent b4f9a11
commit 328a81b
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 5 deletions.
diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go
@@ -526,9 +526,12 @@ func CDCFlowWorkflow(
 			handleError("sync", err)
 		}
 
-		logger.Info("sync finished, finishing normalize")
+		logger.Info("sync finished")
 		syncFlowFuture = nil
 		restart = true
+		if state.SyncFlowOptions.NumberOfSyncs > 0 {
+			state.ActiveSignal = model.PauseSignal
+		}
 	})
 
 	flowSignalChan.AddToSelector(mainLoopSelector, func(val model.CDCFlowSignal, _ bool) {

diff --git a/flow/workflows/sync_flow.go b/flow/workflows/sync_flow.go
@@ -47,7 +47,7 @@ func SyncFlowWorkflow(
 	)
 
 	var stop, syncErr bool
-	currentSyncFlowNum := 0
+	currentSyncFlowNum := int32(0)
 	totalRecordsSynced := int64(0)
 
 	selector := workflow.NewNamedSelector(ctx, "SyncLoop")
@@ -73,7 +73,7 @@ func SyncFlowWorkflow(
 		var syncDone bool
 
 		currentSyncFlowNum += 1
-		logger.Info("executing sync flow", slog.Int("count", currentSyncFlowNum))
+		logger.Info("executing sync flow", slog.Int("count", int(currentSyncFlowNum)))
 
 		var syncFlowFuture workflow.Future
 		if config.System == protos.TypeSystem_Q {
@@ -99,7 +99,7 @@ func SyncFlowWorkflow(
 			selector.Select(ctx)
 		}
 
-		if syncErr || ctx.Err() != nil || workflow.GetInfo(ctx).GetContinueAsNewSuggested() {
+		if currentSyncFlowNum >= options.NumberOfSyncs || syncErr || ctx.Err() != nil || workflow.GetInfo(ctx).GetContinueAsNewSuggested() {
 			break
 		}
 	}
@@ -123,7 +123,7 @@ func SyncFlowWorkflow(
 		logger.Warn("UnmaintainPull failed", slog.Any("error", err))
 	}
 
-	if stop {
+	if stop || currentSyncFlowNum >= options.NumberOfSyncs {
 		return nil
 	} else if _, stop := stopChan.ReceiveAsync(); stop {
 		// if sync flow erroring may outrace receiving stop