From 30bbe965bf6479bd11ae13cff7f7c6a2ca9208b6 Mon Sep 17 00:00:00 2001 From: ryan Date: Wed, 2 Sep 2026 16:11:45 +0800 Subject: [PATCH] fix(task): execute dispatched jobs and persist run records Asynq func handlers now go through ProcessTask so admin execution rows leave pending. The in-process worker resolves admin type identifiers and writes the same w_task_executions table. Remove the no-op admin system_cleanup that shadowed the upload handler. --- backend/core/extpoints/extpoints_test.go | 4 + backend/core/extpoints/task.go | 13 +- backend/plugins/domain/admin/plugin.go | 17 +- backend/plugins/domain/admin/plugin_test.go | 7 +- backend/plugins/domain/domain_test.go | 9 +- .../drivers/driver_asynq_worker/executor.go | 17 +- .../drivers/driver_asynq_worker/plugin.go | 40 +++- .../drivers/driver_inproc_worker/db_helper.go | 39 ++++ .../drivers/driver_inproc_worker/executor.go | 188 +++++++++++++++--- .../drivers/driver_inproc_worker/plugin.go | 16 +- .../driver_inproc_worker/plugin_test.go | 63 ++++++ .../driver_inproc_worker/task_service.go | 88 +++++++- .../drivers/driver_inproc_worker/types.go | 44 ++++ backend/plugins/drivers/drivers_test.go | 67 +++++++ 14 files changed, 530 insertions(+), 82 deletions(-) create mode 100644 backend/plugins/drivers/driver_inproc_worker/db_helper.go create mode 100644 backend/plugins/drivers/driver_inproc_worker/types.go diff --git a/backend/core/extpoints/extpoints_test.go b/backend/core/extpoints/extpoints_test.go index 6d51b999..d0c438f4 100644 --- a/backend/core/extpoints/extpoints_test.go +++ b/backend/core/extpoints/extpoints_test.go @@ -221,6 +221,10 @@ func TestTaskExtension(t *testing.T) { assert.True(t, ok) assert.Equal(t, "order:cancel_timeout", task.Pattern) + byType, ok := tr.Get("cancel_timeout") + assert.True(t, ok, "Get should resolve admin type identifier") + assert.Equal(t, "order:cancel_timeout", byType.Pattern) + _, ok = tr.Get("unknown") assert.False(t, ok) } diff --git a/backend/core/extpoints/task.go b/backend/core/extpoints/task.go index 1217ed64..09270158 100644 --- a/backend/core/extpoints/task.go +++ b/backend/core/extpoints/task.go @@ -276,10 +276,17 @@ func (t *TaskRegistry) Tasks() []TaskDefinition { return res } -// Get retrieves a task definition by its pattern. +// Get retrieves a task definition by its pattern or admin type identifier. func (t *TaskRegistry) Get(pattern string) (TaskDefinition, bool) { t.mu.RLock() defer t.mu.RUnlock() - td, ok := t.lookup[pattern] - return td, ok + if td, ok := t.lookup[pattern]; ok { + return td, true + } + for _, td := range t.tasks { + if td.Type == pattern { + return td, true + } + } + return TaskDefinition{}, false } diff --git a/backend/plugins/domain/admin/plugin.go b/backend/plugins/domain/admin/plugin.go index 3e195233..835e2e80 100644 --- a/backend/plugins/domain/admin/plugin.go +++ b/backend/plugins/domain/admin/plugin.go @@ -181,22 +181,7 @@ func (p *Plugin) Apply(ctx *core.Context) error { return err }, extpoints.WithTaskMeta(service.LogDBSwitchMeta)) - ctx.Task().Register("admin:system_cleanup", func(_ context.Context, _ []byte) error { - return nil - }, - extpoints.WithTaskType("system_cleanup"), - extpoints.WithTaskName("系统垃圾清理"), - extpoints.WithTaskDescription("定期清理未使用上传文件、历史推送记录和过期任务执行日志"), - extpoints.WithTaskCategory("maintenance"), - extpoints.WithTaskRetry(1), - extpoints.WithTaskQueue("default"), - extpoints.WithTaskRetryable(true), - ) - - // 3. Register Cron Schedules - ctx.Schedule().RegisterCron("0 4 * * *", "admin:system_cleanup", map[string]string{"type": "daily"}) - - // 4. Register Settings Schemas + // 3. Register Settings Schemas ctx.Settings().Register(extpoints.SettingSchema{ Key: "admin.system_cleanup_cron", Default: "0 4 * * *", diff --git a/backend/plugins/domain/admin/plugin_test.go b/backend/plugins/domain/admin/plugin_test.go index 5a3a1632..451897be 100644 --- a/backend/plugins/domain/admin/plugin_test.go +++ b/backend/plugins/domain/admin/plugin_test.go @@ -33,14 +33,9 @@ func TestAdminPluginUnit(t *testing.T) { assert.True(t, hasRobots, "admin plugin must register /robots.txt") // Verify tasks - _, ok := ctx.Tasks().Get("admin:system_cleanup") + _, ok := ctx.Tasks().Get("logs:db_switch") require.True(t, ok) - // Verify schedules - sched, ok := ctx.Schedules().Get("admin:system_cleanup") - require.True(t, ok) - assert.Equal(t, "0 4 * * *", sched.Spec) - // Verify settings setting, ok := ctx.Settings().Get("admin.system_cleanup_cron") require.True(t, ok) diff --git a/backend/plugins/domain/domain_test.go b/backend/plugins/domain/domain_test.go index 9f3f1b6a..efa54eb4 100644 --- a/backend/plugins/domain/domain_test.go +++ b/backend/plugins/domain/domain_test.go @@ -380,12 +380,9 @@ func TestAdminPlugin(t *testing.T) { assert.True(t, hasTasks) assert.True(t, hasConfigs) - // 2. Task & Schedule - _, ok := ctx.Tasks().Get("admin:system_cleanup") + // 2. Task + _, ok := ctx.Tasks().Get("logs:db_switch") require.True(t, ok) - sched, ok := ctx.Schedules().Get("admin:system_cleanup") - require.True(t, ok) - assert.Equal(t, "0 4 * * *", sched.Spec) // 3. Settings schema, ok := ctx.Settings().Get("admin.system_cleanup_cron") @@ -455,7 +452,7 @@ func TestAllDomainPluginsCombined(t *testing.T) { // Verify total schedules registered allSchedules := ctx.Schedules().Schedules() - assert.GreaterOrEqual(t, len(allSchedules), 2) + assert.GreaterOrEqual(t, len(allSchedules), 1) // 每个调度指向的任务类型都必须已注册 Handler，否则触发时会投递到无人处理的 // 任务类型，预期的清理逻辑静默失效。 diff --git a/backend/plugins/drivers/driver_asynq_worker/executor.go b/backend/plugins/drivers/driver_asynq_worker/executor.go index fc8459f5..7da4c364 100644 --- a/backend/plugins/drivers/driver_asynq_worker/executor.go +++ b/backend/plugins/drivers/driver_asynq_worker/executor.go @@ -265,8 +265,9 @@ func ProcessTask(ctx context.Context, t *asynq.Task) error { taskID := t.ResultWriter().TaskID() - // 注入 taskID 到 context + // 注入 taskID 到 context，并把原始 asynq.Task 交给适配器 Handler。 ctx = withTaskID(ctx, taskID) + ctx = context.WithValue(ctx, asynqTaskCtxKey{}, t) // 查找处理器 handler, ok := getHandler(t.Type()) @@ -277,18 +278,20 @@ func ProcessTask(ctx context.Context, t *asynq.Task) error { return err } - // 加载或动态创建执行记录 + // 加载或动态创建执行记录。无 DB 时仍执行业务 Handler，避免测试/精简拓扑 panic。 + var execution *TaskExecution now := time.Now() - execution, err := getOrCreateTaskExecution(ctx, taskID, t, taskPayload, now) - if err == nil { - updateExecutionOnStart(ctx, execution, now) + if getDB(ctx) != nil { + var err error + execution, err = getOrCreateTaskExecution(ctx, taskID, t, taskPayload, now) + if err == nil { + updateExecutionOnStart(ctx, execution, now) + } } if execution != nil { AppendLog(ctx, "[系统] 开始执行异步任务 [名称: %s, 类型: %s]，重试次数: %d/%d", execution.TaskName, t.Type(), execution.RetryCount, execution.MaxRetry) - } else { - AppendLog(ctx, "[系统] 开始执行异步任务 [类型: %s]", t.Type()) } // 开始计时 diff --git a/backend/plugins/drivers/driver_asynq_worker/plugin.go b/backend/plugins/drivers/driver_asynq_worker/plugin.go index 05905182..d16b1a6c 100644 --- a/backend/plugins/drivers/driver_asynq_worker/plugin.go +++ b/backend/plugins/drivers/driver_asynq_worker/plugin.go @@ -192,7 +192,7 @@ func (p *Plugin) Start(_ context.Context) error { if p.coreCtx != nil && p.coreCtx.Tasks() != nil { for _, td := range p.coreCtx.Tasks().Tasks() { - handler, err := toAsynqHandler(td.Handler) + handler, err := toAsynqHandler(td.Pattern, td.Handler) if err != nil { return fmt.Errorf("driver_asynq_worker: invalid handler for task pattern %q: %w", td.Pattern, err) } @@ -276,21 +276,30 @@ func (p *Plugin) Mux() *asynq.ServeMux { return p.mux } -func toAsynqHandler(h any) (asynq.Handler, error) { +func toAsynqHandler(pattern string, h any) (asynq.Handler, error) { if h == nil { return nil, errors.New("nil handler") } + if th, ok := h.(TaskHandler); ok { + RegisterHandler(pattern, th) + return asynq.HandlerFunc(ProcessTask), nil + } + + inner, err := toRawAsynqHandler(h) + if err != nil { + return nil, err + } + RegisterHandler(pattern, &asynqHandlerAdapter{inner: inner}) + return asynq.HandlerFunc(ProcessTask), nil +} + +func toRawAsynqHandler(h any) (asynq.Handler, error) { switch fn := h.(type) { case asynq.HandlerFunc: return fn, nil case asynq.Handler: return fn, nil - case TaskHandler: - return asynq.HandlerFunc(func(c context.Context, t *asynq.Task) error { - RegisterHandler(t.Type(), fn) - return ProcessTask(c, t) - }), nil case func(context.Context, *asynq.Task) error: return asynq.HandlerFunc(fn), nil case func(context.Context, []byte) error: @@ -316,6 +325,23 @@ func toAsynqHandler(h any) (asynq.Handler, error) { } } +type asynqTaskCtxKey struct{} + +type asynqHandlerAdapter struct { + inner asynq.Handler +} + +func (a *asynqHandlerAdapter) Execute(ctx context.Context, payload []byte) (*TaskResult, error) { + t, _ := ctx.Value(asynqTaskCtxKey{}).(*asynq.Task) + if t == nil { + t = asynq.NewTask("", payload) + } + if err := a.inner.ProcessTask(ctx, t); err != nil { + return nil, err + } + return &TaskResult{Message: "ok"}, nil +} + type taskServiceImpl struct{} func (s *taskServiceImpl) Dispatch(ctx context.Context, taskType string, payload []byte, triggeredBy string) (string, error) { diff --git a/backend/plugins/drivers/driver_inproc_worker/db_helper.go b/backend/plugins/drivers/driver_inproc_worker/db_helper.go new file mode 100644 index 00000000..d564c491 --- /dev/null +++ b/backend/plugins/drivers/driver_inproc_worker/db_helper.go @@ -0,0 +1,39 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package driver_inproc_worker + +import ( + "Wavelet/core" + "Wavelet/core/contracts" + "context" + "sync" + + "gorm.io/gorm" +) + +var ( + dbMu sync.RWMutex + dbSvc contracts.DBService +) + +func setDBService(s contracts.DBService) { + dbMu.Lock() + defer dbMu.Unlock() + dbSvc = s +} + +func getDB(ctx context.Context) *gorm.DB { + if c, ok := ctx.(*core.Context); ok && c != nil { + if s, err := core.Inject[contracts.DBService](c); err == nil && s != nil { + return s.DB(ctx) + } + } + dbMu.RLock() + s := dbSvc + dbMu.RUnlock() + if s != nil { + return s.DB(ctx) + } + return nil +} diff --git a/backend/plugins/drivers/driver_inproc_worker/executor.go b/backend/plugins/drivers/driver_inproc_worker/executor.go index 7fcc9f2f..fe211612 100644 --- a/backend/plugins/drivers/driver_inproc_worker/executor.go +++ b/backend/plugins/drivers/driver_inproc_worker/executor.go @@ -6,6 +6,7 @@ package driver_inproc_worker import ( "Wavelet/core/extpoints" "Wavelet/pkg/idgen" + "Wavelet/pkg/logger" "Wavelet/pkg/util" "context" "errors" @@ -59,34 +60,56 @@ func NewInprocQueue(concurrency, queueCap int, taskReg extpoints.TaskExtension) } // Enqueue puts a new task into the in-process queue. -func (q *InprocQueue) Enqueue(taskType string, payload []byte, source string) (string, error) { +// taskType may be the registration pattern or the admin type identifier. +func (q *InprocQueue) Enqueue(ctx context.Context, taskType string, payload []byte, source string) (string, error) { if !q.running.Load() { return "", errors.New("driver_inproc_worker: queue is not running") } - taskID := fmt.Sprintf("inproc_%d", idgen.NextUint64ID()) + td, ok := q.lookupTask(taskType) + if !ok { + return "", fmt.Errorf("driver_inproc_worker: unknown task type %q", taskType) + } + + if source == "" { + source = "manual" + } + idType := td.Type + if idType == "" { + idType = td.Pattern + } + taskID := fmt.Sprintf("%s_%s_%d", source, idType, idgen.NextUint64ID()) msg := TaskMessage{ ID: taskID, - TaskType: taskType, + TaskType: td.Pattern, Payload: payload, Source: source, CreatedAt: time.Now(), + RetryLeft: td.Retry, } - if q.taskReg != nil { - if td, ok := q.taskReg.Get(taskType); ok { - msg.RetryLeft = td.Retry - } + if err := q.createExecution(ctx, msg, td); err != nil { + return "", err } + q.appendExecutionLog(ctx, taskID, fmt.Sprintf("[系统] 任务已成功入队，等待调度执行 (最大重试次数: %d)", td.Retry)) + select { case q.queue <- msg: return taskID, nil default: + q.failExecution(ctx, msg, errors.New("queue is full"), 0) return "", errors.New("driver_inproc_worker: queue is full") } } +func (q *InprocQueue) lookupTask(taskType string) (extpoints.TaskDefinition, bool) { + if q.taskReg == nil { + return extpoints.TaskDefinition{}, false + } + return q.taskReg.Get(taskType) +} + // Start begins processing tasks with the worker pool. ctx is the app-lifetime // context used as the parent for per-task execution contexts. func (q *InprocQueue) Start(ctx context.Context) { @@ -145,12 +168,10 @@ func (q *InprocQueue) workerLoop(ctx context.Context) { } func (q *InprocQueue) executeTask(ctx context.Context, msg TaskMessage) { - if q.taskReg == nil { - return - } - - td, ok := q.taskReg.Get(msg.TaskType) + td, ok := q.lookupTask(msg.TaskType) if !ok { + logger.ErrorF(ctx, "driver_inproc_worker: no handler for task %q", msg.TaskType) + q.failExecution(ctx, msg, fmt.Errorf("unregistered task handler: %s", msg.TaskType), 0) return } @@ -162,26 +183,34 @@ func (q *InprocQueue) executeTask(ctx context.Context, msg TaskMessage) { taskCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() + q.markRunning(ctx, msg) + start := time.Now() err := invokeHandler(taskCtx, td.Handler, msg.Payload) - if err != nil && msg.RetryLeft > 0 { - msg.RetryLeft-- - // Retry with backoff - util.Go(func() { - select { - case <-time.After(defaultRetryBackoff): - case <-q.stopCh: - return - case <-ctx.Done(): - return - } - if q.running.Load() { + duration := time.Since(start) + + if err != nil { + q.failExecution(ctx, msg, err, duration) + if msg.RetryLeft > 0 { + msg.RetryLeft-- + util.Go(func() { select { - case q.queue <- msg: - default: + case <-time.After(defaultRetryBackoff): + case <-q.stopCh: + return + case <-ctx.Done(): + return } - } - }) + if q.running.Load() { + select { + case q.queue <- msg: + default: + } + } + }) + } + return } + q.succeedExecution(ctx, msg, duration) } func invokeHandler(ctx context.Context, handler any, payload []byte) error { @@ -202,3 +231,106 @@ func invokeHandler(ctx context.Context, handler any, payload []byte) error { return fmt.Errorf("unsupported handler type: %T", handler) } } + +func (q *InprocQueue) createExecution(ctx context.Context, msg TaskMessage, td extpoints.TaskDefinition) error { + db := getDB(ctx) + if db == nil { + return nil + } + + name := td.Name + if name == "" { + name = td.DisplayName + } + if name == "" { + name = td.Pattern + } + exec := &taskExecution{ + ID: idgen.NextUint64ID(), + TaskID: msg.ID, + TaskType: td.Pattern, + TaskName: name, + Status: taskExecutionStatusPending, + Retryable: td.Retryable || td.Retry > 0, + MaxRetry: td.Retry, + RetryCount: 0, + Payload: string(msg.Payload), + TriggeredBy: msg.Source, + } + if err := db.Create(exec).Error; err != nil { + return fmt.Errorf("driver_inproc_worker: create task execution: %w", err) + } + return nil +} + +func (q *InprocQueue) markRunning(ctx context.Context, msg TaskMessage) { + db := getDB(ctx) + if db == nil { + return + } + now := time.Now() + updates := map[string]any{ + taskExecutionColStatus: taskExecutionStatusRunning, + "started_at": now, + } + if err := db.Model(&taskExecution{}).Where("task_id = ?", msg.ID).Updates(updates).Error; err != nil { + logger.ErrorF(ctx, "driver_inproc_worker: mark running failed taskID=%s: %v", msg.ID, err) + return + } + q.appendExecutionLog(ctx, msg.ID, fmt.Sprintf("[系统] 开始执行异步任务 [类型: %s]", msg.TaskType)) +} + +func (q *InprocQueue) succeedExecution(ctx context.Context, msg TaskMessage, duration time.Duration) { + db := getDB(ctx) + if db == nil { + return + } + now := time.Now() + updates := map[string]any{ + taskExecutionColStatus: taskExecutionStatusSucceeded, + "error_message": "", + "result": "ok", + "finished_at": now, + "duration": duration.Milliseconds(), + } + if err := db.Model(&taskExecution{}).Where("task_id = ?", msg.ID).Updates(updates).Error; err != nil { + logger.ErrorF(ctx, "driver_inproc_worker: mark succeeded failed taskID=%s: %v", msg.ID, err) + return + } + q.appendExecutionLog(ctx, msg.ID, fmt.Sprintf("[系统] 任务执行成功，耗时: %d ms", duration.Milliseconds())) +} + +func (q *InprocQueue) failExecution(ctx context.Context, msg TaskMessage, execErr error, duration time.Duration) { + db := getDB(ctx) + if db == nil { + return + } + now := time.Now() + updates := map[string]any{ + taskExecutionColStatus: taskExecutionStatusFailed, + "error_message": execErr.Error(), + "finished_at": now, + "duration": duration.Milliseconds(), + } + if err := db.Model(&taskExecution{}).Where("task_id = ?", msg.ID).Updates(updates).Error; err != nil { + logger.ErrorF(ctx, "driver_inproc_worker: mark failed failed taskID=%s: %v", msg.ID, err) + return + } + q.appendExecutionLog(ctx, msg.ID, fmt.Sprintf("[系统] 任务执行失败，耗时: %d ms，错误原因: %v", duration.Milliseconds(), execErr)) +} + +func (q *InprocQueue) appendExecutionLog(ctx context.Context, taskID, logLine string) { + db := getDB(ctx) + if db == nil { + return + } + now := time.Now().Format("15:04:05") + line := fmt.Sprintf("[%s] %s\n", now, logLine) + var exec taskExecution + if err := db.Where("task_id = ?", taskID).First(&exec).Error; err != nil { + return + } + if err := db.Model(&exec).Update("log", exec.Log+line).Error; err != nil { + logger.ErrorF(ctx, "driver_inproc_worker: append log failed taskID=%s: %v", taskID, err) + } +} diff --git a/backend/plugins/drivers/driver_inproc_worker/plugin.go b/backend/plugins/drivers/driver_inproc_worker/plugin.go index cd004818..2525955a 100644 --- a/backend/plugins/drivers/driver_inproc_worker/plugin.go +++ b/backend/plugins/drivers/driver_inproc_worker/plugin.go @@ -8,6 +8,7 @@ import ( "Wavelet/core" "Wavelet/core/contracts" "context" + "errors" "sync" "time" ) @@ -24,15 +25,15 @@ var ( ) // DispatchTask enqueues a background task to the active in-process worker queue. -func DispatchTask(_ context.Context, taskType string, payload []byte, source string) (string, error) { +func DispatchTask(ctx context.Context, taskType string, payload []byte, source string) (string, error) { globalMu.RLock() q := globalQueue globalMu.RUnlock() if q == nil { - return "", nil + return "", errors.New("driver_inproc_worker: queue is not running") } - return q.Enqueue(taskType, payload, source) + return q.Enqueue(ctx, taskType, payload, source) } // Option configures the in-process worker driver plugin. @@ -118,10 +119,19 @@ func (p *Plugin) ConfigEnabled(view core.ConfigView) bool { func (p *Plugin) Apply(ctx *core.Context) error { p.coreCtx = ctx + if db, err := core.Inject[contracts.DBService](ctx); err == nil && db != nil { + setDBService(db) + } else { + core.When[contracts.DBService](ctx, func(db contracts.DBService) { + setDBService(db) + }) + } + taskSvc := newInprocTaskService(ctx.Tasks()) core.Provide[contracts.TaskService](ctx, taskSvc) ctx.OnDispose(func() error { + setDBService(nil) shutdownCtx, cancel := context.WithTimeout(context.Background(), p.shutdownTimeout) defer cancel() return p.Stop(shutdownCtx) diff --git a/backend/plugins/drivers/driver_inproc_worker/plugin_test.go b/backend/plugins/drivers/driver_inproc_worker/plugin_test.go index 1107e433..2555fd76 100644 --- a/backend/plugins/drivers/driver_inproc_worker/plugin_test.go +++ b/backend/plugins/drivers/driver_inproc_worker/plugin_test.go @@ -5,8 +5,10 @@ package driver_inproc_worker_test import ( "Wavelet/core" + "Wavelet/core/contracts" "Wavelet/core/extpoints" "Wavelet/pkg/idgen" + "Wavelet/pkg/testhelper" "Wavelet/plugins/drivers/driver_inproc_worker" "context" "sync/atomic" @@ -15,8 +17,19 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "gorm.io/gorm" ) +type testDBService struct { + db *gorm.DB +} + +func (m *testDBService) GORM() *gorm.DB { return m.db } + +func (m *testDBService) DB(ctx context.Context) *gorm.DB { return m.db.WithContext(ctx) } + +func (m *testDBService) Named(_ string) *gorm.DB { return m.db } + func TestInprocWorkerPlugin(t *testing.T) { require.NoError(t, idgen.Init(1)) ctx := core.NewContext(context.Background()) @@ -51,3 +64,53 @@ func TestInprocWorkerPlugin(t *testing.T) { // Stop driver require.NoError(t, p.Stop(context.Background())) } + +func TestInprocWorkerDispatchByTypeTracksExecution(t *testing.T) { + require.NoError(t, idgen.Init(1)) + testDB, _, cleanup := testhelper.SetupTestEnvironment(t) + defer cleanup() + + ctx := core.NewContext(context.Background()) + core.Provide[contracts.DBService](ctx, &testDBService{db: testDB}) + + p := driver_inproc_worker.New( + driver_inproc_worker.WithConcurrency(2), + driver_inproc_worker.WithShutdownTimeout(time.Second), + ) + require.NoError(t, p.Apply(ctx)) + + var executedCount atomic.Int32 + ctx.Tasks().Register("system:cleanup", func(_ context.Context, _ []byte) error { + executedCount.Add(1) + return nil + }, + extpoints.WithTaskType("system_cleanup"), + extpoints.WithTaskName("系统垃圾清理"), + extpoints.WithTaskRetry(1), + extpoints.WithTaskRetryable(true), + ) + + require.NoError(t, p.Start(context.Background())) + t.Cleanup(func() { + _ = p.Stop(context.Background()) + }) + + taskID, err := driver_inproc_worker.DispatchTask(context.Background(), "system_cleanup", []byte("payload"), "manual") + require.NoError(t, err) + assert.NotEmpty(t, taskID) + + require.Eventually(t, func() bool { + return executedCount.Load() == 1 + }, 2*time.Second, 20*time.Millisecond, "inproc worker should execute task dispatched by admin type") + + taskSvc, err := core.Inject[contracts.TaskService](ctx) + require.NoError(t, err) + + require.Eventually(t, func() bool { + execs, total, listErr := taskSvc.ListExecutions(context.Background(), "", "", 1, 10) + if listErr != nil || total == 0 || len(execs) == 0 { + return false + } + return execs[0].TaskID == taskID && execs[0].Status == "succeeded" && execs[0].TaskName == "系统垃圾清理" + }, 2*time.Second, 20*time.Millisecond, "inproc worker should persist a succeeded execution record") +} diff --git a/backend/plugins/drivers/driver_inproc_worker/task_service.go b/backend/plugins/drivers/driver_inproc_worker/task_service.go index 0f3e7d83..a268b444 100644 --- a/backend/plugins/drivers/driver_inproc_worker/task_service.go +++ b/backend/plugins/drivers/driver_inproc_worker/task_service.go @@ -25,8 +25,22 @@ func (s *inprocTaskService) Dispatch(ctx context.Context, taskType string, paylo return DispatchTask(ctx, taskType, payload, triggeredBy) } -func (s *inprocTaskService) Retry(_ context.Context, id uint64) (string, error) { - return fmt.Sprintf("inproc_retry_%d", id), nil +func (s *inprocTaskService) Retry(ctx context.Context, id uint64) (string, error) { + db := getDB(ctx) + if db == nil { + return "", errors.New("driver_inproc_worker: db not initialized") + } + var exec taskExecution + if err := db.Where("id = ?", id).First(&exec).Error; err != nil { + return "", fmt.Errorf("driver_inproc_worker: task execution not found: %w", err) + } + if exec.Status != taskExecutionStatusFailed { + return "", fmt.Errorf("driver_inproc_worker: only failed tasks can be retried, current status: %s", exec.Status) + } + if !exec.Retryable { + return "", errors.New("driver_inproc_worker: task is not retryable") + } + return DispatchTask(ctx, exec.TaskType, []byte(exec.Payload), "retry") } func (s *inprocTaskService) ListTasks() []contracts.TaskMetaDTO { @@ -64,10 +78,72 @@ func (s *inprocTaskService) ReloadScheduler() error { func (s *inprocTaskService) AppendLog(_ context.Context, _ string, _ ...any) { } -func (s *inprocTaskService) ListExecutions(_ context.Context, _, _ string, _, _ int) ([]contracts.TaskExecutionDTO, int64, error) { - return []contracts.TaskExecutionDTO{}, 0, nil +func (s *inprocTaskService) ListExecutions(ctx context.Context, taskType, status string, page, pageSize int) ([]contracts.TaskExecutionDTO, int64, error) { + db := getDB(ctx) + if db == nil { + return []contracts.TaskExecutionDTO{}, 0, nil + } + if page <= 0 { + page = 1 + } + if pageSize <= 0 { + pageSize = 20 + } + query := db.Model(&taskExecution{}) + if taskType != "" { + query = query.Where("task_type = ?", taskType) + } + if status != "" { + query = query.Where("status = ?", status) + } + var total int64 + if err := query.Count(&total).Error; err != nil { + return nil, 0, err + } + var rows []taskExecution + offset := (page - 1) * pageSize + if err := query.Order("id DESC").Offset(offset).Limit(pageSize).Find(&rows).Error; err != nil { + return nil, 0, err + } + res := make([]contracts.TaskExecutionDTO, 0, len(rows)) + for i := range rows { + res = append(res, toExecutionDTO(&rows[i])) + } + return res, total, nil } -func (s *inprocTaskService) GetExecution(_ context.Context, _ uint64) (*contracts.TaskExecutionDTO, error) { - return nil, errors.New("driver_inproc_worker: task executions are not tracked") +func (s *inprocTaskService) GetExecution(ctx context.Context, id uint64) (*contracts.TaskExecutionDTO, error) { + db := getDB(ctx) + if db == nil { + return nil, errors.New("driver_inproc_worker: db not initialized") + } + var exec taskExecution + if err := db.Where("id = ?", id).First(&exec).Error; err != nil { + return nil, err + } + dto := toExecutionDTO(&exec) + return &dto, nil +} + +func toExecutionDTO(exec *taskExecution) contracts.TaskExecutionDTO { + return contracts.TaskExecutionDTO{ + ID: exec.ID, + TaskID: exec.TaskID, + TaskType: exec.TaskType, + TaskName: exec.TaskName, + Status: string(exec.Status), + Retryable: exec.Retryable, + MaxRetry: exec.MaxRetry, + RetryCount: exec.RetryCount, + Log: exec.Log, + ErrorMessage: exec.ErrorMessage, + Result: exec.Result, + StartedAt: exec.StartedAt, + FinishedAt: exec.FinishedAt, + Duration: exec.Duration, + Payload: exec.Payload, + TriggeredBy: exec.TriggeredBy, + CreatedAt: exec.CreatedAt, + UpdatedAt: exec.UpdatedAt, + } } diff --git a/backend/plugins/drivers/driver_inproc_worker/types.go b/backend/plugins/drivers/driver_inproc_worker/types.go new file mode 100644 index 00000000..c3e5f84f --- /dev/null +++ b/backend/plugins/drivers/driver_inproc_worker/types.go @@ -0,0 +1,44 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package driver_inproc_worker + +import "time" + +type taskExecutionStatus string + +const ( + taskExecutionStatusPending taskExecutionStatus = "pending" + taskExecutionStatusRunning taskExecutionStatus = "running" + taskExecutionStatusSucceeded taskExecutionStatus = "succeeded" + taskExecutionStatusFailed taskExecutionStatus = "failed" + + taskExecutionColStatus = "status" +) + +// taskExecution maps to the admin-owned w_task_executions table so the +// console can list in-process runs the same way it lists Asynq runs. +type taskExecution struct { + ID uint64 `gorm:"primaryKey"` + TaskID string `gorm:"size:128;uniqueIndex;not null"` + TaskType string `gorm:"size:64;index;not null"` + TaskName string `gorm:"size:128"` + Status taskExecutionStatus `gorm:"size:32;index;not null"` + Retryable bool `gorm:"not null;default:false"` + MaxRetry int `gorm:"not null;default:0"` + RetryCount int `gorm:"not null;default:0"` + Log string `gorm:"type:text"` + ErrorMessage string `gorm:"type:text"` + Result string `gorm:"type:text"` + StartedAt *time.Time `gorm:"index"` + FinishedAt *time.Time + Duration int64 `gorm:"comment:耗时毫秒"` + Payload string `gorm:"type:text"` + TriggeredBy string `gorm:"size:32;not null;default:system"` + CreatedAt time.Time + UpdatedAt time.Time +} + +func (taskExecution) TableName() string { + return "w_task_executions" +} diff --git a/backend/plugins/drivers/drivers_test.go b/backend/plugins/drivers/drivers_test.go index 2fc891d2..2a323e83 100644 --- a/backend/plugins/drivers/drivers_test.go +++ b/backend/plugins/drivers/drivers_test.go @@ -5,7 +5,10 @@ package drivers_test import ( "Wavelet/core" + "Wavelet/core/contracts" "Wavelet/core/extpoints" + "Wavelet/pkg/idgen" + "Wavelet/pkg/testhelper" "Wavelet/plugins/drivers/driver_asynq_cron" "Wavelet/plugins/drivers/driver_asynq_worker" "Wavelet/plugins/drivers/driver_http" @@ -24,8 +27,19 @@ import ( "github.com/hibiken/asynq" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "gorm.io/gorm" ) +type testDBService struct { + db *gorm.DB +} + +func (m *testDBService) GORM() *gorm.DB { return m.db } + +func (m *testDBService) DB(ctx context.Context) *gorm.DB { return m.db.WithContext(ctx) } + +func (m *testDBService) Named(_ string) *gorm.DB { return m.db } + func init() { gin.SetMode(gin.TestMode) } @@ -214,6 +228,59 @@ func TestAsynqWorkerDriverLifecycle(t *testing.T) { require.NoError(t, err) } +func TestAsynqWorkerDispatchTracksExecution(t *testing.T) { + _ = idgen.Init(1) + testDB, mr, cleanup := testhelper.SetupTestEnvironment(t) + defer cleanup() + + ctx := core.NewContext(context.Background()) + ctx.Config().SetSource(core.NewMapSource(nil)) + require.NoError(t, ctx.Config().Resolve()) + core.Provide[contracts.DBService](ctx, &testDBService{db: testDB}) + + var processed atomic.Bool + ctx.Tasks().Register("system:cleanup", func(_ context.Context, _ []byte) error { + processed.Store(true) + return nil + }, + extpoints.WithTaskType("system_cleanup"), + extpoints.WithTaskName("系统垃圾清理"), + extpoints.WithTaskQueue("default"), + extpoints.WithTaskRetry(1), + extpoints.WithTaskRetryable(true), + ) + + workerPlugin := driver_asynq_worker.New( + driver_asynq_worker.WithRedisOpt(asynq.RedisClientOpt{Addr: mr.Addr()}), + driver_asynq_worker.WithConcurrency(2), + driver_asynq_worker.WithShutdownTimeout(2*time.Second), + ) + require.NoError(t, workerPlugin.Apply(ctx)) + require.NoError(t, workerPlugin.Start(context.Background())) + t.Cleanup(func() { + _ = workerPlugin.Stop(context.Background()) + }) + + taskSvc, err := core.Inject[contracts.TaskService](ctx) + require.NoError(t, err) + + taskID, err := taskSvc.Dispatch(context.Background(), "system_cleanup", []byte(`{}`), "manual") + require.NoError(t, err) + require.NotEmpty(t, taskID) + + require.Eventually(t, func() bool { + return processed.Load() + }, 5*time.Second, 50*time.Millisecond, "asynq worker should execute dispatched func handler") + + require.Eventually(t, func() bool { + execs, _, listErr := taskSvc.ListExecutions(context.Background(), "", "", 1, 10) + if listErr != nil || len(execs) == 0 { + return false + } + return execs[0].TaskID == taskID && execs[0].Status == "succeeded" + }, 5*time.Second, 50*time.Millisecond, "task execution should become succeeded after worker runs") +} + func TestAsynqCronDriverLifecycle(t *testing.T) { mr, err := miniredis.Run() require.NoError(t, err)