统一后台证书任务报告及收尾事务并补齐故障回归
This commit is contained in:
parent
cf3bc7aaa5
commit
657a9c0a36
@ -116,6 +116,8 @@ deployment-results 响应保留旧 items，新增 reports；每次尝试独立
|
||||
- 队列满或关闭时拒绝执行，并尝试保存失败终态；持久化失败保留中断状态，不伪造已完成。
|
||||
- 服务关闭停止接收、发送取消信号并等待工作线程收尾。启动恢复将中断记录标为失败，不自动重放外部签发、部署。
|
||||
- HTTP 异步收尾路径将任务、本次尝试、完整报告、逐目标结果及完成审计放入同一事务；任一写入失败整体回滚并保留中断状态，不伪造终态。其他后台工作流仍需独立核查，不能据此声明全项目所有写入原子化。
|
||||
- 后台证书共用执行器现同样采用事务收尾，覆盖续期、当前证书部署、导入、恢复及校验入口。内部失败重试会保存本次失败尝试及完整报告，父任务保持 retry_wait，不生成完成审计；最终失败审计统一使用 error，与 HTTP 路径一致。
|
||||
- 后台重试等待期间取消，保留此前尝试及报告，只关闭父任务并保存取消审计；尚有活动尝试时拒绝绕过尝试单独收尾。报告、结果或审计落库失败不继续重试外部操作，保留活动状态供恢复。此约束不等于所有后台模块均已完成事务化。
|
||||
- 前端每三秒查询一次，单次周期十秒超时且不重叠；卸载后停止轮询、取消请求。
|
||||
- 切换任务时丢弃旧选择的迟到响应；撤权清除详情及部署结果。一般查询故障保留最后有效状态并提示未知，不改变任务终态。
|
||||
- 三个详情 GET 不是一次原子快照；任务终态与最大 attempt_number 的状态不一致时暂不合并，下一周期重新查询。
|
||||
@ -128,5 +130,6 @@ deployment-results 响应保留旧 items，新增 reports；每次尝试独立
|
||||
- [~] 当前 Linux 自动回归持续执行；竞争检测尚无可靠通过结果。
|
||||
- [x] 完整验证及恢复报告持久化与页面读取已实现，关联回归和前端 256 项通过；未部署。
|
||||
- [x] HTTP 异步完成审计、逐服务器结果与任务终态同事务，已覆盖三类保存失败整体回滚；其他工作流及三库实机复审仍待完成。
|
||||
- [x] 后台证书共用执行器新增五处故障回滚、每次重试报告、等待取消及无活动尝试收尾回归；应用、仓储、HTTP 三包原生 Linux 测试、全后端测试及 go vet 均通过，未正式部署。
|
||||
- [ ] 导入、版本恢复、校验及手动续期等其他长操作尚未全部异步化。
|
||||
- [ ] 正式构建部署后验证超过反代等待窗口的真实操作、断线查询、取消回滚和服务重启；本说明不替代生产验收。
|
||||
|
||||
118
backend/internal/app/certificate_task_completion_test.go
Normal file
118
backend/internal/app/certificate_task_completion_test.go
Normal file
@ -0,0 +1,118 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/deployment"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/repository"
|
||||
)
|
||||
|
||||
func TestCertificateCompletionPersistsFullReportsAcrossRetries(t *testing.T) {
|
||||
w, a := newCertificateTaskFixture(t)
|
||||
calls := 0
|
||||
_, err := w.run(t.Context(), "task-domain", "renewal_run", "certificate_issue", 2, time.Nanosecond,
|
||||
func(context.Context, string) (CertificateWorkflowResult, error) {
|
||||
calls++
|
||||
result := CertificateWorkflowResult{Deployment: deployment.Report{
|
||||
Status: "success", RecoveryStatus: "success", Total: 1, Success: 1,
|
||||
Results: []deployment.Result{{ServerID: "target", Status: "success"}},
|
||||
}}
|
||||
if calls == 1 {
|
||||
result.Deployment.Status = "failed"
|
||||
return result, errors.New("temporary failure")
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
if err != nil || calls != 2 {
|
||||
t.Fatalf("重试执行失败：%d %v", calls, err)
|
||||
}
|
||||
task := assertCertificateTask(t, a, "success", "failed", "success")
|
||||
reports, err := a.Tasks.ListTaskDeploymentReports(t.Context(), task.ID)
|
||||
if err != nil || len(reports) != 2 {
|
||||
t.Fatalf("每次尝试均应保存完整报告：%+v %v", reports, err)
|
||||
}
|
||||
if reports[0].Report.Status != "failed" || reports[1].Report.Status != "success" || reports[0].AttemptID == reports[1].AttemptID || reports[1].Report.RecoveryStatus != "success" {
|
||||
t.Fatalf("重试覆盖报告或丢失恢复状态：%+v", reports)
|
||||
}
|
||||
}
|
||||
|
||||
type cancelAfterCertificateRetry struct {
|
||||
repository.TaskRepository
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
func (r cancelAfterCertificateRetry) FinishTaskAttempt(ctx context.Context, completion repository.TaskCompletion) error {
|
||||
err := r.TaskRepository.FinishTaskAttempt(ctx, completion)
|
||||
if err == nil && completion.Retry {
|
||||
r.cancel()
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func TestCertificateCompletionCancellationDuringRetryWait(t *testing.T) {
|
||||
w, a := newCertificateTaskFixture(t)
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
defer cancel()
|
||||
w.tasks = cancelAfterCertificateRetry{TaskRepository: a.Tasks, cancel: cancel}
|
||||
calls := 0
|
||||
_, err := w.run(ctx, "task-domain", "renewal_run", "certificate_issue", 3, time.Hour,
|
||||
func(context.Context, string) (CertificateWorkflowResult, error) {
|
||||
calls++
|
||||
return CertificateWorkflowResult{}, errors.New("temporary failure")
|
||||
})
|
||||
if !errors.Is(err, context.Canceled) || calls != 1 {
|
||||
t.Fatalf("等待取消后继续重试或丢失取消原因：%d %v", calls, err)
|
||||
}
|
||||
task := assertCertificateTask(t, a, "cancelled", "failed")
|
||||
reports, err := a.Tasks.ListTaskDeploymentReports(t.Context(), task.ID)
|
||||
if err != nil || len(reports) != 1 {
|
||||
t.Fatalf("等待取消丢失上次尝试报告：%+v %v", reports, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCertificateCompletionFailureRollsBackAllRecords(t *testing.T) {
|
||||
for _, table := range []string{"task_deployment_reports", "task_deployment_results", "audit_logs", "task_attempts", "tasks"} {
|
||||
t.Run(table, func(t *testing.T) {
|
||||
w, a := newCertificateTaskFixture(t)
|
||||
calls := 0
|
||||
_, err := w.run(t.Context(), "task-domain", "renewal_run", "certificate_issue", 3, time.Nanosecond,
|
||||
func(context.Context, string) (CertificateWorkflowResult, error) {
|
||||
calls++
|
||||
operation := "INSERT"
|
||||
if table == "tasks" || table == "task_attempts" {
|
||||
operation = "UPDATE"
|
||||
}
|
||||
// 表名只来自本测试的固定枚举，不接受用户输入。
|
||||
if _, err := a.Database.DB.ExecContext(t.Context(), "CREATE TRIGGER IF NOT EXISTS reject_completion BEFORE "+operation+" ON "+table+" BEGIN SELECT RAISE(ABORT, 'completion unavailable'); END"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return CertificateWorkflowResult{Deployment: deployment.Report{Status: "success", Total: 1, Success: 1,
|
||||
Results: []deployment.Result{{ServerID: "target", Status: "success"}}}}, nil
|
||||
})
|
||||
if err == nil || calls != 1 {
|
||||
t.Fatalf("保存失败必须返回错误且不重复外部操作：%d %v", calls, err)
|
||||
}
|
||||
tasks, err := a.Tasks.ListTasks(t.Context(), "task-domain", 10)
|
||||
if err != nil || len(tasks) != 1 || tasks[0].Status != "running" || tasks[0].FinishedAt != nil {
|
||||
t.Fatalf("事务失败必须保留活动任务供恢复：%+v %v", tasks, err)
|
||||
}
|
||||
attempts, err := a.Tasks.ListAttempts(t.Context(), tasks[0].ID)
|
||||
if err != nil || len(attempts) != 1 || attempts[0].Status != "running" || attempts[0].FinishedAt != nil {
|
||||
t.Fatalf("尝试终态未回滚：%+v %v", attempts, err)
|
||||
}
|
||||
for _, stored := range []string{"task_deployment_reports", "task_deployment_results", "audit_logs"} {
|
||||
var count int
|
||||
query := "SELECT COUNT(*) FROM " + stored + " WHERE task_id = ?"
|
||||
if stored == "audit_logs" {
|
||||
query += " AND status <> 'running'"
|
||||
}
|
||||
if err := a.Database.DB.QueryRowContext(t.Context(), query, tasks[0].ID).Scan(&count); err != nil || count != 0 {
|
||||
t.Fatalf("失败后仍残留部分记录：%s %d %v", stored, count, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -101,26 +101,32 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
|
||||
taskCtx = certificate.WithProgressReporter(taskCtx, func(progressCtx context.Context, progressStage, progressMessage string) error {
|
||||
return w.tasks.UpdateTask(progressCtx, task.ID, "running", progressStage, progressMessage, &started, nil)
|
||||
})
|
||||
var activeAttempt string
|
||||
completionFailed := false
|
||||
defer func() {
|
||||
defer func() {
|
||||
if resultErr != nil && w.logError != nil {
|
||||
w.logError(resultErr)
|
||||
}
|
||||
}()
|
||||
// 收尾事务失败后保留中断记录，不能再用分散写入覆盖现场。
|
||||
if completionFailed {
|
||||
return
|
||||
}
|
||||
finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
defer cancel()
|
||||
finished := time.Now().UTC()
|
||||
status, message := certificateTaskOutcome(result, resultErr)
|
||||
auditErr := w.audit.CreateAuditLog(finishCtx, repository.AuditLog{
|
||||
TaskID: task.ID,
|
||||
Action: taskType, ResourceType: "domain", ResourceID: id,
|
||||
OperatorID: task.OperatorID,
|
||||
Status: status, Message: message,
|
||||
var report *deployment.Report
|
||||
if activeAttempt != "" {
|
||||
report = &result.Deployment
|
||||
}
|
||||
recordErr := w.tasks.FinishTaskAttempt(finishCtx, repository.TaskCompletion{
|
||||
TaskID: task.ID, AttemptID: activeAttempt, Status: status, Stage: stage,
|
||||
Message: message, StartedAt: &started, FinishedAt: finished,
|
||||
Report: report, AuditAction: taskType, WithoutAttempt: activeAttempt == "",
|
||||
})
|
||||
if auditErr != nil {
|
||||
status, message = certificateTaskOutcome(result, auditErr)
|
||||
stage = "record"
|
||||
}
|
||||
taskErr := w.tasks.UpdateTask(finishCtx, task.ID, status, stage, message, &started, &finished)
|
||||
resultErr = errors.Join(resultErr, taskErr, auditErr)
|
||||
if resultErr != nil && w.logError != nil {
|
||||
w.logError(resultErr)
|
||||
}
|
||||
resultErr = errors.Join(resultErr, recordErr)
|
||||
}()
|
||||
if err := w.audit.CreateAuditLog(taskCtx, repository.AuditLog{
|
||||
TaskID: task.ID,
|
||||
@ -146,30 +152,30 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
|
||||
if err := w.tasks.CreateAttempt(taskCtx, attempt); err != nil {
|
||||
return result, err
|
||||
}
|
||||
activeAttempt = attempt.ID
|
||||
issueCtx, cancel := context.WithTimeout(taskCtx, 30*time.Minute)
|
||||
issueCtx = deployment.WithTransactionContext(issueCtx, deployment.TransactionContext{
|
||||
TaskID: task.ID, AttemptID: attempt.ID, AttemptNumber: attempt.AttemptNumber,
|
||||
})
|
||||
result, resultErr = execute(issueCtx, id)
|
||||
cancel()
|
||||
// 取消只停止外部操作，结果保存仍使用独立且有限时的收尾上下文。
|
||||
finishCtx, finishCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
if recordErr := w.tasks.RecordTaskDeploymentResults(finishCtx, task.ID, attempt.ID, number, taskDeploymentResults(result.Deployment)); recordErr != nil {
|
||||
resultErr = errors.Join(resultErr, recordErr)
|
||||
}
|
||||
finished := time.Now().UTC()
|
||||
status, message := certificateTaskOutcome(result, resultErr)
|
||||
recordErr := w.tasks.FinishAttempt(finishCtx, attempt.ID, status, message, &finished)
|
||||
finishCancel()
|
||||
if recordErr != nil {
|
||||
return result, errors.Join(resultErr, recordErr)
|
||||
}
|
||||
if resultErr == nil || number == attempts || result.Certificate.Saved.CertificatePath != "" || errors.Is(resultErr, context.Canceled) || errors.Is(resultErr, context.DeadlineExceeded) {
|
||||
if resultErr == nil || number == attempts || result.Certificate.Saved.CertificatePath != "" || result.Deployment.Status == "partial" || result.Deployment.RecoveryStatus == "unconfirmed" || result.Deployment.RecoveryStatus == "failed" || errors.Is(resultErr, context.Canceled) || errors.Is(resultErr, context.DeadlineExceeded) {
|
||||
return result, resultErr
|
||||
}
|
||||
if err := w.tasks.UpdateTask(taskCtx, task.ID, "retry_wait", "retry_wait", "证书申请失败，等待重试", &started, nil); err != nil {
|
||||
return result, errors.Join(resultErr, err)
|
||||
// 取消只停止外部操作，结果保存仍使用独立且有限时的收尾上下文。
|
||||
finishCtx, finishCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
finished := time.Now().UTC()
|
||||
status, message := certificateTaskOutcome(result, resultErr)
|
||||
recordErr := w.tasks.FinishTaskAttempt(finishCtx, repository.TaskCompletion{
|
||||
TaskID: task.ID, AttemptID: attempt.ID, Status: status, Stage: stage,
|
||||
Message: message, StartedAt: &started, FinishedAt: finished, Report: &result.Deployment, Retry: true,
|
||||
})
|
||||
finishCancel()
|
||||
if recordErr != nil {
|
||||
completionFailed = true
|
||||
return result, errors.Join(resultErr, recordErr)
|
||||
}
|
||||
activeAttempt = ""
|
||||
if err := waitForManagementRetry(taskCtx, interval); err != nil {
|
||||
return result, errors.Join(resultErr, err)
|
||||
}
|
||||
|
||||
@ -59,10 +59,14 @@ func assertCertificateTask(t *testing.T, a *App, status string, statuses ...stri
|
||||
}
|
||||
}
|
||||
var count int
|
||||
if err := a.Database.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM audit_logs WHERE action = ? AND resource_id = ? AND status = ?", item.TaskType, item.DomainID, status).Scan(&count); err != nil || count != 1 {
|
||||
auditStatus := status
|
||||
if auditStatus == "failed" {
|
||||
auditStatus = "error"
|
||||
}
|
||||
if err := a.Database.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM audit_logs WHERE action = ? AND resource_id = ? AND status = ?", item.TaskType, item.DomainID, auditStatus).Scan(&count); err != nil || count != 1 {
|
||||
t.Fatalf("审计记录缺失：%d %v", count, err)
|
||||
}
|
||||
if err := a.Database.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM audit_logs WHERE task_id = ? AND scope_type = 'project' AND project_id = ? AND operator_id = '' AND status IN ('running', ?)", item.ID, item.ProjectID, status).Scan(&count); err != nil || count != 2 {
|
||||
if err := a.Database.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM audit_logs WHERE task_id = ? AND scope_type = 'project' AND project_id = ? AND operator_id = '' AND status IN ('running', ?)", item.ID, item.ProjectID, auditStatus).Scan(&count); err != nil || count != 2 {
|
||||
t.Fatalf("后台任务开始与收尾审计关联缺失：%d，%v", count, err)
|
||||
}
|
||||
return item
|
||||
@ -168,11 +172,11 @@ func (r failingCertificateTasks) CreateTask(ctx context.Context, task repository
|
||||
return r.TaskRepository.CreateTask(ctx, task)
|
||||
}
|
||||
|
||||
func (r failingCertificateTasks) FinishAttempt(ctx context.Context, id, status, message string, at *time.Time) error {
|
||||
func (r failingCertificateTasks) FinishTaskAttempt(ctx context.Context, completion repository.TaskCompletion) error {
|
||||
if r.failFinish {
|
||||
return errors.New("attempt-store-unavailable")
|
||||
}
|
||||
return r.TaskRepository.FinishAttempt(ctx, id, status, message, at)
|
||||
return r.TaskRepository.FinishTaskAttempt(ctx, completion)
|
||||
}
|
||||
|
||||
func TestCertificateTaskPersistenceFailuresStopExecutionOrRetries(t *testing.T) {
|
||||
@ -320,11 +324,8 @@ type failingCertificateTerminal struct {
|
||||
failure error
|
||||
}
|
||||
|
||||
func (r failingCertificateTerminal) UpdateTask(ctx context.Context, id, status, stage, message string, started, finished *time.Time) error {
|
||||
if finished != nil {
|
||||
return r.failure
|
||||
}
|
||||
return r.TaskRepository.UpdateTask(ctx, id, status, stage, message, started, finished)
|
||||
func (r failingCertificateTerminal) FinishTaskAttempt(ctx context.Context, completion repository.TaskCompletion) error {
|
||||
return r.failure
|
||||
}
|
||||
|
||||
func TestCertificateAuditFailuresPersistFailedTaskWithoutReissue(t *testing.T) {
|
||||
@ -333,6 +334,13 @@ func TestCertificateAuditFailuresPersistFailedTaskWithoutReissue(t *testing.T) {
|
||||
w, a := newCertificateTaskFixture(t)
|
||||
failure := errors.New("audit-store-unavailable")
|
||||
w.audit = failingCertificateAudit{AuditRepository: w.audit, failStatus: failStatus, failure: failure}
|
||||
if failStatus == "success" {
|
||||
// 完成审计已移入数据库事务，在真实写入点注入失败。
|
||||
if _, err := a.Database.DB.ExecContext(t.Context(), `CREATE TRIGGER reject_success_audit BEFORE INSERT ON audit_logs
|
||||
WHEN NEW.status = 'success' BEGIN SELECT RAISE(ABORT, 'audit-store-unavailable'); END`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
calls := 0
|
||||
var logged error
|
||||
w.logError = func(err error) { logged = err }
|
||||
@ -344,15 +352,19 @@ func TestCertificateAuditFailuresPersistFailedTaskWithoutReissue(t *testing.T) {
|
||||
if failStatus == "running" {
|
||||
expectedCalls = 0
|
||||
}
|
||||
if !errors.Is(err, failure) || !errors.Is(logged, failure) || calls != expectedCalls {
|
||||
if err == nil || logged == nil || !strings.Contains(err.Error(), failure.Error()) || !strings.Contains(logged.Error(), failure.Error()) || calls != expectedCalls {
|
||||
t.Fatalf("calls=%d error=%v logged=%v", calls, err, logged)
|
||||
}
|
||||
tasks, listErr := a.Tasks.ListTasks(t.Context(), "task-domain", 10)
|
||||
if listErr != nil || len(tasks) != 1 || tasks[0].Status != "failed" || tasks[0].FinishedAt == nil {
|
||||
if listErr != nil || len(tasks) != 1 {
|
||||
t.Fatalf("tasks=%+v error=%v", tasks, listErr)
|
||||
}
|
||||
if failStatus == "success" && tasks[0].Stage != "record" {
|
||||
t.Fatalf("stage=%s", tasks[0].Stage)
|
||||
if failStatus == "success" {
|
||||
if tasks[0].Status != "running" || tasks[0].FinishedAt != nil {
|
||||
t.Fatalf("完成审计失败必须回滚终态：%+v", tasks[0])
|
||||
}
|
||||
} else if tasks[0].Status != "failed" || tasks[0].FinishedAt == nil {
|
||||
t.Fatalf("开始审计失败必须停止执行并收尾：%+v", tasks[0])
|
||||
}
|
||||
if strings.Contains(tasks[0].Message, failure.Error()) {
|
||||
t.Fatal("audit-store-unavailable")
|
||||
|
||||
@ -13,15 +13,17 @@ import (
|
||||
|
||||
// TaskCompletion 同时关闭任务与对应尝试，避免只保存其中一方的终态。
|
||||
type TaskCompletion struct {
|
||||
TaskID string
|
||||
AttemptID string
|
||||
Status string
|
||||
Stage string
|
||||
Message string
|
||||
StartedAt *time.Time
|
||||
FinishedAt time.Time
|
||||
Report *deployment.Report
|
||||
AuditAction string
|
||||
TaskID string
|
||||
AttemptID string
|
||||
Status string
|
||||
Stage string
|
||||
Message string
|
||||
StartedAt *time.Time
|
||||
FinishedAt time.Time
|
||||
Report *deployment.Report
|
||||
AuditAction string
|
||||
Retry bool
|
||||
WithoutAttempt bool
|
||||
}
|
||||
|
||||
func (r *SQLTaskRepository) FinishTaskAttempt(ctx context.Context, completion TaskCompletion) error {
|
||||
@ -31,7 +33,7 @@ func (r *SQLTaskRepository) FinishTaskAttempt(ctx context.Context, completion Ta
|
||||
completion.TaskID = strings.TrimSpace(completion.TaskID)
|
||||
completion.AttemptID = strings.TrimSpace(completion.AttemptID)
|
||||
completion.Stage = strings.TrimSpace(completion.Stage)
|
||||
if completion.TaskID == "" || completion.AttemptID == "" || completion.Stage == "" || completion.FinishedAt.IsZero() {
|
||||
if completion.TaskID == "" || (completion.AttemptID == "" && !completion.WithoutAttempt) || completion.Stage == "" || completion.FinishedAt.IsZero() {
|
||||
return errors.New("任务终态记录不完整")
|
||||
}
|
||||
switch completion.Status {
|
||||
@ -39,17 +41,31 @@ func (r *SQLTaskRepository) FinishTaskAttempt(ctx context.Context, completion Ta
|
||||
default:
|
||||
return errors.New("任务终态无效")
|
||||
}
|
||||
if completion.WithoutAttempt && (completion.AttemptID != "" || completion.Report != nil || completion.Retry || (completion.Status != "failed" && completion.Status != "cancelled")) {
|
||||
return errors.New("无活动尝试的收尾参数无效")
|
||||
}
|
||||
if completion.Retry && (completion.Status != "failed" || completion.AuditAction != "") {
|
||||
return errors.New("重试只能保存失败尝试且不能发布完成审计")
|
||||
}
|
||||
tx, err := r.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("开始任务收尾事务失败：%w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
taskStatus, taskStage := completion.Status, completion.Stage
|
||||
var taskFinished any = completion.FinishedAt.UTC()
|
||||
activeStates := "('queued', 'running', 'retry_wait', 'cancelling')"
|
||||
if completion.Retry {
|
||||
taskStatus, taskStage, taskFinished = "retry_wait", "retry_wait", nil
|
||||
// 已受理的取消不能被重试重新激活。
|
||||
activeStates = "('running')"
|
||||
}
|
||||
// 与取消、重试保持任务在前、尝试在后的加锁顺序。旧尝试或跨任务尝试必须整体回滚。
|
||||
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks
|
||||
SET status = ?, stage = ?, message = ?, started_at = COALESCE(?, started_at), finished_at = ?, updated_at = ?
|
||||
WHERE id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`),
|
||||
completion.Status, completion.Stage, completion.Message, nullableTime(completion.StartedAt),
|
||||
completion.FinishedAt.UTC(), completion.FinishedAt.UTC(), completion.TaskID)
|
||||
WHERE id = ? AND status IN `+activeStates),
|
||||
taskStatus, taskStage, completion.Message, nullableTime(completion.StartedAt),
|
||||
taskFinished, completion.FinishedAt.UTC(), completion.TaskID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("保存任务终态失败：%w", err)
|
||||
}
|
||||
@ -58,17 +74,8 @@ func (r *SQLTaskRepository) FinishTaskAttempt(ctx context.Context, completion Ta
|
||||
} else if count != 1 {
|
||||
return sql.ErrNoRows
|
||||
}
|
||||
result, err = tx.ExecContext(ctx, r.query(`UPDATE task_attempts
|
||||
SET status = ?, stage = ?, message = ?, finished_at = ?
|
||||
WHERE id = ? AND task_id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`),
|
||||
completion.Status, completion.Stage, completion.Message, completion.FinishedAt.UTC(), completion.AttemptID, completion.TaskID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("保存任务尝试终态失败：%w", err)
|
||||
}
|
||||
if count, err := result.RowsAffected(); err != nil {
|
||||
return fmt.Errorf("读取任务尝试收尾结果失败：%w", err)
|
||||
} else if count != 1 {
|
||||
return sql.ErrNoRows
|
||||
if err := r.finishCompletionAttempt(ctx, tx, completion); err != nil {
|
||||
return err
|
||||
}
|
||||
if completion.Report != nil {
|
||||
if err := r.recordCompletionReport(ctx, tx, completion); err != nil {
|
||||
@ -85,3 +92,30 @@ func (r *SQLTaskRepository) FinishTaskAttempt(ctx context.Context, completion Ta
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *SQLTaskRepository) finishCompletionAttempt(ctx context.Context, tx *sql.Tx, completion TaskCompletion) error {
|
||||
if completion.WithoutAttempt {
|
||||
var active int
|
||||
if err := tx.QueryRowContext(ctx, r.query(`SELECT COUNT(*) FROM task_attempts
|
||||
WHERE task_id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`), completion.TaskID).Scan(&active); err != nil {
|
||||
return fmt.Errorf("检查活动尝试失败：%w", err)
|
||||
}
|
||||
if active != 0 {
|
||||
return errors.New("任务仍有活动尝试，不能单独收尾")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
result, err := tx.ExecContext(ctx, r.query(`UPDATE task_attempts
|
||||
SET status = ?, stage = ?, message = ?, finished_at = ?
|
||||
WHERE id = ? AND task_id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`),
|
||||
completion.Status, completion.Stage, completion.Message, completion.FinishedAt.UTC(), completion.AttemptID, completion.TaskID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("保存任务尝试终态失败：%w", err)
|
||||
}
|
||||
if count, err := result.RowsAffected(); err != nil {
|
||||
return fmt.Errorf("读取任务尝试收尾结果失败：%w", err)
|
||||
} else if count != 1 {
|
||||
return sql.ErrNoRows
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
102
backend/internal/repository/task_completion_retry_test.go
Normal file
102
backend/internal/repository/task_completion_retry_test.go
Normal file
@ -0,0 +1,102 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/deployment"
|
||||
)
|
||||
|
||||
func TestTaskCompletionRetryKeepsTaskActiveAndPreservesReport(t *testing.T) {
|
||||
repo, db := newTaskRepositoryTest(t)
|
||||
defer db.Close()
|
||||
completion := seedTaskCompletion(t, repo)
|
||||
completion.Status, completion.Retry = "failed", true
|
||||
completion.Report = &deployment.Report{Status: "failed", RecoveryStatus: "success"}
|
||||
if err := repo.FinishTaskAttempt(t.Context(), completion); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task, err := repo.GetTask(t.Context(), completion.TaskID)
|
||||
if err != nil || task.Status != "retry_wait" || task.Stage != "retry_wait" || task.FinishedAt != nil {
|
||||
t.Fatalf("重试不能提前关闭任务：%+v %v", task, err)
|
||||
}
|
||||
attempts, err := repo.ListAttempts(t.Context(), completion.TaskID)
|
||||
if err != nil || len(attempts) != 1 || attempts[0].Status != "failed" || attempts[0].FinishedAt == nil {
|
||||
t.Fatalf("失败尝试未收尾：%+v %v", attempts, err)
|
||||
}
|
||||
reports, err := repo.ListTaskDeploymentReports(t.Context(), completion.TaskID)
|
||||
if err != nil || len(reports) != 1 || reports[0].Report.RecoveryStatus != "success" {
|
||||
t.Fatalf("失败尝试报告未保存：%+v %v", reports, err)
|
||||
}
|
||||
// 重试等待期间取消：只关闭任务和审计，不覆盖已完成的尝试及报告。
|
||||
completion.AttemptID, completion.Status = "", "cancelled"
|
||||
completion.Retry, completion.WithoutAttempt = false, true
|
||||
completion.Report, completion.AuditAction = nil, "certificate_deploy"
|
||||
if err := repo.FinishTaskAttempt(t.Context(), completion); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task, err = repo.GetTask(t.Context(), completion.TaskID)
|
||||
if err != nil || task.Status != "cancelled" || task.FinishedAt == nil {
|
||||
t.Fatalf("等待取消未收尾：%+v %v", task, err)
|
||||
}
|
||||
attempts, err = repo.ListAttempts(t.Context(), completion.TaskID)
|
||||
if err != nil || len(attempts) != 1 || attempts[0].Status != "failed" {
|
||||
t.Fatalf("等待取消覆盖历史尝试：%+v %v", attempts, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskCompletionRejectsUnsafeRetryAndMissingAttempt(t *testing.T) {
|
||||
for _, scenario := range []string{"active_attempt", "retry_cancelled", "retry_success", "retry_audit", "no_attempt_success", "no_attempt_report"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
repo, db := newTaskRepositoryTest(t)
|
||||
defer db.Close()
|
||||
completion := seedTaskCompletion(t, repo)
|
||||
completion.Status = "failed"
|
||||
wantStatus := "running"
|
||||
switch scenario {
|
||||
case "active_attempt":
|
||||
completion.AttemptID, completion.WithoutAttempt = "", true
|
||||
case "retry_cancelled":
|
||||
completion.Retry = true
|
||||
if err := repo.UpdateTask(t.Context(), completion.TaskID, "cancelling", "deployment", "正在取消", nil, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wantStatus = "cancelling"
|
||||
case "retry_success":
|
||||
completion.Retry, completion.Status = true, "success"
|
||||
case "retry_audit":
|
||||
completion.Retry, completion.AuditAction = true, "certificate_deploy"
|
||||
case "no_attempt_success":
|
||||
completion.AttemptID, completion.WithoutAttempt, completion.Status = "", true, "success"
|
||||
case "no_attempt_report":
|
||||
completion.AttemptID, completion.WithoutAttempt = "", true
|
||||
completion.Report = &deployment.Report{}
|
||||
}
|
||||
if err := repo.FinishTaskAttempt(t.Context(), completion); err == nil {
|
||||
t.Fatal("不安全收尾未被拒绝")
|
||||
}
|
||||
task, err := repo.GetTask(t.Context(), completion.TaskID)
|
||||
if err != nil || task.Status != wantStatus || task.FinishedAt != nil {
|
||||
t.Fatalf("拒绝后改变任务状态：%+v %v", task, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskCompletionWithoutAttemptRollsBackOnAuditFailure(t *testing.T) {
|
||||
repo, db := newTaskRepositoryTest(t)
|
||||
defer db.Close()
|
||||
if err := repo.CreateTask(t.Context(), Task{ID: "before-attempt", DomainID: "domain-1", TaskType: "certificate_deploy", Status: "running", Stage: "deployment"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.ExecContext(t.Context(), `CREATE TRIGGER reject_completion_audit BEFORE INSERT ON audit_logs BEGIN SELECT RAISE(ABORT, 'audit unavailable'); END`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := repo.FinishTaskAttempt(t.Context(), TaskCompletion{TaskID: "before-attempt", Status: "failed", Stage: "deployment", FinishedAt: time.Now().UTC(), WithoutAttempt: true, AuditAction: "certificate_deploy"}); err == nil {
|
||||
t.Fatal("审计写入失败没有返回")
|
||||
}
|
||||
task, err := repo.GetTask(t.Context(), "before-attempt")
|
||||
if err != nil || task.Status != "running" || task.FinishedAt != nil {
|
||||
t.Fatalf("无尝试路径未整体回滚：%+v %v", task, err)
|
||||
}
|
||||
}
|
||||
@ -1,5 +1,16 @@
|
||||
# Linux 生产流程测试记录
|
||||
|
||||
## 2026年10月2日18:35 后台原子收尾整改（未部署）
|
||||
|
||||
- [x] 上批 background-full.log 与 background-vet.log 已取回：BACKGROUND_FULL_EXIT=0、BACKGROUND_VET_EXIT=0。第二台既有服务 active，进程检查仅保留系统 MySQL；磁盘剩余约 1014MB，不新增数据库实例。
|
||||
- [!] 第一台 scert-report-race-20261002-1801 的 Result=timeout，运行十分钟退出一；内存上限 512MB、CPU 上限 50% 实际生效。没有竞争检测通过结论，无残留测试／编译进程，Scert、Nginx、MySQL active，可用内存约 903MB。
|
||||
- [x] 修改前已推送 cf3bc7a。后台原子化测试归档 SHA256 为 d32c8d1849ebbd542523006cb131275e2f6ed925401e143ac69d4f1dc5a3ff40，两端一致；旧实现 background-atomic-before.log 退出一，报告缺失、任务／尝试／审计部分提交被复现。结果写入失败场景还触发重复重试后的同名触发器错误，测试已改为幂等安装触发器，不将该错误作为独立产品缺陷。
|
||||
- [x] 修复归档 background-atomic-fix-20261002.tar，SHA256 为 78a140e9f16f83ca3eeac7e86635afe6f199cc1351210b48589f9ece7acb1123，两端校验通过；background-atomic-after.log：应用 21.663 秒、仓储 10.407 秒、HTTP 29.936 秒，BACKGROUND_ATOMIC_AFTER_EXIT=0。
|
||||
- [x] 新测试覆盖五处写入故障整体回滚、每次重试完整报告、重试等待取消、禁止重试重新激活取消任务、无尝试收尾不能绕过活动尝试，以及无尝试路径的审计失败回滚。
|
||||
- [x] background-atomic-full.log 与 background-atomic-vet.log：全后端测试及 go vet 均通过，BACKGROUND_ATOMIC_FULL_EXIT=0、BACKGROUND_ATOMIC_VET_EXIT=0；既有三项服务 active，磁盘剩余约 961MB。外部数据库未启用时的全量测试不当作三库实库回归。
|
||||
- [x] background-atomic-repeat.log：新增收尾和取消专项重复二十次通过，应用 48.982 秒、仓储 49.985 秒，BACKGROUND_ATOMIC_REPEAT_EXIT=0；普通重复测试不替代竞争检测。
|
||||
- [!] 本批仅隔离源码开发及自动回归，不修改正式程序和线上数据库；未运行三库首次 Web 或正式部署，不代替生产验收。
|
||||
|
||||
## 2026年10月2日18:11 后台取消与 MySQL 报告迁移回归（未部署）
|
||||
|
||||
- [x] 新增原生 Go 回归 TestCertificateCancellationPreservesDeploymentOutcome；background-cancel-before.log 中四个场景全部失败，BACKGROUND_CANCEL_BEFORE_EXIT=1。三个场景被错误标为 cancelled，已恢复场景未保存逐服务器结果。
|
||||
|
||||
@ -1,5 +1,21 @@
|
||||
# Scert 整改任务执行清单
|
||||
|
||||
## 2026年10月2日18:35 后台收尾一致性执行入口
|
||||
|
||||
需求沿用已确认的 Linux 全清单，不处理 Windows，不缩减原始范围。先开发及自动回归，再统一正式部署；本批不清线上数据、不创建辅助验收脚本。
|
||||
|
||||
- [x] 修改前基线 cf3bc7a 已提交并推送；第一次推送认证失败，重试后成功。
|
||||
- [x] 已取回上一批后台取消修复的全后端测试与静态检查结果：BACKGROUND_FULL_EXIT=0、BACKGROUND_VET_EXIT=0；第二台 Scert、Nginx、MySQL 均 active，隔离 MySQL 已退出。
|
||||
- [!] 4e8d509 的受限竞争检测达到十分钟限制，Result=timeout、REPORT_BOUNDED_RACE_EXIT=1，不计通过。第一台任务已停止，无残留编译测试进程，既有服务保持 active。
|
||||
- [x] 新增后台完整报告与收尾故障回归；旧实现缺少重试报告，并在报告、结果、尝试或审计写入失败后出现部分保存，BACKGROUND_ATOMIC_BEFORE_EXIT=1。
|
||||
- [x] 将后台任务终态、尝试、完整报告、逐服务器结果和完成审计统一到事务；失败重试保存报告但任务保持 retry_wait，等待取消不覆盖历史尝试，保存失败不继续外部操作。
|
||||
- [x] 增加无活动尝试的失败／取消收尾；存在活动尝试时拒绝单独收尾。应用、仓储和 HTTP 原生 Linux 关联回归通过，BACKGROUND_ATOMIC_AFTER_EXIT=0。
|
||||
- [x] 本批全后端回归、静态检查通过，BACKGROUND_ATOMIC_FULL_EXIT=0、BACKGROUND_ATOMIC_VET_EXIT=0。五处故障注入验证完整回滚，开始审计失败不执行外部操作，完成审计失败保留中断现场。
|
||||
- [x] 新增收尾和取消专项重复二十次通过，BACKGROUND_ATOMIC_REPEAT_EXIT=0；应用 48.982 秒、仓储 49.985 秒。进度文档及接口说明同步更新，本批随代码提交保存。
|
||||
- [!] 其他长操作、升级 502、恢复证据、系统通知重置、三库首次 Web、备份恢复与最终验收仍未关闭。本批尚未部署。
|
||||
|
||||
下方 18:11 的回归运行态已由本节取回的终态替代；历史记录保留，不作为当前阻塞重复计算。
|
||||
|
||||
## 2026年10月2日18:11 当前执行入口
|
||||
|
||||
本节为当前状态，下方时间段保留历史过程。需求沿用已确认的 Linux 全清单：先完成开发和自动回归，再统一正式部署；不清线上数据库，不以测试脚本替代产品验收。
|
||||
|
||||
@ -1,6 +1,16 @@
|
||||
# Scert 整改、合并与优化清单
|
||||
|
||||
## 2026年10月2日18:11 最新开发进度入口
|
||||
## 2026年10月2日18:35 最新开发进度入口
|
||||
|
||||
- [x] 后台取消修复及 MySQL 报告迁移回归基线 cf3bc7a 已提交推送。上一批全后端及 go vet 已确认通过；不等于本批代码或生产验收通过。
|
||||
- [!] 第一台受限竞争检测十分钟超时，退出一，不记为通过；既有服务正常，未重新启动无限制测试。
|
||||
- [x] 新回归已复现后台完整报告缺失，以及分散收尾写入导致的状态不一致。
|
||||
- [x] 后台续期、部署、导入、恢复和校验共用执行器的事务收尾已实现，包含重试报告保留及等待期间取消；应用、仓储、HTTP 三包原生 Linux 回归通过。
|
||||
- [x] 本批全后端测试和 go vet 通过；报告、逐目标结果、审计、尝试与任务的故障注入全部保持原子回滚。失败审计统一使用 error，与 HTTP 契约一致。
|
||||
- [x] 收尾及取消专项重复二十次通过，代码、清单和接口说明同步提交保存；三库实库复核与正式部署仍待后续统一执行。
|
||||
- [!] 原始清单范围不变，P0-10、P0-11 与整体生产验收不关闭。具体执行项及证据见执行清单和 Linux 测试记录的本时段入口。
|
||||
|
||||
## 2026年10月2日18:11 历史开发进度入口
|
||||
|
||||
- [x] P0-10／P1-04 子项：HTTP 异步任务完整报告已持久化，终态、尝试、逐服务器结果与完成审计共用事务；前端按最新尝试读取验证和恢复状态，旧记录不伪造结论。源码 4e8d509 已推送。
|
||||
- [x] 最新仓储、HTTP、数据库关联回归通过；前端 256 项通过、生产构建通过。第一、第二轮失败与修复经过保留在 Linux 测试记录中。
|
||||
|
||||
Loading…
Reference in New Issue
Block a user