失败作业的幂等键把恢复入口锁死,运维修好故障也点不动第二次。 现在按窗口重排失败作业(同窗口有活作业则拒绝并发),CLI/API 都能真恢复。
This commit is contained in:
@@ -390,6 +390,32 @@ func libraryMaintenanceRecoverHandler(w http.ResponseWriter, r *http.Request, co
|
||||
"error": "恢复作业投递失败(或已有同窗口的恢复作业)", "detail": errString(err)})
|
||||
return
|
||||
}
|
||||
// 上一次恢复失败(比如故障还没修完)不该让这个窗口永久卡死:
|
||||
// 运维修好之后可以再点一次,作业会被重新排队;但同一窗口不许并发两份。
|
||||
if queueJob.State == jobs.StateFailed || queueJob.State == jobs.StateCancelled {
|
||||
retrier, ok := config.Store.(interface {
|
||||
RetryLibraryRecovery(ctx context.Context, id, windowJobID string, now time.Time) (bool, error)
|
||||
})
|
||||
if !ok {
|
||||
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_recovery_stuck",
|
||||
"error": "上一次恢复失败,且当前存储不支持重新排队"})
|
||||
return
|
||||
}
|
||||
retried, retryErr := retrier.RetryLibraryRecovery(r.Context(), queueJob.ID, held.OwnerJobID, now)
|
||||
if retryErr != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, map[string]any{"code": "library_recovery_retry_failed",
|
||||
"error": "重新排队恢复作业失败", "detail": errString(retryErr)})
|
||||
return
|
||||
}
|
||||
if !retried {
|
||||
writeJSON(w, http.StatusConflict, map[string]any{"code": "library_recovery_busy",
|
||||
"error": "同一窗口已有恢复作业在跑", "job": queueJob})
|
||||
return
|
||||
}
|
||||
queueJob.State = jobs.StatePending
|
||||
queueJob.Attempt = 0
|
||||
queueJob.LastError = ""
|
||||
}
|
||||
writeJSON(w, http.StatusAccepted, map[string]any{"job": queueJob, "window_job_id": held.OwnerJobID,
|
||||
"scope": held.OwnerScope, "instance": held.Instance})
|
||||
}
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
// retryLibraryRecoverySQL 把失败的恢复作业放回待领取,但只在同一窗口没有其它
|
||||
// 存活恢复作业时允许:恢复本身是破坏性操作,不能并发跑两份。
|
||||
const retryLibraryRecoverySQL = `
|
||||
WITH candidate AS (
|
||||
SELECT id FROM jobs
|
||||
WHERE id = $1 AND type = 'library.recover' AND state = 'failed'
|
||||
AND payload->>'window_job_id' = $2
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM jobs live
|
||||
WHERE live.type = 'library.recover'
|
||||
AND live.payload->>'window_job_id' = $2
|
||||
AND live.state IN ('pending', 'running', 'retry')
|
||||
)
|
||||
FOR UPDATE
|
||||
), retried AS (
|
||||
UPDATE jobs
|
||||
SET state = 'pending', attempt = 0, available_at = $3, lease_owner = '', lease_until = NULL,
|
||||
last_error = '', last_error_code = '', updated_at = $3
|
||||
FROM candidate c
|
||||
WHERE jobs.id = c.id
|
||||
RETURNING jobs.id
|
||||
) SELECT count(*) FROM retried`
|
||||
|
||||
// RetryLibraryRecovery 让"恢复失败后修好故障再点一次"成为真实路径:
|
||||
// 失败作业会被重新排队,而不是让同一个幂等键把入口永久卡死。
|
||||
func (s *Store) RetryLibraryRecovery(ctx context.Context, id, windowJobID string, now time.Time) (bool, error) {
|
||||
if id == "" || windowJobID == "" || now.IsZero() {
|
||||
return false, nil
|
||||
}
|
||||
var count int
|
||||
if err := s.db.QueryRow(ctx, retryLibraryRecoverySQL, id, windowJobID, now).Scan(&count); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return count == 1, nil
|
||||
}
|
||||
Reference in New Issue
Block a user