tasks/service/history/history_test.go at rescue/uncommitted

mirror of https://github.com/hanzoai/tasks.git synced 2026-11-03 03:56:04 +00:00

Files

T

Chetan GowdaandRoey Berman 922cd7445b [Nexus-CHASM] Add workflow methods to handle nexus events (#9546 )

## What changed?
This PR adds methods in workflow component to handle nexus events.

## Why?
Migrating Nexus from HSM to CHASM

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)



---

> [!NOTE]
> **Medium Risk**
> Touches the workflow-task completion path and CHASM workflow wiring
(registry injection and history-event application), so misregistration
or missing context could cause runtime command failures despite largely
being additive/refactor changes.
> 
> **Overview**
> Adds CHASM `Workflow` helpers to *emit and apply* Nexus operation
lifecycle events (started/completed/failed/canceled/timed-out),
including consistent failure wrapping via
`NexusOperationExecutionFailure`.
> 
> Refactors CHASM workflow command/event registration by moving
`workflowregistry` into `chasm/lib/workflow` as `Registry`, injecting it
into CHASM context, and updating Nexus workflow command handlers to use
`AddAndApplyHistoryEvent` so command-emitted history events immediately
run their registered event definitions. Updates wiring/tests/callers
across services to construct `NewLibrary(NewRegistry())` and use the new
types/errors.
> 
> <sup>Written by [Cursor
Bugbot]() for commit
70936c6edabf0b96307738c28d60ac2adbcf8b83. This will update automatically
on new commits. Configure
[here]().</sup>

2026-04-10 11:53:20 -07:00

208 lines

6.5 KiB

Go

