Make sync batch size dynamic (#806)

PeerDB-io · Dec 12, 2023 · de891b0 · de891b0
1 parent 36cc29f
commit de891b0
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 3 deletions.
diff --git a/flow/shared/constants.go b/flow/shared/constants.go
@@ -7,9 +7,10 @@ import (
 )
 
 const (
-	peerFlowTaskQueue     = "peer-flow-task-queue"
-	snapshotFlowTaskQueue = "snapshot-flow-task-queue"
-	CDCFlowSignalName     = "peer-flow-signal"
+	peerFlowTaskQueue      = "peer-flow-task-queue"
+	snapshotFlowTaskQueue  = "snapshot-flow-task-queue"
+	CDCFlowSignalName      = "peer-flow-signal"
+	CDCBatchSizeSignalName = "cdc-batch-size-signal"
 )
 
 const MirrorNameSearchAttribute = "MirrorName"

diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go
@@ -284,6 +284,20 @@ func CDCFlowWorkflowWithConfig(
 		BatchSize: int32(limits.MaxBatchSize),
 	}
 
+	// add a signal to change the batch size
+	batchSizeSignalChan := workflow.GetSignalChannel(ctx, shared.CDCBatchSizeSignalName)
+	batchSizeSelector := workflow.NewSelector(ctx)
+	batchSizeSelector.AddReceive(batchSizeSignalChan, func(c workflow.ReceiveChannel, more bool) {
+		var batchSize int32
+		c.Receive(ctx, &batchSize)
+		w.logger.Info("received batch size signal: ", batchSize)
+		syncFlowOptions.BatchSize = batchSize
+	})
+	batchSizeSelector.AddDefault(func() {
+		w.logger.Info("no batch size signal received, batch size remains: ",
+			syncFlowOptions.BatchSize)
+	})
+
 	currentSyncFlowNum := 0
 	totalRecordsSynced := 0
 
@@ -435,6 +449,7 @@ func CDCFlowWorkflowWithConfig(
 			}
 		})
 		selector.Select(ctx)
+		batchSizeSelector.Select(ctx)
 	}
 
 	state.TruncateProgress()