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

补齐证书任务并发保护和DNS阶段进度

Browse Source 
增加同域名同类活动任务保护、DNS验证阶段进度回调及相关接口冲突响应，补充仓储回归测试。
This commit is contained in:
xuans 2026-09-30 03:56:51 +08:00
parent 69dc6e10dc
commit 604a1bf4f0
6 changed files with 101 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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()

31
backend/internal/certificate/service.go
View File

@ -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)
}

21
backend/internal/httpapi/certificate_api.go
View File

@ -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
}

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

@ -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()

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

@ -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 {

23
backend/internal/repository/task_repository_test.go
View File

@ -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()