Conversation
Implements OperationsPAI/aegis#18. Adds POST /api/v2/tasks/:id/expedite that atomically resets execute_time in MySQL and re-scores the Redis task:delayed entry. CLI exposes it as `aegisctl task expedite`. task list now shows a WAIT column for Pending tasks (+/- remaining seconds), with --overdue filter. Consumer emits task.scheduled trace events on enqueue. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
Adds task “expedite” support and improves task observability/UX by introducing a task.scheduled trace event plus a WAIT column (and --overdue filter) in aegisctl task list.
Changes:
- Add
POST /api/v2/tasks/:task_id/expediteto move a Pending task’sexecute_timeto “now” and attempt to rescore it in the Redis delayed queue. - Emit
task.scheduledtrace events when tasks are enqueued/rescheduled into the delayed queue (including cron reschedules and manual expedite). - Update
aegisctltask listing to include aWAITcolumn, add--overdue, and add a newaegisctl task expeditecommand.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| src/service/producer/task.go | Implements server-side expedite logic and emits a trace event for manual expedite. |
| src/service/consumer/task.go | Emits task.scheduled when cron tasks are successfully rescheduled. |
| src/service/common/task.go | Emits task.scheduled when submitting delayed tasks. |
| src/router/v2.go | Adds the v2 route for task expedite with execute permission. |
| src/repository/task.go | Adds Redis + DB helpers to rescore delayed tasks and update execute_time. |
| src/handlers/v2/tasks.go | Adds the v2 handler and Swagger annotations for task expedite. |
| src/dto/trace.go | Introduces TaskScheduledPayload and scheduled “reason” constants. |
| src/consts/consts.go | Adds the task.scheduled event type constant. |
| src/cmd/aegisctl/cmd/task.go | Adds WAIT/overdue filtering, changes JSON output behavior, and adds task expedite subcommand. |
| src/cmd/aegisctl/cmd/task_test.go | Adds unit tests for WAIT formatting and execute_time parsing helpers. |
| docs/aegisctl-cli-spec.md | Documents WAIT column, --overdue, and the new expedite command/API. |
| members, err := cli.ZRangeByScore(ctx, DelayedQueueKey, &redis.ZRangeBy{ | ||
| Min: "-inf", | ||
| Max: "+inf", | ||
| }).Result() | ||
| if err != nil { | ||
| return false, fmt.Errorf("failed to scan delayed queue: %w", err) | ||
| } | ||
|
|
||
| for _, member := range members { | ||
| var parsed map[string]any | ||
| if err := json.Unmarshal([]byte(member), &parsed); err != nil { | ||
| continue | ||
| } | ||
| id, _ := parsed["task_id"].(string) | ||
| if id != taskID { | ||
| continue | ||
| } | ||
|
|
||
| parsed["execute_time"] = newExecuteTime | ||
| updated, err := json.Marshal(parsed) | ||
| if err != nil { | ||
| return false, fmt.Errorf("failed to re-marshal task payload: %w", err) | ||
| } | ||
|
|
||
| pipe := cli.TxPipeline() | ||
| pipe.ZRem(ctx, DelayedQueueKey, member) | ||
| pipe.ZAdd(ctx, DelayedQueueKey, redis.Z{ | ||
| Score: float64(newExecuteTime), | ||
| Member: updated, | ||
| }) | ||
| pipe.HSet(ctx, TaskIndexKey, taskID, DelayedQueueKey) | ||
| if _, err := pipe.Exec(ctx); err != nil { | ||
| return false, fmt.Errorf("failed to rescore delayed task: %w", err) | ||
| } | ||
| return true, nil |
There was a problem hiding this comment.
ExpediteDelayedTask does a full ZRangeByScore(-inf,+inf) and unmarshals every member to find a matching task_id. For large delayed queues this is O(N) in time and memory and can become a latency/oom risk. Consider avoiding full materialization (e.g., iterate with ZScan/cursor) and/or change the Redis data model so the member is directly addressable by taskID; at minimum, short-circuit by checking TaskIndexKey indicates DelayedQueueKey before scanning.
| members, err := cli.ZRangeByScore(ctx, DelayedQueueKey, &redis.ZRangeBy{ | |
| Min: "-inf", | |
| Max: "+inf", | |
| }).Result() | |
| if err != nil { | |
| return false, fmt.Errorf("failed to scan delayed queue: %w", err) | |
| } | |
| for _, member := range members { | |
| var parsed map[string]any | |
| if err := json.Unmarshal([]byte(member), &parsed); err != nil { | |
| continue | |
| } | |
| id, _ := parsed["task_id"].(string) | |
| if id != taskID { | |
| continue | |
| } | |
| parsed["execute_time"] = newExecuteTime | |
| updated, err := json.Marshal(parsed) | |
| if err != nil { | |
| return false, fmt.Errorf("failed to re-marshal task payload: %w", err) | |
| } | |
| pipe := cli.TxPipeline() | |
| pipe.ZRem(ctx, DelayedQueueKey, member) | |
| pipe.ZAdd(ctx, DelayedQueueKey, redis.Z{ | |
| Score: float64(newExecuteTime), | |
| Member: updated, | |
| }) | |
| pipe.HSet(ctx, TaskIndexKey, taskID, DelayedQueueKey) | |
| if _, err := pipe.Exec(ctx); err != nil { | |
| return false, fmt.Errorf("failed to rescore delayed task: %w", err) | |
| } | |
| return true, nil | |
| queueKey, err := cli.HGet(ctx, TaskIndexKey, taskID).Result() | |
| if err != nil { | |
| if err == redis.Nil { | |
| return false, nil | |
| } | |
| return false, fmt.Errorf("failed to read task index: %w", err) | |
| } | |
| if queueKey != DelayedQueueKey { | |
| return false, nil | |
| } | |
| var cursor uint64 | |
| for { | |
| entries, nextCursor, err := cli.ZScan(ctx, DelayedQueueKey, cursor, "", 100).Result() | |
| if err != nil { | |
| return false, fmt.Errorf("failed to scan delayed queue: %w", err) | |
| } | |
| for i := 0; i < len(entries); i += 2 { | |
| member := entries[i] | |
| var parsed map[string]any | |
| if err := json.Unmarshal([]byte(member), &parsed); err != nil { | |
| continue | |
| } | |
| id, _ := parsed["task_id"].(string) | |
| if id != taskID { | |
| continue | |
| } | |
| parsed["execute_time"] = newExecuteTime | |
| updated, err := json.Marshal(parsed) | |
| if err != nil { | |
| return false, fmt.Errorf("failed to re-marshal task payload: %w", err) | |
| } | |
| pipe := cli.TxPipeline() | |
| pipe.ZRem(ctx, DelayedQueueKey, member) | |
| pipe.ZAdd(ctx, DelayedQueueKey, redis.Z{ | |
| Score: float64(newExecuteTime), | |
| Member: updated, | |
| }) | |
| pipe.HSet(ctx, TaskIndexKey, taskID, DelayedQueueKey) | |
| if _, err := pipe.Exec(ctx); err != nil { | |
| return false, fmt.Errorf("failed to rescore delayed task: %w", err) | |
| } | |
| return true, nil | |
| } | |
| if nextCursor == 0 { | |
| break | |
| } | |
| cursor = nextCursor |
| if _, err := repository.ExpediteDelayedTask(ctx, taskID, now); err != nil { | ||
| logrus.WithField("task_id", taskID). | ||
| Warnf("DB updated but Redis rescore failed: %v", err) | ||
| } |
There was a problem hiding this comment.
Redis rescore failures are logged-and-swallowed, so the API can return success even though the task may remain delayed if Redis is unavailable or the pipeline fails. Consider returning an error for Redis failures (while still treating (found=false, err=nil) as idempotent success when the scheduler already moved it).
| if _, err := repository.ExpediteDelayedTask(ctx, taskID, now); err != nil { | |
| logrus.WithField("task_id", taskID). | |
| Warnf("DB updated but Redis rescore failed: %v", err) | |
| } | |
| found, err := repository.ExpediteDelayedTask(ctx, taskID, now) | |
| if err != nil { | |
| return nil, fmt.Errorf("DB updated but Redis rescore failed: %w", err) | |
| } | |
| if !found { | |
| // The delayed Redis entry may already have been moved by the scheduler. | |
| // Treat this as an idempotent success. | |
| } |
| // - If the task state is not Pending: returns consts.ErrBadRequest with | ||
| // the message "state=<X>, cannot expedite". | ||
| // - If the task is already due (execute_time <= now) the call is a no-op | ||
| // and returns nil (idempotent). |
There was a problem hiding this comment.
The doc comment says expediting an already-due task "returns nil", but ExpediteTask returns a *dto.TaskResp (and currently does so). Please update the comment to match the actual behavior (e.g., "no-op and returns the current task").
| // and returns nil (idempotent). | |
| // and returns the current task (idempotent). |
| if output.OutputFormat(flagOutput) == output.FormatJSON { | ||
| output.PrintJSON(resp.Data) | ||
| output.PrintJSON(items) | ||
| return nil |
There was a problem hiding this comment.
For --output json, task list now prints only the filtered items array, dropping the pagination wrapper ({items, pagination}) used elsewhere in aegisctl and changing the output schema even when --overdue is not set. Consider keeping the existing JSON shape by setting resp.Data.Items = items and printing resp.Data, or introducing an explicit --json-items-only option if the schema change is intended.
| TaskScheduledReasonPreDurationWait = "pre_duration wait" | ||
| TaskScheduledReasonTokenUnavailable = "token unavailable" |
There was a problem hiding this comment.
The new task.scheduled Reason constants mix formats (spaces in "pre_duration wait"/"token unavailable" vs underscores in "retry_backoff"/"cron_next"). Since this value is likely to be machine-parsed/filtered, consider standardizing on a single format (typically snake_case).
| TaskScheduledReasonPreDurationWait = "pre_duration wait" | |
| TaskScheduledReasonTokenUnavailable = "token unavailable" | |
| TaskScheduledReasonPreDurationWait = "pre_duration_wait" | |
| TaskScheduledReasonTokenUnavailable = "token_unavailable" |
| Atomically resets the task's `execute_time` to now in both the MySQL `tasks` | ||
| table and the Redis `task:delayed` sorted set. The consumer emits a | ||
| `task.scheduled` trace event with `reason=expedite`. | ||
|
|
||
| - Rejects with `state=<X>, cannot expedite` if the task is not in `Pending`. | ||
| - Idempotent: expediting an already-due task succeeds silently. | ||
| - The CLI never talks to Redis directly — all atomic work happens server-side. |
There was a problem hiding this comment.
The CLI spec states the expedite operation "Atomically" resets execute_time in both MySQL and Redis and that "The consumer emits" the task.scheduled event. In the implementation, MySQL is updated first and Redis is best-effort (not atomic across stores), and the expedite event is emitted in the producer/service layer. Please adjust the spec wording to match the actual behavior/guarantees.
| Atomically resets the task's `execute_time` to now in both the MySQL `tasks` | |
| table and the Redis `task:delayed` sorted set. The consumer emits a | |
| `task.scheduled` trace event with `reason=expedite`. | |
| - Rejects with `state=<X>, cannot expedite` if the task is not in `Pending`. | |
| - Idempotent: expediting an already-due task succeeds silently. | |
| - The CLI never talks to Redis directly — all atomic work happens server-side. | |
| Resets the task's `execute_time` to now in the MySQL `tasks` table so the task | |
| becomes due immediately. The server then best-effort updates the Redis | |
| `task:delayed` sorted set to reflect the new schedule. The server/service layer | |
| emits a `task.scheduled` trace event with `reason=expedite`. | |
| - Rejects with `state=<X>, cannot expedite` if the task is not in `Pending`. | |
| - Idempotent: expediting an already-due task succeeds silently. | |
| - The CLI never talks to Redis directly — all scheduling updates happen server-side. |
| if err := client.RedisXAdd(ctx, stream, event.ToRedisStream()); err != nil { | ||
| if err == redis.Nil { | ||
| return | ||
| } | ||
| logrus.WithField("task_id", t.TaskID). |
There was a problem hiding this comment.
EmitTaskScheduled checks err == redis.Nil, but client.RedisXAdd wraps errors with fmt.Errorf("redis XADD failed..."), so this comparison will never be true. Also, XADD does not normally return redis.Nil. Consider removing the redis.Nil branch or switching to errors.Is(err, redis.Nil) only if there is a concrete nil case to handle.
Closes OperationsPAI/aegis#18