From 415d504b2b171bdf4d2fe73a9dc8c308f23237be Mon Sep 17 00:00:00 2001 From: xuans Date: Wed, 30 Sep 2026 11:32:43 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=AF=81=E4=B9=A6=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=8F=96=E6=B6=88=E5=90=AF=E5=8A=A8=E7=AB=9E=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/httpapi/task_api.go | 40 ++++++++--- backend/internal/httpapi/task_api_test.go | 81 +++++++++++++++++++++++ 2 files changed, 113 insertions(+), 8 deletions(-) diff --git a/backend/internal/httpapi/task_api.go b/backend/internal/httpapi/task_api.go index 3e633ee..68b0fa6 100644 --- a/backend/internal/httpapi/task_api.go +++ b/backend/internal/httpapi/task_api.go @@ -137,10 +137,20 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { } func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Task, attempt repository.TaskAttempt, auditAction string) { + executionCtx, cancel := context.WithTimeout(c.Request.Context(), certificateTaskTimeout) + s.registerTaskCancel(task.ID, cancel) + defer func() { + cancel() + s.unregisterTaskCancel(task.ID) + }() + now := time.Now().UTC() - if err := s.tasks.UpdateTask(c.Request.Context(), task.ID, "running", "certificate_issue", "正在申请证书并部署", &now, nil); err != nil { + if err := s.tasks.UpdateTask(executionCtx, task.ID, "running", "certificate_issue", "正在申请证书并部署", &now, nil); err != nil { finished := time.Now().UTC() failureMessage := "任务启动失败" + if s.writeCancelledTaskResponse(c, task.ID) { + return + } _ = s.tasks.UpdateTask(context.Background(), task.ID, "failed", "certificate_issue", failureMessage, &now, &finished) _ = s.tasks.FinishAttempt(context.Background(), attempt.ID, "failed", failureMessage, &finished) if s.audit != nil { @@ -154,25 +164,22 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas return } if s.audit != nil { - if err := s.audit.CreateAuditLog(c.Request.Context(), repository.AuditLog{ + if err := s.audit.CreateAuditLog(executionCtx, repository.AuditLog{ TaskID: task.ID, OperatorID: task.OperatorID, Action: auditAction, ResourceType: "domain", ResourceID: task.DomainID, Status: "running", Message: "开始申请证书并部署", }); err != nil { finished := time.Now().UTC() failureMessage := "任务启动审计记录失败" + if s.writeCancelledTaskResponse(c, task.ID) { + 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 } } - executionCtx, cancel := context.WithTimeout(c.Request.Context(), certificateTaskTimeout) - s.registerTaskCancel(task.ID, cancel) - defer func() { - cancel() - s.unregisterTaskCancel(task.ID) - }() executionCtx = certificate.WithProgressReporter(executionCtx, func(progressCtx context.Context, progressStage, progressMessage string) error { return s.tasks.UpdateTask(progressCtx, task.ID, "running", progressStage, progressMessage, &now, nil) }) @@ -250,6 +257,23 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas c.JSON(http.StatusOK, response) } +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 + } + message := persisted.Message + if message == "" { + message = "任务已由用户取消" + } + c.JSON(http.StatusConflict, gin.H{ + "task": persisted, + "task_status": persisted.Status, + "error": message, + }) + return true +} + const certificateTaskTimeout = 30 * time.Minute func (s *Server) registerTaskCancel(taskID string, cancel context.CancelFunc) { diff --git a/backend/internal/httpapi/task_api_test.go b/backend/internal/httpapi/task_api_test.go index af2eebc..3218d26 100644 --- a/backend/internal/httpapi/task_api_test.go +++ b/backend/internal/httpapi/task_api_test.go @@ -64,6 +64,28 @@ func newTaskAPITestServer(t *testing.T, issue func(context.Context, string) (cer return server, taskRepo, auditRepo, func() { _ = db.Close() } } +type blockingTaskAuditRepository struct { + delegate repository.AuditRepository + started chan struct{} + release chan struct{} +} + +func (r *blockingTaskAuditRepository) CreateAuditLog(ctx context.Context, item repository.AuditLog) error { + if item.Status == "running" { + select { + case <-r.started: + default: + close(r.started) + } + select { + case <-r.release: + case <-ctx.Done(): + return ctx.Err() + } + } + return r.delegate.CreateAuditLog(ctx, item) +} + func TestTaskAPIRecordsSuccessfulIssueAndAudit(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{Status: "success"}, nil @@ -363,6 +385,65 @@ func TestTaskAPICanCancelRunningIssue(t *testing.T) { } } +func TestTaskAPICancellationDuringStartupDoesNotExecuteCertificate(t *testing.T) { + issueCalled := make(chan struct{}, 1) + server, taskRepo, _, closeDB := newTaskAPITestServer(t, func(_ context.Context, id string) (certificate.IssueResult, deployment.Report, error) { + issueCalled <- struct{}{} + return certificate.IssueResult{DomainID: id}, deployment.Report{Status: "success"}, nil + }) + defer closeDB() + + blockingAudit := &blockingTaskAuditRepository{ + delegate: server.audit, + started: make(chan struct{}), + release: make(chan struct{}), + } + server.audit = blockingAudit + issueResponse := httptest.NewRecorder() + issueDone := make(chan struct{}) + go func() { + serveAdminHTTP(server, issueResponse, httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)) + close(issueDone) + }() + select { + case <-blockingAudit.started: + case <-time.After(2 * time.Second): + t.Fatal("任务没有进入启动审计阶段") + } + + items, err := taskRepo.ListTasks(context.Background(), "domain-1", 10) + if err != nil || len(items) != 1 || items[0].Status != "running" { + t.Fatalf("启动阶段任务状态错误：%+v %v", items, err) + } + cancelResponse := httptest.NewRecorder() + serveAdminHTTP(server, cancelResponse, httptest.NewRequest(http.MethodDelete, "/api/v1/tasks/"+items[0].ID, nil)) + if cancelResponse.Code != http.StatusOK { + t.Fatalf("启动阶段取消任务失败：%d %s", cancelResponse.Code, cancelResponse.Body.String()) + } + close(blockingAudit.release) + select { + case <-issueDone: + case <-time.After(2 * time.Second): + t.Fatal("取消后的启动任务没有结束") + } + select { + case <-issueCalled: + t.Fatal("任务取消后不应再调用证书申请") + default: + } + if issueResponse.Code != http.StatusConflict { + t.Fatalf("启动阶段取消响应错误：%d %s", issueResponse.Code, issueResponse.Body.String()) + } + item, err := taskRepo.GetTask(context.Background(), items[0].ID) + if err != nil || item.Status != "cancelled" { + t.Fatalf("启动阶段取消状态没有持久化：%+v %v", item, err) + } + attempts, err := taskRepo.ListAttempts(context.Background(), item.ID) + if err != nil || len(attempts) != 1 || attempts[0].Status != "cancelled" { + t.Fatalf("启动阶段取消任务尝试状态错误：%+v %v", attempts, err) + } +} + func TestTaskAPIReplaysIdempotentIssue(t *testing.T) { calls := 0 server, taskRepo, _, closeDB := newTaskAPITestServer(t, func(_ context.Context, id string) (certificate.IssueResult, deployment.Report, error) {