From 13872ea3b11b366eb1862541b81345fad6661ed8 Mon Sep 17 00:00:00 2001 From: xuans Date: Fri, 2 Oct 2026 16:05:40 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=9D=E5=AD=98=E5=BC=82=E6=AD=A5=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=90=8E=E7=AB=AF=E5=BC=80=E5=8F=91=E4=B8=AD=E5=9F=BA?= =?UTF-8?q?=E7=BA=BF=EF=BC=8C=E5=B0=9A=E6=9C=AA=E5=AE=8C=E6=88=90=E5=9B=9E?= =?UTF-8?q?=E5=BD=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/app/app.go | 8 +- backend/internal/httpapi/certificate_api.go | 4 + backend/internal/httpapi/server.go | 12 +- backend/internal/httpapi/task_api.go | 169 ++++++++------- backend/internal/httpapi/task_api_test.go | 99 ++++++--- backend/internal/httpapi/task_executor.go | 92 ++++++++ .../internal/httpapi/task_executor_test.go | 200 ++++++++++++++++++ .../repository/project_task_reader.go | 22 ++ .../internal/repository/task_repository.go | 4 +- 9 files changed, 496 insertions(+), 114 deletions(-) create mode 100644 backend/internal/httpapi/task_executor.go create mode 100644 backend/internal/httpapi/task_executor_test.go diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index 1eca8cf..3e3316c 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -292,9 +292,11 @@ func (a *App) Close() error { <-a.managementDone } if a.HTTP != nil { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - if err := a.HTTP.Shutdown(ctx); err != nil && first == nil { - first = err + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + if err := a.HTTP.Shutdown(ctx); err != nil { + cancel() + // 后台任务可能仍在回滚；不得先关闭其数据库或日志。 + return err } cancel() } diff --git a/backend/internal/httpapi/certificate_api.go b/backend/internal/httpapi/certificate_api.go index 105e2a3..b461f5b 100644 --- a/backend/internal/httpapi/certificate_api.go +++ b/backend/internal/httpapi/certificate_api.go @@ -136,6 +136,10 @@ func (s *Server) deployCertificateCurrent(c *gin.Context) { if !s.authorizeVisibleDomainPermission(c, domain.PermissionDeploymentExecute) { return } + if !taskRepositoryUnavailable(s.tasks) { + s.submitDomainTask(c, "certificate_deploy") + return + } if s.deployCertificate == nil { s.writeError(c, http.StatusNotImplemented, "证书部署接口尚未配置", errors.New("证书部署流程未配置")) return diff --git a/backend/internal/httpapi/server.go b/backend/internal/httpapi/server.go index 85c80d2..abfbce9 100644 --- a/backend/internal/httpapi/server.go +++ b/backend/internal/httpapi/server.go @@ -114,6 +114,10 @@ type Server struct { aliyunChecker aliyunScopeChecker taskMu sync.Mutex taskCancels map[string]context.CancelFunc + taskWorkers sync.WaitGroup + taskWorkersStopped bool + asyncTasks map[string]bool + taskSlots chan struct{} issueTaskMu sync.Mutex setupMu sync.Mutex resetMu sync.Mutex @@ -371,18 +375,20 @@ func isLoopbackHost(host string) bool { } func (s *Server) Shutdown(ctx context.Context) error { + s.stopTaskWorkers() s.httpMu.Lock() s.httpStopped = true server := s.httpServer s.httpMu.Unlock() if server == nil { - return nil + return s.waitTaskWorkers(ctx) } err := server.Shutdown(ctx) + workerErr := s.waitTaskWorkers(ctx) if errors.Is(err, http.ErrServerClosed) { - return nil + return workerErr } - return err + return errors.Join(err, workerErr) } type domainRequest struct { diff --git a/backend/internal/httpapi/task_api.go b/backend/internal/httpapi/task_api.go index 45a2e1d..e5a3a98 100644 --- a/backend/internal/httpapi/task_api.go +++ b/backend/internal/httpapi/task_api.go @@ -19,7 +19,11 @@ import ( ) func (s *Server) issueDomainAsTask(c *gin.Context) { - if s.issueCertificate == nil { + s.submitDomainTask(c, "certificate_issue") +} + +func (s *Server) submitDomainTask(c *gin.Context, taskType string) { + if (taskType == "certificate_issue" && s.issueCertificate == nil) || (taskType == "certificate_deploy" && s.deployCertificateTask == nil) { s.writeError(c, http.StatusNotImplemented, "证书申请接口尚未配置", errors.New("证书申请工作流未配置")) return } @@ -46,13 +50,15 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { s.writeError(c, http.StatusServiceUnavailable, "证书任务权限服务尚未就绪", errors.New("项目授权服务未配置")) return } - if err := authorizer.AuthorizeProject(c.Request.Context(), c.GetString(authenticatedActorKey), resource.ResourceScope.Normalize().ProjectID, domain.PermissionCertificateManage); err != nil { - if errors.Is(err, repository.ErrProjectAccessDenied) { - s.writeError(c, http.StatusForbidden, "无权管理此项目的证书任务", nil) + if taskType == "certificate_issue" { + if err := authorizer.AuthorizeProject(c.Request.Context(), c.GetString(authenticatedActorKey), resource.ResourceScope.Normalize().ProjectID, domain.PermissionCertificateManage); err != nil { + if errors.Is(err, repository.ErrProjectAccessDenied) { + s.writeError(c, http.StatusForbidden, "无权管理此项目的证书任务", nil) + return + } + s.writeError(c, http.StatusInternalServerError, "检查证书任务权限失败", err) return } - s.writeError(c, http.StatusInternalServerError, "检查证书任务权限失败", err) - return } if err := authorizer.AuthorizeProject(c.Request.Context(), c.GetString(authenticatedActorKey), resource.ResourceScope.Normalize().ProjectID, domain.PermissionDeploymentExecute); err != nil { if errors.Is(err, repository.ErrProjectAccessDenied) { @@ -62,7 +68,7 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { s.writeError(c, http.StatusInternalServerError, "检查证书部署权限失败", err) return } - if !s.authorizeDNSCredentialUse(c, resource.ResourceScope.Normalize().ProjectID) { + if taskType == "certificate_issue" && !s.authorizeDNSCredentialUse(c, resource.ResourceScope.Normalize().ProjectID) { return } s.issueTaskMu.Lock() @@ -74,6 +80,11 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { return } if existing, err := s.tasks.FindTaskByIdempotencyKey(c.Request.Context(), domainID, idempotencyKey); err == nil { + if existing.TaskType != taskType { + s.issueTaskMu.Unlock() + s.writeError(c, http.StatusConflict, "幂等键已用于另一种操作", nil) + return + } status := http.StatusOK if isActiveTaskStatus(existing.Status) { status = http.StatusConflict @@ -87,7 +98,7 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { return } } - if active, err := s.tasks.FindActiveTaskByDomain(c.Request.Context(), domainID, "certificate_issue"); err == nil { + if active, err := s.tasks.FindActiveTaskByDomain(c.Request.Context(), domainID, taskType); err == nil { s.issueTaskMu.Unlock() c.JSON(http.StatusConflict, gin.H{"error": "该域名已有正在执行的证书任务", "task": active}) return @@ -100,12 +111,17 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { task := repository.Task{ HistoryScope: repository.HistoryScope{ScopeType: repository.ScopeProject, ProjectID: resource.ResourceScope.Normalize().ProjectID}, OperatorID: s.currentOperatorID(c), - ID: domain.NewID(), DomainID: domainID, TaskType: "certificate_issue", Status: "queued", Stage: "queued", - Message: "等待执行证书申请", IdempotencyKey: idempotencyKey, CreatedAt: now, UpdatedAt: now, + ID: domain.NewID(), DomainID: domainID, TaskType: taskType, Status: "queued", Stage: "queued", + Message: "等待执行证书任务", IdempotencyKey: idempotencyKey, CreatedAt: now, UpdatedAt: now, } if err := s.tasks.CreateTask(c.Request.Context(), task); err != nil { if idempotencyKey != "" { if existing, lookupErr := s.tasks.FindTaskByIdempotencyKey(c.Request.Context(), domainID, idempotencyKey); lookupErr == nil { + if existing.TaskType != taskType { + s.issueTaskMu.Unlock() + s.writeError(c, http.StatusConflict, "幂等键已用于另一种操作", nil) + return + } status := http.StatusOK if isActiveTaskStatus(existing.Status) { status = http.StatusConflict @@ -125,41 +141,42 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { } s.issueTaskMu.Unlock() attempt := repository.TaskAttempt{ - ID: domain.NewID(), TaskID: task.ID, AttemptNumber: 1, Stage: "certificate_issue", Status: "running", - Message: "开始申请证书", StartedAt: &now, CreatedAt: now, + ID: domain.NewID(), TaskID: task.ID, AttemptNumber: 1, Stage: taskType, Status: "running", + Message: "等待执行证书任务", StartedAt: &now, CreatedAt: now, } if err := s.tasks.CreateAttempt(c.Request.Context(), attempt); err != nil { - _ = s.tasks.UpdateTask(context.Background(), task.ID, "failed", "certificate_issue", "创建任务尝试失败", &now, &now) + _ = s.tasks.UpdateTask(context.Background(), task.ID, "failed", taskType, "创建任务尝试失败", &now, &now) s.writeError(c, http.StatusInternalServerError, "创建任务尝试失败", err) return } - s.executeCertificateIssueTask(c, task, attempt, "certificate_issue") + if taskType == "certificate_deploy" { + s.executeCertificateDeployTask(c, task, attempt, taskType) + return + } + s.executeCertificateIssueTask(c, task, attempt, taskType) } 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", "正在申请证书并部署", "证书申请和部署完成") + s.enqueueCertificateTask(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.deployCertificateTask, "certificate_deploy", "正在部署证书", "证书部署完成") + s.enqueueCertificateTask(c, task, attempt, auditAction, s.deployCertificateTask, "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 = deployment.WithTransactionContext(executionCtx, deployment.TransactionContext{ +func (s *Server) runCertificateTask(ctx context.Context, task repository.Task, attempt repository.TaskAttempt, auditAction string, operation func(context.Context, string) (certificate.IssueResult, deployment.Report, error), stage, runningMessage, successMessage string) { + executionCtx := deployment.WithTransactionContext(ctx, deployment.TransactionContext{ TaskID: task.ID, AttemptID: attempt.ID, AttemptNumber: attempt.AttemptNumber, }) - s.registerTaskCancel(task.ID, cancel) - defer func() { - cancel() - s.unregisterTaskCancel(task.ID) - }() - now := time.Now().UTC() + if err := executionCtx.Err(); err != nil { + s.finishTaskBeforeExecution(task, attempt, "cancelled", "任务在开始前已停止") + return + } if err := s.tasks.UpdateTask(executionCtx, task.ID, "running", stage, runningMessage, &now, nil); err != nil { finished := time.Now().UTC() failureMessage := "任务启动失败" - if s.writeCancelledTaskResponse(c, task.ID) { + if s.finishCancelledTask(task, attempt, executionCtx) { return } _ = s.tasks.UpdateTask(context.Background(), task.ID, "failed", stage, failureMessage, &now, &finished) @@ -171,38 +188,40 @@ func (s *Server) executeCertificateTask(c *gin.Context, task repository.Task, at Status: "error", Message: failureMessage, }) } - s.writeError(c, http.StatusInternalServerError, "更新任务状态失败", err) return } if s.audit != nil { if err := s.audit.CreateAuditLog(executionCtx, repository.AuditLog{ TaskID: task.ID, OperatorID: task.OperatorID, Action: auditAction, ResourceType: "domain", ResourceID: task.DomainID, - Status: "running", Message: "开始申请证书并部署", + Status: "running", Message: runningMessage, }); err != nil { finished := time.Now().UTC() failureMessage := "任务启动审计记录失败" - if s.writeCancelledTaskResponse(c, task.ID) { + if s.finishCancelledTask(task, attempt, executionCtx) { return } _ = s.tasks.UpdateTask(context.Background(), task.ID, "failed", "record", failureMessage, &now, &finished) _ = s.tasks.FinishAttempt(context.Background(), attempt.ID, "failed", failureMessage, &finished) - s.writeError(c, http.StatusInternalServerError, failureMessage, err) return } } + if s.finishCancelledTask(task, attempt, executionCtx) { + return + } executionCtx = certificate.WithProgressReporter(executionCtx, func(progressCtx context.Context, progressStage, progressMessage string) error { return s.tasks.UpdateTask(progressCtx, task.ID, "running", progressStage, progressMessage, &now, nil) }) - certificateResult, deploymentReport, issueErr := operation(executionCtx, task.DomainID) + _, deploymentReport, issueErr := operation(executionCtx, task.DomainID) + finishCtx, finishCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer finishCancel() 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(finishCtx, task.ID, attempt.ID, attempt.AttemptNumber, taskDeploymentResults(deploymentReport)); recordErr != nil { issueErr = errors.Join(issueErr, recordErr) } finished := time.Now().UTC() status := "success" message := successMessage - cancelled := false partial := deploymentReport.Status == "partial" if issueErr != nil { status = "failed" @@ -220,18 +239,18 @@ func (s *Server) executeCertificateTask(c *gin.Context, task repository.Task, at } else if partial { status = "partial_success" } - if persisted, err := s.tasks.GetTask(context.Background(), task.ID); err == nil && persisted.Status == "cancelled" { + s.taskMu.Lock() + defer s.taskMu.Unlock() + if persisted, err := s.tasks.GetTask(finishCtx, task.ID); err == nil && (persisted.Status == "cancelled" || persisted.Status == "cancelling") { status = "cancelled" message = persisted.Message - cancelled = true if issueErr == nil { issueErr = errors.New(message) } } - if !cancelled { - _ = s.tasks.UpdateTask(context.Background(), task.ID, status, stage, message, &now, &finished) + if err := s.tasks.FinishAttempt(finishCtx, attempt.ID, status, message, &finished); err != nil { + status, message = "failed", "任务尝试结果保存失败，请检查任务恢复状态" } - _ = s.tasks.FinishAttempt(context.Background(), attempt.ID, status, message, &finished) task.Status = status task.Stage = stage task.Message = message @@ -243,46 +262,25 @@ func (s *Server) executeCertificateTask(c *gin.Context, task repository.Task, at if issueErr != nil && status != "cancelled" && status != "partial_success" { auditStatus = "error" } - _ = s.audit.CreateAuditLog(context.Background(), repository.AuditLog{ + if err := s.audit.CreateAuditLog(finishCtx, repository.AuditLog{ TaskID: task.ID, OperatorID: task.OperatorID, Action: auditAction, ResourceType: "domain", ResourceID: task.DomainID, Status: auditStatus, Message: message, - }) - } - response := gin.H{ - "task": task, "certificate": certificateResult, "deployment": deploymentReport, - "task_status": status, "finished_at": finished, - } - if issueErr != nil { - response["error"] = message - responseStatus := http.StatusBadGateway - if status == "cancelled" { - responseStatus = http.StatusConflict - } else if status == "partial_success" { - responseStatus = http.StatusMultiStatus - } else if errors.Is(issueErr, context.DeadlineExceeded) { - responseStatus = http.StatusGatewayTimeout + }); err != nil { + status, message = "failed", "任务完成审计保存失败，请检查任务恢复状态" } - c.JSON(responseStatus, response) - return } - c.JSON(http.StatusOK, response) + if err := s.tasks.UpdateTask(finishCtx, task.ID, status, stage, message, &now, &finished); err != nil && s.logger != nil { + s.logger.Error("保存任务终态失败", "task_id", task.ID, "error", err) + } } -func (s *Server) writeCancelledTaskResponse(c *gin.Context, taskID string) bool { - persisted, err := s.tasks.GetTask(context.Background(), taskID) - if err != nil || persisted.Status != "cancelled" { - return false +func (s *Server) finishCancelledTask(task repository.Task, attempt repository.TaskAttempt, ctx context.Context) bool { + persisted, err := s.tasks.GetTask(context.Background(), task.ID) + if (err == nil && (persisted.Status == "cancelled" || persisted.Status == "cancelling")) || errors.Is(ctx.Err(), context.Canceled) { + s.finishTaskBeforeExecution(task, attempt, "cancelled", "任务已取消，未继续执行") + return true } - message := persisted.Message - if message == "" { - message = "任务已由用户取消" - } - c.JSON(http.StatusConflict, gin.H{ - "task": persisted, - "task_status": persisted.Status, - "error": message, - }) - return true + return false } const certificateTaskTimeout = 30 * time.Minute @@ -300,6 +298,7 @@ func (s *Server) unregisterTaskCancel(taskID string) { s.taskMu.Lock() defer s.taskMu.Unlock() delete(s.taskCancels, taskID) + delete(s.asyncTasks, taskID) } func (s *Server) taskWorkflowContext(c *gin.Context) context.Context { @@ -365,6 +364,27 @@ func (s *Server) cancelTask(c *gin.Context) { if !s.authorizeProjectPermissions(c, item.ProjectID, domain.PermissionTaskCancel) { return } + s.taskMu.Lock() + if s.asyncTasks[taskID] { + if err := reader.RequestCancelVisibleTask(c.Request.Context(), c.GetString(authenticatedActorKey), taskID); err != nil { + s.taskMu.Unlock() + s.writeError(c, http.StatusConflict, "任务状态已变化，无法取消", err) + return + } + s.taskCancels[taskID]() + s.taskMu.Unlock() + item.Status = "cancelling" + item.Message = "正在取消任务并等待部署收尾" + if s.audit != nil { + _ = s.audit.CreateAuditLog(c.Request.Context(), repository.AuditLog{ + TaskID: taskID, OperatorID: s.currentOperatorID(c), Action: "task_cancel", + ResourceType: "task", ResourceID: taskID, Status: "success", Message: item.Message, + }) + } + c.JSON(http.StatusAccepted, gin.H{"task": item}) + return + } + s.taskMu.Unlock() finished := time.Now().UTC() if err := reader.CancelVisibleTask(c.Request.Context(), c.GetString(authenticatedActorKey), taskID, "任务已由用户取消", &finished); err != nil { if errors.Is(err, repository.ErrProjectAccessDenied) || errors.Is(err, sql.ErrNoRows) { @@ -409,6 +429,13 @@ func (s *Server) retryTask(c *gin.Context) { s.writeError(c, http.StatusInternalServerError, "读取任务失败", err) return } + s.taskMu.Lock() + workerActive := s.asyncTasks[taskID] + s.taskMu.Unlock() + if workerActive { + s.writeError(c, http.StatusConflict, "任务仍在执行或收尾，暂时不能重试", nil) + return + } if item.TaskType != "certificate_issue" && item.TaskType != "certificate_deploy" { c.JSON(http.StatusConflict, gin.H{"error": "当前任务类型不支持手动重试", "task": item}) return diff --git a/backend/internal/httpapi/task_api_test.go b/backend/internal/httpapi/task_api_test.go index 83e4c0f..64f2f13 100644 --- a/backend/internal/httpapi/task_api_test.go +++ b/backend/internal/httpapi/task_api_test.go @@ -61,7 +61,24 @@ func newTaskAPITestServer(t *testing.T, issue func(context.Context, string) (cer ProjectResources: repository.NewSQLProjectMemberRepository(db.DB, cfg.Database.Driver), Users: repository.NewSQLUserRepository(db.DB, cfg.Database.Driver), }) - return server, taskRepo, auditRepo, func() { _ = db.Close() } + return server, taskRepo, auditRepo, func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := server.Shutdown(ctx); err != nil { + t.Errorf("测试任务未能停止：%v", err) + return + } + _ = db.Close() + } +} + +func waitTaskWorkersForTest(t *testing.T, server *Server) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := server.waitTaskWorkers(ctx); err != nil { + t.Fatalf("等待后台任务结束失败：%v", err) + } } type blockingTaskAuditRepository struct { @@ -94,9 +111,10 @@ func TestTaskAPIRecordsSuccessfulIssueAndAudit(t *testing.T) { response := httptest.NewRecorder() serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) - if response.Code != http.StatusOK { - t.Fatalf("成功任务响应错误：%d %s", response.Code, response.Body.String()) + if response.Code != http.StatusAccepted { + t.Fatalf("提交任务响应错误：%d %s", response.Code, response.Body.String()) } + waitTaskWorkersForTest(t, server) items, err := taskRepo.ListTasks(context.Background(), "", 10) if err != nil || len(items) != 1 || items[0].Status != "success" { t.Fatalf("任务没有进入成功状态：%+v %v", items, err) @@ -215,9 +233,10 @@ func TestTaskAPIClosesTaskWhenStartAuditFails(t *testing.T) { response := httptest.NewRecorder() serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) - if response.Code != http.StatusInternalServerError || !strings.Contains(response.Body.String(), "任务启动审计记录失败") { + if response.Code != http.StatusAccepted { t.Fatalf("任务启动审计失败响应错误：%d %s", response.Code, response.Body.String()) } + waitTaskWorkersForTest(t, server) items, err := taskRepo.ListTasks(context.Background(), "", 10) if err != nil || len(items) != 1 || items[0].Status != "failed" || items[0].Stage != "record" { t.Fatalf("任务启动审计失败后任务未收敛：%+v %v", items, err) @@ -241,9 +260,10 @@ func TestTaskAPIClosesTaskWhenStartStateUpdateFails(t *testing.T) { response := httptest.NewRecorder() serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) - if response.Code != http.StatusInternalServerError || !strings.Contains(response.Body.String(), "更新任务状态失败") { + if response.Code != http.StatusAccepted { t.Fatalf("任务启动状态写入失败响应错误：%d %s", response.Code, response.Body.String()) } + waitTaskWorkersForTest(t, server) items, err := taskRepo.ListTasks(context.Background(), "", 10) if err != nil || len(items) != 1 || items[0].Status != "failed" || items[0].Message != "任务启动失败" { t.Fatalf("任务启动失败后未正确收敛任务状态：%+v %v", items, err) @@ -269,30 +289,30 @@ func TestTaskAPIReturnsFailureAndKeepsTaskTrace(t *testing.T) { response := httptest.NewRecorder() serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) - if response.Code != http.StatusBadGateway || !strings.Contains(response.Body.String(), "证书申请或部署失败") { + if response.Code != http.StatusAccepted { t.Fatalf("失败任务响应错误：%d %s", response.Code, response.Body.String()) } + waitTaskWorkersForTest(t, server) + location := response.Header().Get("Location") + if location == "" { + t.Fatal("任务提交未返回查询地址") + } + response = httptest.NewRecorder() + serveAdminHTTP(server, response, httptest.NewRequest(http.MethodGet, location, nil)) + if response.Code != http.StatusOK { + t.Fatalf("查询失败任务错误：%d %s", response.Code, response.Body.String()) + } if strings.Contains(response.Body.String(), "/etc/scert/secret/config.json") || strings.Contains(response.Body.String(), "/etc/nginx/private/key.pem") || strings.Contains(response.Body.String(), "permission denied") { t.Fatalf("失败任务响应泄露底层错误：%s", response.Body.String()) } - var payload struct { - Task struct { - Status string `json:"status"` - Stage string `json:"stage"` - Message string `json:"message"` - FinishedAt string `json:"finished_at"` - UpdatedAt string `json:"updated_at"` - } `json:"task"` - TaskStatus string `json:"task_status"` - FinishedAt string `json:"finished_at"` - } + var payload repository.Task if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil { t.Fatalf("失败任务响应不是有效 JSON：%v；响应：%s", err, response.Body.String()) } - if payload.Task.Status != "failed" || payload.Task.Stage != "certificate_issue" || payload.Task.Message != "证书申请或部署失败" { - t.Fatalf("失败任务嵌套状态不一致：%+v", payload.Task) + if payload.Status != "failed" || payload.Stage != "certificate_issue" || payload.Message != "证书申请或部署失败" { + t.Fatalf("失败任务状态不一致：%+v", payload) } - if payload.TaskStatus != "failed" || payload.FinishedAt == "" || payload.Task.FinishedAt == "" || payload.Task.UpdatedAt == "" { + if payload.FinishedAt == nil || payload.UpdatedAt.IsZero() { t.Fatalf("失败任务顶层状态或时间字段不完整：%+v", payload) } items, err := taskRepo.ListTasks(context.Background(), "", 10) @@ -359,7 +379,7 @@ func TestTaskAPICanCancelRunningIssue(t *testing.T) { } cancelResponse := httptest.NewRecorder() serveAdminHTTP(server, cancelResponse, httptest.NewRequest(http.MethodDelete, "/api/v1/tasks/"+items[0].ID, nil)) - if cancelResponse.Code != http.StatusOK || !strings.Contains(cancelResponse.Body.String(), "cancelled") { + if cancelResponse.Code != http.StatusAccepted || !strings.Contains(cancelResponse.Body.String(), "cancelling") { t.Fatalf("取消任务响应错误：%d %s", cancelResponse.Code, cancelResponse.Body.String()) } select { @@ -367,9 +387,10 @@ func TestTaskAPICanCancelRunningIssue(t *testing.T) { case <-time.After(2 * time.Second): t.Fatal("取消后证书任务没有结束") } - if issueResponse.Code != http.StatusConflict { - t.Fatalf("取消任务的申请请求应返回 409：%d %s", issueResponse.Code, issueResponse.Body.String()) + if issueResponse.Code != http.StatusAccepted { + t.Fatalf("申请请求应先返回 202：%d %s", issueResponse.Code, issueResponse.Body.String()) } + waitTaskWorkersForTest(t, server) item, err := taskRepo.GetTask(context.Background(), items[0].ID) if err != nil || item.Status != "cancelled" { t.Fatalf("任务取消状态没有持久化：%+v %v", item, err) @@ -417,7 +438,7 @@ func TestTaskAPICancellationDuringStartupDoesNotExecuteCertificate(t *testing.T) } cancelResponse := httptest.NewRecorder() serveAdminHTTP(server, cancelResponse, httptest.NewRequest(http.MethodDelete, "/api/v1/tasks/"+items[0].ID, nil)) - if cancelResponse.Code != http.StatusOK { + if cancelResponse.Code != http.StatusAccepted { t.Fatalf("启动阶段取消任务失败：%d %s", cancelResponse.Code, cancelResponse.Body.String()) } close(blockingAudit.release) @@ -426,12 +447,13 @@ func TestTaskAPICancellationDuringStartupDoesNotExecuteCertificate(t *testing.T) case <-time.After(2 * time.Second): t.Fatal("取消后的启动任务没有结束") } + waitTaskWorkersForTest(t, server) select { case <-issueCalled: t.Fatal("任务取消后不应再调用证书申请") default: } - if issueResponse.Code != http.StatusConflict { + if issueResponse.Code != http.StatusAccepted { t.Fatalf("启动阶段取消响应错误：%d %s", issueResponse.Code, issueResponse.Body.String()) } item, err := taskRepo.GetTask(context.Background(), items[0].ID) @@ -456,9 +478,10 @@ func TestTaskAPIReplaysIdempotentIssue(t *testing.T) { firstRequest.Header.Set("Idempotency-Key", "issue-key-20260830") firstResponse := httptest.NewRecorder() serveAdminHTTP(server, firstResponse, firstRequest) - if firstResponse.Code != http.StatusOK { + if firstResponse.Code != http.StatusAccepted { t.Fatalf("首次幂等请求失败：%d %s", firstResponse.Code, firstResponse.Body.String()) } + waitTaskWorkersForTest(t, server) secondRequest := httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil) secondRequest.Header.Set("Idempotency-Key", "issue-key-20260830") secondResponse := httptest.NewRecorder() @@ -658,9 +681,10 @@ func TestTaskAPIRetriesFailedCertificateIssue(t *testing.T) { first := httptest.NewRecorder() serveAdminHTTP(server, first, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) - if first.Code != http.StatusBadGateway { + if first.Code != http.StatusAccepted { t.Fatalf("initial failure response: %d %s", first.Code, first.Body.String()) } + waitTaskWorkersForTest(t, server) tasks, err := taskRepo.ListTasks(context.Background(), "domain-1", 10) if err != nil || len(tasks) != 1 || tasks[0].Status != "failed" { t.Fatalf("failed task was not persisted: %+v %v", tasks, err) @@ -668,9 +692,10 @@ func TestTaskAPIRetriesFailedCertificateIssue(t *testing.T) { retry := httptest.NewRecorder() serveAdminHTTP(server, retry, httptest.NewRequest(http.MethodPost, "/api/v1/tasks/"+tasks[0].ID+"/retry", nil)) - if retry.Code != http.StatusOK { + if retry.Code != http.StatusAccepted { t.Fatalf("retry response: %d %s", retry.Code, retry.Body.String()) } + waitTaskWorkersForTest(t, server) updated, err := taskRepo.GetTask(context.Background(), tasks[0].ID) if err != nil || updated.Status != "success" || updated.OperatorID != "test-admin" { t.Fatalf("retried task did not converge: %+v %v", updated, err) @@ -699,9 +724,10 @@ func TestTaskAPIRecordsPartialDeploymentSuccess(t *testing.T) { 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()) + if response.Code != http.StatusAccepted { + t.Fatalf("部分成功任务提交响应错误：%d %s", response.Code, response.Body.String()) } + waitTaskWorkersForTest(t, server) 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) @@ -726,9 +752,10 @@ func TestTaskAPIRecordsPartialDeploymentFailureAsPartialSuccess(t *testing.T) { response := httptest.NewRecorder() serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) - if response.Code != http.StatusMultiStatus { + if response.Code != http.StatusAccepted { t.Fatalf("partial failure response: %d %s", response.Code, response.Body.String()) } + waitTaskWorkersForTest(t, server) items, err := taskRepo.ListTasks(context.Background(), "domain-1", 10) if err != nil || len(items) != 1 || items[0].Status != "partial_success" { t.Fatalf("partial failure task status: %+v %v", items, err) @@ -772,9 +799,10 @@ func TestTaskAPIRetriesFailedCertificateDeploymentWithoutDNSCredential(t *testin } 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 response.Code != http.StatusAccepted { + t.Fatalf("重试提交响应错误：%d %s", response.Code, response.Body.String()) } + waitTaskWorkersForTest(t, server) if calls != 1 { t.Fatalf("deployment workflow calls: %d", calls) } @@ -849,9 +877,10 @@ func TestTaskAPIRetriesPartialCertificateDeployment(t *testing.T) { } 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()) + if response.Code != http.StatusAccepted { + t.Fatalf("部分成功重试提交响应错误：%d %s", response.Code, response.Body.String()) } + waitTaskWorkersForTest(t, server) 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) diff --git a/backend/internal/httpapi/task_executor.go b/backend/internal/httpapi/task_executor.go new file mode 100644 index 0000000..9ebdf8c --- /dev/null +++ b/backend/internal/httpapi/task_executor.go @@ -0,0 +1,92 @@ +package httpapi + +import ( + "context" + "net/http" + "time" + + "github.com/gin-gonic/gin" + + "git.pcux.cn/xuans/Scert/backend/internal/certificate" + "git.pcux.cn/xuans/Scert/backend/internal/deployment" + "git.pcux.cn/xuans/Scert/backend/internal/repository" +) + +// 持久化任务与尝试后才调用此入口；后台执行不得持有请求或响应对象。 +// 进程中断由启动恢复标记失败，不自动重放可能已经签发或部署的外部操作。 +func (s *Server) enqueueCertificateTask(c *gin.Context, task repository.Task, attempt repository.TaskAttempt, action string, operation func(context.Context, string) (certificate.IssueResult, deployment.Report, error), stage, runningMessage, successMessage string) { + s.taskMu.Lock() + if s.taskWorkersStopped || len(s.asyncTasks) >= 32 { + s.taskMu.Unlock() + s.finishTaskBeforeExecution(task, attempt, "failed", "服务正在停止或任务队列已满，任务未执行") + s.writeError(c, http.StatusServiceUnavailable, "服务正在停止或任务队列已满，请稍后重试", nil) + return + } + ctx, cancel := context.WithTimeout(context.Background(), certificateTaskTimeout) + if s.taskCancels == nil { + s.taskCancels = make(map[string]context.CancelFunc) + } + if s.asyncTasks == nil { + s.asyncTasks = make(map[string]bool) + s.taskSlots = make(chan struct{}, 4) + } + s.taskCancels[task.ID] = cancel + s.asyncTasks[task.ID] = true + s.taskWorkers.Add(1) + s.taskMu.Unlock() + go func() { + defer s.taskWorkers.Done() + defer s.unregisterTaskCancel(task.ID) + defer cancel() + defer func() { + if recover() != nil { + s.finishTaskBeforeExecution(task, attempt, "failed", "任务执行异常中断，请检查部署恢复状态后重试") + } + }() + select { + case s.taskSlots <- struct{}{}: + defer func() { <-s.taskSlots }() + case <-ctx.Done(): + s.finishTaskBeforeExecution(task, attempt, "cancelled", "任务在开始前已停止") + return + } + s.runCertificateTask(ctx, task, attempt, action, operation, stage, runningMessage, successMessage) + }() + c.Header("Location", "/api/v1/tasks/"+task.ID) + c.JSON(http.StatusAccepted, gin.H{"task": task, "accepted": true}) +} + +func (s *Server) finishTaskBeforeExecution(task repository.Task, attempt repository.TaskAttempt, status, message string) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + finished := time.Now().UTC() + if err := s.tasks.FinishAttempt(ctx, attempt.ID, status, message, &finished); err != nil && s.logger != nil { + s.logger.Error("保存任务尝试失败", "task_id", task.ID, "error", err) + } + if err := s.tasks.UpdateTask(ctx, task.ID, status, attempt.Stage, message, nil, &finished); err != nil && s.logger != nil { + s.logger.Error("保存任务终态失败", "task_id", task.ID, "error", err) + } +} + +func (s *Server) stopTaskWorkers() { + s.taskMu.Lock() + defer s.taskMu.Unlock() + s.taskWorkersStopped = true + for _, cancel := range s.taskCancels { + cancel() + } +} + +func (s *Server) waitTaskWorkers(ctx context.Context) error { + done := make(chan struct{}) + go func() { + s.taskWorkers.Wait() + close(done) + }() + select { + case <-done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} diff --git a/backend/internal/httpapi/task_executor_test.go b/backend/internal/httpapi/task_executor_test.go new file mode 100644 index 0000000..0a00028 --- /dev/null +++ b/backend/internal/httpapi/task_executor_test.go @@ -0,0 +1,200 @@ +package httpapi + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/certificate" + "git.pcux.cn/xuans/Scert/backend/internal/deployment" + "git.pcux.cn/xuans/Scert/backend/internal/repository" +) + +func submittedTask(t *testing.T, response *httptest.ResponseRecorder) repository.Task { + t.Helper() + if response.Code != http.StatusAccepted { + t.Fatalf("应快速返回已接收：%d %s", response.Code, response.Body.String()) + } + var payload struct { + Task repository.Task `json:"task"` + } + if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil || payload.Task.ID == "" { + t.Fatalf("任务编号缺失：%v %s", err, response.Body.String()) + } + return payload.Task +} + +func TestAsyncTaskSurvivesRequestDisconnect(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + server, tasks, _, cleanup := newTaskAPITestServer(t, func(ctx context.Context, id string) (certificate.IssueResult, deployment.Report, error) { + close(started) + select { + case <-release: + return certificate.IssueResult{DomainID: id}, deployment.Report{Status: "success"}, nil + case <-ctx.Done(): + return certificate.IssueResult{}, deployment.Report{}, ctx.Err() + } + }) + defer cleanup() + ctx, disconnect := context.WithCancel(t.Context()) + defer disconnect() + response := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil).WithContext(ctx) + request.Header.Set("Idempotency-Key", "disconnect-request") + done := make(chan struct{}) + go func() { serveAdminHTTP(server, response, request); close(done) }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("请求仍等待工作流，未异步返回") + } + task := submittedTask(t, response) + if response.Header().Get("Location") != "/api/v1/tasks/"+task.ID { + t.Fatal("任务查询地址不正确") + } + disconnect() + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("任务未开始") + } + stored, err := tasks.GetTask(t.Context(), task.ID) + if err != nil || stored.Status != "running" { + t.Fatalf("断线影响任务：%+v %v", stored, err) + } + duplicate := httptest.NewRecorder() + replay := httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil) + replay.Header.Set("Idempotency-Key", "disconnect-request") + serveAdminHTTP(server, duplicate, replay) + if duplicate.Code != http.StatusConflict { + t.Fatal("活动任务没有拒绝重复提交") + } + close(release) + waitTaskWorkersForTest(t, server) + stored, err = tasks.GetTask(t.Context(), task.ID) + if err != nil || stored.Status != "success" { + t.Fatalf("后台任务未完成：%+v %v", stored, err) + } +} + +func TestAsyncCancellationBlocksRetryUntilCleanup(t *testing.T) { + started, cancelled, release := make(chan struct{}), make(chan struct{}), make(chan struct{}) + server, tasks, _, cleanup := newTaskAPITestServer(t, func(ctx context.Context, _ string) (certificate.IssueResult, deployment.Report, error) { + close(started) + <-ctx.Done() + close(cancelled) + <-release + return certificate.IssueResult{}, deployment.Report{}, ctx.Err() + }) + defer cleanup() + defer close(release) + response := httptest.NewRecorder() + serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) + task := submittedTask(t, response) + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("未开始执行") + } + response = httptest.NewRecorder() + serveAdminHTTP(server, response, httptest.NewRequest(http.MethodDelete, "/api/v1/tasks/"+task.ID, nil)) + if response.Code != http.StatusAccepted { + t.Fatal(response.Body.String()) + } + select { + case <-cancelled: + case <-time.After(2 * time.Second): + t.Fatal("取消信号未送达") + } + stored, err := tasks.GetTask(t.Context(), task.ID) + if err != nil || stored.Status != "cancelling" { + t.Fatalf("过早发布取消终态：%+v %v", stored, err) + } + if err := tasks.UpdateTask(t.Context(), task.ID, "running", "deployment", "迟到进度", nil, nil); err == nil { + t.Fatal("迟到进度覆盖取消状态") + } + response = httptest.NewRecorder() + serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/tasks/"+task.ID+"/retry", nil)) + if response.Code != http.StatusConflict { + t.Fatal("回滚期间允许了重试") + } + response = httptest.NewRecorder() + serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) + if response.Code != http.StatusConflict { + t.Fatal("回滚期间允许了新任务") + } +} + +func TestAsyncShutdownWaitsForCleanupAndRejectsNewWork(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + server, tasks, _, cleanup := newTaskAPITestServer(t, func(ctx context.Context, _ string) (certificate.IssueResult, deployment.Report, error) { + close(started) + <-ctx.Done() + <-release + return certificate.IssueResult{}, deployment.Report{}, ctx.Err() + }) + defer cleanup() + response := httptest.NewRecorder() + serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) + task := submittedTask(t, response) + select { + case <-started: + case <-time.After(2 * time.Second): + close(release) + t.Fatal("任务未开始") + } + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond) + defer cancel() + if err := server.Shutdown(ctx); !errors.Is(err, context.DeadlineExceeded) { + close(release) + t.Fatalf("停服没有等待收尾：%v", err) + } + stored, err := tasks.GetTask(t.Context(), task.ID) + if err != nil || !isActiveTaskStatus(stored.Status) { + close(release) + t.Fatalf("收尾前终态错误：%+v %v", stored, err) + } + close(release) + waitTaskWorkersForTest(t, server) + response = httptest.NewRecorder() + serveAdminHTTP(server, response, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) + if response.Code != http.StatusServiceUnavailable { + t.Fatalf("停止后仍接收任务：%d", response.Code) + } +} + +func TestAsyncDeploymentUsesExistingTaskAndRejectsCrossOperationKey(t *testing.T) { + var calls atomic.Int32 + server, tasks, _, cleanup := newTaskAPITestServer(t, nil) + defer cleanup() + server.deployCertificateTask = func(ctx context.Context, id string) (certificate.IssueResult, deployment.Report, error) { + calls.Add(1) + return certificate.IssueResult{DomainID: id}, deployment.Report{Status: "success"}, nil + } + response := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/deploy", nil) + request.Header.Set("Idempotency-Key", "shared-operation-key") + serveAdminHTTP(server, response, request) + task := submittedTask(t, response) + waitTaskWorkersForTest(t, server) + items, err := tasks.ListTasks(t.Context(), "domain-1", 10) + if err != nil || len(items) != 1 || items[0].ID != task.ID || items[0].Status != "success" || calls.Load() != 1 { + t.Fatalf("部署任务重复创建或未完成：%+v %v", items, err) + } + server.issueCertificate = func(context.Context, string) (certificate.IssueResult, deployment.Report, error) { + calls.Add(1) + return certificate.IssueResult{}, deployment.Report{}, nil + } + request = httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil) + request.Header.Set("Idempotency-Key", "shared-operation-key") + response = httptest.NewRecorder() + serveAdminHTTP(server, response, request) + if response.Code != http.StatusConflict || calls.Load() != 1 { + t.Fatal("跨操作复用了幂等键") + } +} diff --git a/backend/internal/repository/project_task_reader.go b/backend/internal/repository/project_task_reader.go index fd92630..e2af962 100644 --- a/backend/internal/repository/project_task_reader.go +++ b/backend/internal/repository/project_task_reader.go @@ -18,6 +18,7 @@ type ProjectTaskReader interface { ListVisibleTaskAttempts(context.Context, string, string) ([]TaskAttempt, error) ListVisibleTaskDeploymentResults(context.Context, string, string) ([]TaskDeploymentResult, error) CancelVisibleTask(context.Context, string, string, string, *time.Time) error + RequestCancelVisibleTask(context.Context, string, string) error RetryVisibleTask(context.Context, string, string, *time.Time) (Task, TaskAttempt, error) } @@ -164,6 +165,27 @@ func (r *SQLTaskRepository) ListVisibleTaskDeploymentResults(ctx context.Context return items, nil } +// 请求取消仍保留活动状态，直到执行器完成回滚收尾，禁止提前重试。 +func (r *SQLTaskRepository) RequestCancelVisibleTask(ctx context.Context, actorID, taskID string) error { + predicate, predicateArgs := taskVisibilityPredicate("tasks", domain.PermissionTaskCancel) + predicateArgs[0] = strings.TrimSpace(actorID) + args := []any{"正在取消任务并等待部署收尾", time.Now().UTC(), strings.TrimSpace(taskID)} + args = append(args, predicateArgs...) + result, err := r.db.ExecContext(ctx, r.query(`UPDATE tasks SET status = 'cancelling', message = ?, updated_at = ? + WHERE id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling') AND `+predicate), args...) + if err != nil { + return fmt.Errorf("请求取消任务失败：%w", err) + } + count, err := result.RowsAffected() + if err != nil { + return err + } + if count != 1 { + return ErrProjectAccessDenied + } + return nil +} + func (r *SQLTaskRepository) CancelVisibleTask(ctx context.Context, actorID, taskID, message string, finishedAt *time.Time) error { if r == nil || r.db == nil { return errors.New("数据库连接不能为空") diff --git a/backend/internal/repository/task_repository.go b/backend/internal/repository/task_repository.go index de5d30a..e94b4e0 100644 --- a/backend/internal/repository/task_repository.go +++ b/backend/internal/repository/task_repository.go @@ -264,8 +264,8 @@ 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', 'retry_wait', 'cancelling')`), - status, stage, message, nullableTime(startedAt), nullableTime(finishedAt), time.Now().UTC(), id) + 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') AND NOT (status = 'cancelling' AND ? IN ('queued', 'running', 'retry_wait'))`), + status, stage, message, nullableTime(startedAt), nullableTime(finishedAt), time.Now().UTC(), id, status) if err != nil { return fmt.Errorf("更新任务失败：%w", err) }