完善任务取消状态一致性保护
This commit is contained in:
parent
8b1d1853e3
commit
8158cefc05
@ -144,7 +144,7 @@ func (r *SQLTaskRepository) UpdateTask(ctx context.Context, id, status, stage, m
|
||||
if !validTaskStatus(status) || stage == "" {
|
||||
return errors.New("任务状态或阶段无效")
|
||||
}
|
||||
result, err := r.db.ExecContext(ctx, `UPDATE tasks SET status = ?, stage = ?, message = ?, started_at = ?, finished_at = ?, updated_at = ? WHERE id = ?`,
|
||||
result, err := r.db.ExecContext(ctx, `UPDATE tasks SET status = ?, stage = ?, message = ?, started_at = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`,
|
||||
status, stage, message, nullableTime(startedAt), nullableTime(finishedAt), time.Now().UTC(), id)
|
||||
if err != nil {
|
||||
return fmt.Errorf("更新任务失败:%w", err)
|
||||
@ -167,24 +167,45 @@ func (r *SQLTaskRepository) CancelTask(ctx context.Context, id, message string,
|
||||
if id == "" {
|
||||
return errors.New("任务 ID 不能为空")
|
||||
}
|
||||
result, err := r.db.ExecContext(ctx, `UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`, message, nullableTime(finishedAt), time.Now().UTC(), id)
|
||||
tx, err := r.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("开始取消任务事务失败:%w", err)
|
||||
}
|
||||
rollback := func() {
|
||||
_ = tx.Rollback()
|
||||
}
|
||||
result, err := tx.ExecContext(ctx, `UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`, message, nullableTime(finishedAt), time.Now().UTC(), id)
|
||||
if err != nil {
|
||||
rollback()
|
||||
return fmt.Errorf("取消任务失败:%w", err)
|
||||
}
|
||||
count, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
rollback()
|
||||
return fmt.Errorf("读取任务取消结果失败:%w", err)
|
||||
}
|
||||
if count == 0 {
|
||||
item, getErr := r.GetTask(ctx, id)
|
||||
item, getErr := scanTask(tx.QueryRowContext(ctx, taskSelect+` WHERE id = ?`, id))
|
||||
if getErr != nil {
|
||||
rollback()
|
||||
return getErr
|
||||
}
|
||||
if item.Status == "cancelled" {
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("提交重复取消任务事务失败:%w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
rollback()
|
||||
return errors.New("任务当前无法取消")
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `UPDATE task_attempts SET status = 'cancelled', message = ?, finished_at = ? WHERE task_id = ? AND status IN ('queued', 'running')`, message, nullableTime(finishedAt), id); err != nil {
|
||||
rollback()
|
||||
return fmt.Errorf("取消任务尝试失败:%w", err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("提交取消任务事务失败:%w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (r *SQLTaskRepository) FindActiveTaskByDomain(ctx context.Context, domainID, taskType string) (Task, error) {
|
||||
@ -247,7 +268,7 @@ func (r *SQLTaskRepository) FinishAttempt(ctx context.Context, id, status, messa
|
||||
if id == "" || !validTaskStatus(status) {
|
||||
return errors.New("任务尝试 ID 或状态无效")
|
||||
}
|
||||
result, err := r.db.ExecContext(ctx, `UPDATE task_attempts SET status = ?, message = ?, finished_at = ? WHERE id = ?`, status, message, nullableTime(finishedAt), id)
|
||||
result, err := r.db.ExecContext(ctx, `UPDATE task_attempts SET status = ?, message = ?, finished_at = ? WHERE id = ? AND (status IN ('queued', 'running') OR status = ?)`, status, message, nullableTime(finishedAt), id, status)
|
||||
if err != nil {
|
||||
return fmt.Errorf("完成任务尝试失败:%w", err)
|
||||
}
|
||||
@ -256,6 +277,13 @@ func (r *SQLTaskRepository) FinishAttempt(ctx context.Context, id, status, messa
|
||||
return fmt.Errorf("读取任务尝试更新结果失败:%w", err)
|
||||
}
|
||||
if count == 0 {
|
||||
var currentStatus string
|
||||
if err := r.db.QueryRowContext(ctx, `SELECT status FROM task_attempts WHERE id = ?`, id).Scan(¤tStatus); err != nil {
|
||||
return err
|
||||
}
|
||||
if currentStatus == "cancelled" {
|
||||
return nil
|
||||
}
|
||||
return sql.ErrNoRows
|
||||
}
|
||||
return nil
|
||||
|
||||
@ -91,3 +91,65 @@ func TestSQLTaskRepositoryRejectsInvalidStatus(t *testing.T) {
|
||||
t.Fatal("无效任务状态应被拒绝")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSQLTaskRepositoryCancelMarksRunningAttempt(t *testing.T) {
|
||||
repository, db := newTaskRepositoryTest(t)
|
||||
defer db.Close()
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
if err := repository.CreateTask(ctx, Task{
|
||||
ID: "task-cancel", DomainID: "domain-1", TaskType: "certificate_issue",
|
||||
Status: "running", Stage: "certificate_issue", Message: "正在执行", CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("创建取消测试任务失败:%v", err)
|
||||
}
|
||||
if err := repository.CreateAttempt(ctx, TaskAttempt{
|
||||
ID: "attempt-cancel", TaskID: "task-cancel", AttemptNumber: 1,
|
||||
Stage: "certificate_issue", Status: "running", Message: "正在执行", StartedAt: &now, CreatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("创建取消测试尝试失败:%v", err)
|
||||
}
|
||||
finished := now.Add(time.Second)
|
||||
if err := repository.CancelTask(ctx, "task-cancel", "任务已由用户取消", &finished); err != nil {
|
||||
t.Fatalf("取消任务失败:%v", err)
|
||||
}
|
||||
task, err := repository.GetTask(ctx, "task-cancel")
|
||||
if err != nil || task.Status != "cancelled" || task.FinishedAt == nil {
|
||||
t.Fatalf("取消后的任务状态错误:%+v %v", task, err)
|
||||
}
|
||||
attempts, err := repository.ListAttempts(ctx, "task-cancel")
|
||||
if err != nil || len(attempts) != 1 || attempts[0].Status != "cancelled" || attempts[0].FinishedAt == nil {
|
||||
t.Fatalf("取消后的任务尝试状态错误:%+v %v", attempts, err)
|
||||
}
|
||||
if err := repository.FinishAttempt(ctx, "attempt-cancel", "success", "晚到的成功结果", &finished); err != nil {
|
||||
t.Fatalf("重复完成已取消尝试不应失败:%v", err)
|
||||
}
|
||||
attempts, err = repository.ListAttempts(ctx, "task-cancel")
|
||||
if err != nil || attempts[0].Status != "cancelled" {
|
||||
t.Fatalf("已取消尝试不应被晚到结果覆盖:%+v %v", attempts, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSQLTaskRepositoryDoesNotOverwriteCancelledTask(t *testing.T) {
|
||||
repository, db := newTaskRepositoryTest(t)
|
||||
defer db.Close()
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
if err := repository.CreateTask(ctx, Task{
|
||||
ID: "task-state", DomainID: "domain-1", TaskType: "certificate_issue",
|
||||
Status: "running", Stage: "certificate_issue", Message: "正在执行", CreatedAt: now, UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("创建状态测试任务失败:%v", err)
|
||||
}
|
||||
finished := now.Add(time.Second)
|
||||
if err := repository.CancelTask(ctx, "task-state", "任务已由用户取消", &finished); err != nil {
|
||||
t.Fatalf("取消状态测试任务失败:%v", err)
|
||||
}
|
||||
if err := repository.UpdateTask(ctx, "task-state", "success", "certificate_issue", "晚到的成功结果", &now, &finished); err == nil {
|
||||
t.Fatal("已取消任务不应被更新为成功")
|
||||
}
|
||||
task, err := repository.GetTask(ctx, "task-state")
|
||||
if err != nil || task.Status != "cancelled" {
|
||||
t.Fatalf("已取消任务状态被覆盖:%+v %v", task, err)
|
||||
}
|
||||
}
|
||||
|
||||
@ -189,7 +189,9 @@ h2 {
|
||||
justify-content: flex-end;
|
||||
gap: 10px;
|
||||
flex-wrap: wrap;
|
||||
}.record-row > div {
|
||||
}
|
||||
|
||||
.record-row > div {
|
||||
display: grid;
|
||||
gap: 5px;
|
||||
min-width: 0;
|
||||
|
||||
@ -50,7 +50,8 @@ export function useDashboard() {
|
||||
showForm.value ||
|
||||
showServerForm.value ||
|
||||
showSSHKeyForm.value ||
|
||||
showSettings.value,
|
||||
showSettings.value ||
|
||||
showCertificateImport.value,
|
||||
);
|
||||
let previousBodyOverflow = "";
|
||||
|
||||
@ -98,7 +99,8 @@ export function useDashboard() {
|
||||
success: "已完成",
|
||||
failed: "失败",
|
||||
cancelled: "已取消",
|
||||
}; const domainStatusLabels = {
|
||||
};
|
||||
const domainStatusLabels = {
|
||||
pending: "等待申请",
|
||||
active: "正常",
|
||||
error: "错误",
|
||||
|
||||
Loading…
Reference in New Issue
Block a user