mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
zeekay
40351cce6f
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.
263 lines
7.9 KiB
Go
263 lines
7.9 KiB
Go
package deletemanager
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/stretchr/testify/suite"
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
"go.temporal.io/server/chasm"
|
|
"go.temporal.io/server/common/clock"
|
|
"go.temporal.io/server/common/cluster"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/common/persistence/visibility/manager"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
"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"
|
|
)
|
|
|
|
type (
|
|
deleteManagerWorkflowSuite struct {
|
|
suite.Suite
|
|
*require.Assertions
|
|
|
|
controller *gomock.Controller
|
|
mockCache *wcache.MockCache
|
|
mockShardContext *historyi.MockShardContext
|
|
mockClock *clock.EventTimeSource
|
|
mockNamespaceRegistry *namespace.MockRegistry
|
|
mockMetadata *cluster.MockMetadata
|
|
mockVisibilityManager *manager.MockVisibilityManager
|
|
|
|
deleteManager DeleteManager
|
|
}
|
|
)
|
|
|
|
func TestDeleteManagerSuite(t *testing.T) {
|
|
s := &deleteManagerWorkflowSuite{}
|
|
suite.Run(t, s)
|
|
}
|
|
|
|
func (s *deleteManagerWorkflowSuite) SetupSuite() {
|
|
|
|
}
|
|
|
|
func (s *deleteManagerWorkflowSuite) TearDownSuite() {
|
|
|
|
}
|
|
|
|
func (s *deleteManagerWorkflowSuite) SetupTest() {
|
|
s.Assertions = require.New(s.T())
|
|
|
|
s.controller = gomock.NewController(s.T())
|
|
s.mockCache = wcache.NewMockCache(s.controller)
|
|
s.mockClock = clock.NewEventTimeSource()
|
|
s.mockNamespaceRegistry = namespace.NewMockRegistry(s.controller)
|
|
s.mockMetadata = cluster.NewMockMetadata(s.controller)
|
|
s.mockVisibilityManager = manager.NewMockVisibilityManager(s.controller)
|
|
s.mockVisibilityManager.EXPECT().GetIndexName().Return("").AnyTimes()
|
|
|
|
config := tests.NewDynamicConfig()
|
|
s.mockShardContext = historyi.NewMockShardContext(s.controller)
|
|
s.mockShardContext.EXPECT().GetMetricsHandler().Return(metrics.NoopMetricsHandler).AnyTimes()
|
|
s.mockShardContext.EXPECT().GetNamespaceRegistry().Return(s.mockNamespaceRegistry).AnyTimes()
|
|
s.mockShardContext.EXPECT().GetClusterMetadata().Return(s.mockMetadata).AnyTimes()
|
|
|
|
s.deleteManager = NewDeleteManager(
|
|
s.mockShardContext,
|
|
s.mockCache,
|
|
config,
|
|
s.mockClock,
|
|
s.mockVisibilityManager,
|
|
)
|
|
}
|
|
|
|
func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution() {
|
|
we := commonpb.WorkflowExecution{
|
|
WorkflowId: tests.WorkflowID,
|
|
RunId: tests.RunID,
|
|
}
|
|
|
|
mockWeCtx := historyi.NewMockWorkflowContext(s.controller)
|
|
mockMutableState := historyi.NewMockMutableState(s.controller)
|
|
mockMutableState.EXPECT().GetCurrentBranchToken().Return([]byte{22, 8, 78}, nil)
|
|
closeExecutionVisibilityTaskID := int64(39)
|
|
mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
CloseVisibilityTaskId: closeExecutionVisibilityTaskID,
|
|
})
|
|
mockMutableState.EXPECT().ChasmTree().Return(workflow.NoopChasmTree).AnyTimes()
|
|
stage := tasks.DeleteWorkflowExecutionStageNone
|
|
|
|
s.mockShardContext.EXPECT().DeleteWorkflowExecution(
|
|
gomock.Any(),
|
|
definition.WorkflowKey{
|
|
NamespaceID: tests.NamespaceID.String(),
|
|
WorkflowID: tests.WorkflowID,
|
|
RunID: tests.RunID,
|
|
},
|
|
chasm.WorkflowArchetypeID,
|
|
[]byte{22, 8, 78},
|
|
closeExecutionVisibilityTaskID,
|
|
time.Unix(0, 0).UTC(),
|
|
&stage,
|
|
).Return(nil)
|
|
mockWeCtx.EXPECT().Clear()
|
|
|
|
err := s.deleteManager.DeleteWorkflowExecution(
|
|
context.Background(),
|
|
tests.NamespaceID,
|
|
&we,
|
|
mockWeCtx,
|
|
mockMutableState,
|
|
&stage,
|
|
)
|
|
s.NoError(err)
|
|
}
|
|
|
|
func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution_Error() {
|
|
we := commonpb.WorkflowExecution{
|
|
WorkflowId: tests.WorkflowID,
|
|
RunId: tests.RunID,
|
|
}
|
|
|
|
mockWeCtx := historyi.NewMockWorkflowContext(s.controller)
|
|
mockMutableState := historyi.NewMockMutableState(s.controller)
|
|
mockMutableState.EXPECT().GetCurrentBranchToken().Return([]byte{22, 8, 78}, nil)
|
|
closeExecutionVisibilityTaskID := int64(39)
|
|
mockMutableState.EXPECT().GetExecutionInfo().MinTimes(1).Return(&persistencespb.WorkflowExecutionInfo{
|
|
CloseVisibilityTaskId: closeExecutionVisibilityTaskID,
|
|
})
|
|
mockMutableState.EXPECT().ChasmTree().Return(workflow.NoopChasmTree).AnyTimes()
|
|
stage := tasks.DeleteWorkflowExecutionStageNone
|
|
|
|
s.mockShardContext.EXPECT().DeleteWorkflowExecution(
|
|
gomock.Any(),
|
|
definition.WorkflowKey{
|
|
NamespaceID: tests.NamespaceID.String(),
|
|
WorkflowID: tests.WorkflowID,
|
|
RunID: tests.RunID,
|
|
},
|
|
chasm.WorkflowArchetypeID,
|
|
[]byte{22, 8, 78},
|
|
closeExecutionVisibilityTaskID,
|
|
time.Unix(0, 0).UTC(),
|
|
&stage,
|
|
).Return(serviceerror.NewInternal("test error"))
|
|
|
|
err := s.deleteManager.DeleteWorkflowExecution(
|
|
context.Background(),
|
|
tests.NamespaceID,
|
|
&we,
|
|
mockWeCtx,
|
|
mockMutableState,
|
|
&stage,
|
|
)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecutionByRetention_SkipsReplication() {
|
|
we := commonpb.WorkflowExecution{
|
|
WorkflowId: tests.WorkflowID,
|
|
RunId: tests.RunID,
|
|
}
|
|
|
|
mockWeCtx := historyi.NewMockWorkflowContext(s.controller)
|
|
mockMutableState := historyi.NewMockMutableState(s.controller)
|
|
mockMutableState.EXPECT().GetCurrentBranchToken().Return([]byte{22, 8, 78}, nil)
|
|
closeExecutionVisibilityTaskID := int64(39)
|
|
mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
CloseVisibilityTaskId: closeExecutionVisibilityTaskID,
|
|
})
|
|
mockMutableState.EXPECT().ChasmTree().Return(workflow.NoopChasmTree).AnyTimes()
|
|
stage := tasks.DeleteWorkflowExecutionStageNone
|
|
|
|
s.mockShardContext.EXPECT().DeleteWorkflowExecution(
|
|
gomock.Any(),
|
|
definition.WorkflowKey{
|
|
NamespaceID: tests.NamespaceID.String(),
|
|
WorkflowID: tests.WorkflowID,
|
|
RunID: tests.RunID,
|
|
},
|
|
workflow.NoopChasmTree.ArchetypeID(),
|
|
[]byte{22, 8, 78},
|
|
closeExecutionVisibilityTaskID,
|
|
time.Unix(0, 0).UTC(),
|
|
gomock.Any(),
|
|
).DoAndReturn(func(
|
|
_ context.Context,
|
|
_ definition.WorkflowKey,
|
|
_ chasm.ArchetypeID,
|
|
_ []byte,
|
|
_ int64,
|
|
_ time.Time,
|
|
stagePtr *tasks.DeleteWorkflowExecutionStage,
|
|
) error {
|
|
// Verify that replication stage is already marked as processed before DeleteWorkflowExecution is called.
|
|
s.True(stagePtr.IsProcessed(tasks.DeleteWorkflowExecutionStageReplication),
|
|
"Replication stage should be pre-marked as processed for retention-based deletion")
|
|
return nil
|
|
})
|
|
mockWeCtx.EXPECT().Clear()
|
|
|
|
err := s.deleteManager.DeleteWorkflowExecutionByRetention(
|
|
context.Background(),
|
|
tests.NamespaceID,
|
|
&we,
|
|
mockWeCtx,
|
|
mockMutableState,
|
|
&stage,
|
|
)
|
|
s.NoError(err)
|
|
}
|
|
|
|
func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecution_OpenWorkflow() {
|
|
we := commonpb.WorkflowExecution{
|
|
WorkflowId: tests.WorkflowID,
|
|
RunId: tests.RunID,
|
|
}
|
|
|
|
mockWeCtx := historyi.NewMockWorkflowContext(s.controller)
|
|
mockMutableState := historyi.NewMockMutableState(s.controller)
|
|
closeExecutionVisibilityTaskID := int64(39)
|
|
mockMutableState.EXPECT().GetCurrentBranchToken().Return([]byte{22, 8, 78}, nil)
|
|
mockMutableState.EXPECT().GetExecutionInfo().MinTimes(1).Return(&persistencespb.WorkflowExecutionInfo{
|
|
CloseVisibilityTaskId: closeExecutionVisibilityTaskID,
|
|
})
|
|
mockMutableState.EXPECT().ChasmTree().Return(workflow.NoopChasmTree).AnyTimes()
|
|
stage := tasks.DeleteWorkflowExecutionStageNone
|
|
|
|
s.mockShardContext.EXPECT().DeleteWorkflowExecution(
|
|
gomock.Any(),
|
|
definition.WorkflowKey{
|
|
NamespaceID: tests.NamespaceID.String(),
|
|
WorkflowID: tests.WorkflowID,
|
|
RunID: tests.RunID,
|
|
},
|
|
chasm.WorkflowArchetypeID,
|
|
[]byte{22, 8, 78},
|
|
closeExecutionVisibilityTaskID,
|
|
time.Unix(0, 0).UTC(),
|
|
&stage,
|
|
).Return(nil)
|
|
mockWeCtx.EXPECT().Clear()
|
|
|
|
err := s.deleteManager.DeleteWorkflowExecution(
|
|
context.Background(),
|
|
tests.NamespaceID,
|
|
&we,
|
|
mockWeCtx,
|
|
mockMutableState,
|
|
&stage,
|
|
)
|
|
s.NoError(err)
|
|
}
|