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

修复证书任务取消启动竞态

Browse Source
This commit is contained in:
xuans 2026-09-30 11:32:43 +08:00
parent b7b37cf614
commit 415d504b2b
2 changed files with 113 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

81
backend/internal/httpapi/task_api_test.go
View File

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