diff --git a/backend/internal/httpapi/task_api.go b/backend/internal/httpapi/task_api.go index 72b0109..faa4ed8 100644 --- a/backend/internal/httpapi/task_api.go +++ b/backend/internal/httpapi/task_api.go @@ -432,21 +432,14 @@ func validTaskIdempotencyKey(value string) bool { } return true } + func isActiveTaskStatus(status string) bool { - -switch status { - -case "queued", "running", "retry_wait", "cancelling": - - -return true - -default: - - -return false - -} + switch status { + case "queued", "running", "retry_wait", "cancelling": + return true + default: + return false + } } func (s *Server) listTasks(c *gin.Context) { diff --git a/backend/internal/httpapi/task_api_test.go b/backend/internal/httpapi/task_api_test.go index 1ec09e3..fc38c19 100644 --- a/backend/internal/httpapi/task_api_test.go +++ b/backend/internal/httpapi/task_api_test.go @@ -394,6 +394,33 @@ func TestTaskAPIReplaysIdempotentIssue(t *testing.T) { } } +func TestTaskAPIRejectsIdempotentIssueForAllActiveStatuses(t *testing.T) { + server, taskRepo, _, closeDB := newTaskAPITestServer(t, func(context.Context, string) (certificate.IssueResult, deployment.Report, error) { + t.Fatal("活动幂等任务不应重复执行证书申请") + return certificate.IssueResult{}, deployment.Report{}, nil + }) + defer closeDB() + + for _, status := range []string{"retry_wait", "cancelling"} { + key := "issue-active-" + status + task := repository.Task{ + HistoryScope: repository.HistoryScope{ScopeType: repository.ScopeProject, ProjectID: domain.DefaultProjectID}, + ID: "task-active-" + status, DomainID: "domain-1", TaskType: "certificate_issue", + Status: status, Stage: "certificate_issue", IdempotencyKey: key, + } + if err := taskRepo.CreateTask(t.Context(), task); err != nil { + t.Fatalf("创建%s幂等任务失败:%v", status, err) + } + request := httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil) + request.Header.Set("Idempotency-Key", key) + response := httptest.NewRecorder() + serveAdminHTTP(server, response, request) + if response.Code != http.StatusConflict || !strings.Contains(response.Body.String(), task.ID) { + t.Fatalf("%s幂等任务应返回冲突:%d %s", status, response.Code, response.Body.String()) + } + } +} + func TestTaskAPIReturnsNotImplementedWhenTaskDependenciesAreMissing(t *testing.T) { cfg := readyTestConfig(t.TempDir()) server := New(Dependencies{ diff --git a/backend/internal/repository/project_domain_repository_test.go b/backend/internal/repository/project_domain_repository_test.go index 57fac4f..4379ea8 100644 --- a/backend/internal/repository/project_domain_repository_test.go +++ b/backend/internal/repository/project_domain_repository_test.go @@ -105,7 +105,7 @@ func TestProjectDomainRepositorySQLite(t *testing.T) { t.Fatal(err) } activeStatuses := []string{"queued", "running", "retry_wait", "cancelling"} - taskTypes := []string{"certificate_issue", "certificate_validate", "certificate_deploy", "certificate_import", "certificate_restore"} + taskTypes := []string{"certificate_issue", "certificate_validate", "certificate_deploy", "certificate_import", "certificate_restore", "renewal_run"} for _, taskType := range taskTypes { for _, status := range activeStatuses { taskID := "domain-matrix-task-" + taskType + "-" + status diff --git a/backend/internal/repository/task_repository.go b/backend/internal/repository/task_repository.go index 14efab6..e0b44c1 100644 --- a/backend/internal/repository/task_repository.go +++ b/backend/internal/repository/task_repository.go @@ -97,11 +97,30 @@ func (r *SQLTaskRepository) CreateTask(ctx context.Context, item Task) error { if item.DomainID != ManagementResourceID || !item.HistoryScope.matches(HistoryScope{ScopeType: ScopeSystem}) { return errors.New("管理证书任务范围无效") } - result, err := r.db.ExecContext(ctx, r.query(insertQuery+`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'system', '')`), args...) + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("开始管理证书任务事务失败:%w", err) + } + defer tx.Rollback() + var activeTaskID string + if err := tx.QueryRowContext(ctx, r.query(`SELECT id FROM tasks + WHERE domain_id = ? AND task_type = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling') + ORDER BY created_at DESC LIMIT 1`), item.DomainID, item.TaskType).Scan(&activeTaskID); err == nil { + return fmt.Errorf("%w:%s", ErrTaskAlreadyRunning, activeTaskID) + } else if !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("检查管理证书活动任务失败:%w", err) + } + result, err := tx.ExecContext(ctx, r.query(insertQuery+`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'system', '')`), args...) if err != nil { return fmt.Errorf("创建任务失败:%w", err) } - return validateTaskInsertResult(result) + if err := validateTaskInsertResult(result); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("提交管理证书任务事务失败:%w", err) + } + return nil } if item.DomainID == ManagementResourceID || (item.ScopeType != "" && item.ScopeType != ScopeProject) { return errors.New("业务任务必须归属于实际域名项目") @@ -345,7 +364,11 @@ func (r *SQLTaskRepository) RecoverInterruptedTasks(ctx context.Context, finishe } if _, err := tx.ExecContext(ctx, r.query(`UPDATE task_attempts SET status = 'failed', message = ?, finished_at = ? - WHERE status IN ('queued', 'running', 'retry_wait', 'cancelling')`), recoveryMessage, finishedAt); err != nil { + WHERE status IN ('queued', 'running', 'retry_wait', 'cancelling') + AND task_id IN ( + SELECT id FROM tasks + WHERE status = 'failed' AND stage = 'service_restart' AND finished_at = ? + )`), recoveryMessage, finishedAt, finishedAt); err != nil { rollback() return 0, fmt.Errorf("\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u5c1d\u8bd5\u5931\u8d25\uff1a%w", err) } diff --git a/backend/internal/repository/task_repository_test.go b/backend/internal/repository/task_repository_test.go index a6b1368..6f0febd 100644 --- a/backend/internal/repository/task_repository_test.go +++ b/backend/internal/repository/task_repository_test.go @@ -122,6 +122,28 @@ func TestSQLTaskRepositoryRejectsConcurrentActiveTaskOfSameType(t *testing.T) { } } +func TestSQLTaskRepositoryRejectsConcurrentManagementRenewal(t *testing.T) { + repository, db := newTaskRepositoryTest(t) + defer db.Close() + ctx := context.Background() + if err := repository.CreateTask(ctx, Task{ + ID: "management-renewal-active", DomainID: ManagementResourceID, TaskType: "management_renewal", + Status: "running", Stage: "certificate_issue", + }); err != nil { + t.Fatalf("创建管理证书活动任务失败:%v", err) + } + err := repository.CreateTask(ctx, Task{ + ID: "management-renewal-duplicate", DomainID: ManagementResourceID, TaskType: "management_renewal", + Status: "queued", Stage: "queued", + }) + if !errors.Is(err, ErrTaskAlreadyRunning) { + t.Fatalf("管理证书活动任务应被拒绝:%v", err) + } + if _, err := repository.GetTask(ctx, "management-renewal-duplicate"); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("拒绝管理证书并发任务后不应留下重复任务:%v", err) + } +} + func TestSQLTaskRepositoryCancelMarksRunningAttempt(t *testing.T) { repository, db := newTaskRepositoryTest(t) defer db.Close() @@ -208,6 +230,11 @@ func TestSQLTaskRepositoryRecoversInterruptedTasksAndAttempts(t *testing.T) { }); err != nil { t.Fatalf("\u521b\u5efa\u5df2\u5b8c\u6210\u6d4b\u8bd5\u5c1d\u8bd5\u5931\u8d25\uff1a%v", err) } + if err := repository.CreateAttempt(ctx, TaskAttempt{ + ID: "attempt-success-stale", TaskID: "task-success", AttemptNumber: 2, Stage: "certificate_issue", Status: "running", Message: "\u5386\u53f2\u5c1d\u8bd5", CreatedAt: created, + }); err != nil { + t.Fatalf("\u521b\u5efa\u5386\u53f2\u6d3b\u52a8\u5c1d\u8bd5\u5931\u8d25\uff1a%v", err) + } finished := time.Date(2026, 8, 30, 10, 0, 0, 0, time.UTC) count, err := repository.RecoverInterruptedTasks(ctx, finished) if err != nil || count != 2 { @@ -228,7 +255,7 @@ func TestSQLTaskRepositoryRecoversInterruptedTasksAndAttempts(t *testing.T) { t.Fatalf("\u8fd0\u884c\u4e2d\u4efb\u52a1\u5c1d\u8bd5\u6062\u590d\u7ed3\u679c\u9519\u8bef\uff1a%+v %v", attempts, err) } completedAttempts, err := repository.ListAttempts(ctx, "task-success") - if err != nil || len(completedAttempts) != 1 || completedAttempts[0].Status != "success" { + if err != nil || len(completedAttempts) != 2 || completedAttempts[0].Status != "success" || completedAttempts[1].Status != "running" { t.Fatalf("\u5df2\u5b8c\u6210\u4efb\u52a1\u5c1d\u8bd5\u4e0d\u5e94\u88ab\u6062\u590d\u903b\u8f91\u4fee\u6539\uff1a%+v %v", completedAttempts, err) } var auditCount int