补充证书部署任务重试流程

支持证书部署失败、取消和部分成功任务重试,补充任务执行阶段、权限判断、前端操作入口及回归测试。
This commit is contained in:
xuans 2026-10-01 15:42:24 +08:00
parent cfe8f35e42
commit 9edb7c11be
5 changed files with 157 additions and 34 deletions

View File

@ -137,6 +137,14 @@ func (s *Server) issueDomainAsTask(c *gin.Context) {
} }
func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Task, attempt repository.TaskAttempt, auditAction string) { func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Task, attempt repository.TaskAttempt, auditAction string) {
s.executeCertificateTask(c, task, attempt, auditAction, s.issueCertificate, "certificate_issue", "正在申请证书并部署", "证书申请和部署完成")
}
func (s *Server) executeCertificateDeployTask(c *gin.Context, task repository.Task, attempt repository.TaskAttempt, auditAction string) {
s.executeCertificateTask(c, task, attempt, auditAction, s.deployCertificate, "certificate_deploy", "正在部署证书", "证书部署完成")
}
func (s *Server) executeCertificateTask(c *gin.Context, task repository.Task, attempt repository.TaskAttempt, auditAction string, operation func(context.Context, string) (certificate.IssueResult, deployment.Report, error), stage, runningMessage, successMessage string) {
executionCtx, cancel := context.WithTimeout(c.Request.Context(), certificateTaskTimeout) executionCtx, cancel := context.WithTimeout(c.Request.Context(), certificateTaskTimeout)
executionCtx = deployment.WithTransactionContext(executionCtx, deployment.TransactionContext{ executionCtx = deployment.WithTransactionContext(executionCtx, deployment.TransactionContext{
TaskID: task.ID, AttemptID: attempt.ID, AttemptNumber: attempt.AttemptNumber, TaskID: task.ID, AttemptID: attempt.ID, AttemptNumber: attempt.AttemptNumber,
@ -148,13 +156,13 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas
}() }()
now := time.Now().UTC() now := time.Now().UTC()
if err := s.tasks.UpdateTask(executionCtx, task.ID, "running", "certificate_issue", "正在申请证书并部署", &now, nil); err != nil { if err := s.tasks.UpdateTask(executionCtx, task.ID, "running", stage, runningMessage, &now, nil); err != nil {
finished := time.Now().UTC() finished := time.Now().UTC()
failureMessage := "任务启动失败" failureMessage := "任务启动失败"
if s.writeCancelledTaskResponse(c, task.ID) { if s.writeCancelledTaskResponse(c, task.ID) {
return return
} }
_ = s.tasks.UpdateTask(context.Background(), task.ID, "failed", "certificate_issue", failureMessage, &now, &finished) _ = s.tasks.UpdateTask(context.Background(), task.ID, "failed", stage, failureMessage, &now, &finished)
_ = s.tasks.FinishAttempt(context.Background(), attempt.ID, "failed", failureMessage, &finished) _ = s.tasks.FinishAttempt(context.Background(), attempt.ID, "failed", failureMessage, &finished)
if s.audit != nil { if s.audit != nil {
_ = s.audit.CreateAuditLog(context.Background(), repository.AuditLog{ _ = s.audit.CreateAuditLog(context.Background(), repository.AuditLog{
@ -186,14 +194,14 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas
executionCtx = certificate.WithProgressReporter(executionCtx, func(progressCtx context.Context, progressStage, progressMessage string) error { executionCtx = certificate.WithProgressReporter(executionCtx, func(progressCtx context.Context, progressStage, progressMessage string) error {
return s.tasks.UpdateTask(progressCtx, task.ID, "running", progressStage, progressMessage, &now, nil) return s.tasks.UpdateTask(progressCtx, task.ID, "running", progressStage, progressMessage, &now, nil)
}) })
certificateResult, deploymentReport, issueErr := s.issueCertificate(executionCtx, task.DomainID) certificateResult, deploymentReport, issueErr := operation(executionCtx, task.DomainID)
deploymentReport = sanitizeDeploymentReport(deploymentReport) deploymentReport = sanitizeDeploymentReport(deploymentReport)
if recordErr := s.tasks.RecordTaskDeploymentResults(context.Background(), task.ID, attempt.ID, attempt.AttemptNumber, taskDeploymentResults(deploymentReport)); recordErr != nil { if recordErr := s.tasks.RecordTaskDeploymentResults(context.Background(), task.ID, attempt.ID, attempt.AttemptNumber, taskDeploymentResults(deploymentReport)); recordErr != nil {
issueErr = errors.Join(issueErr, recordErr) issueErr = errors.Join(issueErr, recordErr)
} }
finished := time.Now().UTC() finished := time.Now().UTC()
status := "success" status := "success"
message := "证书申请和部署完成" message := successMessage
cancelled := false cancelled := false
partial := deploymentReport.Status == "partial" partial := deploymentReport.Status == "partial"
if issueErr != nil { if issueErr != nil {
@ -221,11 +229,11 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas
} }
} }
if !cancelled { if !cancelled {
_ = s.tasks.UpdateTask(context.Background(), task.ID, status, "certificate_issue", message, &now, &finished) _ = s.tasks.UpdateTask(context.Background(), task.ID, status, stage, message, &now, &finished)
} }
_ = s.tasks.FinishAttempt(context.Background(), attempt.ID, status, message, &finished) _ = s.tasks.FinishAttempt(context.Background(), attempt.ID, status, message, &finished)
task.Status = status task.Status = status
task.Stage = "certificate_issue" task.Stage = stage
task.Message = message task.Message = message
task.StartedAt = &now task.StartedAt = &now
task.FinishedAt = &finished task.FinishedAt = &finished
@ -386,10 +394,6 @@ func (s *Server) cancelTask(c *gin.Context) {
} }
func (s *Server) retryTask(c *gin.Context) { func (s *Server) retryTask(c *gin.Context) {
if s.issueCertificate == nil {
s.writeError(c, http.StatusNotImplemented, "证书申请接口尚未配置", errors.New("证书申请工作流未配置"))
return
}
reader, ok := s.projectTaskReader(c) reader, ok := s.projectTaskReader(c)
if !ok { if !ok {
return return
@ -405,14 +409,32 @@ func (s *Server) retryTask(c *gin.Context) {
s.writeError(c, http.StatusInternalServerError, "读取任务失败", err) s.writeError(c, http.StatusInternalServerError, "读取任务失败", err)
return return
} }
if item.TaskType != "certificate_issue" { if item.TaskType != "certificate_issue" && item.TaskType != "certificate_deploy" {
c.JSON(http.StatusConflict, gin.H{"error": "当前任务类型不支持手动重试", "task": item}) c.JSON(http.StatusConflict, gin.H{"error": "当前任务类型不支持手动重试", "task": item})
return return
} }
if item.Status != "failed" && item.Status != "cancelled" { retryable := item.Status == "failed" || item.Status == "cancelled"
c.JSON(http.StatusConflict, gin.H{"error": "仅失败或已取消的证书申请任务可以重试", "task": item}) if item.TaskType == "certificate_deploy" && item.Status == "partial_success" {
retryable = true
}
if !retryable {
c.JSON(http.StatusConflict, gin.H{"error": "仅失败、已取消或部分成功的证书任务可以重试", "task": item})
return return
} }
var runTask func(*gin.Context, repository.Task, repository.TaskAttempt, string)
if item.TaskType == "certificate_deploy" {
if s.deployCertificate == nil {
s.writeError(c, http.StatusNotImplemented, "证书部署接口尚未配置", errors.New("证书部署工作流未配置"))
return
}
runTask = s.executeCertificateDeployTask
} else {
if s.issueCertificate == nil {
s.writeError(c, http.StatusNotImplemented, "证书申请接口尚未配置", errors.New("证书申请工作流未配置"))
return
}
runTask = s.executeCertificateIssueTask
}
authorizer, ok := s.projectMembers.(repository.ProjectAuthorizer) authorizer, ok := s.projectMembers.(repository.ProjectAuthorizer)
if !ok { if !ok {
s.writeError(c, http.StatusServiceUnavailable, "任务重试权限服务尚未就绪", errors.New("项目授权服务未配置")) s.writeError(c, http.StatusServiceUnavailable, "任务重试权限服务尚未就绪", errors.New("项目授权服务未配置"))
@ -433,7 +455,7 @@ func (s *Server) retryTask(c *gin.Context) {
return return
} }
} }
if !s.authorizeDNSCredentialUse(c, projectID) { if item.TaskType == "certificate_issue" && !s.authorizeDNSCredentialUse(c, projectID) {
return return
} }
startedAt := time.Now().UTC() startedAt := time.Now().UTC()
@ -449,9 +471,8 @@ func (s *Server) retryTask(c *gin.Context) {
} }
return return
} }
s.executeCertificateIssueTask(c, retryTask, attempt, "certificate_issue_retry") runTask(c, retryTask, attempt, item.TaskType+"_retry")
} }
func validTaskIdempotencyKey(value string) bool { func validTaskIdempotencyKey(value string) bool {
if len(value) < 8 || len(value) > 128 { if len(value) < 8 || len(value) > 128 {
return false return false

View File

@ -742,3 +742,88 @@ func TestTaskAPIRecordsPartialDeploymentFailureAsPartialSuccess(t *testing.T) {
t.Fatalf("partial failure audit status: %+v %v", logs, err) t.Fatalf("partial failure audit status: %+v %v", logs, err)
} }
} }
func TestTaskAPIRetriesFailedCertificateDeploymentWithoutDNSCredential(t *testing.T) {
calls := 0
server, taskRepo, auditRepo, closeDB := newTaskAPITestServer(t, func(context.Context, string) (certificate.IssueResult, deployment.Report, error) {
return certificate.IssueResult{}, deployment.Report{}, nil
})
defer closeDB()
server.deployCertificate = func(_ context.Context, id string) (certificate.IssueResult, deployment.Report, error) {
calls++
return certificate.IssueResult{DomainID: id}, deployment.Report{Status: "success"}, nil
}
finished := time.Now().UTC()
task := repository.Task{
HistoryScope: repository.HistoryScope{ScopeType: repository.ScopeProject, ProjectID: domain.DefaultProjectID},
ID: "deploy-retry-failed", DomainID: "domain-1", TaskType: "certificate_deploy",
Status: "failed", Stage: "certificate_deploy", Message: "deployment failed", FinishedAt: &finished,
}
if err := taskRepo.CreateTask(t.Context(), task); err != nil {
t.Fatalf("create deployment task: %v", err)
}
attempt := repository.TaskAttempt{
HistoryScope: repository.HistoryScope{ScopeType: repository.ScopeProject, ProjectID: domain.DefaultProjectID},
ID: "deploy-retry-failed-attempt-1", TaskID: task.ID, AttemptNumber: 1,
Stage: "certificate_deploy", Status: "failed", Message: "deployment failed", FinishedAt: &finished,
}
if err := taskRepo.CreateAttempt(t.Context(), attempt); err != nil {
t.Fatalf("create deployment attempt: %v", err)
}
response := httptest.NewRecorder()
serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/tasks/"+task.ID+"/retry", nil))
if response.Code != http.StatusOK {
t.Fatalf("retry response: %d %s", response.Code, response.Body.String())
}
if calls != 1 {
t.Fatalf("deployment workflow calls: %d", calls)
}
updated, err := taskRepo.GetTask(t.Context(), task.ID)
if err != nil || updated.Status != "success" || updated.Stage != "certificate_deploy" {
t.Fatalf("deployment retry did not converge: %+v %v", updated, err)
}
attempts, err := taskRepo.ListAttempts(t.Context(), task.ID)
if err != nil || len(attempts) != 2 || attempts[1].Stage != "certificate_deploy" || attempts[1].Status != "success" {
t.Fatalf("deployment retry attempts: %+v %v", attempts, err)
}
logs, err := auditRepo.ListAuditLogs(t.Context(), 10)
if err != nil || len(logs) != 2 || logs[0].Action != "certificate_deploy_retry" || logs[0].Status != "success" {
t.Fatalf("deployment retry audit: %+v %v", logs, err)
}
}
func TestTaskAPIRetriesPartialCertificateDeployment(t *testing.T) {
server, taskRepo, _, closeDB := newTaskAPITestServer(t, func(context.Context, string) (certificate.IssueResult, deployment.Report, error) {
return certificate.IssueResult{}, deployment.Report{}, nil
})
defer closeDB()
server.deployCertificate = func(_ context.Context, id string) (certificate.IssueResult, deployment.Report, error) {
return certificate.IssueResult{DomainID: id}, deployment.Report{Status: "success"}, nil
}
finished := time.Now().UTC()
task := repository.Task{
HistoryScope: repository.HistoryScope{ScopeType: repository.ScopeProject, ProjectID: domain.DefaultProjectID},
ID: "deploy-retry-partial", DomainID: "domain-1", TaskType: "certificate_deploy",
Status: "partial_success", Stage: "certificate_deploy", Message: "partial deployment", FinishedAt: &finished,
}
if err := taskRepo.CreateTask(t.Context(), task); err != nil {
t.Fatalf("create partial deployment task: %v", err)
}
attempt := repository.TaskAttempt{
HistoryScope: repository.HistoryScope{ScopeType: repository.ScopeProject, ProjectID: domain.DefaultProjectID},
ID: "deploy-retry-partial-attempt-1", TaskID: task.ID, AttemptNumber: 1,
Stage: "certificate_deploy", Status: "partial_success", Message: "partial deployment", FinishedAt: &finished,
}
if err := taskRepo.CreateAttempt(t.Context(), attempt); err != nil {
t.Fatalf("create partial deployment attempt: %v", err)
}
response := httptest.NewRecorder()
serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/tasks/"+task.ID+"/retry", nil))
if response.Code != http.StatusOK {
t.Fatalf("partial retry response: %d %s", response.Code, response.Body.String())
}
updated, err := taskRepo.GetTask(t.Context(), task.ID)
if err != nil || updated.Status != "success" {
t.Fatalf("partial deployment retry did not converge: %+v %v", updated, err)
}
}

View File

@ -242,12 +242,16 @@ func (r *SQLTaskRepository) RetryVisibleTask(ctx context.Context, actorID, taskI
predicate, predicateArgs := taskVisibilityPredicate("tasks", domain.PermissionTaskRetry) predicate, predicateArgs := taskVisibilityPredicate("tasks", domain.PermissionTaskRetry)
predicateArgs[0] = actorID predicateArgs[0] = actorID
updateArgs := []any{ updateArgs := []any{
actorID, "certificate_issue", "正在重试证书申请", nullableTime(startedAt), time.Now().UTC(), taskID, actorID, nullableTime(startedAt), time.Now().UTC(), taskID,
} }
updateArgs = append(updateArgs, predicateArgs...) updateArgs = append(updateArgs, predicateArgs...)
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks SET operator_id = ?, status = 'running', stage = ?, message = ?, result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks SET operator_id = ?, status = 'running',
stage = CASE task_type WHEN 'certificate_deploy' THEN 'certificate_deploy' ELSE 'certificate_issue' END,
message = CASE task_type WHEN 'certificate_deploy' THEN '正在重试证书部署' ELSE '正在重试证书申请' END,
started_at = ?, finished_at = NULL, updated_at = ? started_at = ?, finished_at = NULL, updated_at = ?
WHERE id = ? AND task_type = 'certificate_issue' AND status IN ('failed', 'cancelled') AND `+predicate), updateArgs...) WHERE id = ? AND task_type IN ('certificate_issue', 'certificate_deploy')
AND (status IN ('failed', 'cancelled') OR (task_type = 'certificate_deploy' AND status = 'partial_success'))
AND `+predicate), updateArgs...)
if err != nil { if err != nil {
rollback() rollback()
return Task{}, TaskAttempt{}, fmt.Errorf("更新重试任务状态失败:%w", err) return Task{}, TaskAttempt{}, fmt.Errorf("更新重试任务状态失败:%w", err)
@ -266,25 +270,37 @@ func (r *SQLTaskRepository) RetryVisibleTask(ctx context.Context, actorID, taskI
if getErr != nil { if getErr != nil {
return Task{}, TaskAttempt{}, getErr return Task{}, TaskAttempt{}, getErr
} }
if item.Status != "failed" && item.Status != "cancelled" { if !retryableTaskStatus(item) {
return Task{}, TaskAttempt{}, ErrTaskNotRetryable return Task{}, TaskAttempt{}, ErrTaskNotRetryable
} }
return Task{}, TaskAttempt{}, ErrProjectAccessDenied return Task{}, TaskAttempt{}, ErrProjectAccessDenied
} }
task, err := scanTask(tx.QueryRowContext(ctx, r.query(taskSelect+` WHERE id = ?`), taskID))
if err != nil {
rollback()
return Task{}, TaskAttempt{}, err
}
stage := "certificate_issue"
message := "开始重试证书申请"
if task.TaskType == "certificate_deploy" {
stage = "certificate_deploy"
message = "开始重试证书部署"
}
var attemptNumber int var attemptNumber int
if err := tx.QueryRowContext(ctx, r.query(`SELECT COALESCE(MAX(attempt_number), 0) + 1 FROM task_attempts WHERE task_id = ?`), taskID).Scan(&attemptNumber); err != nil { if err := tx.QueryRowContext(ctx, r.query(`SELECT COALESCE(MAX(attempt_number), 0) + 1 FROM task_attempts WHERE task_id = ?`), taskID).Scan(&attemptNumber); err != nil {
rollback() rollback()
return Task{}, TaskAttempt{}, fmt.Errorf("读取任务重试次数失败:%w", err) return Task{}, TaskAttempt{}, fmt.Errorf("读取任务重试次数失败:%w", err)
} }
attempt := TaskAttempt{ attempt := TaskAttempt{
HistoryScope: HistoryScope{ScopeType: ScopeProject}, HistoryScope: task.HistoryScope,
ID: domain.NewID(), ID: domain.NewID(),
TaskID: taskID, TaskID: taskID,
AttemptNumber: attemptNumber, AttemptNumber: attemptNumber,
Stage: "certificate_issue", Stage: stage,
Status: "running", Status: "running",
Message: "开始重试证书申请", Message: message,
StartedAt: startedAt, StartedAt: startedAt,
CreatedAt: startedAt.UTC(), CreatedAt: startedAt.UTC(),
} }
@ -305,14 +321,15 @@ func (r *SQLTaskRepository) RetryVisibleTask(ctx context.Context, actorID, taskI
return Task{}, TaskAttempt{}, ErrTaskNotRetryable return Task{}, TaskAttempt{}, ErrTaskNotRetryable
} }
task, err := scanTask(tx.QueryRowContext(ctx, r.query(taskSelect+` WHERE id = ?`), taskID))
if err != nil {
rollback()
return Task{}, TaskAttempt{}, err
}
attempt.HistoryScope = task.HistoryScope
if err := tx.Commit(); err != nil { if err := tx.Commit(); err != nil {
return Task{}, TaskAttempt{}, fmt.Errorf("提交重试任务事务失败:%w", err) return Task{}, TaskAttempt{}, fmt.Errorf("提交重试任务事务失败:%w", err)
} }
return task, attempt, nil return task, attempt, nil
} }
func retryableTaskStatus(item Task) bool {
if item.Status == "failed" || item.Status == "cancelled" {
return item.TaskType == "certificate_issue" || item.TaskType == "certificate_deploy"
}
return item.TaskType == "certificate_deploy" && item.Status == "partial_success"
}

View File

@ -1096,7 +1096,7 @@ test("workflowTasks.renderOnlySupportedActions", async () => {
assert.deepEqual(compiled.errors, []); assert.deepEqual(compiled.errors, []);
const render = new Function("Vue", compiled.code)(Vue); const render = new Function("Vue", compiled.code)(Vue);
for (const taskType of ["certificate_issue", "renewal_run", "certificate_deploy", "management_renewal", "unknown"]) { for (const taskType of ["certificate_issue", "renewal_run", "certificate_deploy", "management_renewal", "unknown"]) {
for (const status of ["queued", "running", "success", "failed", "cancelled"]) { for (const status of ["queued", "running", "success", "failed", "cancelled", "partial_success"]) {
const html = await renderToString(Vue.createSSRApp({ const html = await renderToString(Vue.createSSRApp({
render, render,
setup: () => ({ setup: () => ({
@ -1112,8 +1112,8 @@ test("workflowTasks.renderOnlySupportedActions", async () => {
retryTask: () => {}, retryTask: () => {},
}), }),
})); }));
const canCancel = taskType === "certificate_issue" && ["queued", "running"].includes(status); const canCancel = ["certificate_issue", "certificate_deploy"].includes(taskType) && ["queued", "running"].includes(status);
const canRetry = taskType === "certificate_issue" && ["failed", "cancelled"].includes(status); const canRetry = (taskType === "certificate_issue" && ["failed", "cancelled"].includes(status)) || (taskType === "certificate_deploy" && ["failed", "cancelled", "partial_success"].includes(status));
const buttonCount = (html.match(/<button\b/g) || []).length; const buttonCount = (html.match(/<button\b/g) || []).length;
assert.equal(buttonCount, canCancel || canRetry ? 2 : 1, `${taskType}:${status}`); assert.equal(buttonCount, canCancel || canRetry ? 2 : 1, `${taskType}:${status}`);
assert.equal(canCancel && canRetry, false, `${taskType}:${status} must not show cancel and retry together`); assert.equal(canCancel && canRetry, false, `${taskType}:${status} must not show cancel and retry together`);

View File

@ -92,12 +92,12 @@ const { detailView, summary, visibleTasks, taskDomain } = useRuntimeOverview(pro
<td> <td>
<button type="button" @click="openTaskDetails(item)">查看详情</button> <button type="button" @click="openTaskDetails(item)">查看详情</button>
<button <button
v-if="item.task_type === 'certificate_issue' && (item.status === 'queued' || item.status === 'running')" v-if="(item.task_type === 'certificate_issue' || item.task_type === 'certificate_deploy') && (item.status === 'queued' || item.status === 'running')"
type="button" type="button"
@click="cancelTask(item)" @click="cancelTask(item)"
>取消任务</button> >取消任务</button>
<button <button
v-if="item.task_type === 'certificate_issue' && (item.status === 'failed' || item.status === 'cancelled')" v-if="(item.task_type === 'certificate_issue' && (item.status === 'failed' || item.status === 'cancelled')) || (item.task_type === 'certificate_deploy' && (item.status === 'failed' || item.status === 'cancelled' || item.status === 'partial_success'))"
type="button" type="button"
@click="retryTask(item)" @click="retryTask(item)"
>重试任务</button> >重试任务</button>