206 lines
8.4 KiB
Go
206 lines
8.4 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/hex"
|
||
"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/service"
|
||
)
|
||
|
||
// recoveryTaskIdentity 认证后按应用额度→管理身份→队列→任务锁定。
|
||
// 回执核对不重新执行核查/重试，仍检查当前凭据与队列范围。
|
||
func recoveryTaskIdentity(ctx context.Context, tx *sql.Tx, t auth.Token, p auth.Permissions, id core.ID, operation, key string, in any) (executionTask, core.ID, []byte, error) {
|
||
v, err := readTask(ctx, tx, t.NamespaceID, id, false)
|
||
if err != nil {
|
||
return executionTask{}, "", nil, err
|
||
}
|
||
if !p.AllowsQueue(v.QueueID) {
|
||
return executionTask{}, "", nil, fault.NotFound
|
||
}
|
||
if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil {
|
||
return executionTask{}, "", nil, err
|
||
}
|
||
identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, operation, string(id), key, "task", id, in)
|
||
if err != nil || cached != nil {
|
||
return executionTask{}, identity, cached, err
|
||
}
|
||
if _, _, err := lockQueueAfterApplication(ctx, tx, t.NamespaceID, v.QueueID); err != nil {
|
||
return executionTask{}, "", nil, err
|
||
}
|
||
task, err := readExecutionTask(ctx, tx, t.NamespaceID, id)
|
||
return task, identity, nil, err
|
||
}
|
||
|
||
func allowTaskRecovery(ctx context.Context, tx *sql.Tx, ns, id core.ID, allowSchedule bool) error {
|
||
var owner sql.NullString
|
||
var ownerID, gate *core.ID
|
||
if err := tx.QueryRowContext(ctx, `SELECT owner_kind,owner_id,gate_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, id).Scan(&owner, &ownerID, &gate); err != nil {
|
||
return err
|
||
}
|
||
if gate != nil {
|
||
return fault.Conflict
|
||
}
|
||
if owner.Valid {
|
||
if !allowSchedule || owner.String != "schedule" || ownerID == nil {
|
||
return fault.Conflict
|
||
}
|
||
var matched int
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_schedules s JOIN pulse_schedule_occurrences o ON o.namespace_id=s.namespace_id AND o.schedule_id=s.id WHERE s.namespace_id=? AND s.id=? AND o.task_id=?`, ns, *ownerID, id).Scan(&matched); err != nil {
|
||
return err
|
||
}
|
||
if matched != 1 {
|
||
return fault.Conflict
|
||
}
|
||
} else if ownerID != nil {
|
||
return fault.Conflict
|
||
}
|
||
return taskExecutionSettled(ctx, tx, ns, id)
|
||
}
|
||
|
||
// taskExecutionSettled 仅检查平台内已授权尝试和许可；调用者仍须提供外部执行已结束的证据。
|
||
func taskExecutionSettled(ctx context.Context, tx *sql.Tx, ns, id core.ID) error {
|
||
var active int
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_attempts WHERE namespace_id=? AND task_id=? AND state IN ('LEASED','RUNNING')`, ns, id).Scan(&active); err != nil {
|
||
return err
|
||
}
|
||
if active != 0 {
|
||
return fault.Conflict
|
||
}
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_permits p JOIN pulse_attempts a ON a.namespace_id=p.namespace_id AND a.id=p.attempt_id WHERE a.namespace_id=? AND a.task_id=? AND p.released_at IS NULL`, ns, id).Scan(&active); err != nil {
|
||
return err
|
||
}
|
||
if active != 0 {
|
||
return fault.Conflict
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *TaskStore) ReviewTask(ctx context.Context, t auth.Token, id core.ID, key string, in service.ReviewInput) (service.ReviewResult, error) {
|
||
var out service.ReviewResult
|
||
err := s.transaction(ctx, t, "tasks:review", func(tx *sql.Tx, p auth.Permissions) error {
|
||
task, identity, cached, err := recoveryTaskIdentity(ctx, tx, t, p, id, "task.review", key, in)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if cached != nil {
|
||
if err := json.Unmarshal(cached, &out); err != nil {
|
||
return err
|
||
}
|
||
out.Deduplicated = true
|
||
return nil
|
||
}
|
||
v := task.view
|
||
if v.State != pulse.TaskNeedsReview || v.Revision != in.ExpectedRevision || task.lease != nil {
|
||
return fault.Conflict
|
||
}
|
||
if err := allowTaskRecovery(ctx, tx, t.NamespaceID, id, true); err != nil {
|
||
return err
|
||
}
|
||
now, err := dbNow(ctx, tx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
out, err = applyTaskReview(ctx, tx, t, task, identity, in, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := saveIdentity(ctx, tx, t.NamespaceID, identity, out); err != nil {
|
||
return err
|
||
}
|
||
// 结果流锁最后取得；冻结新的终态事件，历史NEEDS_REVIEW事件/尝试保持原样。
|
||
return emitEvent(ctx, tx, t.NamespaceID, id, out.Task.Revision, "task", resultEvent(in.Outcome), map[string]any{"task_id": id, "state": in.Outcome, "effect_uncertain": false, "decision": "MANUAL_REVIEW", "audit_id": out.AuditID})
|
||
})
|
||
return out, err
|
||
}
|
||
|
||
func (s *TaskStore) RetryTask(ctx context.Context, t auth.Token, id core.ID, key string, in service.ManualRetryInput) (service.ManualRetryResult, error) {
|
||
var out service.ManualRetryResult
|
||
err := s.transaction(ctx, t, "tasks:retry", func(tx *sql.Tx, p auth.Permissions) error {
|
||
task, identity, cached, err := recoveryTaskIdentity(ctx, tx, t, p, id, "task.retry", key, in)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if cached != nil {
|
||
if err := json.Unmarshal(cached, &out); err != nil {
|
||
return err
|
||
}
|
||
out.Deduplicated = true
|
||
return nil
|
||
}
|
||
v := task.view
|
||
if (v.State != pulse.TaskFailed && v.State != pulse.TaskCancelled) || v.EffectUncertain || v.Revision != in.ExpectedRevision || task.lease != nil {
|
||
return fault.Conflict
|
||
}
|
||
if err := allowTaskRecovery(ctx, tx, t.NamespaceID, id, false); err != nil {
|
||
return err
|
||
}
|
||
var children int
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=? AND retry_of=?`, t.NamespaceID, id).Scan(&children); err != nil {
|
||
return err
|
||
}
|
||
if children != 0 {
|
||
return fault.Conflict
|
||
}
|
||
input, err := service.NormalizeSubmit(service.SubmitInput{QueueID: v.QueueID, Handler: v.Handler, HandlerVersion: v.HandlerVersion, PayloadVersion: v.PayloadVersion, Payload: v.Payload, BusinessKey: v.BusinessKey, ConcurrencyKey: v.ConcurrencyKey, Priority: v.Priority, RecoverySafety: v.RecoverySafety, Retry: v.Retry, AvailableAt: in.AvailableAt, StartDeadlineAt: in.StartDeadlineAt, TaskTimeoutMS: in.TaskTimeoutMS})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
newID, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
out.SourceTaskID = id
|
||
out.Task, err = insertAcceptedTask(ctx, tx, t.NamespaceID, newID, input, &id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
out.AuditID, err = appendAuditWithEvidence(ctx, tx, t, identity, id, "task", "task.retry", in.Reason, in.EvidenceRef, map[string]any{"state": v.State, "revision": fmt.Sprint(v.Revision)}, map[string]any{"source_task_id": id, "retry_task_id": newID, "safe_to_retry": true, "task_timeout_ms": in.TaskTimeoutMS})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := emitEvent(ctx, tx, t.NamespaceID, newID, 1, "task", "task.accepted", map[string]any{"task_id": newID, "state": pulse.TaskWaiting, "retry_of": id, "audit_id": out.AuditID}); err != nil {
|
||
return err
|
||
}
|
||
return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
|
||
})
|
||
return out, err
|
||
}
|
||
|
||
// applyTaskReview 核查状态、额度与任务审计同事务。调用方已校验归属并持锁，须在最后发布结果事件。
|
||
func applyTaskReview(ctx context.Context, tx *sql.Tx, t auth.Token, task executionTask, identity core.ID, in service.ReviewInput, now time.Time) (service.ReviewResult, error) {
|
||
var out service.ReviewResult
|
||
v := task.view
|
||
if v.State != pulse.TaskNeedsReview || v.Revision != in.ExpectedRevision || task.lease != nil {
|
||
return out, fault.Conflict
|
||
}
|
||
changed, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET state=?,result=?,effect_uncertain=FALSE,completed_at=?,updated_at=?,revision=revision+1 WHERE namespace_id=? AND id=? AND revision=? AND state='NEEDS_REVIEW'`, in.Outcome, string(in.Result), now, now, t.NamespaceID, v.ID, in.ExpectedRevision)
|
||
if err := exactlyOne(changed, err); err != nil {
|
||
return out, err
|
||
}
|
||
if err := releaseQueueTask(ctx, tx, t.NamespaceID, v.QueueID); err != nil {
|
||
return out, err
|
||
}
|
||
out.Task, err = readTask(ctx, tx, t.NamespaceID, v.ID, false)
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
digest, err := requestDigest(in.Result)
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
after := map[string]any{"state": in.Outcome, "revision": fmt.Sprint(out.Task.Revision), "effect_uncertain": false, "execution_settled": true, "result_digest": hex.EncodeToString(digest[:])}
|
||
out.AuditID, err = appendAuditWithEvidence(ctx, tx, t, identity, v.ID, "task", "task.review", in.Reason, in.EvidenceRef, map[string]any{"state": v.State, "revision": fmt.Sprint(v.Revision), "effect_uncertain": v.EffectUncertain}, after)
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
return out, nil
|
||
}
|