diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index ff08747..0b4f3b6 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -87,6 +87,13 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string } } taskRepo := repository.NewSQLTaskRepository(db.DB) + if recovered, err := taskRepo.RecoverInterruptedTasks(ctx, time.Now().UTC()); err != nil { + _ = db.Close() + _ = logger.Close() + return nil, fmt.Errorf("\u6062\u590d\u670d\u52a1\u91cd\u542f\u524d\u672a\u5b8c\u6210\u4efb\u52a1\u5931\u8d25\uff1a%w", err) + } else if recovered > 0 { + logger.Warn("\u670d\u52a1\u91cd\u542f\u524d\u672a\u5b8c\u6210\u4efb\u52a1\u5df2\u6807\u8bb0\u4e3a\u5931\u8d25", "count", recovered) + } auditRepo := repository.NewSQLAuditRepository(db.DB) sshKeys := sshkey.NewService(secretStore) accounts := certificate.NewAccountRepository(secretStore) diff --git a/backend/internal/repository/task_repository.go b/backend/internal/repository/task_repository.go index 03b2139..f51ad44 100644 --- a/backend/internal/repository/task_repository.go +++ b/backend/internal/repository/task_repository.go @@ -45,6 +45,7 @@ type TaskRepository interface { ListTasks(context.Context, string, int) ([]Task, error) UpdateTask(context.Context, string, string, string, string, *time.Time, *time.Time) error CancelTask(context.Context, string, string, *time.Time) error + RecoverInterruptedTasks(context.Context, time.Time) (int, error) FindActiveTaskByDomain(context.Context, string, string) (Task, error) FindTaskByIdempotencyKey(context.Context, string, string) (Task, error) CreateAttempt(context.Context, TaskAttempt) error @@ -208,6 +209,48 @@ func (r *SQLTaskRepository) CancelTask(ctx context.Context, id, message string, } return nil } + +// RecoverInterruptedTasks 将服务重启前未完成的任务标记为失败,避免任务永久停留在执行状态。 +func (r *SQLTaskRepository) RecoverInterruptedTasks(ctx context.Context, finishedAt time.Time) (int, error) { + if r == nil || r.db == nil { + return 0, errors.New("\u6570\u636e\u5e93\u8fde\u63a5\u4e0d\u80fd\u4e3a\u7a7a") + } + if finishedAt.IsZero() { + finishedAt = time.Now().UTC() + } else { + finishedAt = finishedAt.UTC() + } + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("\u5f00\u59cb\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u4e8b\u52a1\u5931\u8d25\uff1a%w", err) + } + rollback := func() { + _ = tx.Rollback() + } + result, err := tx.ExecContext(ctx, `UPDATE tasks + SET status = 'failed', stage = 'service_restart', message = ?, finished_at = ?, updated_at = ? + WHERE status IN ('queued', 'running')`, "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt, finishedAt) + if err != nil { + rollback() + return 0, fmt.Errorf("\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u5931\u8d25\uff1a%w", err) + } + count, err := result.RowsAffected() + if err != nil { + rollback() + return 0, fmt.Errorf("\u8bfb\u53d6\u4e2d\u65ad\u4efb\u52a1\u6570\u91cf\u5931\u8d25\uff1a%w", err) + } + if _, err := tx.ExecContext(ctx, `UPDATE task_attempts + SET status = 'failed', message = ?, finished_at = ? + WHERE status IN ('queued', 'running')`, "\u670d\u52a1\u91cd\u542f\u5bfc\u81f4\u4efb\u52a1\u4e2d\u65ad\uff0c\u8bf7\u91cd\u65b0\u6267\u884c", finishedAt); err != nil { + rollback() + return 0, fmt.Errorf("\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u5c1d\u8bd5\u5931\u8d25\uff1a%w", err) + } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("\u63d0\u4ea4\u4e2d\u65ad\u4efb\u52a1\u6062\u590d\u4e8b\u52a1\u5931\u8d25\uff1a%w", err) + } + return int(count), nil +} + func (r *SQLTaskRepository) FindActiveTaskByDomain(ctx context.Context, domainID, taskType string) (Task, error) { if r == nil || r.db == nil { return Task{}, errors.New("数据库连接不能为空") diff --git a/backend/internal/repository/task_repository_test.go b/backend/internal/repository/task_repository_test.go index 4430655..7e74f17 100644 --- a/backend/internal/repository/task_repository_test.go +++ b/backend/internal/repository/task_repository_test.go @@ -153,3 +153,52 @@ func TestSQLTaskRepositoryDoesNotOverwriteCancelledTask(t *testing.T) { t.Fatalf("已取消任务状态被覆盖:%+v %v", task, err) } } + +func TestSQLTaskRepositoryRecoversInterruptedTasksAndAttempts(t *testing.T) { + repository, db := newTaskRepositoryTest(t) + defer db.Close() + ctx := context.Background() + created := time.Date(2026, 8, 30, 9, 0, 0, 0, time.UTC) + for _, item := range []Task{ + {ID: "task-queued", DomainID: "domain-1", TaskType: "certificate_issue", Status: "queued", Stage: "queued", Message: "\u7b49\u5f85\u6267\u884c", CreatedAt: created, UpdatedAt: created}, + {ID: "task-running", DomainID: "domain-2", TaskType: "certificate_issue", Status: "running", Stage: "certificate_issue", Message: "\u6b63\u5728\u6267\u884c", CreatedAt: created, UpdatedAt: created}, + {ID: "task-success", DomainID: "domain-3", TaskType: "certificate_issue", Status: "success", Stage: "certificate_issue", Message: "\u5df2\u5b8c\u6210", CreatedAt: created, UpdatedAt: created}, + } { + if err := repository.CreateTask(ctx, item); err != nil { + t.Fatalf("\u521b\u5efa\u6062\u590d\u6d4b\u8bd5\u4efb\u52a1\u5931\u8d25\uff1a%v", err) + } + } + if err := repository.CreateAttempt(ctx, TaskAttempt{ + ID: "attempt-running", TaskID: "task-running", AttemptNumber: 1, Stage: "certificate_issue", Status: "running", Message: "\u6b63\u5728\u6267\u884c", CreatedAt: created, + }); err != nil { + t.Fatalf("\u521b\u5efa\u6062\u590d\u6d4b\u8bd5\u5c1d\u8bd5\u5931\u8d25\uff1a%v", err) + } + if err := repository.CreateAttempt(ctx, TaskAttempt{ + ID: "attempt-success", TaskID: "task-success", AttemptNumber: 1, Stage: "certificate_issue", Status: "success", Message: "\u5df2\u5b8c\u6210", CreatedAt: created, + }); err != nil { + t.Fatalf("\u521b\u5efa\u5df2\u5b8c\u6210\u6d4b\u8bd5\u5c1d\u8bd5\u5931\u8d25\uff1a%v", err) + } + finished := time.Date(2026, 8, 30, 10, 0, 0, 0, time.UTC) + count, err := repository.RecoverInterruptedTasks(ctx, finished) + if err != nil || count != 2 { + t.Fatalf("\u6062\u590d\u4e2d\u65ad\u4efb\u52a1\u6570\u91cf\u9519\u8bef\uff1a%d %v", count, err) + } + for _, id := range []string{"task-queued", "task-running"} { + item, getErr := repository.GetTask(ctx, id) + if getErr != nil || item.Status != "failed" || item.Stage != "service_restart" || item.FinishedAt == nil || !item.FinishedAt.Equal(finished) { + t.Fatalf("\u4e2d\u65ad\u4efb\u52a1\u6062\u590d\u7ed3\u679c\u9519\u8bef\uff1a%+v %v", item, getErr) + } + } + completed, err := repository.GetTask(ctx, "task-success") + if err != nil || completed.Status != "success" { + t.Fatalf("\u5df2\u5b8c\u6210\u4efb\u52a1\u4e0d\u5e94\u88ab\u6062\u590d\u903b\u8f91\u4fee\u6539\uff1a%+v %v", completed, err) + } + attempts, err := repository.ListAttempts(ctx, "task-running") + if err != nil || len(attempts) != 1 || attempts[0].Status != "failed" || attempts[0].FinishedAt == nil || !attempts[0].FinishedAt.Equal(finished) { + t.Fatalf("\u8fd0\u884c\u4e2d\u4efb\u52a1\u5c1d\u8bd5\u6062\u590d\u7ed3\u679c\u9519\u8bef\uff1a%+v %v", attempts, err) + } + completedAttempts, err := repository.ListAttempts(ctx, "task-success") + if err != nil || len(completedAttempts) != 1 || completedAttempts[0].Status != "success" { + t.Fatalf("\u5df2\u5b8c\u6210\u4efb\u52a1\u5c1d\u8bd5\u4e0d\u5e94\u88ab\u6062\u590d\u903b\u8f91\u4fee\u6539\uff1a%+v %v", completedAttempts, err) + } +} diff --git a/desktop/frontend/dist/index.html b/desktop/frontend/dist/index.html index ea6053c..e73311e 100644 --- a/desktop/frontend/dist/index.html +++ b/desktop/frontend/dist/index.html @@ -5,7 +5,7 @@