[temporal] adjust threshold for workflows to ContinueAsNew

PeerDB-io · Dec 6, 2024 · af25514 · af25514
1 parent 8e87695
commit af25514
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 2 deletions.
diff --git a/flow/shared/workflow.go b/flow/shared/workflow.go
@@ -6,6 +6,7 @@ import (
 	"log/slog"
 
 	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/workflow"
 
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 )
@@ -25,3 +26,9 @@ func GetWorkflowStatus(ctx context.Context, temporalClient client.Client, workfl
 	}
 	return state, nil
 }
+
+func ShouldWorkflowContinueAsNew(ctx workflow.Context) bool {
+	info := workflow.GetInfo(ctx)
+	return info.GetContinueAsNewSuggested() &&
+		(info.GetCurrentHistoryLength() > 40960 || info.GetCurrentHistorySize() > 40*1024*1024)
+}
diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go
@@ -610,7 +610,7 @@ func CDCFlowWorkflow(
 			return state, err
 		}
 
-		if state.ActiveSignal == model.PauseSignal || workflow.GetInfo(ctx).GetContinueAsNewSuggested() {
+		if state.ActiveSignal == model.PauseSignal || shared.ShouldWorkflowContinueAsNew(ctx) {
 			restart = true
 			if syncFlowFuture != nil {
 				err := model.SyncStopSignal.SignalChildWorkflow(ctx, syncFlowFuture, struct{}{}).Get(ctx, nil)
@@ -620,6 +620,7 @@ func CDCFlowWorkflow(
 				}
 			}
 		}
+		workflow.GetInfo(ctx).GetCurrentHistorySize()
 
 		if restart {
 			if state.ActiveSignal == model.PauseSignal {

diff --git a/flow/workflows/sync_flow.go b/flow/workflows/sync_flow.go
@@ -161,7 +161,7 @@ func SyncFlowWorkflow(
 			break
 		}
 
-		restart := syncErr || workflow.GetInfo(ctx).GetContinueAsNewSuggested()
+		restart := syncErr || shared.ShouldWorkflowContinueAsNew(ctx)
 		if !stop && !syncErr && mustWait {
 			waitSelector.Select(ctx)
 			if restart {