完善P0-10任务并发与恢复保护

补齐幂等活动状态、管理证书并发保护、服务重启尝试范围保护及对应回归测试。
This commit is contained in:
xuans 2026-09-30 04:20:34 +08:00
parent 569499e54d
commit 402d282022
5 changed files with 89 additions and 19 deletions

View File

@ -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) {

View File

@ -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{

View File

@ -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

View File

@ -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)
}

View File

@ -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