Raw Permalink Blame History

 package history
 import (
 	"context"
 	"errors"
 	"testing"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/server/api/adminservice/v1"
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	historyspb "go.temporal.io/server/api/history/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
 	chasmworkflow "go.temporal.io/server/chasm/lib/workflow"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/persistence/visibility/manager"
 	"go.temporal.io/server/service/history/api/forcedeleteworkflowexecution"
 	"go.temporal.io/server/service/history/tests"
 	"go.uber.org/mock/gomock"
 )
 type (
 	historyAPISuite struct {
 		suite.Suite
 		*require.Assertions
 		controller         *gomock.Controller
 		mockExecutionMgr   *persistence.MockExecutionManager
 		mockVisibilityMgr  *manager.MockVisibilityManager
 		mockNamespaceCache *namespace.MockRegistry
 		chasmRegistry *chasm.Registry
 		logger        log.Logger
 	}
 )
 func TestHistoryAPISuite(t *testing.T) {
 	s := new(historyAPISuite)
 	suite.Run(t, s)
 }
 func (s *historyAPISuite) SetupSuite() {
 }
 func (s *historyAPISuite) TearDownSuite() {
 }
 func (s *historyAPISuite) SetupTest() {
 	s.Assertions = require.New(s.T())
 	s.controller = gomock.NewController(s.T())
 	s.mockExecutionMgr = persistence.NewMockExecutionManager(s.controller)
 	s.mockVisibilityMgr = manager.NewMockVisibilityManager(s.controller)
 	s.mockNamespaceCache = namespace.NewMockRegistry(s.controller)
 	s.mockNamespaceCache.EXPECT().GetNamespaceByID(tests.NamespaceID).Return(tests.LocalNamespaceEntry, nil).AnyTimes()
 	s.mockNamespaceCache.EXPECT().GetNamespace(tests.Namespace).Return(tests.LocalNamespaceEntry, nil).AnyTimes()
 	s.logger = log.NewTestLogger()
 	s.chasmRegistry = chasm.NewRegistry(s.logger)
 	err := s.chasmRegistry.Register(chasmworkflow.NewLibrary(chasmworkflow.NewRegistry()))
 	s.NoError(err)
 }
 func (s *historyAPISuite) TearDownTest() {
 	s.controller.Finish()
 }
 func (s *historyAPISuite) TestDeleteWorkflowExecution_DeleteCurrentExecution() {
 	execution := commonpb.WorkflowExecution{
 		WorkflowId: "workflowID",
 	}
 	shardID := common.WorkflowIDToHistoryShard(
 		tests.NamespaceID.String(),
 		execution.GetWorkflowId(),
 ,
 	)
 	request := &historyservice.ForceDeleteWorkflowExecutionRequest{
 		NamespaceId: tests.NamespaceID.String(),
 		ArchetypeId: chasm.WorkflowArchetypeID,
 		Request: &adminservice.DeleteWorkflowExecutionRequest{
 			Execution: &execution,
 			Archetype: chasm.WorkflowArchetype,
 		},
 	}
 	s.mockNamespaceCache.EXPECT().GetNamespaceID(tests.Namespace).Return(tests.NamespaceID, nil).AnyTimes()
 	s.mockVisibilityMgr.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).AnyTimes()
 	s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), gomock.Any()).Return(nil, errors.New("some random error"))
 	resp, err := forcedeleteworkflowexecution.Invoke(
 		context.Background(),
 		request,
 		shardID,
 		s.chasmRegistry,
 		s.mockExecutionMgr,
 		s.mockVisibilityMgr,
 		s.logger,
 	)
 	s.Nil(resp)
 	s.Error(err)
 	mutableState := &persistencespb.WorkflowMutableState{
 		ExecutionInfo: &persistencespb.WorkflowExecutionInfo{
 			VersionHistories: &historyspb.VersionHistories{
 				CurrentVersionHistoryIndex: 0,
 				Histories: []*historyspb.VersionHistory{
 					{BranchToken: []byte("branch1")},
 					{BranchToken: []byte("branch2")},
 					{BranchToken: []byte("branch3")},
 					{BranchToken: []byte{}},
 				},
 			},
 		},
 	}
 	runID := uuid.NewString()
 	s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), gomock.Any()).Return(&persistence.GetCurrentExecutionResponse{
 		StartRequestID: uuid.NewString(),
 		RunID:          runID,
 		State:          enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
 		Status:         enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
 	}, nil)
 	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), &persistence.GetWorkflowExecutionRequest{
 		ShardID:     shardID,
 		NamespaceID: tests.NamespaceID.String(),
 		WorkflowID:  execution.WorkflowId,
 		RunID:       runID,
 		ArchetypeID: chasm.WorkflowArchetypeID,
 	}).Return(&persistence.GetWorkflowExecutionResponse{State: mutableState}, nil)
 	s.mockExecutionMgr.EXPECT().DeleteCurrentWorkflowExecution(gomock.Any(), &persistence.DeleteCurrentWorkflowExecutionRequest{
 		ShardID:     shardID,
 		NamespaceID: tests.NamespaceID.String(),
 		WorkflowID:  execution.WorkflowId,
 		RunID:       runID,
 		ArchetypeID: chasm.WorkflowArchetypeID,
 	}).Return(nil)
 	s.mockExecutionMgr.EXPECT().DeleteWorkflowExecution(gomock.Any(), &persistence.DeleteWorkflowExecutionRequest{
 		ShardID:     shardID,
 		NamespaceID: tests.NamespaceID.String(),
 		WorkflowID:  execution.WorkflowId,
 		RunID:       runID,
 		ArchetypeID: chasm.WorkflowArchetypeID,
 	}).Return(nil)
 	s.mockExecutionMgr.EXPECT().DeleteHistoryBranch(gomock.Any(), gomock.Any()).Times(3)
 	_, err = forcedeleteworkflowexecution.Invoke(
 		context.Background(),
 		request,
 		shardID,
 		s.chasmRegistry,
 		s.mockExecutionMgr,
 		s.mockVisibilityMgr,
 		s.logger,
 	)
 	s.NoError(err)
 }
 func (s *historyAPISuite) TestDeleteWorkflowExecution_LoadMutableStateFailed() {
 	execution := commonpb.WorkflowExecution{
 		WorkflowId: "workflowID",
 		RunId:      uuid.NewString(),
 	}
 	shardID := common.WorkflowIDToHistoryShard(
 		tests.NamespaceID.String(),
 		execution.GetWorkflowId(),
 ,
 	)
 	request := &historyservice.ForceDeleteWorkflowExecutionRequest{
 		NamespaceId: tests.NamespaceID.String(),
 		ArchetypeId: chasm.WorkflowArchetypeID,
 		Request: &adminservice.DeleteWorkflowExecutionRequest{
 			Execution: &execution,
 			Archetype: chasm.WorkflowArchetype,
 		},
 	}
 	s.mockNamespaceCache.EXPECT().GetNamespaceID(tests.Namespace).Return(tests.NamespaceID, nil).AnyTimes()
 	s.mockVisibilityMgr.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).AnyTimes()
 	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, errors.New("some random error"))
 	s.mockExecutionMgr.EXPECT().DeleteCurrentWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil)
 	s.mockExecutionMgr.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil)
 	_, err := forcedeleteworkflowexecution.Invoke(
 		context.Background(),
 		request,
 		shardID,
 		s.chasmRegistry,
 		s.mockExecutionMgr,
 		s.mockVisibilityMgr,
 		s.logger,
 	)
 	s.NoError(err)
 }

208 lines 6.5 KiB Go Raw Permalink Blame History

208 lines

6.5 KiB

Go

Raw Permalink Blame History