125 lines
4.7 KiB
Go
125 lines
4.7 KiB
Go
// Package retry 以冻结策略和持久化预算决定下一状态，不执行任务或维护内存定时器。
|
||
package retry
|
||
|
||
import (
|
||
"crypto/sha256"
|
||
"encoding/binary"
|
||
"errors"
|
||
"time"
|
||
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||
)
|
||
|
||
// Input 描述本次尝试结束时数据库已锁定的事实，不能从客户端覆盖预算和时间。
|
||
type Input struct {
|
||
Policy pulse.RetryPolicy // Policy 受理时冻结的有限策略。
|
||
Safety pulse.RecoverySafety // Safety 仅 IDEMPOTENT 允许未知效果自动恢复。
|
||
Consumed uint32 // Consumed 已领取次数，包含本次及未获批开始的尝试。
|
||
Outcome string // Outcome SUCCEEDED、FAILED、UNKNOWN、CANCELLED 或 NOT_STARTED。
|
||
Retryable bool // Retryable 仅对明确失败有效，Worker 声明可安全重试。
|
||
Uncertain bool // Uncertain 之前仍未核实的外部效果，失败或超时不能抹掉。
|
||
CancelRequested bool // CancelRequested 已保存取消意图时禁止后续尝试。
|
||
Now time.Time // Now 完成资源锁定后读取的数据库时间。
|
||
Deadline time.Time // Deadline 任务总截止时间。
|
||
StartDeadline *time.Time // StartDeadline 所有新尝试的最晚开始时间，可为空。
|
||
AttemptID core.ID // AttemptID 持久化尝试身份，用作可重现抖动的种子。
|
||
}
|
||
|
||
// Decision 必须与尝试结果、许可释放、事件及操作回执在同一事务提交。
|
||
type Decision struct {
|
||
State pulse.TaskState // State 下一任务状态，WAITING 不是业务终态。
|
||
Uncertain bool // Uncertain 跨尝试保留的不确定性；权威成功才自动清除。
|
||
NextAt *time.Time // NextAt 仅 WAITING 有值，写入 available_at。
|
||
Reason string // Reason 稳定决策分类，用于事件和审计。
|
||
DelayMS uint64 // DelayMS 实际等待毫秒数，非重试时为零。
|
||
}
|
||
|
||
func Decide(in Input) (Decision, error) {
|
||
out := Decision{State: pulse.TaskFailed, Uncertain: in.Uncertain, Reason: "NON_RETRYABLE_FAILURE"}
|
||
if err := in.Policy.Validate(); err != nil {
|
||
return out, err
|
||
}
|
||
if in.Consumed == 0 || in.Consumed > in.Policy.MaxAttempts || in.AttemptID.Validate() != nil || in.Now.IsZero() || in.Deadline.IsZero() {
|
||
return out, errors.New("invalid retry facts")
|
||
}
|
||
canRetry := false
|
||
switch in.Outcome {
|
||
case "SUCCEEDED":
|
||
return Decision{State: pulse.TaskSucceeded, Reason: "SUCCEEDED"}, nil
|
||
case "CANCELLED":
|
||
if !in.CancelRequested {
|
||
return out, errors.New("cancelled result requires persisted intent")
|
||
}
|
||
out.State, out.Reason = pulse.TaskCancelled, "CANCEL_CONFIRMED"
|
||
if in.Uncertain {
|
||
out.State, out.Reason = pulse.TaskNeedsReview, "RECONCILIATION_REQUIRED"
|
||
}
|
||
return out, nil
|
||
case "FAILED":
|
||
canRetry = in.Retryable
|
||
if canRetry {
|
||
out.Reason = "RETRYABLE_FAILURE"
|
||
}
|
||
case "UNKNOWN":
|
||
out.Uncertain = true
|
||
canRetry = in.Safety.CanRecoverAutomatically()
|
||
out.Reason = "RECONCILIATION_REQUIRED"
|
||
if canRetry {
|
||
out.Reason = "IDEMPOTENT_UNKNOWN"
|
||
}
|
||
case "NOT_STARTED":
|
||
canRetry = true
|
||
out.Reason = "UNSTARTED_LEASE_EXPIRED"
|
||
default:
|
||
return out, errors.New("invalid retry outcome")
|
||
}
|
||
if out.Uncertain {
|
||
out.State = pulse.TaskNeedsReview
|
||
if !in.Safety.CanRecoverAutomatically() {
|
||
canRetry = false
|
||
out.Reason = "RECONCILIATION_REQUIRED"
|
||
}
|
||
}
|
||
if in.CancelRequested {
|
||
out.Reason = "CANCEL_REQUESTED"
|
||
return out, nil
|
||
}
|
||
if !canRetry {
|
||
return out, nil
|
||
}
|
||
if in.Consumed >= in.Policy.MaxAttempts {
|
||
out.Reason = "ATTEMPT_LIMIT"
|
||
return out, nil
|
||
}
|
||
delay := DelayMS(in.Policy, in.Consumed, in.AttemptID)
|
||
next := in.Now.Add(time.Duration(delay) * time.Millisecond)
|
||
if !next.Before(in.Deadline) || (in.StartDeadline != nil && !next.Before(*in.StartDeadline)) {
|
||
out.Reason = "DEADLINE_EXCEEDED"
|
||
return out, nil
|
||
}
|
||
out.State = pulse.TaskWaiting
|
||
out.DelayMS = delay
|
||
out.NextAt = &next
|
||
return out, nil
|
||
}
|
||
|
||
// DelayMS 使用有上限的指数退避和 [ceil(base/2),base] 确定性抖动；策略须先 Validate。
|
||
// 相同尝试重复计算不改变结果，倍增先截断，避免次数或 Duration 溢出。
|
||
func DelayMS(policy pulse.RetryPolicy, consumed uint32, attempt core.ID) uint64 {
|
||
base := policy.InitialDelayMS
|
||
for n := uint32(1); n < consumed && base < policy.MaxDelayMS; n++ {
|
||
if base > policy.MaxDelayMS/2 {
|
||
base = policy.MaxDelayMS
|
||
break
|
||
}
|
||
base *= 2
|
||
}
|
||
if base > policy.MaxDelayMS {
|
||
base = policy.MaxDelayMS
|
||
}
|
||
lower := (base + 1) / 2
|
||
sum := sha256.Sum256([]byte("orbit.retry.v1:" + string(attempt)))
|
||
return lower + binary.BigEndian.Uint64(sum[:8])%(base-lower+1)
|
||
}
|