Retry transient visibility count failures

loadgen/helpers.go

@@ -57,57 +57,67 @@ func MinVisibilityCountEventually(
 	request *workflowservice.CountWorkflowExecutionsRequest,
 	minCount int,
 	waitAtMost time.Duration,
+) error {
+	lastPrint := time.Now()
+	return retryVisibilityCount(
+		ctx,
+		int64(minCount),
+		waitAtMost,
+		3*time.Second,
+		func(countCtx context.Context) (int64, error) {
+			visibilityCount, err := info.Client.CountWorkflow(countCtx, request)
+			if err != nil {
+				return 0, err
+			}
+			if time.Since(lastPrint) >= 30*time.Second {
+				info.Logger.Infof("current visibility count: %d (expected at least: %d)\n",
+					visibilityCount.Count, minCount)
+				lastPrint = time.Now()
+			}
+			return visibilityCount.Count, nil
+		},
+	)
+}
+
+func retryVisibilityCount(
+	ctx context.Context,
+	minCount int64,
+	waitAtMost time.Duration,
+	retryInterval time.Duration,
+	count func(context.Context) (int64, error),
 ) error {
 	timeoutCtx, cancel := context.WithTimeout(ctx, waitAtMost)
 	defer cancel()
-
-	countTicker := time.NewTicker(3 * time.Second)
-	defer countTicker.Stop()
-
-	printTicker := time.NewTicker(30 * time.Second)
-	defer printTicker.Stop()
-
-	var lastVisibilityCount int64
-	done := false
-
-	check := func() error {
-		visibilityCount, err := info.Client.CountWorkflow(timeoutCtx, request)
-		if err != nil {
-			return fmt.Errorf("failed to count workflows in visibility: %w", err)
-		}
-		lastVisibilityCount = visibilityCount.Count
-		if lastVisibilityCount >= int64(minCount) {
-			done = true
+	var lastCount int64
+	var lastErr error
+	for {
+		current, err := count(timeoutCtx)
+		if err == nil {
+			lastCount = current
+			lastErr = nil
+			if current >= minCount {
+				return nil
+			}
+		} else {
+			lastErr = err
 		}
-		return nil
-	}
-
-	// Initial check before entering the loop.
-	if err := check(); err != nil {
-		return err
-	}
-
-	// Loop until we reach the desired count or timeout.
-	for !done {
+		timer := time.NewTimer(retryInterval)
 		select {
 		case <-timeoutCtx.Done():
+			timer.Stop()
+			if lastErr != nil {
+				return fmt.Errorf(
+					"expected at least %d workflows in visibility, got %d after waiting %v; last count error: %w",
+					minCount, lastCount, waitAtMost, lastErr,
+				)
+			}
 			return fmt.Errorf(
 				"expected at least %d workflows in visibility, got %d after waiting %v",
-				minCount, lastVisibilityCount, waitAtMost,
+				minCount, lastCount, waitAtMost,
 			)
-
-		case <-printTicker.C:
-			info.Logger.Infof("current visibility count: %d (expected at least: %d)\n",
-				lastVisibilityCount, minCount)
-
-		case <-countTicker.C:
-			if err := check(); err != nil {
-				return err
-			}
+		case <-timer.C:
 		}
 	}
-
-	return nil
 }
 
 // VerifyNoFailedWorkflows verifies that there are no failed or terminated workflows for the given search attribute.

loadgen/helpers_visibility_test.go

@@ -0,0 +1,36 @@
+package loadgen
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestRetryVisibilityCountIgnoresTransientErrorsUntilMinimumObserved(t *testing.T) {
+	attempts := 0
+	err := retryVisibilityCount(
+		context.Background(),
+		3,
+		100*time.Millisecond,
+		time.Millisecond,
+		func(context.Context) (int64, error) {
+			attempts++
+			switch attempts {
+			case 1:
+				return 0, errors.New("transient deadline")
+			case 2:
+				return 1, nil
+			default:
+				return 3, nil
+			}
+		},
+	)
+
+	if err != nil {
+		t.Fatal(err)
+	}
+	if attempts != 3 {
+		t.Fatalf("expected three attempts, got %d", attempts)
+	}
+}

loadgen/kitchen_sink_executor_test.go

@@ -48,8 +48,8 @@ type testCase struct {
 // TestKitchenSink tests specific kitchensink features across SDKs.
 // Use the `SDK` environment variable to run only a specific SDK.
 func TestKitchenSink(t *testing.T) {
-	if os.Getenv("CI") != "" && onlySDK == "" {
-		t.Skip("Skipping kitchensink test in CI without specific SDK set")
+	if onlySDK == "" && os.Getenv("OMES_RUN_ALL_SDKS") == "" {
+		t.Skip("Skipping kitchensink test without specific SDK set; set SDK or OMES_RUN_ALL_SDKS=1")
 	}
 	env := SetupTestEnvironment(t)
 

scenarios/ebb_and_flow.go

@@ -49,6 +49,7 @@ type ebbAndFlowConfig struct {
 	MaxConsecutiveErrors          int
 	BacklogLogInterval            time.Duration
 	VisibilityVerificationTimeout time.Duration
+	VerifyVisibility              bool
 	SleepActivityConfig           *loadgen.SleepActivityConfig
 }
 
@@ -97,6 +98,7 @@ func (e *ebbAndFlowExecutor) Configure(info loadgen.ScenarioInfo) error {
 		MaxConsecutiveErrors:          info.ScenarioOptionInt(MaxConsecutiveErrorsFlag, 10),
 		BacklogLogInterval:            info.ScenarioOptionDuration(BacklogLogIntervalFlag, 30*time.Second),
 		VisibilityVerificationTimeout: info.ScenarioOptionDuration(VisibilityVerificationTimeoutFlag, 30*time.Second),
+		VerifyVisibility:              info.ScenarioOptionBool(VerifyVisibilityFlag, true),
 	}
 
 	config.MinBacklog = int64(info.ScenarioOptionInt(MinBacklogFlag, 0))
@@ -224,23 +226,26 @@ func (e *ebbAndFlowExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo)
 		return errors.New("No iterations completed. Either the scenario never ran, or it failed to resume correctly.")
 	}
 
-	// Post-scenario: verify reported workflow completion count from Visibility.
-	if err := loadgen.MinVisibilityCountEventually(
-		ctx,
-		e.ScenarioInfo,
-		&workflowservice.CountWorkflowExecutionsRequest{
-			Namespace: e.Namespace,
-			Query: fmt.Sprintf("%s='%s'",
-				loadgen.OmesExecutionIDSearchAttribute, e.ExecutionID),
-		},
-		totalCompletedWorkflows,
-		config.VisibilityVerificationTimeout,
-	); err != nil {
-		return err
-	}
+	if config.VerifyVisibility {
+		// Post-scenario: verify reported workflow completion count from Visibility.
+		if err := loadgen.MinVisibilityCountEventually(
+			ctx,
+			e.ScenarioInfo,
+			&workflowservice.CountWorkflowExecutionsRequest{
+				Namespace: e.Namespace,
+				Query: fmt.Sprintf("%s='%s'",
+					loadgen.OmesExecutionIDSearchAttribute, e.ExecutionID),
+			},
+			totalCompletedWorkflows,
+			config.VisibilityVerificationTimeout,
+		); err != nil {
+			return err
+		}
 
-	// Post-scenario: ensure there are no failed or terminated workflows for this run.
-	return loadgen.VerifyNoFailedWorkflows(ctx, e.ScenarioInfo, loadgen.OmesExecutionIDSearchAttribute, e.ExecutionID)
+		// Post-scenario: ensure there are no failed or terminated workflows for this run.
+		return loadgen.VerifyNoFailedWorkflows(ctx, e.ScenarioInfo, loadgen.OmesExecutionIDSearchAttribute, e.ExecutionID)
+	}
+	return nil
 }
 
 // Snapshot returns a snapshot of the current state.

scenarios/ebb_and_flow_test.go

@@ -114,3 +114,13 @@ func TestEbbAndFlow(t *testing.T) {
 		require.NoError(t, err, "Executor should complete successfully")
 	})
 }
+
+func TestEbbAndFlowCanDisableVisibilityVerification(t *testing.T) {
+	executor := newEbbAndFlowExecutor()
+	info := loadgen.ScenarioInfo{
+		ScenarioOptions: map[string]string{VerifyVisibilityFlag: "false"},
+	}
+
+	require.NoError(t, executor.Configure(info))
+	require.False(t, executor.config.VerifyVisibility)
+}

scenarios/external_event.go

@@ -0,0 +1,26 @@
+package scenarios
+
+import (
+	"github.com/temporalio/omes/loadgen"
+	. "github.com/temporalio/omes/loadgen/kitchensink"
+)
+
+func init() {
+	loadgen.MustRegisterScenario(loadgen.Scenario{
+		Description: "Each iteration starts one workflow, sends one signal, waits for the signaled state, and completes.",
+		ExecutorFn: func() loadgen.Executor {
+			return loadgen.KitchenSinkExecutor{
+				TestInput: &TestInput{
+					ClientSequence: ClientActions(NewSignalActionsWithIDs(1)...),
+					WorkflowInput: &WorkflowInput{
+						ExpectedSignalCount: 1,
+						InitialActions: ListActionSet(
+							NewAwaitWorkflowStateAction("signal_0", "received"),
+							NewEmptyReturnResultAction(),
+						),
+					},
+				},
+			}
+		},
+	})
+}

scenarios/failure_retry.go

@@ -0,0 +1,26 @@
+package scenarios
+
+import (
+	"time"
+
+	"github.com/temporalio/omes/loadgen"
+	. "github.com/temporalio/omes/loadgen/kitchensink"
+)
+
+func init() {
+	loadgen.MustRegisterScenario(loadgen.Scenario{
+		Description: "Each iteration starts one workflow with one retryable activity that fails once, retries, and completes.",
+		ExecutorFn: func() loadgen.Executor {
+			return loadgen.KitchenSinkExecutor{
+				TestInput: &TestInput{
+					WorkflowInput: &WorkflowInput{
+						InitialActions: ListActionSet(
+							RetryableErrorActivity(1, RemoteActivityWithRetry(5*time.Second, 2, 500*time.Millisecond, 1.0)),
+							NewEmptyReturnResultAction(),
+						),
+					},
+				},
+			}
+		},
+	})
+}

scenarios/failure_retry_test.go

@@ -0,0 +1,26 @@
+package scenarios
+
+import (
+	"testing"
+	"time"
+
+	"github.com/temporalio/omes/loadgen"
+)
+
+func TestFailureRetryAllowsOneFailureAndOneRetry(t *testing.T) {
+	scenario := loadgen.GetScenario("failure_retry")
+	executor, ok := scenario.ExecutorFn().(loadgen.KitchenSinkExecutor)
+	if !ok {
+		t.Fatal("failure_retry does not use KitchenSinkExecutor")
+	}
+	activity := executor.TestInput.WorkflowInput.InitialActions[0].Actions[0].GetExecActivity()
+	if got := activity.StartToCloseTimeout.AsDuration(); got != 5*time.Second {
+		t.Fatalf("got StartToClose timeout %v, want 5s", got)
+	}
+	if got := activity.RetryPolicy.MaximumAttempts; got != 2 {
+		t.Fatalf("got %d maximum attempts, want 2", got)
+	}
+	if got := activity.GetRetryableError().FailAttempts; got != 1 {
+		t.Fatalf("got %d failed attempts, want 1", got)
+	}
+}

scenarios/matched_fixed_resource_consumption.go

@@ -0,0 +1,24 @@
+package scenarios
+
+import (
+	"github.com/temporalio/omes/loadgen"
+	. "github.com/temporalio/omes/loadgen/kitchensink"
+)
+
+func init() {
+	loadgen.MustRegisterScenario(loadgen.Scenario{
+		Description: "Each iteration runs one bounded 64-KB, 25,000-write resource activity.",
+		ExecutorFn: func() loadgen.Executor {
+			return loadgen.KitchenSinkExecutor{
+				TestInput: &TestInput{
+					WorkflowInput: &WorkflowInput{
+						InitialActions: ListActionSet(
+							GenericActivity("matched_resource", DefaultRemoteActivity),
+							NewEmptyReturnResultAction(),
+						),
+					},
+				},
+			}
+		},
+	})
+}

scenarios/matched_fixed_resource_consumption_test.go

@@ -0,0 +1,29 @@
+package scenarios
+
+import (
+	"testing"
+
+	"github.com/temporalio/omes/loadgen"
+)
+
+func TestMatchedFixedResourceConsumptionUsesOneMatchedActivity(t *testing.T) {
+	scenario := loadgen.GetScenario("matched_fixed_resource_consumption")
+	if scenario == nil {
+		t.Fatal("matched_fixed_resource_consumption is not registered")
+	}
+	executor, ok := scenario.ExecutorFn().(loadgen.KitchenSinkExecutor)
+	if !ok {
+		t.Fatal("matched_fixed_resource_consumption does not use KitchenSinkExecutor")
+	}
+	actions := executor.TestInput.WorkflowInput.InitialActions
+	if len(actions) != 1 || len(actions[0].Actions) != 2 {
+		t.Fatalf("got action sets %#v, want one activity and one return", actions)
+	}
+	generic := actions[0].Actions[0].GetExecActivity().GetGeneric()
+	if generic == nil || generic.Type != "matched_resource" {
+		t.Fatalf("got generic activity %#v, want matched_resource", generic)
+	}
+	if actions[0].Actions[1].GetReturnResult() == nil {
+		t.Fatal("matched resource workflow does not return a result")
+	}
+}

scenarios/matched_fuzzer.go

@@ -0,0 +1,31 @@
+package scenarios
+
+import (
+	"context"
+
+	"github.com/temporalio/omes/loadgen"
+	"github.com/temporalio/omes/loadgen/kitchensink"
+)
+
+func init() {
+	loadgen.MustRegisterScenario(loadgen.Scenario{
+		Description: "Each iteration follows a deterministic one-, two-, or three-activity path.",
+		ExecutorFn: func() loadgen.Executor {
+			return loadgen.KitchenSinkExecutor{
+				TestInput: &kitchensink.TestInput{
+					WorkflowInput: &kitchensink.WorkflowInput{},
+				},
+				UpdateWorkflowOptions: func(_ context.Context, run *loadgen.Run, options *loadgen.KitchenSinkWorkflowOptions) error {
+					pathLength := run.Iteration%3 + 1
+					actionSet := &kitchensink.ActionSet{}
+					for i := 0; i < pathLength; i++ {
+						actionSet.Actions = append(actionSet.Actions, kitchensink.NoOpSingleActivityActionSet().Actions[0])
+					}
+					actionSet.Actions = append(actionSet.Actions, kitchensink.NoOpSingleActivityActionSet().Actions[1])
+					options.Params.WorkflowInput.InitialActions = []*kitchensink.ActionSet{actionSet}
+					return nil
+				},
+			}
+		},
+	})
+}