xuans/Scert
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

补齐任务部分成功与重试状态

Browse Source 
支持批量部署部分成功任务状态，补齐重试等待、取消和服务重启恢复的任务状态范围，并增加 HTTP 回归测试。
This commit is contained in:
xuans 2026-09-29 23:58:24 +08:00
parent 8c80a7261f
commit c9a82e6953
6 changed files with 44 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/internal/app/certificate_tasks.go
View File

@ -104,7 +104,7 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
if resultErr == nil || number == attempts || result.Certificate.Saved.CertificatePath != "" || errors.Is(resultErr, context.Canceled) || errors.Is(resultErr, context.DeadlineExceeded) {
return result, resultErr
}
if err := w.tasks.UpdateTask(ctx, task.ID, "running", "retry_wait", "证书申请失败，等待重试", &started, nil); err != nil {
if err := w.tasks.UpdateTask(ctx, task.ID, "retry_wait", "retry_wait", "证书申请失败，等待重试", &started, nil); err != nil {
return result, errors.Join(resultErr, err)
}
if err := waitForManagementRetry(ctx, interval); err != nil {

2
backend/internal/app/management_renewal.go
View File

@ -142,7 +142,7 @@ func (m *managementRenewal) renew(ctx context.Context, state httpapi.ManagementC
return saved, resultErr
}
stage = "retry_wait"
if err := m.tasks.UpdateTask(ctx, task.ID, "running", stage, "管理证书申请失败，等待重试", &started, nil); err != nil {
if err := m.tasks.UpdateTask(ctx, task.ID, "retry_wait", stage, "管理证书申请失败，等待重试", &started, nil); err != nil {
return saved, errors.Join(resultErr, err)
}
wait := m.wait

3
backend/internal/httpapi/task_api.go
View File

@ -148,6 +148,7 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas
status := "success"
message := "证书申请和部署完成"
cancelled := false
partial := issueErr == nil && deploymentReport.Status == "partial"
if issueErr != nil {
status = "failed"
message = certificateTaskErrorMessage(issueErr)
@ -157,6 +158,8 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas
if errors.Is(issueErr, context.Canceled) {
status = "cancelled"
}
} else if partial {
status = "partial_success"
}
if persisted, err := s.tasks.GetTask(context.Background(), task.ID); err == nil && persisted.Status == "cancelled" {
status = "cancelled"

31
backend/internal/httpapi/task_api_test.go
View File

@ -532,3 +532,34 @@ func TestTaskAPIRetriesFailedCertificateIssue(t *testing.T) {
t.Fatalf("retry audit was not recorded correctly: %+v %v", logs, err)
}
}
func TestTaskAPIRecordsPartialDeploymentSuccess(t *testing.T) {
server, taskRepo, auditRepo, closeDB := newTaskAPITestServer(t, func(_ context.Context, id string) (certificate.IssueResult, deployment.Report, error) {
return certificate.IssueResult{DomainID: id}, deployment.Report{
Total: 2, Success: 1, Failed: 1, Status: "partial",
Results: []deployment.Result{
{ServerID: "server-success", Status: "success"},
{ServerID: "server-failed", Status: "failed", Message: "deployment failed"},
},
}, nil
})
defer closeDB()
response := httptest.NewRecorder()
serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil))
if response.Code != http.StatusOK {
t.Fatalf("partial response: %d %s", response.Code, response.Body.String())
}
items, err := taskRepo.ListTasks(context.Background(), "domain-1", 10)
if err != nil || len(items) != 1 || items[0].Status != "partial_success" {
t.Fatalf("partial task status: %+v %v", items, err)
}
attempts, err := taskRepo.ListAttempts(context.Background(), items[0].ID)
if err != nil || len(attempts) != 1 || attempts[0].Status != "partial_success" {
t.Fatalf("partial attempt status: %+v %v", attempts, err)
}
logs, err := auditRepo.ListAuditLogs(context.Background(), 10)
if err != nil || len(logs) != 1 || logs[0].Status != "partial_success" {
t.Fatalf("partial audit status: %+v %v", logs, err)
}
}

4
backend/internal/repository/project_task_reader.go
View File

@ -145,7 +145,7 @@ func (r *SQLTaskRepository) CancelVisibleTask(ctx context.Context, actorID, task
args := []any{message, nullableTime(finishedAt), time.Now().UTC(), strings.TrimSpace(taskID)}
args = append(args, predicateArgs...)
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ?
WHERE id = ? AND status IN ('queued', 'running') AND `+predicate), args...)
WHERE id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling') AND `+predicate), args...)
if err != nil {
rollback()
return fmt.Errorf("取消任务失败：%w", err)
@ -175,7 +175,7 @@ func (r *SQLTaskRepository) CancelVisibleTask(ctx context.Context, actorID, task
return ErrProjectAccessDenied
}
if _, err := tx.ExecContext(ctx, r.query(`UPDATE task_attempts SET status = 'cancelled', message = ?, finished_at = ?
WHERE task_id = ? AND status IN ('queued', 'running')`), message, nullableTime(finishedAt), strings.TrimSpace(taskID)); err != nil {
WHERE task_id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`), message, nullableTime(finishedAt), strings.TrimSpace(taskID)); err != nil {
rollback()
return fmt.Errorf("取消任务尝试失败：%w", err)
}

12
backend/internal/repository/task_repository.go
View File

@ -208,7 +208,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, r.query(`UPDATE tasks SET status = ?, stage = ?, message = ?, started_at = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`),
result, err := r.db.ExecContext(ctx, r.query(`UPDATE tasks SET status = ?, stage = ?, message = ?, started_at = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`),
status, stage, message, nullableTime(startedAt), nullableTime(finishedAt), time.Now().UTC(), id)
if err != nil {
return fmt.Errorf("更新任务失败：%w", err)
@ -238,7 +238,7 @@ func (r *SQLTaskRepository) CancelTask(ctx context.Context, id, message string,
rollback := func() {
_ = tx.Rollback()
}
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`), message, nullableTime(finishedAt), time.Now().UTC(), id)
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`), message, nullableTime(finishedAt), time.Now().UTC(), id)
if err != nil {
rollback()
return fmt.Errorf("取消任务失败：%w", err)
@ -292,7 +292,7 @@ func (r *SQLTaskRepository) RecoverInterruptedTasks(ctx context.Context, finishe
}
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks
SET status = 'failed', stage = 'service_restart', message = ?, finished_at = ?, updated_at = ?
WHERE status IN ('queued', 'running')`), "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt, finishedAt)
WHERE status IN ('queued', 'running', 'retry_wait', 'cancelling')`), "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt, finishedAt)
if err != nil {
rollback()
return 0, fmt.Errorf("\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u5931\u8d25\uff1a%w", err)
@ -304,7 +304,7 @@ 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')`), "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt); err != nil {
WHERE status IN ('queued', 'running', 'retry_wait', 'cancelling')`), "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt); err != nil {
rollback()
return 0, fmt.Errorf("\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u5c1d\u8bd5\u5931\u8d25\uff1a%w", err)
}
@ -323,7 +323,7 @@ func (r *SQLTaskRepository) FindActiveTaskByDomain(ctx context.Context, domainID
if domainID == "" || taskType == "" {
return Task{}, errors.New("域名 ID 和任务类型不能为空")
}
return scanTask(r.db.QueryRowContext(ctx, r.query(taskSelect+` WHERE domain_id = ? AND task_type = ? AND status IN ('queued', 'running') ORDER BY created_at DESC LIMIT 1`), domainID, taskType))
return scanTask(r.db.QueryRowContext(ctx, r.query(taskSelect+` WHERE domain_id = ? AND task_type = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling') ORDER BY created_at DESC LIMIT 1`), domainID, taskType))
}
func (r *SQLTaskRepository) FindTaskByIdempotencyKey(ctx context.Context, domainID, key string) (Task, error) {
@ -494,7 +494,7 @@ func validateTask(item Task) error {
func validTaskStatus(status string) bool {
switch status {
case "queued", "running", "success", "failed", "cancelled":
case "queued", "running", "success", "failed", "cancelling", "cancelled", "retry_wait", "partial_success":
return true
default:
return false