Verify rollback ownership across continue-as-new

chasm/lib/scheduler/migration_handoff_test.go

@@ -14,6 +14,7 @@ import (
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/historyservicemock/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler"
@@ -34,6 +35,7 @@ type migrationCloseFaultEngine struct {
 	chasm.Engine
 	afterCommit bool
 	armed       bool
+	skip        int
 }
 
 func requireMigrationCounterexamples(t *testing.T) {
@@ -46,6 +48,10 @@ func requireMigrationCounterexamples(t *testing.T) {
 func (e *migrationCloseFaultEngine) UpdateComponent(
 	ctx context.Context, ref chasm.ComponentRef, update func(chasm.MutableContext, chasm.Component) error, opts ...chasm.TransitionOption,
 ) ([]byte, error) {
+	if e.skip > 0 {
+		e.skip--
+		return e.Engine.UpdateComponent(ctx, ref, update, opts...)
+	}
 	if !e.armed {
 		return e.Engine.UpdateComponent(ctx, ref, update, opts...)
 	}
@@ -105,7 +111,7 @@ func TestMigrationScenario_RollbackRetryBoundaries(t *testing.T) {
 	for _, boundary := range []string{"success", "before_destination", "after_destination", "before_source_close", "after_source_close"} {
 		t.Run(boundary, func(t *testing.T) {
 			e, handler, historyClient := newRollbackScenario(t)
-			faultEngine := &migrationCloseFaultEngine{Engine: e.engine,
+			faultEngine := &migrationCloseFaultEngine{Engine: e.engine, skip: 1,
 				armed: boundary == "before_source_close" || boundary == "after_source_close", afterCommit: boundary == "after_source_close"}
 			var destination *schedulespb.StartScheduleArgs
 			var destinationRequestID string
@@ -132,7 +138,26 @@ func TestMigrationScenario_RollbackRetryBoundaries(t *testing.T) {
 					}
 					return &historyservice.StartWorkflowExecutionResponse{RunId: "destination-run"}, nil
 				})
+			historyClient.EXPECT().GetMutableState(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
+				func(context.Context, *historyservice.GetMutableStateRequest, ...grpc.CallOption) (*historyservice.GetMutableStateResponse, error) {
+					if destination == nil {
+						return nil, serviceerror.NewNotFound("no destination")
+					}
+					return &historyservice.GetMutableStateResponse{FirstExecutionRunId: "destination-run"}, nil
+				})
+			historyClient.EXPECT().DescribeMutableState(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
+				func(context.Context, *historyservice.DescribeMutableStateRequest, ...grpc.CallOption) (*historyservice.DescribeMutableStateResponse, error) {
+					return &historyservice.DescribeMutableStateResponse{DatabaseMutableState: &persistencespb.WorkflowMutableState{ExecutionState: &persistencespb.WorkflowExecutionState{RunId: "destination-run", CreateRequestId: destinationRequestID}}}, nil
+				})
 			dropped, err := executeRollbackTask(t, e, handler, faultEngine)
+			if boundary == "before_destination" {
+				require.Error(t, err)
+				_, err = executeRollbackTask(t, e, handler, faultEngine)
+				require.Error(t, err)
+				require.Zero(t, creations)
+				return
+			}
+
 			require.False(t, dropped)
 			if boundary == "success" {
 				require.NoError(t, err)

chasm/lib/scheduler/migration_rollback_chain_test.go

@@ -0,0 +1,106 @@
+package scheduler_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/api/historyservice/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/chasm/lib/scheduler"
+	"go.uber.org/mock/gomock"
+)
+
+type rollbackReceiptFaultEngine struct {
+	chasm.Engine
+	skip        int
+	afterCommit bool
+	armed       bool
+}
+
+func (e *rollbackReceiptFaultEngine) UpdateComponent(ctx context.Context, ref chasm.ComponentRef, update func(chasm.MutableContext, chasm.Component) error, opts ...chasm.TransitionOption) ([]byte, error) {
+	if e.armed && e.skip == 0 {
+		e.armed = false
+		if e.afterCommit {
+			if _, err := e.Engine.UpdateComponent(ctx, ref, update, opts...); err != nil {
+				return nil, err
+			}
+		}
+		return nil, serviceerror.NewUnavailable("injected receipt/close failure")
+	}
+	e.skip--
+	return e.Engine.UpdateComponent(ctx, ref, update, opts...)
+}
+
+func TestRollbackChainOwnership(t *testing.T) {
+	for _, scenario := range []string{"owned_current", "owned_descendant", "foreign_chain", "first_run_removed", "response_loss", "source_close_loss", "receipt_response_loss", "closed_descendant"} {
+		t.Run(scenario, func(t *testing.T) {
+			e, handler, history := newRollbackScenario(t)
+			engine := &rollbackReceiptFaultEngine{Engine: e.engine, armed: scenario == "source_close_loss" || scenario == "receipt_response_loss", afterCommit: scenario == "receipt_response_loss"}
+			if scenario == "closed_descendant" {
+				require.NoError(t, e.updateScheduler(func(s *scheduler.Scheduler, _ chasm.MutableContext) error {
+					s.WorkflowMigration.StartPending = false
+					return nil
+				}))
+			}
+			if scenario == "source_close_loss" {
+				engine.skip = 2
+			}
+			if scenario == "receipt_response_loss" {
+				engine.skip = 1
+			}
+			if scenario == "response_loss" {
+				history.EXPECT().StartWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, serviceerror.NewUnavailable("committed start response lost")).Times(1)
+				_, err := executeRollbackTask(t, e, handler, engine)
+				require.Error(t, err)
+			}
+			if scenario == "source_close_loss" || scenario == "receipt_response_loss" {
+				history.EXPECT().StartWorkflowExecution(gomock.Any(), gomock.Any()).Return(&historyservice.StartWorkflowExecutionResponse{RunId: "first-run", FirstExecutionRunId: "first-run"}, nil).Times(1)
+			} else {
+				owner, run := "descendant-request", "descendant-run"
+				if scenario == "owned_current" {
+					owner, run = "rollback", "first-run"
+				}
+				if scenario != "response_loss" && scenario != "closed_descendant" {
+					history.EXPECT().StartWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, serviceerror.NewWorkflowExecutionAlreadyStarted("collision", owner, run)).Times(1)
+				} else {
+					run = ""
+				}
+				if scenario != "owned_current" {
+					history.EXPECT().GetMutableState(gomock.Any(), &historyservice.GetMutableStateRequest{NamespaceId: namespaceID, Execution: &commonpb.WorkflowExecution{WorkflowId: "temporal-sys-scheduler:" + scheduleID, RunId: run}}).Return(&historyservice.GetMutableStateResponse{FirstExecutionRunId: "first-run"}, nil).Times(1)
+					call := history.EXPECT().DescribeMutableState(gomock.Any(), &historyservice.DescribeMutableStateRequest{NamespaceId: namespaceID, Execution: &commonpb.WorkflowExecution{WorkflowId: "temporal-sys-scheduler:" + scheduleID, RunId: "first-run"}}).Times(1)
+					if scenario == "first_run_removed" {
+						call.Return(nil, serviceerror.NewNotFound("first run expired"))
+					} else {
+						firstOwner := "rollback"
+						if scenario == "foreign_chain" {
+							firstOwner = "foreign-owner"
+						}
+						call.Return(&historyservice.DescribeMutableStateResponse{DatabaseMutableState: &persistencespb.WorkflowMutableState{ExecutionState: &persistencespb.WorkflowExecutionState{RunId: "first-run", CreateRequestId: firstOwner}}}, nil)
+					}
+				}
+			}
+			_, err := executeRollbackTask(t, e, handler, engine)
+			if scenario == "source_close_loss" || scenario == "receipt_response_loss" {
+				require.Error(t, err)
+				_, err = executeRollbackTask(t, e, handler, engine)
+			}
+			conflict := scenario == "foreign_chain" || scenario == "first_run_removed"
+			if conflict {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			require.NoError(t, e.readScheduler(func(s *scheduler.Scheduler, _ chasm.Context) error {
+				require.Equal(t, !conflict, s.Closed)
+				if conflict {
+					require.NotNil(t, s.WorkflowMigration)
+				}
+				return nil
+			}))
+		})
+	}
+}

chasm/lib/scheduler/proto/v1/message.proto

@@ -71,6 +71,12 @@ message WorkflowMigrationState {
 
   // Stable request ID used to start and reconcile the destination workflow.
   string request_id = 3;
+
+  // Durable acceptance receipt; prevents recreating a destination after it closes.
+  string destination_first_run_id = 4;
+
+  // Consumed durably before the only destination start attempt. Missing means reconcile only.
+  bool start_pending = 5;
 }
 
 // CHASM scheduler's Generator internal state.

chasm/lib/scheduler/scheduler.go

@@ -984,6 +984,7 @@ func (s *Scheduler) MigrateToWorkflow(
 	// Save pre-migration paused state, mark migration as pending, then pause.
 	s.WorkflowMigration = &schedulerpb.WorkflowMigrationState{
 		PreMigrationPaused: s.Schedule.State.Paused,
+		StartPending:       true,
 		PreMigrationNotes:  s.Schedule.State.Notes,
 		RequestId:          req.GetRequestId(),
 	}