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-08-30 21:22:51 +08:00
parent 794734c1df
commit 8b1d1853e3
7 changed files with 307 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
backend/internal/httpapi/server.go
View File

@ -83,6 +83,8 @@ type Server struct {
resetStatePath string
audit repository.AuditRepository
tasks repository.TaskRepository
taskMu sync.Mutex
taskCancels map[string]context.CancelFunc
resetMu sync.Mutex
resetClosedHash string
resetLimiter *resetRateLimiter
@ -120,6 +122,7 @@ func New(deps Dependencies) *Server {
resetStatePath: deps.ResetStatePath,
audit: deps.Audit,
tasks: deps.Tasks,
taskCancels: make(map[string]context.CancelFunc),
resetLimiter: newResetRateLimiter(),
authSessions: make(map[string]authSession),
loginAttempts: make(map[string]loginAttempt),
@ -155,6 +158,7 @@ func (s *Server) registerRoutes() {
adminAPI.POST("/domains/:id/issue", s.issueDomain)
adminAPI.GET("/tasks", s.listTasks)
adminAPI.GET("/tasks/:id", s.getTask)
adminAPI.DELETE("/tasks/:id", s.cancelTask)
adminAPI.GET("/tasks/:id/attempts", s.listTaskAttempts)
adminAPI.GET("/audit-logs", s.listAuditLogs)
adminAPI.GET("/settings", s.getSettings)

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

@ -29,6 +29,24 @@ func (s *Server) issueDomainAsTask(c *gin.Context) {
s.writeError(c, http.StatusInternalServerError, "读取域名失败", err)
return
}
idempotencyKey := strings.TrimSpace(c.GetHeader("Idempotency-Key"))
if idempotencyKey != "" {
if !validTaskIdempotencyKey(idempotencyKey) {
s.writeError(c, http.StatusBadRequest, "幂等键格式无效", errors.New("幂等键长度或字符无效"))
return
}
if existing, err := s.tasks.FindTaskByIdempotencyKey(c.Request.Context(), domainID, idempotencyKey); err == nil {
status := http.StatusOK
if existing.Status == "queued" || existing.Status == "running" {
status = http.StatusConflict
}
c.JSON(status, gin.H{"task": existing, "replayed": true})
return
} else if !errors.Is(err, sql.ErrNoRows) {
s.writeError(c, http.StatusInternalServerError, "检查幂等任务失败", err)
return
}
}
if active, err := s.tasks.FindActiveTaskByDomain(c.Request.Context(), domainID, "certificate_issue"); err == nil {
c.JSON(http.StatusConflict, gin.H{"error": "该域名已有正在执行的证书任务", "task": active})
return
@ -38,28 +56,16 @@ func (s *Server) issueDomainAsTask(c *gin.Context) {
}
now := time.Now().UTC()
task := repository.Task{
ID: domain.NewID(),
DomainID: domainID,
TaskType: "certificate_issue",
Status: "queued",
Stage: "queued",
Message: "等待执行证书申请",
CreatedAt: now,
UpdatedAt: now,
ID: domain.NewID(), DomainID: domainID, TaskType: "certificate_issue", Status: "queued", Stage: "queued",
Message: "等待执行证书申请", IdempotencyKey: idempotencyKey, CreatedAt: now, UpdatedAt: now,
}
if err := s.tasks.CreateTask(c.Request.Context(), task); err != nil {
s.writeError(c, http.StatusInternalServerError, "创建证书任务失败", err)
return
}
attempt := repository.TaskAttempt{
ID: domain.NewID(),
TaskID: task.ID,
AttemptNumber: 1,
Stage: "certificate_issue",
Status: "running",
Message: "开始申请证书",
StartedAt: &now,
CreatedAt: now,
ID: domain.NewID(), TaskID: task.ID, AttemptNumber: 1, Stage: "certificate_issue", Status: "running",
Message: "开始申请证书", StartedAt: &now, CreatedAt: now,
}
if err := s.tasks.CreateAttempt(c.Request.Context(), attempt); err != nil {
_ = s.tasks.UpdateTask(context.Background(), task.ID, "failed", "certificate_issue", "创建任务尝试失败", &now, &now)
@ -70,15 +76,39 @@ func (s *Server) issueDomainAsTask(c *gin.Context) {
s.writeError(c, http.StatusInternalServerError, "更新任务状态失败", err)
return
}
certificateResult, deploymentReport, issueErr := s.issueCertificate(c.Request.Context(), domainID)
executionCtx, cancel := context.WithTimeout(c.Request.Context(), certificateTaskTimeout)
s.registerTaskCancel(task.ID, cancel)
defer func() {
cancel()
s.unregisterTaskCancel(task.ID)
}()
certificateResult, deploymentReport, issueErr := s.issueCertificate(executionCtx, domainID)
finished := time.Now().UTC()
status := "success"
message := "证书申请和部署完成"
cancelled := false
if issueErr != nil {
status = "failed"
message = issueErr.Error()
if errors.Is(issueErr, context.Canceled) {
status = "cancelled"
message = "证书任务已取消"
}
if errors.Is(issueErr, context.DeadlineExceeded) {
message = "证书任务执行超时"
}
}
if persisted, err := s.tasks.GetTask(context.Background(), task.ID); err == nil && persisted.Status == "cancelled" {
status = "cancelled"
message = persisted.Message
cancelled = true
if issueErr == nil {
issueErr = errors.New(message)
}
}
if !cancelled {
_ = s.tasks.UpdateTask(context.Background(), task.ID, status, "certificate_issue", message, &now, &finished)
}
_ = s.tasks.UpdateTask(context.Background(), task.ID, status, "certificate_issue", message, &now, &finished)
_ = s.tasks.FinishAttempt(context.Background(), attempt.ID, status, message, &finished)
task.Status = status
task.Stage = "certificate_issue"
@ -88,33 +118,101 @@ func (s *Server) issueDomainAsTask(c *gin.Context) {
task.UpdatedAt = finished
if s.audit != nil {
auditStatus := status
if issueErr != nil {
if issueErr != nil && status != "cancelled" {
auditStatus = "error"
}
_ = s.audit.CreateAuditLog(context.Background(), repository.AuditLog{
Action: "certificate_issue",
ResourceType: "domain",
ResourceID: domainID,
Status: auditStatus,
Message: message,
Action: "certificate_issue", ResourceType: "domain", ResourceID: domainID, Status: auditStatus, Message: message,
})
}
response := gin.H{
"task": task,
"certificate": certificateResult,
"deployment": deploymentReport,
"task_status": status,
"finished_at": finished,
"task": task, "certificate": certificateResult, "deployment": deploymentReport,
"task_status": status, "finished_at": finished,
}
if issueErr != nil {
response["error"] = issueErr.Error()
c.JSON(http.StatusBadGateway, response)
response["error"] = message
responseStatus := http.StatusBadGateway
if status == "cancelled" {
responseStatus = http.StatusConflict
} else if errors.Is(issueErr, context.DeadlineExceeded) {
responseStatus = http.StatusGatewayTimeout
}
c.JSON(responseStatus, response)
return
}
response["task"] = task
c.JSON(http.StatusOK, response)
}
const certificateTaskTimeout = 30 * time.Minute
func (s *Server) registerTaskCancel(taskID string, cancel context.CancelFunc) {
s.taskMu.Lock()
defer s.taskMu.Unlock()
if s.taskCancels == nil {
s.taskCancels = make(map[string]context.CancelFunc)
}
s.taskCancels[taskID] = cancel
}
func (s *Server) unregisterTaskCancel(taskID string) {
s.taskMu.Lock()
defer s.taskMu.Unlock()
delete(s.taskCancels, taskID)
}
func (s *Server) cancelTask(c *gin.Context) {
if s.tasks == nil {
s.writeError(c, http.StatusNotImplemented, "任务管理接口尚未配置", errors.New("任务仓储未配置"))
return
}
taskID := strings.TrimSpace(c.Param("id"))
item, err := s.tasks.GetTask(c.Request.Context(), taskID)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
s.writeError(c, http.StatusNotFound, "任务不存在", err)
return
}
s.writeError(c, http.StatusInternalServerError, "读取任务失败", err)
return
}
if item.Status != "queued" && item.Status != "running" {
c.JSON(http.StatusConflict, gin.H{"error": "任务当前无法取消", "task": item})
return
}
finished := time.Now().UTC()
if err := s.tasks.CancelTask(c.Request.Context(), taskID, "任务已由用户取消", &finished); err != nil {
s.writeError(c, http.StatusConflict, "取消任务失败", err)
return
}
s.taskMu.Lock()
cancel := s.taskCancels[taskID]
s.taskMu.Unlock()
if cancel != nil {
cancel()
}
item.Status = "cancelled"
item.Message = "任务已由用户取消"
item.FinishedAt = &finished
item.UpdatedAt = finished
if s.audit != nil {
_ = s.audit.CreateAuditLog(context.Background(), repository.AuditLog{
Action: "task_cancel", ResourceType: "task", ResourceID: taskID, Status: "success", Message: item.Message,
})
}
c.JSON(http.StatusOK, gin.H{"task": item})
}
func validTaskIdempotencyKey(value string) bool {
if len(value) < 8 || len(value) > 128 {
return false
}
for _, char := range value {
if (char < 'a' || char > 'z') && (char < 'A' || char > 'Z') && (char < '0' || char > '9') && char != '-' && char != '_' && char != '.' {
return false
}
}
return true
}
func (s *Server) listTasks(c *gin.Context) {
if s.tasks == nil {
s.writeError(c, http.StatusNotImplemented, "任务管理接口尚未配置", errors.New("任务仓储未配置"))

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

@ -8,6 +8,7 @@ import (
"net/http/httptest"
"strings"
"testing"
"time"
"git.pcux.cn/xuans/Scert/backend/internal/certificate"
"git.pcux.cn/xuans/Scert/backend/internal/database"
@ -149,3 +150,77 @@ func TestTaskAPIRejectsUnknownTask(t *testing.T) {
t.Fatalf("未知任务应返回 404：%d %s", response.Code, response.Body.String())
}
}
func TestTaskAPICanCancelRunningIssue(t *testing.T) {
started := make(chan struct{})
server, taskRepo, _, closeDB := newTaskAPITestServer(t, func(ctx context.Context, id string) (certificate.IssueResult, deployment.Report, error) {
close(started)
<-ctx.Done()
return certificate.IssueResult{DomainID: id}, deployment.Report{}, ctx.Err()
})
defer closeDB()
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 <-started:
case <-time.After(2 * time.Second):
t.Fatal("证书任务没有进入执行状态")
}
items, err := taskRepo.ListTasks(context.Background(), "", 10)
if err != nil || len(items) != 1 {
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 || !strings.Contains(cancelResponse.Body.String(), "cancelled") {
t.Fatalf("取消任务响应错误：%d %s", cancelResponse.Code, cancelResponse.Body.String())
}
select {
case <-issueDone:
case <-time.After(2 * time.Second):
t.Fatal("取消后证书任务没有结束")
}
if issueResponse.Code != http.StatusConflict {
t.Fatalf("取消任务的申请请求应返回 409：%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)
}
}
func TestTaskAPIReplaysIdempotentIssue(t *testing.T) {
calls := 0
server, taskRepo, _, closeDB := newTaskAPITestServer(t, func(_ context.Context, id string) (certificate.IssueResult, deployment.Report, error) {
calls++
return certificate.IssueResult{DomainID: id}, deployment.Report{Status: "success"}, nil
})
defer closeDB()
firstRequest := httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)
firstRequest.Header.Set("Idempotency-Key", "issue-key-20260830")
firstResponse := httptest.NewRecorder()
serveAdminHTTP(server, firstResponse, firstRequest)
if firstResponse.Code != http.StatusOK {
t.Fatalf("首次幂等请求失败：%d %s", firstResponse.Code, firstResponse.Body.String())
}
secondRequest := httptest.NewRequest(http.MethodPost, "/api/v1/domains/domain-1/issue", nil)
secondRequest.Header.Set("Idempotency-Key", "issue-key-20260830")
secondResponse := httptest.NewRecorder()
serveAdminHTTP(server, secondResponse, secondRequest)
if secondResponse.Code != http.StatusOK || !strings.Contains(secondResponse.Body.String(), "replayed") {
t.Fatalf("重复幂等请求响应错误：%d %s", secondResponse.Code, secondResponse.Body.String())
}
if calls != 1 {
t.Fatalf("幂等请求重复执行了证书申请：%d", calls)
}
items, err := taskRepo.ListTasks(context.Background(), "", 10)
if err != nil || len(items) != 1 || items[0].IdempotencyKey != "issue-key-20260830" {
t.Fatalf("幂等键没有正确保存：%+v %v", items, err)
}
}

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

@ -44,7 +44,9 @@ type TaskRepository interface {
GetTask(context.Context, string) (Task, error)
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
FindActiveTaskByDomain(context.Context, string, string) (Task, error)
FindTaskByIdempotencyKey(context.Context, string, string) (Task, error)
CreateAttempt(context.Context, TaskAttempt) error
FinishAttempt(context.Context, string, string, string, *time.Time) error
ListAttempts(context.Context, string) ([]TaskAttempt, error)
@ -157,6 +159,34 @@ func (r *SQLTaskRepository) UpdateTask(ctx context.Context, id, status, stage, m
return nil
}
func (r *SQLTaskRepository) CancelTask(ctx context.Context, id, message string, finishedAt *time.Time) error {
if r == nil || r.db == nil {
return errors.New("数据库连接不能为空")
}
id = strings.TrimSpace(id)
if id == "" {
return errors.New("任务 ID 不能为空")
}
result, err := r.db.ExecContext(ctx, `UPDATE tasks SET status = 'cancelled', message = ?, finished_at = ?, updated_at = ? WHERE id = ? AND status IN ('queued', 'running')`, message, nullableTime(finishedAt), time.Now().UTC(), id)
if err != nil {
return fmt.Errorf("取消任务失败：%w", err)
}
count, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("读取任务取消结果失败：%w", err)
}
if count == 0 {
item, getErr := r.GetTask(ctx, id)
if getErr != nil {
return getErr
}
if item.Status == "cancelled" {
return nil
}
return errors.New("任务当前无法取消")
}
return nil
}
func (r *SQLTaskRepository) FindActiveTaskByDomain(ctx context.Context, domainID, taskType string) (Task, error) {
if r == nil || r.db == nil {
return Task{}, errors.New("数据库连接不能为空")
@ -169,6 +199,18 @@ func (r *SQLTaskRepository) FindActiveTaskByDomain(ctx context.Context, domainID
return scanTask(r.db.QueryRowContext(ctx, taskSelect+` WHERE domain_id = ? AND task_type = ? AND status IN ('queued', 'running') ORDER BY created_at DESC LIMIT 1`, domainID, taskType))
}
func (r *SQLTaskRepository) FindTaskByIdempotencyKey(ctx context.Context, domainID, key string) (Task, error) {
if r == nil || r.db == nil {
return Task{}, errors.New("数据库连接不能为空")
}
domainID = strings.TrimSpace(domainID)
key = strings.TrimSpace(key)
if domainID == "" || key == "" {
return Task{}, errors.New("域名 ID 和幂等键不能为空")
}
return scanTask(r.db.QueryRowContext(ctx, taskSelect+` WHERE domain_id = ? AND idempotency_key = ? ORDER BY created_at DESC LIMIT 1`, domainID, key))
}
func (r *SQLTaskRepository) CreateAttempt(ctx context.Context, item TaskAttempt) error {
if r == nil || r.db == nil {
return errors.New("数据库连接不能为空")

8
frontend/src/views/Dashboard/Dashboard.css
View File

@ -183,7 +183,13 @@ h2 {
border-radius: 11px;
background: #fbfcfe;
}
.record-row > div {
.record-row-actions {
display: flex;
align-items: center;
justify-content: flex-end;
gap: 10px;
flex-wrap: wrap;
}.record-row > div {
display: grid;
gap: 5px;
min-width: 0;

41
frontend/src/views/Dashboard/Dashboard.js
View File

@ -92,7 +92,13 @@ export function useDashboard() {
wildcard: "泛域名",
san: "SAN 多域名",
};
const domainStatusLabels = {
const taskStatusLabels = {
queued: "排队中",
running: "执行中",
success: "已完成",
failed: "失败",
cancelled: "已取消",
}; const domainStatusLabels = {
pending: "等待申请",
active: "正常",
error: "错误",
@ -484,9 +490,12 @@ export function useDashboard() {
error.value = "";
notice.value = "";
try {
const response = await fetchAdmin(`/api/v1/domains/${item.id}/${action}`, {
method: "POST",
});
const options = { method: "POST" };
if (action === "issue") {
const suffix = crypto.randomUUID ? crypto.randomUUID() : createSSHKeyRandomSuffix();
options.headers = { "Idempotency-Key": `issue-${item.id}-${suffix}` };
}
const response = await fetchAdmin(`/api/v1/domains/${item.id}/${action}`, options);
const result = await response.json().catch(() => ({}));
if (!response.ok) throw new Error(result.error || "域名操作失败");
notice.value = successMessage;
@ -495,7 +504,6 @@ export function useDashboard() {
error.value = err.message || "域名操作失败";
}
}
function issueDomain(item) {
return runDomainAction(item, "issue", `已提交 ${item.name} 的证书申请`);
}
@ -508,6 +516,20 @@ export function useDashboard() {
return runDomainAction(item, "deploy", `已提交 ${item.name} 的证书部署`);
}
async function cancelTask(item) {
if (!window.confirm("确定取消这项正在执行的证书任务吗？")) return;
error.value = "";
notice.value = "";
try {
const response = await fetchAdmin(`/api/v1/tasks/${item.id}`, { method: "DELETE" });
const result = await response.json().catch(() => ({}));
if (!response.ok) throw new Error(result.error || "取消任务失败");
notice.value = "证书任务已取消";
await loadData();
} catch (err) {
error.value = err.message || "取消任务失败";
}
}
async function runRenewalNow() {
error.value = "";
notice.value = "";
@ -821,6 +843,7 @@ export function useDashboard() {
sshKeyFormError,
certificateTypeLabels,
domainStatusLabels,
taskStatusLabels,
httpsModeLabels,
environmentLabels,
webServerLabels,
@ -870,6 +893,14 @@ export function useDashboard() {
deleteServer,
saveDomain,
deleteDomain,
issueDomain,
validateDomain,
deployDomain,
cancelTask,
runRenewalNow,
openSettings,
closeSettings,
saveSettings,
loadData,
displayLabel,
formatDate,

14
frontend/src/views/Dashboard/Dashboard.vue
View File

@ -30,6 +30,7 @@ const {
sshKeyFormError,
certificateTypeLabels,
domainStatusLabels,
taskStatusLabels,
httpsModeLabels,
environmentLabels,
webServerLabels,
@ -78,6 +79,7 @@ const {
validateDomain,
deployDomain,
runRenewalNow,
cancelTask,
displayLabel,
formatDate,
daysRemaining,
@ -199,7 +201,17 @@ const {
<strong>{{ item.task_type }}</strong>
<span class="muted">{{ item.message || "无附加信息" }}</span>
</div>
<span class="tag" :class="statusClass(item.status)">{{ item.status }}</span>
<div class="record-row-actions">
<span class="tag" :class="statusClass(item.status)">{{ displayLabel(taskStatusLabels, item.status) }}</span>
<button
v-if="item.status === 'queued' || item.status === 'running'"
class="text-button"
type="button"
@click="cancelTask(item)"
>
取消任务
</button>
</div>
</article>
</div>
</div>