[temporal] adjust threshold for workflows to ContinueAsNew (#2328)

Co-authored-by: Philip Dubé <[email protected]>

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)
+}

flow/workflows/cdc_flow.go

@@ -552,7 +552,7 @@ func CDCFlowWorkflow(
 			return state, err
 		}
 
-		if workflow.GetInfo(ctx).GetContinueAsNewSuggested() {
+		if shared.ShouldWorkflowContinueAsNew(ctx) {
 			restart = true
 			if syncFlowFuture != nil {
 				if err := model.SyncStopSignal.SignalChildWorkflow(ctx, syncFlowFuture, struct{}{}).Get(ctx, nil); err != nil {

flow/workflows/sync_flow.go

@@ -108,7 +108,7 @@ func SyncFlowWorkflow(
 		}
 
 		if (options.NumberOfSyncs > 0 && currentSyncFlowNum >= options.NumberOfSyncs) ||
-			syncErr || ctx.Err() != nil || workflow.GetInfo(ctx).GetContinueAsNewSuggested() {
+			syncErr || ctx.Err() != nil || shared.ShouldWorkflowContinueAsNew(ctx) {
 			break
 		}
 	}