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).
106 lines
2.8 KiB
Go
106 lines
2.8 KiB
Go
package matching
|
|
|
|
import (
|
|
"context"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/suite"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
persistencespb "github.com/hanzoai/tasks/api/persistence/v1"
|
|
"github.com/hanzoai/tasks/common/dynamicconfig"
|
|
"github.com/hanzoai/tasks/common/log"
|
|
"github.com/hanzoai/tasks/common/metrics"
|
|
"github.com/hanzoai/tasks/common/testing/testlogger"
|
|
"github.com/hanzoai/tasks/common/tqid"
|
|
"go.uber.org/mock/gomock"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
)
|
|
|
|
type PriMatcherSuite struct {
|
|
suite.Suite
|
|
controller *gomock.Controller
|
|
logger log.Logger
|
|
}
|
|
|
|
func TestPriMatcherSuite(t *testing.T) {
|
|
t.Parallel()
|
|
suite.Run(t, new(PriMatcherSuite))
|
|
}
|
|
|
|
func (s *PriMatcherSuite) SetupTest() {
|
|
s.controller = gomock.NewController(s.T())
|
|
s.logger = testlogger.NewTestLogger(s.T(), testlogger.FailOnAnyUnexpectedError)
|
|
}
|
|
|
|
// TestValidatorWorksOnRoot tests that the validator goroutine can pick up tasks
|
|
// on a root partition (where there is no forwarder).
|
|
func (s *PriMatcherSuite) TestValidatorWorksOnRoot() {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
cfg := newTaskQueueConfig(
|
|
tqid.UnsafeTaskQueueFamily("nsid", "tq").TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW),
|
|
NewConfig(dynamicconfig.NewNoopCollection()),
|
|
"nsname",
|
|
)
|
|
|
|
partition := tqid.UnsafeTaskQueueFamily("nsid", "tq").
|
|
TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW).
|
|
RootPartition()
|
|
|
|
var validatorValidatedTask atomic.Bool
|
|
|
|
// record validator calls
|
|
mockValidator := NewMocktaskValidator(s.controller)
|
|
mockValidator.EXPECT().maybeValidate(gomock.Any(), gomock.Any()).DoAndReturn(func(task *persistencespb.AllocatedTaskInfo, taskType enumspb.TaskQueueType) bool {
|
|
validatorValidatedTask.Store(true)
|
|
return true // task is valid
|
|
})
|
|
|
|
rateLimitManager := newRateLimitManager(&mockUserDataManager{}, cfg, enumspb.TASK_QUEUE_TYPE_WORKFLOW)
|
|
|
|
tm := newPriTaskMatcher(
|
|
ctx,
|
|
cfg,
|
|
partition,
|
|
nil, // nil forwarder = root partition
|
|
nil, // no client needed for this test
|
|
mockValidator,
|
|
s.logger,
|
|
metrics.NoopMetricsHandler,
|
|
rateLimitManager,
|
|
func() {},
|
|
)
|
|
|
|
// start the matcher
|
|
tm.Start()
|
|
defer tm.Stop()
|
|
|
|
completionCalled := make(chan taskResponse)
|
|
task := newInternalTaskFromBacklog(&persistencespb.AllocatedTaskInfo{
|
|
TaskId: 1,
|
|
Data: &persistencespb.TaskInfo{
|
|
CreateTime: timestamppb.Now(),
|
|
},
|
|
}, func(t *internalTask, res taskResponse) {
|
|
completionCalled <- res
|
|
})
|
|
|
|
// add the task
|
|
task.resetMatcherState()
|
|
_ = tm.AddTask(task)
|
|
|
|
// validator should pick up and check task
|
|
select {
|
|
case res := <-completionCalled:
|
|
// error should be errReprocessTask
|
|
s.ErrorIs(res.err(), errReprocessTask) //nolint:testifylint
|
|
case <-time.After(2 * time.Second):
|
|
s.Fail("Timeout waiting for validator to process task")
|
|
}
|
|
|
|
s.True(validatorValidatedTask.Load(), "Validator should have called maybeValidate")
|
|
}
|