From 604a1bf4f038f437b43e3635d013d21fd56bb121 Mon Sep 17 00:00:00 2001 From: xuans Date: Wed, 30 Sep 2026 03:56:51 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E9=BD=90=E8=AF=81=E4=B9=A6=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=B9=B6=E5=8F=91=E4=BF=9D=E6=8A=A4=E5=92=8CDNS?= =?UTF-8?q?=E9=98=B6=E6=AE=B5=E8=BF=9B=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 增加同域名同类活动任务保护、DNS验证阶段进度回调及相关接口冲突响应，补充仓储回归测试。 --- backend/internal/app/certificate_tasks.go | 10 ++++++ backend/internal/certificate/service.go | 31 +++++++++++++++++++ backend/internal/httpapi/certificate_api.go | 21 +++++++++++-- backend/internal/httpapi/task_api.go | 8 +++++ .../internal/repository/task_repository.go | 11 +++++++ .../repository/task_repository_test.go | 23 ++++++++++++++ 6 files changed, 101 insertions(+), 3 deletions(-) diff --git a/backend/internal/app/certificate_tasks.go b/backend/internal/app/certificate_tasks.go index e779497..461ed83 100644 --- a/backend/internal/app/certificate_tasks.go +++ b/backend/internal/app/certificate_tasks.go @@ -2,7 +2,9 @@ package app import ( "context" + "database/sql" "errors" + "fmt" "time" "git.pcux.cn/xuans/Scert/backend/internal/certificate" @@ -61,6 +63,11 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s if w.tasks == nil || w.audit == nil { return result, errors.New("证书任务记录服务未初始化") } + if active, err := w.tasks.FindActiveTaskByDomain(ctx, id, taskType); err == nil { + return result, fmt.Errorf("%w：%s", repository.ErrTaskAlreadyRunning, active.ID) + } else if !errors.Is(err, sql.ErrNoRows) { + return result, fmt.Errorf("检查活动证书任务失败：%w", err) + } started := time.Now().UTC() task := repository.Task{ ID: domain.NewID(), DomainID: id, TaskType: taskType, @@ -77,6 +84,9 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s defer hooks.Unregister(task.ID) } defer taskCancel() + taskCtx = certificate.WithProgressReporter(taskCtx, func(progressCtx context.Context, progressStage, progressMessage string) error { + return w.tasks.UpdateTask(progressCtx, task.ID, "running", progressStage, progressMessage, &started, nil) + }) defer func() { finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) defer cancel() diff --git a/backend/internal/certificate/service.go b/backend/internal/certificate/service.go index edf1517..78f6a31 100644 --- a/backend/internal/certificate/service.go +++ b/backend/internal/certificate/service.go @@ -39,6 +39,28 @@ type DNSAuthorizationCacheInvalidator interface { InvalidateDomain(providerID, requestedDomain string, credentials map[string]string) } +// ProgressReporter 用于把长耗时证书流程的阶段同步到任务中心。 +type ProgressReporter func(context.Context, string, string) error + +type progressReporterContextKey struct{} + +// WithProgressReporter 将任务进度回调绑定到证书工作流上下文。 +func WithProgressReporter(ctx context.Context, reporter ProgressReporter) context.Context { + if reporter == nil { + return ctx + } + return context.WithValue(ctx, progressReporterContextKey{}, reporter) +} + +// ReportProgress 报告证书工作流的当前阶段；未绑定任务时不执行任何操作。 +func ReportProgress(ctx context.Context, stage, message string) error { + reporter, _ := ctx.Value(progressReporterContextKey{}).(ProgressReporter) + if reporter == nil { + return nil + } + return reporter(ctx, strings.TrimSpace(stage), strings.TrimSpace(message)) +} + type AccountManager interface { LoadOrCreate(ctx context.Context, environment, email string) (*AccountUser, error) Save(ctx context.Context, environment string, user *AccountUser) error @@ -155,9 +177,15 @@ func (s *Service) Issue(ctx context.Context, domainID string) (IssueResult, erro if err != nil { return fail(fmt.Errorf("读取 DNS 凭据失败：%w", err)) } + if err := ReportProgress(ctx, "dns_validation", "正在验证 DNS 凭据权限"); err != nil { + return fail(err) + } if err := s.verifyDNSAuthorization(ctx, providerID, domains, credentials); err != nil { return fail(err) } + if err := ReportProgress(ctx, "certificate_issue", "DNS 凭据验证通过，正在申请证书"); err != nil { + return fail(err) + } provider, err := s.providers.Create(providerID, credentials) if err != nil { return fail(fmt.Errorf("创建 DNS Provider 失败：%w", err)) @@ -297,6 +325,9 @@ func (s *Service) verifyDNSAuthorization(ctx context.Context, providerID string, return nil } for _, requestedDomain := range domains { + if err := ReportProgress(ctx, "dns_validation", fmt.Sprintf("正在验证域名 %s 的 DNS 权限", requestedDomain)); err != nil { + return err + } if err := s.authorization.VerifyDomain(ctx, providerID, requestedDomain, credentials); err != nil { return fmt.Errorf("DNS 凭据无权操作域名 %s：%w", requestedDomain, err) } diff --git a/backend/internal/httpapi/certificate_api.go b/backend/internal/httpapi/certificate_api.go index c737405..105e2a3 100644 --- a/backend/internal/httpapi/certificate_api.go +++ b/backend/internal/httpapi/certificate_api.go @@ -103,7 +103,11 @@ func (s *Server) validateCertificate(c *gin.Context) { if s.validateCertificateWorkflow != nil { result, validation, err := s.validateCertificateWorkflow(s.taskWorkflowContext(c), strings.TrimSpace(c.Param("id"))) if err != nil { - s.writeError(c, http.StatusUnprocessableEntity, "\u8bc1\u4e66\u6821\u9a8c\u5931\u8d25", err) + status := http.StatusUnprocessableEntity + if errors.Is(err, repository.ErrTaskAlreadyRunning) { + status = http.StatusConflict + } + s.writeError(c, status, "证书校验失败", err) return } c.JSON(http.StatusOK, gin.H{"certificate": result, "validation": validation}) @@ -144,7 +148,11 @@ func (s *Server) deployCertificateCurrent(c *gin.Context) { s.logger.Error("证书部署失败", "error", err) } response["error"] = "证书部署失败" - c.JSON(http.StatusBadGateway, response) + status := http.StatusBadGateway + if errors.Is(err, repository.ErrTaskAlreadyRunning) { + status = http.StatusConflict + } + c.JSON(status, response) return } c.JSON(http.StatusOK, response) @@ -180,7 +188,11 @@ func (s *Server) importCertificateFiles(c *gin.Context) { s.logger.Error(message, "error", err) } response["error"] = message - c.JSON(http.StatusBadGateway, response) + status := http.StatusBadGateway + if errors.Is(err, repository.ErrTaskAlreadyRunning) { + status = http.StatusConflict + } + c.JSON(status, response) return } c.JSON(http.StatusOK, response) @@ -287,6 +299,9 @@ func saveOptionalMultipartFile(form *multipart.Form, field string) (string, func } func certificateErrorStatus(err error) int { + if errors.Is(err, repository.ErrTaskAlreadyRunning) { + return http.StatusConflict + } if repository.IsNotFound(err) || errors.Is(err, os.ErrNotExist) { return http.StatusNotFound } diff --git a/backend/internal/httpapi/task_api.go b/backend/internal/httpapi/task_api.go index e9af3d5..b9b4e15 100644 --- a/backend/internal/httpapi/task_api.go +++ b/backend/internal/httpapi/task_api.go @@ -12,6 +12,7 @@ import ( "github.com/gin-gonic/gin" + "git.pcux.cn/xuans/Scert/backend/internal/certificate" "git.pcux.cn/xuans/Scert/backend/internal/domain" "git.pcux.cn/xuans/Scert/backend/internal/repository" ) @@ -114,6 +115,10 @@ func (s *Server) issueDomainAsTask(c *gin.Context) { } } s.issueTaskMu.Unlock() + if errors.Is(err, repository.ErrTaskAlreadyRunning) { + c.JSON(http.StatusConflict, gin.H{"error": "该域名已有正在执行的同类证书任务"}) + return + } s.writeError(c, http.StatusInternalServerError, "创建证书任务失败", err) return } @@ -167,6 +172,9 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas 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) + }) certificateResult, deploymentReport, issueErr := s.issueCertificate(executionCtx, task.DomainID) deploymentReport = sanitizeDeploymentReport(deploymentReport) finished := time.Now().UTC() diff --git a/backend/internal/repository/task_repository.go b/backend/internal/repository/task_repository.go index dcc443f..14efab6 100644 --- a/backend/internal/repository/task_repository.go +++ b/backend/internal/repository/task_repository.go @@ -41,6 +41,9 @@ type TaskAttempt struct { CreatedAt time.Time `json:"created_at"` } +// ErrTaskAlreadyRunning 表示同一域名的同类任务仍处于活动状态。 +var ErrTaskAlreadyRunning = errors.New("同一域名的同类任务正在执行") + type TaskRepository interface { CreateTask(context.Context, Task) error GetTask(context.Context, string) (Task, error) @@ -119,6 +122,14 @@ func (r *SQLTaskRepository) CreateTask(ctx context.Context, item Task) error { } return fmt.Errorf("读取任务所属域名失败：%w", err) } + 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) + } args = append(args, projectID) result, err := tx.ExecContext(ctx, r.query(insertQuery+`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'project', ?)`), args...) if err != nil { diff --git a/backend/internal/repository/task_repository_test.go b/backend/internal/repository/task_repository_test.go index b16356f..a6b1368 100644 --- a/backend/internal/repository/task_repository_test.go +++ b/backend/internal/repository/task_repository_test.go @@ -3,6 +3,7 @@ package repository import ( "context" "database/sql" + "errors" "path/filepath" "testing" "time" @@ -99,6 +100,28 @@ func TestSQLTaskRepositoryRejectsInvalidStatus(t *testing.T) { } } +func TestSQLTaskRepositoryRejectsConcurrentActiveTaskOfSameType(t *testing.T) { + repository, db := newTaskRepositoryTest(t) + defer db.Close() + ctx := context.Background() + if err := repository.CreateTask(ctx, Task{ + ID: "task-active", DomainID: "domain-1", TaskType: "certificate_validate", + Status: "running", Stage: "certificate_validate", Message: "正在校验证书", + }); err != nil { + t.Fatalf("创建活动任务失败：%v", err) + } + err := repository.CreateTask(ctx, Task{ + ID: "task-duplicate", DomainID: "domain-1", TaskType: "certificate_validate", + Status: "queued", Stage: "queued", Message: "等待校验证书", + }) + if !errors.Is(err, ErrTaskAlreadyRunning) { + t.Fatalf("同类活动任务应被拒绝：%v", err) + } + if _, err := repository.GetTask(ctx, "task-duplicate"); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("拒绝并发任务后不应留下重复任务：%v", err) + } +} + func TestSQLTaskRepositoryCancelMarksRunningAttempt(t *testing.T) { repository, db := newTaskRepositoryTest(t) defer db.Close()