接入证书任务取消上下文

为证书导入、恢复、校验和当前部署工作流绑定任务取消回调,补充取消接口与回归测试,并同步四份整改清单进度。
This commit is contained in:
xuans 2026-09-30 02:32:58 +08:00
parent b8997512f9
commit 4cf9fb6076
11 changed files with 162 additions and 14 deletions

View File

@ -70,6 +70,12 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
if err := w.tasks.CreateTask(ctx, task); err != nil {
return result, err
}
taskCtx, taskCancel := context.WithCancel(ctx)
if hooks, ok := repository.TaskCancellationHooksFromContext(ctx); ok {
hooks.Register(task.ID, taskCancel)
defer hooks.Unregister(task.ID)
}
defer taskCancel()
defer func() {
finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer cancel()
@ -90,7 +96,7 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
w.logError(resultErr)
}
}()
if err := w.audit.CreateAuditLog(ctx, repository.AuditLog{
if err := w.audit.CreateAuditLog(taskCtx, repository.AuditLog{
TaskID: task.ID,
Action: taskType, ResourceType: "domain", ResourceID: id,
Status: "running", Message: "开始执行证书工作流",
@ -98,10 +104,10 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
return result, err
}
for number := 1; number <= attempts; number++ {
if err := ctx.Err(); err != nil {
if err := taskCtx.Err(); err != nil {
return result, err
}
if err := w.tasks.UpdateTask(ctx, task.ID, "running", stage, "正在执行证书工作流", &started, nil); err != nil {
if err := w.tasks.UpdateTask(taskCtx, task.ID, "running", stage, "正在执行证书工作流", &started, nil); err != nil {
return result, err
}
attemptStarted := time.Now().UTC()
@ -110,10 +116,10 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
Stage: stage, Status: "running", Message: "正在执行证书工作流",
StartedAt: &attemptStarted, CreatedAt: attemptStarted,
}
if err := w.tasks.CreateAttempt(ctx, attempt); err != nil {
if err := w.tasks.CreateAttempt(taskCtx, attempt); err != nil {
return result, err
}
issueCtx, cancel := context.WithTimeout(ctx, 30*time.Minute)
issueCtx, cancel := context.WithTimeout(taskCtx, 30*time.Minute)
result, resultErr = execute(issueCtx, id)
cancel()
finishCtx, finishCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
@ -127,10 +133,10 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
if resultErr == nil || number == attempts || result.Certificate.Saved.CertificatePath != "" || errors.Is(resultErr, context.Canceled) || errors.Is(resultErr, context.DeadlineExceeded) {
return result, resultErr
}
if err := w.tasks.UpdateTask(ctx, task.ID, "retry_wait", "retry_wait", "证书申请失败,等待重试", &started, nil); err != nil {
if err := w.tasks.UpdateTask(taskCtx, task.ID, "retry_wait", "retry_wait", "证书申请失败,等待重试", &started, nil); err != nil {
return result, errors.Join(resultErr, err)
}
if err := waitForManagementRetry(ctx, interval); err != nil {
if err := waitForManagementRetry(taskCtx, interval); err != nil {
return result, errors.Join(resultErr, err)
}
}

View File

@ -259,6 +259,38 @@ func TestCertificateTerminalPersistenceFailureIsReturnedWithoutReissue(t *testin
}
}
func TestCertificateTasksBindCancellationHook(t *testing.T) {
w, a := newCertificateTaskFixture(t)
registerCh := make(chan context.CancelFunc, 1)
unregisterCh := make(chan string, 1)
ctx := repository.WithTaskCancellationHooks(t.Context(), repository.TaskCancellationHooks{
Register: func(_ string, cancel context.CancelFunc) { registerCh <- cancel },
Unregister: func(taskID string) { unregisterCh <- taskID },
})
started := make(chan struct{})
finished := make(chan error, 1)
go func() {
_, err := w.run(ctx, "task-domain", "certificate_validate", "certificate_validate", 1, 0, func(ctx context.Context, _ string) (CertificateWorkflowResult, error) {
close(started)
<-ctx.Done()
return CertificateWorkflowResult{}, ctx.Err()
})
finished <- err
}()
<-started
cancel := <-registerCh
cancel()
if err := <-finished; !errors.Is(err, context.Canceled) {
t.Fatalf("task cancellation error = %v", err)
}
select {
case <-unregisterCh:
case <-time.After(time.Second):
t.Fatal("task cancellation hook was not unregistered")
}
assertCertificateTask(t, a, "cancelled", "cancelled")
}
func TestCertificateTasksRecordPartialDeploymentFailure(t *testing.T) {
w, a := newCertificateTaskFixture(t)
_, err := w.run(t.Context(), "task-domain", "renewal_run", "certificate_issue", 1, 0, func(context.Context, string) (CertificateWorkflowResult, error) {

View File

@ -96,13 +96,22 @@ func (s *Server) validateCertificate(c *gin.Context) {
if !s.authorizeVisibleDomainPermission(c, domain.PermissionPrivateKeyRead) {
return
}
if s.currentCertificate == nil {
if s.currentCertificate == nil && s.validateCertificateWorkflow == nil {
s.writeError(c, http.StatusNotImplemented, "证书校验接口尚未配置", errors.New("证书服务未配置"))
return
}
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)
return
}
c.JSON(http.StatusOK, gin.H{"certificate": result, "validation": validation})
return
}
result, err := s.currentCertificate.Current(c.Request.Context(), strings.TrimSpace(c.Param("id")))
if err != nil {
s.writeError(c, certificateErrorStatus(err), "读取当前证书失败", err)
s.writeError(c, certificateErrorStatus(err), "\u8bfb\u53d6\u5f53\u524d\u8bc1\u4e66\u5931\u8d25", err)
return
}
validation, err := (certificate.LocalValidator{VerifyChain: true}).ValidateFiles(
@ -127,7 +136,7 @@ func (s *Server) deployCertificateCurrent(c *gin.Context) {
s.writeError(c, http.StatusNotImplemented, "证书部署接口尚未配置", errors.New("证书部署流程未配置"))
return
}
result, report, err := s.deployCertificate(c.Request.Context(), strings.TrimSpace(c.Param("id")))
result, report, err := s.deployCertificate(s.taskWorkflowContext(c), strings.TrimSpace(c.Param("id")))
report = sanitizeDeploymentReport(report)
response := gin.H{"certificate": result, "deployment": report}
if err != nil {
@ -156,7 +165,7 @@ func (s *Server) importCertificateFiles(c *gin.Context) {
return
}
defer cleanup()
result, report, err := s.importCertificate(c.Request.Context(), request)
result, report, err := s.importCertificate(s.taskWorkflowContext(c), request)
report = sanitizeDeploymentReport(report)
// 导入保存与远端部署是不同阶段;部署失败不能让用户误以为证书尚未保存。
imported := err == nil || (result.DomainID == request.DomainID &&

View File

@ -20,7 +20,7 @@ func (s *Server) restoreCertificateVersion(c *gin.Context) {
return
}
version := strings.TrimSpace(c.Param("version"))
result, err := s.restoreVersion(c.Request.Context(), strings.TrimSpace(c.Param("id")), version)
result, err := s.restoreVersion(s.taskWorkflowContext(c), strings.TrimSpace(c.Param("id")), version)
if err != nil {
if s.logger != nil {
s.logger.Error("恢复证书历史版本失败", "domain_id", c.Param("id"), "version", version, "error", err)

View File

@ -241,6 +241,13 @@ func (s *Server) unregisterTaskCancel(taskID string) {
delete(s.taskCancels, taskID)
}
func (s *Server) taskWorkflowContext(c *gin.Context) context.Context {
return repository.WithTaskCancellationHooks(c.Request.Context(), repository.TaskCancellationHooks{
Register: s.registerTaskCancel,
Unregister: s.unregisterTaskCancel,
})
}
func taskRepositoryUnavailable(tasks repository.TaskRepository) bool {
if tasks == nil {
return true
@ -285,7 +292,7 @@ func (s *Server) cancelTask(c *gin.Context) {
c.JSON(http.StatusConflict, gin.H{"error": "任务当前无法取消", "task": item})
return
}
if item.TaskType == "renewal_run" || item.TaskType == "certificate_deploy" {
if item.TaskType == "renewal_run" {
c.JSON(http.StatusConflict, gin.H{"error": "该任务由证书工作流控制,不能单独修改取消状态", "task": item})
return
}

View File

@ -4,12 +4,40 @@ import (
"net/http"
"net/http/httptest"
"testing"
"time"
"git.pcux.cn/xuans/Scert/backend/internal/domain"
"git.pcux.cn/xuans/Scert/backend/internal/repository"
)
func TestTaskAPICancelsTrackedCertificateWorkflowTask(t *testing.T) {
server, tasks, _, closeDB := newTaskAPITestServer(t, nil)
defer closeDB()
item := repository.Task{
ID: "tracked-workflow-task", DomainID: "domain-1",
HistoryScope: repository.HistoryScope{ScopeType: repository.ScopeProject, ProjectID: domain.DefaultProjectID},
TaskType: "certificate_deploy", Status: "running", Stage: "certificate_deploy",
}
if err := tasks.CreateTask(t.Context(), item); err != nil {
t.Fatal(err)
}
cancelled := make(chan struct{}, 1)
server.registerTaskCancel(item.ID, func() { cancelled <- struct{}{} })
response := httptest.NewRecorder()
serveAdminHTTP(server, response, httptest.NewRequest(http.MethodDelete, "/api/v1/tasks/"+item.ID, nil))
stored, err := tasks.GetTask(t.Context(), item.ID)
if response.Code != http.StatusOK || err != nil || stored.Status != "cancelled" {
t.Fatalf("tracked workflow task was not cancelled: %d %+v %v", response.Code, stored, err)
}
select {
case <-cancelled:
case <-time.After(time.Second):
t.Fatal("tracked workflow cancellation callback was not called")
}
}
func TestTaskAPICannotFalselyCancelWorkflowTasks(t *testing.T) {
for _, kind := range []string{"renewal_run", "certificate_deploy"} {
for _, kind := range []string{"renewal_run"} {
t.Run(kind, func(t *testing.T) {
server, tasks, _, closeDB := newTaskAPITestServer(t, nil)
defer closeDB()

View File

@ -0,0 +1,22 @@
package repository
import "context"
type TaskCancellationHooks struct {
Register func(string, context.CancelFunc)
Unregister func(string)
}
type taskCancellationHooksContextKey struct{}
func WithTaskCancellationHooks(ctx context.Context, hooks TaskCancellationHooks) context.Context {
return context.WithValue(ctx, taskCancellationHooksContextKey{}, hooks)
}
func TaskCancellationHooksFromContext(ctx context.Context) (TaskCancellationHooks, bool) {
hooks, ok := ctx.Value(taskCancellationHooksContextKey{}).(TaskCancellationHooks)
if !ok || hooks.Register == nil || hooks.Unregister == nil {
return TaskCancellationHooks{}, false
}
return hooks, true
}

View File

@ -1,5 +1,16 @@
# Scert 开发任务清单
## 最新执行摘要(2026-09-29 继续开发)
本轮继续保持总体验收位置为 **P0-03 完成,待复审**,没有新增复审通过项;当前仍按“先开发、后统一测试”执行。
- [x] 已提交证书导入、恢复、校验任务入口接入代码:`b899751`;远程推送因仓库认证失败未成功,未阻塞本地继续开发。
- [x] 已将导入、恢复、校验及当前证书部署工作流接入统一任务取消上下文,任务中心取消可以触发实际执行上下文取消。
- [x] 已补充任务取消绑定回归用例;专项测试和后端全量 `go test ./...` 已通过,`git diff --check` 已通过。
- [x] 已核对导入、恢复、校验入口的任务状态、审计、尝试记录和取消后的终态一致性;本轮开发项等待复审。
- [ ] 三数据库矩阵、浏览器清缓存、密码 SSH、远端部署和生产验收仍未执行。
- [!] 总体验收继续保持 **P0-03 完成,待复审**;P0-10 其余任务链和 P0-12 仍未收尾。
## 最新执行摘要(2026-09-29 23:48)
本节为最新执行口径,下方按日期保留历史记录。总体验收仍为 **P0-03 完成,待复审**,没有新增复审通过项。

View File

@ -1,5 +1,16 @@
# Scert 整改任务执行清单
## 最新任务批次(2026-09-29 继续开发)
总体验收仍保持 **P0-03 完成,待复审**;本批代码已完成,统一测试已完成本地自动化部分。
- [x] 证书导入、恢复、校验入口已接入应用任务工作流,提交 `b899751`。
- [x] 当前证书部署、导入、恢复、校验工作流已接入任务取消上下文,可由任务中心取消实际执行上下文。
- [x] 已新增任务取消绑定回归用例;专项测试、后端全量 `go test ./...` 和 `git diff --check` 已通过。
- [x] 已核对任务创建、尝试记录、审计收尾、取消状态和 HTTP 响应的一致性;本批等待复审。
- [ ] 三数据库矩阵、浏览器、密码 SSH、远端部署和生产验收尚未执行。
- [!] 不升级总体验收位置;P0-10 仍在开发中,P0-12 尚未开始。
## 最新任务批次(2026-09-29 23:48)
本节优先于下方历史批次;总体验收保持 **P0-03 完成,待复审**。

View File

@ -1,5 +1,16 @@
# Scert 整改、合并与优化清单
## 最新进度摘要(2026-09-29 继续开发)
**总体验收位置仍为 P0-03 完成,待复审。** 本轮只推进 P0-10 任务工作流,不把局部代码完成或专项测试通过视为生产验收通过。
- [x] 证书导入、恢复、校验任务入口接入应用层,提交 `b899751`。
- [x] 当前证书部署、导入、恢复、校验入口接入任务取消上下文,任务中心取消可触发执行上下文取消。
- [x] 已补充取消绑定回归用例;专项测试、后端全量 `go test ./...` 和 `git diff --check` 已通过,当前批次等待复审。
- [~] P0-10 继续检查剩余任务状态、审计、尝试记录、取消、重试与崩溃恢复的一致性。
- [ ] P0-12 备份恢复、三数据库实库矩阵、浏览器、密码 SSH、远端部署及最终生产门禁尚未执行。
- [!] 不升级总体验收,不新增 `[✓]` 复审通过项。
## 最新进度摘要(2026-09-29 23:48)
**总体验收位置:P0-03 完成,待复审。** 本节为最新摘要,下方保留历史批次。开发子项完成不等于整项复审或生产验收通过,原始 44 项范围不缩减。

View File

@ -1,5 +1,16 @@
# 基础整改任务规格
## 最新执行与补充约束(2026-09-29 继续开发)
总体验收位置继续保持 **P0-03 完成,待复审**。本轮开发不改变原始 44 项范围,也不提前进入生产验收。
- [x] 证书导入、恢复、校验已接入统一任务工作流,提交 `b899751`。
- [x] 当前证书部署、导入、恢复、校验任务已接入取消上下文,取消任务可以中止实际执行。
- [x] 已新增取消绑定回归用例;专项测试、后端全量 `go test ./...` 和 `git diff --check` 已通过,仍需复审。
- [~] 继续核对任务尝试、审计、取消终态、重试幂等和异常收尾。
- [ ] P0-12 备份恢复、三数据库、浏览器、密码 SSH、远端部署和生产门禁仍未验收。
- [!] 不得把本轮开发结果标记为 `[✓]` 复审通过。
## 最新执行与补充约束(2026-09-29 23:48)
总体验收位置保持 **P0-03 完成,待复审**;下方各时间点内容为历史记录,不以开发子项升级整项验收结论。