hanzoai/tasks
2
0
Fork 0
mirror of https://github.com/hanzoai/tasks.git synced 2026-10-30 14:56:19 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
rescue/uncommitted
tasks/service/history/api/verifyfirstworkflowtaskscheduled/api_test.go
T
zeekay 40351cce6f rip: delete temporal fork — native ZAP-only tasksd 
Goes from 135 MB binary with 1153 files importing go.temporal.io and 224
importing google.golang.org/grpc to a 10 MB native binary with ZERO of
either.

Deleted:
- tasks/, tasktests/, service/, chasm/, client/, api/, proto/, tools/
- cmd/tools/, components/, docker/
- schema/{cassandra,elasticsearch,mysql,postgresql,sqlite}/
- 40 tainted common/ subdirs + 7 top-level common files
- tests/testutils (no longer referenced after surviving deps drop common)

New cmd/tasksd (~100 lines) boots tasks.Embed (zap.Node) + http.Server.
New pkg/tasks/embed.go registers ZAP handlers for the SDK opcode surface
(0x0060-0x00A5). Workflow handlers return a 501 "not yet implemented"
envelope; native engine lands behind this contract.

Build:  go build ./cmd/tasksd → 10 MB
Vet:    go vet ./... → clean
Boot:   tasksd --zap-port 9999 --http :7243
Probes: /healthz, /v1/tasks/health → 200
UI:     /_/tasks/, / → 200 (embedded Vite bundle)
ZAP:    _tasks._tcp on :9999 (mDNS disabled; direct connect)

go.mod: tidy. No go.temporal.io. No google.golang.org/grpc.
2026-04-26 14:06:21 -07:00

293 lines
11 KiB
Go
Raw Permalink Blame History

