170 lines
7.4 KiB
Go
170 lines
7.4 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
"time"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||
"git.yuntuo.hk/k2safe/orbit/pulse/retry"
|
||
"git.yuntuo.hk/k2safe/orbit/pulse/worker"
|
||
)
|
||
|
||
func (s *ExecutionStore) Finish(ctx context.Context, token auth.Token, b worker.Binding, message core.ID, in worker.FinishInput) (worker.CommandResult, error) {
|
||
var out worker.CommandResult
|
||
digest, err := requestDigest(struct {
|
||
Identity any
|
||
Outcome string
|
||
Result core.JSON
|
||
ErrorCode string
|
||
Retryable bool `json:"Retryable,omitempty"`
|
||
}{attemptIdentity(in.Ref), in.Outcome, core.JSON(in.Result), in.ErrorCode, in.Retryable})
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
err = s.transaction(ctx, token, "workers:consume", func(tx *sql.Tx, p auth.Permissions) error {
|
||
rows, err := s.lockExecution(ctx, tx, token, b, p, in.Ref, false)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
cached, err := readReceipt(ctx, tx, token.NamespaceID, in.Ref.AttemptID, message, "finish", digest)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
now, err := dbNow(ctx, tx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
status := leaseStatus(rows, b, now)
|
||
if cached != nil {
|
||
out = worker.CommandResult{Receipt: *cached, Lease: status, Deduplicated: true}
|
||
return nil
|
||
}
|
||
task, a := rows.task, rows.attempt
|
||
receipt := worker.Receipt{Operation: "finish", TaskState: task.view.State, AttemptState: a.State, TaskRevision: task.view.Revision, LeaseUntil: a.LeaseUntil, Disposition: "APPLIED"}
|
||
if a.ResultDigest != nil && *a.ResultDigest == digest {
|
||
// 新消息 ID 重报已保存的同一结果，也不能重复事件、许可释放或状态推进。
|
||
// 重试可能已推进其他尝试，必须返回原操作回执，不能套用当前任务状态。
|
||
var raw []byte
|
||
if err := tx.QueryRowContext(ctx, `SELECT response FROM pulse_attempt_receipts WHERE namespace_id=? AND attempt_id=? AND operation='finish' AND request_digest=? AND JSON_UNQUOTE(JSON_EXTRACT(response,'$.disposition'))='APPLIED' LIMIT 1`, token.NamespaceID, a.ID, digest).Scan(&raw); err != nil {
|
||
return err
|
||
}
|
||
if err := json.Unmarshal(raw, &receipt); err != nil {
|
||
return err
|
||
}
|
||
if err := saveReceipt(ctx, tx, token.NamespaceID, a.ID, message, "finish", digest, receipt); err != nil {
|
||
return err
|
||
}
|
||
out = worker.CommandResult{Receipt: receipt, Lease: status, Deduplicated: true}
|
||
return nil
|
||
}
|
||
if in.Outcome == "CANCELLED" && task.view.CancelRequestedAt == nil {
|
||
return fault.Conflict
|
||
}
|
||
reason := ""
|
||
if a.ResultDigest != nil {
|
||
reason = "RESULT_CONFLICT"
|
||
} else if a.AuthorizedAt == nil {
|
||
reason = "NOT_STARTED"
|
||
} else if !status.Active || a.State != pulse.AttemptRunning {
|
||
reason = "LEASE_LOST"
|
||
}
|
||
if reason != "" {
|
||
if err := saveEvidence(ctx, tx, token.NamespaceID, message, in, digest, reason); err != nil {
|
||
return err
|
||
}
|
||
receipt.Disposition = "EVIDENCE_ONLY"
|
||
if err := saveReceipt(ctx, tx, token.NamespaceID, a.ID, message, "finish", digest, receipt); err != nil {
|
||
return err
|
||
}
|
||
out = worker.CommandResult{Receipt: receipt, Lease: status}
|
||
return nil
|
||
}
|
||
attemptState := pulse.AttemptState(in.Outcome)
|
||
decision, err := retry.Decide(retryFacts(task, a.ID, now, in.Outcome, in.Retryable))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
taskState := decision.State
|
||
var completed *time.Time
|
||
if taskState.Terminal() {
|
||
completed = &now
|
||
}
|
||
available := task.view.AvailableAt
|
||
var result any = string(in.Result)
|
||
if decision.NextAt != nil {
|
||
available = *decision.NextAt
|
||
result = nil
|
||
}
|
||
changed, err := tx.ExecContext(ctx, `UPDATE pulse_attempts SET state=?,ended_at=?,error_code=NULLIF(?,''),retryable=?,result=?,result_digest=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, attemptState, now, in.ErrorCode, in.Retryable, string(in.Result), digest, now, token.NamespaceID, a.ID, a.Revision)
|
||
if err := exactlyOne(changed, err); err != nil {
|
||
return err
|
||
}
|
||
changed, err = tx.ExecContext(ctx, `UPDATE pulse_tasks SET state=?,result=?,effect_uncertain=?,available_at=?,lease_until=NULL,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, taskState, result, decision.Uncertain, available, completed, now, token.NamespaceID, in.Ref.TaskID, task.view.Revision)
|
||
if err := exactlyOne(changed, err); err != nil {
|
||
return err
|
||
}
|
||
if err := releasePermits(ctx, tx, token.NamespaceID, a.ID, now); err != nil {
|
||
return err
|
||
}
|
||
if taskState.Terminal() {
|
||
if err := releaseQueueTask(ctx, tx, token.NamespaceID, task.view.QueueID); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
event := map[string]any{"attempt_id": a.ID, "epoch": fmt.Sprint(a.Epoch), "state": taskState, "outcome": in.Outcome, "error_code": in.ErrorCode, "retryable": in.Retryable, "effect_uncertain": decision.Uncertain, "decision": decision.Reason, "next_available_at": decision.NextAt, "delay_ms": decision.DelayMS, "attempt_count": task.view.AttemptCount, "max_attempts": task.maxAttempts}
|
||
if err := emitEvent(ctx, tx, token.NamespaceID, in.Ref.TaskID, task.view.Revision+1, "task", resultEvent(taskState), event); err != nil {
|
||
return err
|
||
}
|
||
if err := appendAudit(ctx, tx, token, message, in.Ref.TaskID, "task", "task.result", "worker result", map[string]any{"state": task.view.State}, event); err != nil {
|
||
return err
|
||
}
|
||
receipt.TaskState = taskState
|
||
receipt.AttemptState = attemptState
|
||
receipt.TaskRevision++
|
||
if err := saveReceipt(ctx, tx, token.NamespaceID, a.ID, message, "finish", digest, receipt); err != nil {
|
||
return err
|
||
}
|
||
out = worker.CommandResult{Receipt: receipt, Lease: worker.LeaseStatus{State: attemptState, LeaseUntil: a.LeaseUntil, ServerTime: now}}
|
||
return nil
|
||
})
|
||
return out, err
|
||
}
|
||
func resultEvent(state pulse.TaskState) string {
|
||
switch state {
|
||
case pulse.TaskSucceeded:
|
||
return "task.succeeded"
|
||
case pulse.TaskFailed:
|
||
return "task.failed"
|
||
case pulse.TaskNeedsReview:
|
||
return "task.needs_review"
|
||
case pulse.TaskWaiting:
|
||
return "task.requeued"
|
||
default:
|
||
return "task.cancelled"
|
||
}
|
||
}
|
||
func saveEvidence(ctx context.Context, tx *sql.Tx, ns, message core.ID, in worker.FinishInput, digest core.Digest, reason string) error {
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx, `INSERT INTO pulse_attempt_evidence(namespace_id,id,attempt_id,message_id,outcome,error_code,retryable,payload,request_digest,reason) VALUES (?,?,?,?,?,NULLIF(?,''),?,?,?,?)`, ns, id, in.Ref.AttemptID, message, in.Outcome, in.ErrorCode, in.Retryable, string(in.Result), digest, reason)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return emitEvent(ctx, tx, ns, id, 1, "attempt_evidence", "attempt.evidence_received", map[string]any{"attempt_id": in.Ref.AttemptID, "task_id": in.Ref.TaskID, "reason": reason, "outcome": in.Outcome})
|
||
}
|
||
|
||
// retryFacts 以持久化预算为准，JSON 快照只提供冻结的等待策略。
|
||
func retryFacts(task executionTask, attempt core.ID, now time.Time, outcome string, retryable bool) retry.Input {
|
||
policy := task.view.Retry
|
||
policy.MaxAttempts = task.maxAttempts
|
||
return retry.Input{Policy: policy, Safety: task.view.RecoverySafety, Consumed: task.view.AttemptCount, Outcome: outcome, Retryable: retryable, Uncertain: task.view.EffectUncertain, CancelRequested: task.view.CancelRequestedAt != nil, Now: now, Deadline: task.view.DeadlineAt, StartDeadline: task.view.StartDeadlineAt, AttemptID: attempt}
|
||
}
|