mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
zigeHandGitHub
0e742bb64e
## What changed? This PR refactors the WaitCatchup workflow by generalizing its input parameters. The goal is to make the workflow more flexible and reusable across different use cases where similar catch-up logic is needed. e.g. catchup process in passive migration. ## Why? <img width="825" alt="Screenshot 2025-05-08 at 12 39 23 PM" src="https://github.com/user-attachments/assets/993156ba-3cd1-48b6-9d72-6005229b98ba" /> In the previous setup, we ran catchup on passive replica A after T2 to ensure that cluster C could catch up to replica B’s (the current active) last AckID on A, guaranteeing that C wouldn't miss any data present on A. However, this logic was previously hardcoded to use B the active as catchup target. This PR generalizes the input, allowing the same logic to be reused during migration scenarios—enabling the source passive replica to catch up to the new replica’s ack level when the migration passive is in progress. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks parameter name change. no potential risk
60 lines
1.3 KiB
Go
60 lines
1.3 KiB
Go
package migration
|
|
|
|
import (
|
|
"time"
|
|
|
|
"go.temporal.io/sdk/temporal"
|
|
"go.temporal.io/sdk/workflow"
|
|
)
|
|
|
|
const (
|
|
catchupWorkflowName = "catchup"
|
|
)
|
|
|
|
type (
|
|
CatchUpParams struct {
|
|
Namespace string
|
|
CatchupCluster string
|
|
TargetCluster string
|
|
}
|
|
|
|
CatchUpOutput struct{}
|
|
)
|
|
|
|
func CatchupWorkflow(ctx workflow.Context, params CatchUpParams) (CatchUpOutput, error) {
|
|
if err := validateCatchupParams(¶ms); err != nil {
|
|
return CatchUpOutput{}, err
|
|
}
|
|
|
|
retryPolicy := &temporal.RetryPolicy{
|
|
InitialInterval: time.Second,
|
|
MaximumInterval: time.Second,
|
|
BackoffCoefficient: 1,
|
|
}
|
|
ao := workflow.ActivityOptions{
|
|
StartToCloseTimeout: time.Hour,
|
|
HeartbeatTimeout: time.Second * 10,
|
|
RetryPolicy: retryPolicy,
|
|
}
|
|
ctx1 := workflow.WithActivityOptions(ctx, ao)
|
|
|
|
var a *activities
|
|
err := workflow.ExecuteActivity(ctx1, a.WaitCatchup, params).Get(ctx, nil)
|
|
if err != nil {
|
|
return CatchUpOutput{}, err
|
|
}
|
|
|
|
return CatchUpOutput{}, err
|
|
}
|
|
|
|
func validateCatchupParams(params *CatchUpParams) error {
|
|
if len(params.Namespace) == 0 {
|
|
return temporal.NewNonRetryableApplicationError("InvalidArgument: Namespace is required", "InvalidArgument", nil)
|
|
}
|
|
if len(params.CatchupCluster) == 0 {
|
|
return temporal.NewNonRetryableApplicationError("InvalidArgument: RemoteCluster is required", "InvalidArgument", nil)
|
|
}
|
|
|
|
return nil
|
|
}
|