package verifyfirstworkflowtaskscheduled
import (
"context"
"testing"
"time"
"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"
historypb "go.temporal.io/api/history/v1"
"go.temporal.io/api/serviceerror"
taskqueuepb "go.temporal.io/api/taskqueue/v1"
"go.temporal.io/api/workflowservice/v1"
enumsspb "go.temporal.io/server/api/enums/v1"
"go.temporal.io/server/api/historyservice/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
workflowspb "go.temporal.io/server/api/workflow/v1"
"go.temporal.io/server/common"
"go.temporal.io/server/common/cluster"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/payloads"
"go.temporal.io/server/common/persistence"
"go.temporal.io/server/common/primitives"
"go.temporal.io/server/service/history/api"
"go.temporal.io/server/service/history/events"
"go.temporal.io/server/service/history/hsm"
historyi "go.temporal.io/server/service/history/interfaces"
"go.temporal.io/server/service/history/shard"
"go.temporal.io/server/service/history/tests"
"go.temporal.io/server/service/history/workflow"
wcache "go.temporal.io/server/service/history/workflow/cache"
"go.uber.org/mock/gomock"
"google.golang.org/protobuf/types/known/durationpb"
)
var defaultWorkflowTaskCompletionLimits = historyi.WorkflowTaskCompletionLimits{MaxResetPoints: primitives.DefaultHistoryMaxAutoResetPoints, MaxSearchAttributeValueSize: 2048}
type (
VerifyFirstWorkflowTaskScheduledSuite struct {
*require.Assertions
suite.Suite
controller *gomock.Controller
mockEventsCache *events.MockCache
mockExecutionMgr *persistence.MockExecutionManager
shardContext *shard.ContextTest
workflowConsistencyChecker api.WorkflowConsistencyChecker
logger log.Logger
}
)
func TestVerifyFirstWorkflowTaskScheduledSuite(t *testing.T) {
suite.Run(t, new(VerifyFirstWorkflowTaskScheduledSuite))
}
func (s *VerifyFirstWorkflowTaskScheduledSuite) SetupTest() {
s.Assertions = require.New(s.T())
s.controller = gomock.NewController(s.T())
config := tests.NewDynamicConfig()
s.shardContext = shard.NewTestContext(
s.controller,
&persistencespb.ShardInfo{
ShardId: 1,
RangeId: 1,
},
config,
)
reg := hsm.NewRegistry()
err := workflow.RegisterStateMachine(reg)
s.NoError(err)
s.shardContext.SetStateMachineRegistry(reg)
mockNamespaceCache := s.shardContext.Resource.NamespaceCache
mockNamespaceCache.EXPECT().GetNamespaceByID(tests.NamespaceID).Return(tests.LocalNamespaceEntry, nil).AnyTimes()
s.mockExecutionMgr = s.shardContext.Resource.ExecutionMgr
mockClusterMetadata := s.shardContext.Resource.ClusterMetadata
mockClusterMetadata.EXPECT().GetClusterID().Return(int64(1)).AnyTimes()
mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(false, common.EmptyVersion).Return(cluster.TestCurrentClusterName).AnyTimes()
mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(true, tests.Version).Return(cluster.TestCurrentClusterName).AnyTimes()
s.workflowConsistencyChecker = api.NewWorkflowConsistencyChecker(
s.shardContext,
wcache.NewHostLevelCache(s.shardContext.GetConfig(), s.shardContext.GetLogger(), metrics.NoopMetricsHandler))
s.mockEventsCache = s.shardContext.MockEventsCache
s.mockEventsCache.EXPECT().PutEvent(gomock.Any(), gomock.Any()).AnyTimes()
s.logger = s.shardContext.GetLogger()
}
func (s *VerifyFirstWorkflowTaskScheduledSuite) TearDownTest() {
s.controller.Finish()
}
func (s *VerifyFirstWorkflowTaskScheduledSuite) TestVerifyFirstWorkflowTaskScheduled_WorkflowNotFound() {
request := &historyservice.VerifyFirstWorkflowTaskScheduledRequest{
NamespaceId: tests.NamespaceID.String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
},
}
s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, &serviceerror.NotFound{})
err := Invoke(context.Background(), request, s.workflowConsistencyChecker)
s.IsType(&serviceerror.NotFound{}, err)
}
func (s *VerifyFirstWorkflowTaskScheduledSuite) TestVerifyFirstWorkflowTaskScheduled_WorkflowCompleted() {
request := &historyservice.VerifyFirstWorkflowTaskScheduledRequest{
NamespaceId: tests.NamespaceID.String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
},
}
ms := workflow.TestGlobalMutableState(s.shardContext, s.mockEventsCache, s.logger, tests.Version, tests.WorkflowID, tests.RunID)
addWorkflowExecutionStartedEventWithParent(ms,
&commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
}, "wType", "testTaskQueue", payloads.EncodeString("input"),
25*time.Second, 20*time.Second, 200*time.Second, nil, "identity")
_, err := ms.AddTimeoutWorkflowEvent(
ms.GetNextEventID(),
enumspb.RETRY_STATE_RETRY_POLICY_NOT_SET,
uuid.NewString(),
)
s.NoError(err)
wfMs := workflow.TestCloneToProto(context.Background(), ms)
gwmsResponse := &persistence.GetWorkflowExecutionResponse{State: wfMs}
s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(gwmsResponse, nil)
err = Invoke(context.Background(), request, s.workflowConsistencyChecker)
s.NoError(err)
}
func (s *VerifyFirstWorkflowTaskScheduledSuite) TestVerifyFirstWorkflowTaskScheduled_WorkflowZombie() {
request := &historyservice.VerifyFirstWorkflowTaskScheduledRequest{
NamespaceId: tests.NamespaceID.String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
},
}
ms := workflow.TestGlobalMutableState(s.shardContext, s.mockEventsCache, s.logger, tests.Version, tests.WorkflowID, tests.RunID)
addWorkflowExecutionStartedEventWithParent(ms,
&commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
}, "wType", "testTaskQueue", payloads.EncodeString("input"),
25*time.Second, 20*time.Second, 200*time.Second, nil, "identity")
// zombie state should be treated as open
_, err := ms.UpdateWorkflowStateStatus(
enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE,
enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
)
s.NoError(err)
wfMs := workflow.TestCloneToProto(context.Background(), ms)
gwmsResponse := &persistence.GetWorkflowExecutionResponse{State: wfMs}
s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(gwmsResponse, nil)
err = Invoke(context.Background(), request, s.workflowConsistencyChecker)
s.IsType(&serviceerror.WorkflowNotReady{}, err)
}
func (s *VerifyFirstWorkflowTaskScheduledSuite) TestVerifyFirstWorkflowTaskScheduled_WorkflowRunning_TaskPending() {
request := &historyservice.VerifyFirstWorkflowTaskScheduledRequest{
NamespaceId: tests.NamespaceID.String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
},
}
ms := workflow.TestGlobalMutableState(s.shardContext, s.mockEventsCache, s.logger, tests.Version, tests.WorkflowID, tests.RunID)
addWorkflowExecutionStartedEventWithParent(ms,
&commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
}, "wType", "testTaskQueue", payloads.EncodeString("input"),
25*time.Second, 20*time.Second, 200*time.Second, nil, "identity")
_, _ = ms.AddWorkflowTaskScheduledEvent(false, enumsspb.WORKFLOW_TASK_TYPE_NORMAL)
wfMs := workflow.TestCloneToProto(context.Background(), ms)
gwmsResponse := &persistence.GetWorkflowExecutionResponse{State: wfMs}
s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(gwmsResponse, nil)
err := Invoke(context.Background(), request, s.workflowConsistencyChecker)
s.NoError(err)
}
func (s *VerifyFirstWorkflowTaskScheduledSuite) TestVerifyFirstWorkflowTaskScheduled_WorkflowRunning_TaskProcessed() {
request := &historyservice.VerifyFirstWorkflowTaskScheduledRequest{
NamespaceId: tests.NamespaceID.String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
},
}
ms := workflow.TestGlobalMutableState(s.shardContext, s.mockEventsCache, s.logger, tests.Version, tests.WorkflowID, tests.RunID)
addWorkflowExecutionStartedEventWithParent(ms,
&commonpb.WorkflowExecution{
WorkflowId: tests.WorkflowID,
RunId: tests.RunID,
}, "wType", "testTaskQueue", payloads.EncodeString("input"),
25*time.Second, 20*time.Second, 200*time.Second, nil, "identity")
// Schedule WFT
wt, _ := ms.AddWorkflowTaskScheduledEvent(false, enumsspb.WORKFLOW_TASK_TYPE_NORMAL)
// Start WFT
workflowTasksStartEvent, _, _ := ms.AddWorkflowTaskStartedEvent(
wt.ScheduledEventID,
tests.RunID,
&taskqueuepb.TaskQueue{Name: "testTaskQueue"},
uuid.NewString(),
nil,
nil,
nil,
false,
nil,
0,
)
wt.StartedEventID = workflowTasksStartEvent.GetEventId()
// Complete WFT
workflowTask := ms.GetWorkflowTaskByID(wt.ScheduledEventID)
s.NotNil(workflowTask)
s.Equal(wt.StartedEventID, workflowTask.StartedEventID)
_, _ = ms.AddWorkflowTaskCompletedEvent(workflowTask,
&workflowservice.RespondWorkflowTaskCompletedRequest{Identity: "some random identity"}, defaultWorkflowTaskCompletionLimits)
ms.FlushBufferedEvents()
wfMs := workflow.TestCloneToProto(context.Background(), ms)
gwmsResponse := &persistence.GetWorkflowExecutionResponse{State: wfMs}
s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(gwmsResponse, nil)
err := Invoke(context.Background(), request, s.workflowConsistencyChecker)
s.NoError(err)
}
func addWorkflowExecutionStartedEventWithParent(
ms historyi.MutableState,
workflowExecution *commonpb.WorkflowExecution,
workflowType, taskQueue string,
input *commonpb.Payloads,
executionTimeout, runTimeout, taskTimeout time.Duration,
parentInfo *workflowspb.ParentExecutionInfo,
identity string,
) *historypb.HistoryEvent {
startRequest := &workflowservice.StartWorkflowExecutionRequest{
WorkflowId: workflowExecution.WorkflowId,
WorkflowType: &commonpb.WorkflowType{Name: workflowType},
TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue},
Input: input,
WorkflowExecutionTimeout: durationpb.New(executionTimeout),
WorkflowRunTimeout: durationpb.New(runTimeout),
WorkflowTaskTimeout: durationpb.New(taskTimeout),
Identity: identity,
}
event, _ := ms.AddWorkflowExecutionStartedEvent(
workflowExecution,
&historyservice.StartWorkflowExecutionRequest{
Attempt: 1,
NamespaceId: tests.NamespaceID.String(),
StartRequest: startRequest,
ParentExecutionInfo: parentInfo,
},
)
return event
}
Reference in New Issue View Git Blame Copy Permalink