package mysql import ( "context" "database/sql" "errors" "time" core "git.yuntuo.hk/k2safe/orbit/model" pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" "git.yuntuo.hk/k2safe/orbit/pulse/service" ) // cancelExecutionTask 由已验证所有权的调用方复用;调用前必须持有应用、队列和任务锁。 // 仅已知未开始的任务立即结清,在途执行保留许可并等待结果,历史未知效果不能变成已取消。 func cancelExecutionTask(ctx context.Context, tx *sql.Tx, ns core.ID, task executionTask, now time.Time, reason string) (service.TaskView, error) { var out service.TaskView view := task.view if view.State.Terminal() || view.CancelRequestedAt != nil { return view, nil } id := view.ID var err error var attempt *pulse.Attempt if view.State == pulse.TaskLeased || view.State == pulse.TaskRunning { if task.current == nil { return out, errors.New("active task lacks attempt") } a, err := readAttempt(ctx, tx, ns, *task.current) if err != nil { return out, err } if a.TaskID != id || a.Epoch != task.epoch || string(a.State) != string(view.State) { return out, errors.New("task attempt identity or state mismatch") } attempt = &a } next := view.State lease := task.lease var completed *time.Time if view.State == pulse.TaskWaiting || view.State == pulse.TaskLeased { next = pulse.TaskCancelled completed = &now lease = nil if view.EffectUncertain { next, completed = pulse.TaskNeedsReview, nil } if attempt != nil { if attempt.AuthorizedAt != nil { return out, errors.New("leased attempt already authorized") } result, err := tx.ExecContext(ctx, `UPDATE pulse_attempts SET state='CANCELLED',ended_at=?,error_code='CANCELLED_BEFORE_START',revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, now, now, ns, attempt.ID, attempt.Revision) if err := exactlyOne(result, err); err != nil { return out, err } if err := releasePermits(ctx, tx, ns, attempt.ID, now); err != nil { return out, err } } } result, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET state=?,cancel_requested_at=?,lease_until=?,completed_at=?,updated_at=?,revision=revision+1 WHERE namespace_id=? AND id=? AND revision=?`, next, now, lease, completed, now, ns, id, view.Revision) if err := exactlyOne(result, err); err != nil { return out, err } if next.Terminal() { if err := releaseQueueTask(ctx, tx, ns, view.QueueID); err != nil { return out, err } } out, err = readTask(ctx, tx, ns, id, false) if err != nil { return out, err } event := resultEvent(next) if next == view.State { event = "task.cancel_requested" } if err := emitEvent(ctx, tx, ns, id, out.Revision, "task", event, map[string]any{"task_id": id, "state": next, "reason": reason, "effect_uncertain": view.EffectUncertain}); err != nil { return out, err } return out, nil }