mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
zeekay
5a601d081b
Full module path rebrand from Temporal to Hanzo Tasks: - go.mod module: github.com/hanzoai/tasks - All internal Go imports updated (1906 files) - cmd/server renamed to cmd/tasksd - Binary name: tasksd (was temporal-server) - Docker images: hanzoai/tasks (was temporalio/server) - README rewritten with Hanzo Tasks branding - LLM.md created with integration notes - Makefile, goreleaser, CI workflows updated - go.temporal.io/sdk and go.temporal.io/api imports preserved (separate repos) Fork of Temporal.io (MIT license).
131 lines
4.4 KiB
Go
131 lines
4.4 KiB
Go
package replication_test
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
enumsspb "github.com/hanzoai/tasks/api/enums/v1"
|
|
persistencespb "github.com/hanzoai/tasks/api/persistence/v1"
|
|
"github.com/hanzoai/tasks/common/log"
|
|
"github.com/hanzoai/tasks/common/metrics"
|
|
"github.com/hanzoai/tasks/common/metrics/metricstest"
|
|
"github.com/hanzoai/tasks/common/namespace"
|
|
"github.com/hanzoai/tasks/common/persistence"
|
|
"github.com/hanzoai/tasks/common/persistence/serialization"
|
|
"github.com/hanzoai/tasks/service/history/queues"
|
|
"github.com/hanzoai/tasks/service/history/queues/queuestest"
|
|
"github.com/hanzoai/tasks/service/history/replication"
|
|
"github.com/hanzoai/tasks/service/history/tests"
|
|
"go.uber.org/mock/gomock"
|
|
)
|
|
|
|
type (
|
|
fakeExecutionManager struct {
|
|
requests []*persistence.PutReplicationTaskToDLQRequest
|
|
}
|
|
)
|
|
|
|
func TestNewExecutionManagerDLQWriter(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
executionManager := &fakeExecutionManager{}
|
|
writer := replication.NewExecutionManagerDLQWriter(executionManager)
|
|
replicationTaskInfo := &persistencespb.ReplicationTaskInfo{
|
|
TaskId: 21,
|
|
}
|
|
err := writer.WriteTaskToDLQ(context.Background(), replication.DLQWriteRequest{
|
|
SourceShardID: 13,
|
|
TargetShardID: 26,
|
|
SourceCluster: "test-source-cluster",
|
|
ReplicationTaskInfo: replicationTaskInfo,
|
|
})
|
|
require.NoError(t, err)
|
|
require.Len(t, executionManager.requests, 1)
|
|
request := executionManager.requests[0]
|
|
assert.Equal(t, 26, int(request.ShardID))
|
|
assert.Equal(t, "test-source-cluster", request.SourceClusterName)
|
|
assert.Equal(t, replicationTaskInfo, request.TaskInfo)
|
|
}
|
|
|
|
func TestNewDLQWriterAdapter(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
taskType enumsspb.TaskType
|
|
expectErr bool
|
|
}{
|
|
{
|
|
name: "history replication task",
|
|
taskType: enumsspb.TASK_TYPE_REPLICATION_HISTORY,
|
|
expectErr: false,
|
|
},
|
|
{
|
|
name: "unspecified task type",
|
|
taskType: enumsspb.TASK_TYPE_UNSPECIFIED,
|
|
expectErr: true,
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
controller := gomock.NewController(t)
|
|
queueWriter := &queuestest.FakeQueueWriter{}
|
|
taskSerializer := serialization.NewSerializer()
|
|
namespaceRegistry := namespace.NewMockRegistry(controller)
|
|
namespaceRegistry.EXPECT().GetNamespaceByID(gomock.Any()).Return(&namespace.Namespace{}, nil).AnyTimes()
|
|
metricsHandler := metricstest.NewCaptureHandler()
|
|
capture := metricsHandler.StartCapture()
|
|
writer := replication.NewDLQWriterAdapter(
|
|
queues.NewDLQWriter(queueWriter, metricsHandler, log.NewTestLogger(), namespaceRegistry),
|
|
taskSerializer,
|
|
"test-current-cluster",
|
|
)
|
|
|
|
replicationTaskInfo := &persistencespb.ReplicationTaskInfo{
|
|
NamespaceId: string(tests.NamespaceID),
|
|
WorkflowId: tests.WorkflowID,
|
|
RunId: tests.RunID,
|
|
TaskType: tc.taskType,
|
|
TaskId: 21,
|
|
}
|
|
err := writer.WriteTaskToDLQ(context.Background(), replication.DLQWriteRequest{
|
|
SourceShardID: 13,
|
|
SourceCluster: "test-source-cluster",
|
|
ReplicationTaskInfo: replicationTaskInfo,
|
|
})
|
|
if tc.expectErr {
|
|
require.Error(t, err)
|
|
assert.Contains(t, strings.ToLower(err.Error()), "unknown replication task type")
|
|
assert.Empty(t, queueWriter.EnqueueTaskRequests)
|
|
} else {
|
|
require.NoError(t, err)
|
|
require.Len(t, queueWriter.EnqueueTaskRequests, 1)
|
|
request := queueWriter.EnqueueTaskRequests[0]
|
|
assert.Equal(t, string(tests.NamespaceID), request.Task.GetNamespaceID())
|
|
assert.Equal(t, tests.WorkflowID, request.Task.GetWorkflowID())
|
|
assert.Equal(t, tests.RunID, request.Task.GetRunID())
|
|
assert.Equal(t, 21, int(request.Task.GetTaskID()))
|
|
assert.Equal(t, "test-source-cluster", request.SourceCluster)
|
|
assert.Equal(t, "test-current-cluster", request.TargetCluster)
|
|
snapshot := capture.Snapshot()
|
|
recordings := snapshot[metrics.DLQWrites.Name()]
|
|
assert.Len(t, recordings, 1)
|
|
assert.Len(t, recordings[0].Tags, 2)
|
|
assert.Equal(t, "replication", recordings[0].Tags[metrics.TaskCategoryTagName])
|
|
namespaceStateTag := metrics.NamespaceStateTag(metrics.PassiveNamespaceStateTagValue)
|
|
assert.Equal(t, metrics.PassiveNamespaceStateTagValue, recordings[0].Tags[namespaceStateTag.Key])
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func (f *fakeExecutionManager) PutReplicationTaskToDLQ(
|
|
_ context.Context,
|
|
request *persistence.PutReplicationTaskToDLQRequest,
|
|
) error {
|
|
f.requests = append(f.requests, request)
|
|
return nil
|
|
}
|