From 24744dee48c92e49d8f2f4a4951de420b9e88db6 Mon Sep 17 00:00:00 2001 From: lilydoar Date: Mon, 26 Oct 2026 13:50:27 -0700 Subject: [PATCH 1/3] Retain original throughput starts across iteration retries --- loadgen/kitchen_sink_executor.go | 7 + loadgen/kitchen_sink_start.go | 18 + loadgen/kitchen_sink_start_test.go | 58 +++ loadgen/kitchensink/client_action_executor.go | 48 ++- loadgen/kitchensink/original_start.go | 73 ++++ loadgen/kitchensink/original_start_test.go | 336 ++++++++++++++++++ loadgen/scenario.go | 14 +- .../throughput_start_compatibility_test.go | 37 ++ scenarios/throughput_stress.go | 1 + 9 files changed, 589 insertions(+), 3 deletions(-) create mode 100644 loadgen/kitchen_sink_start.go create mode 100644 loadgen/kitchen_sink_start_test.go create mode 100644 loadgen/kitchensink/original_start.go create mode 100644 loadgen/kitchensink/original_start_test.go create mode 100644 scenarios/throughput_start_compatibility_test.go diff --git a/loadgen/kitchen_sink_executor.go b/loadgen/kitchen_sink_executor.go index 3bd4ce88f..e8bbf5abd 100644 --- a/loadgen/kitchen_sink_executor.go +++ b/loadgen/kitchen_sink_executor.go @@ -10,6 +10,9 @@ import ( type KitchenSinkExecutor struct { TestInput *kitchensink.TestInput + // RecoverOriginalStart retains the iteration request and acknowledged execution across retries. + RecoverOriginalStart bool + // Called once on start PrepareTestInput func(context.Context, ScenarioInfo, *kitchensink.TestInput) error @@ -27,7 +30,11 @@ func (k KitchenSinkExecutor) Run(ctx context.Context, info ScenarioInfo) error { // Create generic executor and run it ge := &GenericExecutor{ Execute: func(ctx context.Context, run *Run) error { + if k.RecoverOriginalStart && run.kitchenSinkExecution != nil { + return run.ExecuteKitchenSinkWorkflow(ctx, &run.kitchenSinkExecution.options) + } options := run.DefaultKitchenSinkWorkflowOptions() + options.recoverOriginalStart = k.RecoverOriginalStart testInputClone, ok := proto.Clone(k.TestInput).(*kitchensink.TestInput) if !ok { panic("failed to clone test input") diff --git a/loadgen/kitchen_sink_start.go b/loadgen/kitchen_sink_start.go new file mode 100644 index 000000000..6ca26f954 --- /dev/null +++ b/loadgen/kitchen_sink_start.go @@ -0,0 +1,18 @@ +package loadgen + +import ( + "github.com/golang/protobuf/proto" + "github.com/temporalio/omes/loadgen/kitchensink" +) + +type kitchenSinkExecution struct { + options KitchenSinkWorkflowOptions + executor *kitchensink.ClientActionsExecutor +} + +func freezeKitchenSinkExecution(options KitchenSinkWorkflowOptions, executor *kitchensink.ClientActionsExecutor) *kitchenSinkExecution { + options.Params = proto.Clone(options.Params).(*kitchensink.TestInput) + executor.WorkflowInput = options.Params.GetWorkflowInput() + executor.RecoverOriginalStart = true + return &kitchenSinkExecution{options: options, executor: executor} +} diff --git a/loadgen/kitchen_sink_start_test.go b/loadgen/kitchen_sink_start_test.go new file mode 100644 index 000000000..2ef7fe5b8 --- /dev/null +++ b/loadgen/kitchen_sink_start_test.go @@ -0,0 +1,58 @@ +package loadgen + +import ( + "context" + "reflect" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "github.com/temporalio/omes/loadgen/kitchensink" + "go.temporal.io/api/serviceerror" + "go.temporal.io/sdk/client" + "go.uber.org/zap" +) + +type startInputsClient struct { + client.Client + ids []string + inputs []int32 +} + +func (c *startInputsClient) ExecuteWorkflow(ctx context.Context, opts client.StartWorkflowOptions, _ interface{}, args ...interface{}) (client.WorkflowRun, error) { + c.ids = append(c.ids, opts.ID) + c.inputs = append(c.inputs, args[0].(*kitchensink.WorkflowInput).ExpectedSignalCount) + if len(c.ids) == 1 { + return nil, serviceerror.NewDeadlineExceeded("start response unavailable") + } + return completedOriginalRun{id: opts.ID}, nil +} + +type completedOriginalRun struct { + client.WorkflowRun + id string +} + +func (r completedOriginalRun) GetID() string { return r.id } +func (r completedOriginalRun) GetRunID() string { return "original-run" } +func (r completedOriginalRun) Get(context.Context, interface{}) error { return nil } + +func TestKitchenSinkOriginalStartFreezesRetryConstruction(t *testing.T) { + cl := &startInputsClient{} + generation := int32(0) + executor := KitchenSinkExecutor{TestInput: &kitchensink.TestInput{WorkflowInput: &kitchensink.WorkflowInput{}}, UpdateWorkflowOptions: func(ctx context.Context, r *Run, o *KitchenSinkWorkflowOptions) error { + generation++ + o.StartOptions.ID = uuid.NewString() + o.Params.WorkflowInput.ExpectedSignalCount = generation + return nil + }} + // Keep the regression executable on the base, where this capability is absent. + if f := reflect.ValueOf(&executor).Elem().FieldByName("RecoverOriginalStart"); f.IsValid() { + f.SetBool(true) + } + info := ScenarioInfo{Client: cl, Namespace: "default", RunID: "frozen", ExecutionID: "same", Logger: zap.NewNop().Sugar(), MetricsHandler: client.MetricsNopHandler, Configuration: RunConfiguration{Iterations: 1, MaxConcurrent: 1, MaxIterationAttempts: 2, DoNotRegisterSearchAttributes: true}} + require.NoError(t, executor.Run(t.Context(), info)) + require.Len(t, cl.ids, 2) + require.Equal(t, cl.ids[0], cl.ids[1]) + require.Equal(t, []int32{1, 1}, cl.inputs) +} diff --git a/loadgen/kitchensink/client_action_executor.go b/loadgen/kitchensink/client_action_executor.go index 29f9c2456..1de707c96 100644 --- a/loadgen/kitchensink/client_action_executor.go +++ b/loadgen/kitchensink/client_action_executor.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" enumspb "go.temporal.io/api/enums/v1" + "go.temporal.io/api/serviceerror" "go.temporal.io/sdk/client" "go.temporal.io/sdk/converter" "go.temporal.io/sdk/workflow" @@ -22,6 +23,9 @@ type ClientActionsExecutor struct { WorkflowInput *WorkflowInput Handle client.WorkflowRun runID string + // RecoverOriginalStart rejects duplicate starts and retains acknowledged run/update handles. + RecoverOriginalStart bool + originalStart *originalStart } func (e *ClientActionsExecutor) Start( @@ -29,6 +33,30 @@ func (e *ClientActionsExecutor) Start( withStartAction *WithStartClientAction, ) error { var err error + if e.RecoverOriginalStart { + if err := ctx.Err(); err != nil { + return err + } + if e.originalStart == nil { + if withStartAction.GetDoSignal() != nil { + return errors.New("original start recovery does not support Signal-with-Start") + } + e.freezeStart(withStartAction) + } + withStartAction = e.originalStart.action + if e.originalStart.update != nil { + if err := e.originalStart.update.Get(ctx, nil); err != nil { + return err + } + } + if e.Handle != nil { + e.runID = e.Handle.GetRunID() + return nil + } + if e.originalStart.withStartErr != nil { + return e.originalStart.withStartErr + } + } if withStartAction == nil { e.Handle, err = e.Client.ExecuteWorkflow(ctx, e.WorkflowOptions, e.WorkflowType, e.WorkflowInput) } else if sig := withStartAction.GetDoSignal(); sig != nil { @@ -38,6 +66,16 @@ func (e *ClientActionsExecutor) Start( } else { return fmt.Errorf("unsupported with_start_action: %v", withStartAction.String()) } + if err != nil && e.RecoverOriginalStart { + if withStartAction == nil { + var already *serviceerror.WorkflowExecutionAlreadyStarted + if errors.As(err, &already) && already.RunId != "" { + err = e.attachOriginalStart(ctx, already.RunId) + } + } else { + e.originalStart.withStartErr = err + } + } if err != nil { return fmt.Errorf("failed to start kitchen sink workflow: %w", err) } @@ -170,7 +208,11 @@ func (e *ClientActionsExecutor) executeUpdateAction(ctx context.Context, upd *Do var handle client.WorkflowUpdateHandle if upd.WithStart { workflowOpts := e.WorkflowOptions - workflowOpts.WorkflowIDConflictPolicy = enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING + if !e.RecoverOriginalStart { + workflowOpts.WorkflowIDConflictPolicy = enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING + } else { + updateOpts.WaitForStage = client.WorkflowUpdateStageAccepted + } op := e.Client.NewWithStartWorkflowOperation(workflowOpts, e.WorkflowType, e.WorkflowInput) handle, err = e.Client.UpdateWithStartWorkflow(ctx, client.UpdateWithStartWorkflowOptions{ StartWorkflowOperation: op, @@ -181,6 +223,10 @@ func (e *ClientActionsExecutor) executeUpdateAction(ctx context.Context, upd *Do } if err == nil { + if upd.WithStart && e.RecoverOriginalStart { + e.originalStart.update = handle + run = e.Client.GetWorkflow(ctx, handle.WorkflowID(), handle.RunID()) + } err = handle.Get(ctx, nil) if upd.WithStart { run = e.Client.GetWorkflow(ctx, handle.WorkflowID(), handle.RunID()) diff --git a/loadgen/kitchensink/original_start.go b/loadgen/kitchensink/original_start.go new file mode 100644 index 000000000..0e0bdac59 --- /dev/null +++ b/loadgen/kitchensink/original_start.go @@ -0,0 +1,73 @@ +package kitchensink + +import ( + "context" + "errors" + "fmt" + + "github.com/google/uuid" + enumspb "go.temporal.io/api/enums/v1" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/converter" + "google.golang.org/protobuf/proto" +) + +const startIntentMemo = "omes.original-start" + +type originalStart struct { + intent string + action *WithStartClientAction + update client.WorkflowUpdateHandle + withStartErr error +} + +func (e *ClientActionsExecutor) freezeStart(action *WithStartClientAction) { + e.originalStart = &originalStart{intent: uuid.NewString()} + memo := make(map[string]interface{}, len(e.WorkflowOptions.Memo)+1) + for k, v := range e.WorkflowOptions.Memo { + memo[k] = v + } + memo[startIntentMemo] = e.originalStart.intent + e.WorkflowOptions.Memo = memo + e.WorkflowOptions.WorkflowExecutionErrorWhenAlreadyStarted = true + e.WorkflowOptions.WorkflowIDReusePolicy = enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE + e.WorkflowOptions.WorkflowIDConflictPolicy = enumspb.WORKFLOW_ID_CONFLICT_POLICY_FAIL + if upd := action.GetDoUpdate(); upd != nil && upd.UpdateId == "" { + upd.UpdateId = uuid.NewString() + } + if action != nil { + e.originalStart.action = proto.Clone(action).(*WithStartClientAction) + } +} + +func (e *ClientActionsExecutor) attachOriginalStart(ctx context.Context, runID string) error { + if err := ctx.Err(); err != nil { + return err + } + history := e.Client.GetWorkflowHistory(ctx, e.WorkflowOptions.ID, runID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT) + event, err := history.Next() + if err != nil { + return err + } + attrs := event.GetWorkflowExecutionStartedEventAttributes() + if attrs == nil { + return errors.New("original start history is missing its start event") + } + if attrs.GetMemo().GetFields()[startIntentMemo] == nil { + return errors.New("existing workflow has no original start identity") + } + var intent string + if err := converter.GetDefaultDataConverter().FromPayload(attrs.GetMemo().GetFields()[startIntentMemo], &intent); err != nil { + return fmt.Errorf("reading original start identity: %w", err) + } + input, err := converter.GetDefaultDataConverter().ToPayloads(e.WorkflowInput) + if err != nil { + return err + } + if intent != e.originalStart.intent || attrs.GetWorkflowType().GetName() != e.WorkflowType || attrs.GetTaskQueue().GetName() != e.WorkflowOptions.TaskQueue || !proto.Equal(attrs.GetInput(), input) { + return errors.New("existing workflow does not match the original start intent") + } + // Bind the history-verified run, never a later execution selected by workflow ID. + e.Handle = e.Client.GetWorkflow(ctx, e.WorkflowOptions.ID, runID) + return nil +} diff --git a/loadgen/kitchensink/original_start_test.go b/loadgen/kitchensink/original_start_test.go new file mode 100644 index 000000000..3915d5834 --- /dev/null +++ b/loadgen/kitchensink/original_start_test.go @@ -0,0 +1,336 @@ +package kitchensink + +import ( + "context" + "errors" + "os" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "github.com/temporalio/omes/devserver" + "github.com/temporalio/omes/internal/versions" + enumspb "go.temporal.io/api/enums/v1" + "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/temporal" + "go.temporal.io/sdk/worker" + "go.temporal.io/sdk/workflow" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// Optional by reflection so the same regression executes against the base implementation. +func enableOriginalStart(executor *ClientActionsExecutor) { + if field := reflect.ValueOf(executor).Elem().FieldByName("RecoverOriginalStart"); field.IsValid() { + field.SetBool(true) + } +} + +type responseLossClient struct { + client.Client + starts int + multis int + original client.WorkflowRun + loseMulti bool + loseOutcome bool + updateID string +} + +func (c *responseLossClient) ExecuteWorkflow(ctx context.Context, opts client.StartWorkflowOptions, wf interface{}, args ...interface{}) (client.WorkflowRun, error) { + c.starts++ + h, err := c.Client.ExecuteWorkflow(ctx, opts, wf, args...) + if err == nil && c.starts == 1 { + c.original = h + return nil, serviceerror.NewDeadlineExceeded("lost accepted start reply") + } + return h, err +} +func (c *responseLossClient) UpdateWithStartWorkflow(ctx context.Context, opts client.UpdateWithStartWorkflowOptions) (client.WorkflowUpdateHandle, error) { + c.multis++ + c.updateID = opts.UpdateOptions.UpdateID + h, err := c.Client.UpdateWithStartWorkflow(ctx, opts) + if err != nil { + return nil, err + } + if c.loseMulti { + return nil, serviceerror.NewDeadlineExceeded("lost multi-operation reply") + } + if c.loseOutcome { + return &outcomeLossHandle{WorkflowUpdateHandle: h}, nil + } + return h, nil +} + +type outcomeLossHandle struct { + client.WorkflowUpdateHandle + calls atomic.Int32 +} + +func (h *outcomeLossHandle) Get(ctx context.Context, v interface{}) error { + if h.calls.Add(1) == 1 { + return serviceerror.NewDeadlineExceeded("lost update outcome reply") + } + return h.WorkflowUpdateHandle.Get(ctx, v) +} + +func originalStartWorkflow(ctx workflow.Context, input *WorkflowInput) error { + updates := 0 + if err := workflow.SetQueryHandler(ctx, "updates", func() (int, error) { return updates, nil }); err != nil { + return err + } + if err := workflow.SetUpdateHandler(ctx, "do_actions_update", func(ctx workflow.Context, _ *DoActionsUpdate) error { + updates++ + if input.ExpectedSignalCount == 42 { + return temporal.NewApplicationError("update failed", "UpdateFailure") + } + return workflow.Sleep(ctx, 50*time.Millisecond) + }); err != nil { + return err + } + workflow.GetSignalChannel(ctx, "finish").Receive(ctx, nil) + if input.ExpectedSignalCount < 0 { + return temporal.NewNonRetryableApplicationError("original target failure", "TargetFailure", nil) + } + return nil +} + +func TestOriginalStartProtocol(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 4*time.Minute) + defer cancel() + ref, err := versions.Get("SERVER_VERSION") + require.NoError(t, err) + options := devserver.Options{Ref: ref, OutputDir: os.Getenv("OMES_START_TEST_SERVER_CACHE")} + if source := os.Getenv("OMES_START_TEST_SERVER_SOURCE"); source != "" { + options.Ref = "" + options.SourceDir = source + } + server, err := devserver.Start(ctx, options) + require.NoError(t, err) + defer func() { require.NoError(t, server.Stop()) }() + c, err := client.Dial(client.Options{HostPort: server.FrontendHostPort(), Namespace: "default"}) + require.NoError(t, err) + defer c.Close() + queue := "original-start-" + uuid.NewString() + w := worker.New(c, queue, worker.Options{}) + w.RegisterWorkflowWithOptions(originalStartWorkflow, workflow.RegisterOptions{Name: "kitchenSink"}) + require.NoError(t, w.Start()) + defer w.Stop() + newExecutor := func(cl client.Client) *ClientActionsExecutor { + e := &ClientActionsExecutor{Client: cl, Namespace: "default", WorkflowType: "kitchenSink", WorkflowInput: &WorkflowInput{}, WorkflowOptions: client.StartWorkflowOptions{ID: uuid.NewString(), TaskQueue: queue, WorkflowExecutionErrorWhenAlreadyStarted: true}} + enableOriginalStart(e) + return e + } + for _, state := range []string{"running", "completed", "failed"} { + t.Run("lost_response_"+state, func(t *testing.T) { + lost := &responseLossClient{Client: c} + e := newExecutor(lost) + if state == "failed" { + e.WorkflowInput.ExpectedSignalCount = -1 + } + require.Error(t, e.Start(ctx, nil)) + original := lost.original + require.NotNil(t, original) + if state != "running" { + require.NoError(t, c.SignalWorkflow(ctx, original.GetID(), original.GetRunID(), "finish", nil)) + getErr := original.Get(ctx, nil) + if state == "failed" { + require.Error(t, getErr) + } else { + require.NoError(t, getErr) + } + } + require.NoError(t, e.Start(ctx, nil)) + require.Equal(t, original.GetRunID(), e.Handle.GetRunID(), "must never replace the original closed execution") + if state == "running" { + require.NoError(t, c.SignalWorkflow(ctx, original.GetID(), original.GetRunID(), "finish", nil)) + } + getErr := e.Handle.Get(ctx, nil) + if state == "failed" { + var failure *temporal.ApplicationError + require.ErrorAs(t, getErr, &failure) + require.Equal(t, "TargetFailure", failure.Type()) + require.NoError(t, e.Start(ctx, nil)) + require.Error(t, e.Handle.Get(ctx, nil)) + } else { + require.NoError(t, getErr) + } + }) + } + t.Run("foreign_reuse_and_mismatched_input", func(t *testing.T) { + lost := &responseLossClient{Client: c} + e := newExecutor(lost) + require.Error(t, e.Start(ctx, nil)) + original := lost.original + require.NoError(t, c.SignalWorkflow(ctx, original.GetID(), original.GetRunID(), "finish", nil)) + require.NoError(t, original.Get(ctx, nil)) + opts := e.WorkflowOptions + opts.WorkflowIDReusePolicy = enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE + foreign, err := c.ExecuteWorkflow(ctx, opts, "kitchenSink", &WorkflowInput{ExpectedSignalCount: 99}) + require.NoError(t, err) + require.Error(t, e.Start(ctx, nil)) + require.Nil(t, e.Handle) + require.NoError(t, c.SignalWorkflow(ctx, foreign.GetID(), foreign.GetRunID(), "finish", nil)) + require.NoError(t, foreign.Get(ctx, nil)) + }) + t.Run("foreign_existing", func(t *testing.T) { + e := newExecutor(c) + foreign, err := c.ExecuteWorkflow(ctx, e.WorkflowOptions, "kitchenSink", e.WorkflowInput) + require.NoError(t, err) + require.Error(t, e.Start(ctx, nil)) + require.Nil(t, e.Handle) + require.NoError(t, c.SignalWorkflow(ctx, foreign.GetID(), foreign.GetRunID(), "finish", nil)) + require.NoError(t, foreign.Get(ctx, nil)) + }) + t.Run("canceled", func(t *testing.T) { + e := newExecutor(c) + canceled, stop := context.WithCancel(ctx) + stop() + require.Error(t, e.Start(canceled, nil)) + require.Nil(t, e.Handle) + }) + action := func() *WithStartClientAction { + return &WithStartClientAction{Variant: &WithStartClientAction_DoUpdate{DoUpdate: &DoUpdate{WithStart: true, Variant: &DoUpdate_DoActions{DoActions: &DoActionsUpdate{Variant: &DoActionsUpdate_DoActions{}}}}}} + } + t.Run("uws_confirmed_update_outcome", func(t *testing.T) { + lost := &responseLossClient{Client: c, loseOutcome: true} + e := newExecutor(lost) + a := action() + require.Error(t, e.Start(ctx, a)) + require.NoError(t, e.Start(ctx, a)) + require.Equal(t, 1, lost.multis) + require.NotEmpty(t, a.GetDoUpdate().GetUpdateId()) + require.Equal(t, lost.updateID, a.GetDoUpdate().GetUpdateId()) + require.NoError(t, c.SignalWorkflow(ctx, e.Handle.GetID(), e.Handle.GetRunID(), "finish", nil)) + require.NoError(t, e.Handle.Get(ctx, nil)) + }) + t.Run("uws_unknown_acceptance_is_not_replayed", func(t *testing.T) { + lost := &responseLossClient{Client: c, loseMulti: true} + e := newExecutor(lost) + a := action() + require.Error(t, e.Start(ctx, a)) + require.Error(t, e.Start(ctx, a)) + require.Equal(t, 1, lost.multis) + require.NotEmpty(t, a.GetDoUpdate().GetUpdateId()) + canceled, stop := context.WithCancel(ctx) + stop() + require.True(t, errors.Is(e.Start(canceled, a), context.Canceled)) + require.NoError(t, c.SignalWorkflow(ctx, e.WorkflowOptions.ID, "", "finish", nil)) + require.NoError(t, c.GetWorkflow(ctx, e.WorkflowOptions.ID, "").Get(ctx, nil)) + }) + + t.Run("uws_closed_target_failure_is_not_replaced", func(t *testing.T) { + observed := &responseLossClient{Client: c} + e := newExecutor(observed) + e.WorkflowInput.ExpectedSignalCount = -1 + a := action() + require.NoError(t, e.Start(ctx, a)) + original := e.Handle.GetRunID() + require.NoError(t, c.SignalWorkflow(ctx, e.Handle.GetID(), original, "finish", nil)) + require.Error(t, e.Handle.Get(ctx, nil)) + require.NoError(t, e.Start(ctx, a)) + require.Equal(t, original, e.Handle.GetRunID()) + require.Equal(t, 1, observed.multis) + require.Error(t, e.Handle.Get(ctx, nil)) + }) + t.Run("uws_failed_update_outcome_is_preserved", func(t *testing.T) { + observed := &responseLossClient{Client: c} + e := newExecutor(observed) + e.WorkflowInput.ExpectedSignalCount = 42 + a := action() + require.Error(t, e.Start(ctx, a)) + require.Error(t, e.Start(ctx, a)) + require.Equal(t, 1, observed.multis) + require.NoError(t, c.SignalWorkflow(ctx, e.WorkflowOptions.ID, "", "finish", nil)) + require.NoError(t, c.GetWorkflow(ctx, e.WorkflowOptions.ID, "").Get(ctx, nil)) + }) + t.Run("uws_foreign_closed_failed_is_not_replaced", func(t *testing.T) { + e := newExecutor(c) + e.WorkflowInput.ExpectedSignalCount = -1 + foreign, err := c.ExecuteWorkflow(ctx, e.WorkflowOptions, "kitchenSink", e.WorkflowInput) + require.NoError(t, err) + require.NoError(t, c.SignalWorkflow(ctx, foreign.GetID(), foreign.GetRunID(), "finish", nil)) + require.Error(t, foreign.Get(ctx, nil)) + require.Error(t, e.Start(ctx, action())) + require.Nil(t, e.Handle) + }) + t.Run("uws_wire_reply_loss_and_external_reuse", func(t *testing.T) { + var lost atomic.Bool + var mu sync.Mutex + requests := map[string]bool{} + foreignRuns := make(chan client.WorkflowRun, 1) + intercept := func(callCtx context.Context, method string, req, reply interface{}, conn *grpc.ClientConn, invoke grpc.UnaryInvoker, opts ...grpc.CallOption) error { + if !strings.HasSuffix(method, "/ExecuteMultiOperation") { + return invoke(callCtx, method, req, reply, conn, opts...) + } + request := req.(*workflowservice.ExecuteMultiOperationRequest) + start := request.GetOperations()[0].GetStartWorkflow() + mu.Lock() + requests[start.GetRequestId()] = true + mu.Unlock() + err := invoke(callCtx, method, req, reply, conn, opts...) + if err == nil && !lost.Swap(true) { + originalRun := reply.(*workflowservice.ExecuteMultiOperationResponse).GetResponses()[0].GetStartWorkflow().GetRunId() + if err := c.SignalWorkflow(ctx, start.GetWorkflowId(), originalRun, "finish", nil); err != nil { + return err + } + if err := c.GetWorkflow(ctx, start.GetWorkflowId(), originalRun).Get(ctx, nil); err != nil { + return err + } + foreign, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ID: start.GetWorkflowId(), TaskQueue: queue, WorkflowIDReusePolicy: enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE}, "kitchenSink", &WorkflowInput{ExpectedSignalCount: 99}) + if err != nil { + return err + } + foreignRuns <- foreign + return status.Error(codes.DeadlineExceeded, "lost real multi-operation reply after external ID reuse") + } + return err + } + wire, err := client.Dial(client.Options{HostPort: server.FrontendHostPort(), Namespace: "default", ConnectionOptions: client.ConnectionOptions{DialOptions: []grpc.DialOption{grpc.WithChainUnaryInterceptor(intercept)}}}) + require.NoError(t, err) + defer wire.Close() + e := newExecutor(wire) + a := action() + firstErr := e.Start(ctx, a) + secondErr := e.Start(ctx, a) + require.True(t, lost.Load(), "real successful multi-operation response must have been discarded") + var foreign client.WorkflowRun + select { + case foreign = <-foreignRuns: + case <-time.After(time.Second): + t.Fatal("external reuse was not established") + } + value, err := c.QueryWorkflow(ctx, foreign.GetID(), foreign.GetRunID(), "updates") + require.NoError(t, err) + var updates int + require.NoError(t, value.Get(&updates)) + require.Zero(t, updates, "recovery must not update a foreign replacement run") + mu.Lock() + requestCount := len(requests) + mu.Unlock() + require.Equal(t, 1, requestCount, "outer recovery must not reconstruct a new multi-operation request") + if firstErr != nil { + require.Error(t, secondErr) + } + require.NoError(t, c.SignalWorkflow(ctx, foreign.GetID(), foreign.GetRunID(), "finish", nil)) + require.NoError(t, foreign.Get(ctx, nil)) + }) + + t.Run("uws_foreign_is_not_updated", func(t *testing.T) { + e := newExecutor(c) + foreign, err := c.ExecuteWorkflow(ctx, e.WorkflowOptions, "kitchenSink", e.WorkflowInput) + require.NoError(t, err) + err = e.Start(ctx, action()) + require.Error(t, err) + require.Nil(t, e.Handle) + require.NoError(t, c.SignalWorkflow(ctx, foreign.GetID(), foreign.GetRunID(), "finish", nil)) + require.NoError(t, foreign.Get(ctx, nil)) + }) +} diff --git a/loadgen/scenario.go b/loadgen/scenario.go index 16318c3ec..1860af544 100644 --- a/loadgen/scenario.go +++ b/loadgen/scenario.go @@ -396,6 +396,8 @@ type Run struct { // tracks how many attempts have been made for this iteration attemptCount int + + kitchenSinkExecution *kitchenSinkExecution } // NewRun creates a new run. @@ -534,8 +536,9 @@ func (r *Run) ShouldRetry(err error) (time.Duration, bool) { } type KitchenSinkWorkflowOptions struct { - Params *kitchensink.TestInput - StartOptions client.StartWorkflowOptions + recoverOriginalStart bool + Params *kitchensink.TestInput + StartOptions client.StartWorkflowOptions } // ExecuteKitchenSinkWorkflow starts the generic "kitchen sink" workflow and waits for its @@ -553,6 +556,13 @@ func (r *Run) ExecuteKitchenSinkWorkflow(ctx context.Context, options *KitchenSi WorkflowType: "kitchenSink", WorkflowInput: options.Params.GetWorkflowInput(), } + if options.recoverOriginalStart { + if r.kitchenSinkExecution == nil { + r.kitchenSinkExecution = freezeKitchenSinkExecution(*options, executor) + } + options = &r.kitchenSinkExecution.options + executor = r.kitchenSinkExecution.executor + } startErr := executor.Start(ctx, options.Params.WithStartAction) if startErr != nil { return fmt.Errorf("failed to start kitchen sink workflow: %w", startErr) diff --git a/scenarios/throughput_start_compatibility_test.go b/scenarios/throughput_start_compatibility_test.go new file mode 100644 index 000000000..38808729c --- /dev/null +++ b/scenarios/throughput_start_compatibility_test.go @@ -0,0 +1,37 @@ +package scenarios + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/temporalio/omes/clioptions" + "github.com/temporalio/omes/internal/workertest" + "github.com/temporalio/omes/loadgen" + enumspb "go.temporal.io/api/enums/v1" +) + +func TestThroughputLegacyStartAndSnapshotResume(t *testing.T) { + env := workertest.SetupTestEnvironment(t, workertest.WithExecutorTimeout(time.Minute)) + id := fmt.Sprintf("legacy-start-%d", time.Now().UnixNano()) + info := loadgen.ScenarioInfo{RunID: id, ExecutionID: id, Configuration: loadgen.RunConfiguration{Iterations: 1, IgnoreAlreadyStarted: true}, Options: loadgen.MustResolveScenarioOptions("throughput_stress", map[string]string{IterFlag: "1", ContinueAsNewAfterIterFlag: "0", NexusEnabledFlag: "false", SleepTimeFlag: "1ms", VisibilityVerificationTimeoutFlag: "10s"})} + first := newThroughputStressExecutor() + _, err := env.RunExecutorTest(t, first, info, clioptions.LangGo) + require.NoError(t, err) + workflowID := fmt.Sprintf("w-%s-%s-1", id, id) + checkUnmarked := func() { + history := env.TemporalClient().GetWorkflowHistory(t.Context(), workflowID, "", false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT) + event, err := history.Next() + require.NoError(t, err) + require.NotContains(t, event.GetWorkflowExecutionStartedEventAttributes().GetMemo().GetFields(), "omes.original-start") + } + checkUnmarked() + resumed := newThroughputStressExecutor() + require.NoError(t, resumed.LoadState(func(v any) error { v.(*tpsState).CompletedIterations = 0; return nil })) + info.Configuration.IgnoreAlreadyStarted = false + _, err = env.RunExecutorTest(t, resumed, info, clioptions.LangGo) + require.NoError(t, err) + require.Equal(t, 1, resumed.Snapshot().(tpsState).CompletedIterations) + checkUnmarked() +} diff --git a/scenarios/throughput_stress.go b/scenarios/throughput_stress.go index d4baab0d4..2bae05d37 100644 --- a/scenarios/throughput_stress.go +++ b/scenarios/throughput_stress.go @@ -391,6 +391,7 @@ func (t *tpsExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) error mainResult.finishedAt = time.Now() } else { ksExec := &loadgen.KitchenSinkExecutor{ + RecoverOriginalStart: !isResuming && !info.Configuration.IgnoreAlreadyStarted, TestInput: &TestInput{ WorkflowInput: &WorkflowInput{ InitialActions: []*ActionSet{}, From 0df74b4abd4560cb3d38e942b6556b6b6f0bca45 Mon Sep 17 00:00:00 2001 From: lilydoar Date: Mon, 26 Oct 2026 14:03:50 -0700 Subject: [PATCH 2/3] Test original-start recovery without baseline compatibility scaffolding --- loadgen/kitchen_sink_start_test.go | 7 ++---- loadgen/kitchensink/original_start_test.go | 26 +++++----------------- 2 files changed, 7 insertions(+), 26 deletions(-) diff --git a/loadgen/kitchen_sink_start_test.go b/loadgen/kitchen_sink_start_test.go index 2ef7fe5b8..93b0a938c 100644 --- a/loadgen/kitchen_sink_start_test.go +++ b/loadgen/kitchen_sink_start_test.go @@ -2,7 +2,6 @@ package loadgen import ( "context" - "reflect" "testing" "github.com/google/uuid" @@ -46,10 +45,8 @@ func TestKitchenSinkOriginalStartFreezesRetryConstruction(t *testing.T) { o.Params.WorkflowInput.ExpectedSignalCount = generation return nil }} - // Keep the regression executable on the base, where this capability is absent. - if f := reflect.ValueOf(&executor).Elem().FieldByName("RecoverOriginalStart"); f.IsValid() { - f.SetBool(true) - } + executor.RecoverOriginalStart = true + info := ScenarioInfo{Client: cl, Namespace: "default", RunID: "frozen", ExecutionID: "same", Logger: zap.NewNop().Sugar(), MetricsHandler: client.MetricsNopHandler, Configuration: RunConfiguration{Iterations: 1, MaxConcurrent: 1, MaxIterationAttempts: 2, DoNotRegisterSearchAttributes: true}} require.NoError(t, executor.Run(t.Context(), info)) require.Len(t, cl.ids, 2) diff --git a/loadgen/kitchensink/original_start_test.go b/loadgen/kitchensink/original_start_test.go index 3915d5834..e0516d95d 100644 --- a/loadgen/kitchensink/original_start_test.go +++ b/loadgen/kitchensink/original_start_test.go @@ -2,9 +2,6 @@ package kitchensink import ( "context" - "errors" - "os" - "reflect" "strings" "sync" "sync/atomic" @@ -27,13 +24,6 @@ import ( "google.golang.org/grpc/status" ) -// Optional by reflection so the same regression executes against the base implementation. -func enableOriginalStart(executor *ClientActionsExecutor) { - if field := reflect.ValueOf(executor).Elem().FieldByName("RecoverOriginalStart"); field.IsValid() { - field.SetBool(true) - } -} - type responseLossClient struct { client.Client starts int @@ -107,12 +97,7 @@ func TestOriginalStartProtocol(t *testing.T) { defer cancel() ref, err := versions.Get("SERVER_VERSION") require.NoError(t, err) - options := devserver.Options{Ref: ref, OutputDir: os.Getenv("OMES_START_TEST_SERVER_CACHE")} - if source := os.Getenv("OMES_START_TEST_SERVER_SOURCE"); source != "" { - options.Ref = "" - options.SourceDir = source - } - server, err := devserver.Start(ctx, options) + server, err := devserver.Start(ctx, devserver.Options{Ref: ref}) require.NoError(t, err) defer func() { require.NoError(t, server.Stop()) }() c, err := client.Dial(client.Options{HostPort: server.FrontendHostPort(), Namespace: "default"}) @@ -125,7 +110,7 @@ func TestOriginalStartProtocol(t *testing.T) { defer w.Stop() newExecutor := func(cl client.Client) *ClientActionsExecutor { e := &ClientActionsExecutor{Client: cl, Namespace: "default", WorkflowType: "kitchenSink", WorkflowInput: &WorkflowInput{}, WorkflowOptions: client.StartWorkflowOptions{ID: uuid.NewString(), TaskQueue: queue, WorkflowExecutionErrorWhenAlreadyStarted: true}} - enableOriginalStart(e) + e.RecoverOriginalStart = true return e } for _, state := range []string{"running", "completed", "failed"} { @@ -221,7 +206,7 @@ func TestOriginalStartProtocol(t *testing.T) { require.NotEmpty(t, a.GetDoUpdate().GetUpdateId()) canceled, stop := context.WithCancel(ctx) stop() - require.True(t, errors.Is(e.Start(canceled, a), context.Canceled)) + require.ErrorIs(t, e.Start(canceled, a), context.Canceled) require.NoError(t, c.SignalWorkflow(ctx, e.WorkflowOptions.ID, "", "finish", nil)) require.NoError(t, c.GetWorkflow(ctx, e.WorkflowOptions.ID, "").Get(ctx, nil)) }) @@ -316,9 +301,8 @@ func TestOriginalStartProtocol(t *testing.T) { requestCount := len(requests) mu.Unlock() require.Equal(t, 1, requestCount, "outer recovery must not reconstruct a new multi-operation request") - if firstErr != nil { - require.Error(t, secondErr) - } + require.Error(t, firstErr) + require.Error(t, secondErr) require.NoError(t, c.SignalWorkflow(ctx, foreign.GetID(), foreign.GetRunID(), "finish", nil)) require.NoError(t, foreign.Get(ctx, nil)) }) From 9e32184e2246477c281717979adf49205e2142cf Mon Sep 17 00:00:00 2001 From: lilydoar Date: Thu, 29 Oct 2026 10:58:37 -0700 Subject: [PATCH 3/3] Separate original-start protocol budget from server setup --- loadgen/kitchensink/original_start_test.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/loadgen/kitchensink/original_start_test.go b/loadgen/kitchensink/original_start_test.go index e0516d95d..08b5b9649 100644 --- a/loadgen/kitchensink/original_start_test.go +++ b/loadgen/kitchensink/original_start_test.go @@ -93,14 +93,14 @@ func originalStartWorkflow(ctx workflow.Context, input *WorkflowInput) error { } func TestOriginalStartProtocol(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 4*time.Minute) - defer cancel() ref, err := versions.Get("SERVER_VERSION") require.NoError(t, err) - server, err := devserver.Start(ctx, devserver.Options{Ref: ref}) + server, err := devserver.Start(t.Context(), devserver.Options{Ref: ref}) require.NoError(t, err) - defer func() { require.NoError(t, server.Stop()) }() - c, err := client.Dial(client.Options{HostPort: server.FrontendHostPort(), Namespace: "default"}) + t.Cleanup(func() { require.NoError(t, server.Stop()) }) + ctx, cancel := context.WithTimeout(t.Context(), 4*time.Minute) + defer cancel() + c, err := client.DialContext(ctx, client.Options{HostPort: server.FrontendHostPort(), Namespace: "default"}) require.NoError(t, err) defer c.Close() queue := "original-start-" + uuid.NewString()