orbit/internal/storage/mysql/worker_recovery.go

package mysql

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	core "git.yuntuo.hk/k2safe/orbit/model"
	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
	"git.yuntuo.hk/k2safe/orbit/pulse/retry"
	"git.yuntuo.hk/k2safe/orbit/pulse/worker"
)

// Recover 是受控后台扫描入口，每批最多 100 项、每项一个短事务，可由多个节点竞争。
// 已授权执行的失联保留效果未知，只有冻结的幂等恢复声明与剩余预算允许自动重试。
func (s *ExecutionStore) Recover(ctx context.Context, limit int) (worker.RecoveryStats, error) {
	var out worker.RecoveryStats
	if limit < 1 || limit > 100 {
		return out, fault.Invalid("recovery batch limit must be 1 to 100")
	}
	type candidate struct{ ns, id, queue core.ID }
	candidates := []candidate{}
	seen := map[string]bool{}
	// 分别使用租约、总截止及开始截止索引，避免对整个 WAITING 队列做 COALESCE 扫描。
	for _, q := range []string{
		`SELECT namespace_id,id,queue_id FROM pulse_tasks WHERE state IN ('LEASED','RUNNING') AND lease_until<=UTC_TIMESTAMP(6) ORDER BY lease_until,namespace_id,id LIMIT ?`,
		`SELECT namespace_id,id,queue_id FROM pulse_tasks WHERE state='WAITING' AND deadline_at<=UTC_TIMESTAMP(6) ORDER BY deadline_at,namespace_id,id LIMIT ?`,
		`SELECT namespace_id,id,queue_id FROM pulse_tasks WHERE state='WAITING' AND start_deadline_at<=UTC_TIMESTAMP(6) ORDER BY start_deadline_at,namespace_id,id LIMIT ?`,
	} {
		if len(candidates) >= limit {
			break
		}
		rows, err := s.db.QueryContext(ctx, q, limit-len(candidates))
		if err != nil {
			return out, err
		}
		for rows.Next() {
			var c candidate
			if err := rows.Scan(&c.ns, &c.id, &c.queue); err != nil {
				rows.Close()
				return out, err
			}
			key := string(c.ns) + string(c.id)
			if !seen[key] {
				candidates = append(candidates, c)
				seen[key] = true
			}
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			return out, err
		}
	}
	for _, c := range candidates {
		out.Examined++
		var changed pulse.TaskState
		err := WithTx(ctx, s.db, func(tx *sql.Tx) error {
			// 与 Worker 命令一致先锁队列；有其他节点正在处理时跳过，本批不占用等待线程。
			var queue core.ID
			err := tx.QueryRowContext(ctx, `SELECT id FROM pulse_queues WHERE namespace_id=? AND id=? FOR UPDATE SKIP LOCKED`, c.ns, c.queue).Scan(&queue)
			if errors.Is(err, sql.ErrNoRows) {
				return nil
			}
			if err != nil {
				return err
			}
			var taskID core.ID
			err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_tasks WHERE namespace_id=? AND id=? FOR UPDATE SKIP LOCKED`, c.ns, c.id).Scan(&taskID)
			if errors.Is(err, sql.ErrNoRows) {
				return nil
			}
			if err != nil {
				return err
			}
			task, err := readExecutionTask(ctx, tx, c.ns, c.id)
			if err != nil {
				return err
			}
			if task.view.State == pulse.TaskWaiting {
				now, err := dbNow(ctx, tx)
				if err != nil {
					return err
				}
				if task.view.DeadlineAt.After(now) && (task.view.StartDeadlineAt == nil || task.view.StartDeadlineAt.After(now)) {
					return nil
				}
				changed = pulse.TaskFailed
				var completed *time.Time = &now
				if task.view.EffectUncertain {
					changed = pulse.TaskNeedsReview
					completed = nil
				}
				result, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET state=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, changed, completed, now, c.ns, c.id, task.view.Revision)
				if err := exactlyOne(result, err); err != nil {
					return err
				}
				return emitEvent(ctx, tx, c.ns, c.id, task.view.Revision+1, "task", resultEvent(changed), map[string]any{"reason": "START_DEADLINE_EXCEEDED", "node_id": s.node, "effect_uncertain": task.view.EffectUncertain})
			}
			if task.view.State != pulse.TaskLeased && task.view.State != pulse.TaskRunning {
				return nil
			}
			if task.current == nil || task.lease == nil {
				return errors.New("active task lacks attempt or lease")
			}
			attempt, err := readAttempt(ctx, tx, c.ns, *task.current)
			if err != nil {
				return err
			}
			now, err := dbNow(ctx, tx)
			if err != nil {
				return err
			}
			if task.lease.After(now) || attempt.LeaseUntil.After(now) {
				return nil
			}
			if attempt.Epoch != task.epoch || attempt.TaskID != c.id {
				return errors.New("task attempt identity mismatch")
			}
			if attempt.State != pulse.AttemptLeased && attempt.State != pulse.AttemptRunning {
				return errors.New("active task references ended attempt")
			}
			attemptState := pulse.AttemptExpired
			outcome := "NOT_STARTED"
			reason := "LEASE_EXPIRED_BEFORE_START"
			if attempt.AuthorizedAt != nil {
				attemptState = pulse.AttemptUnknown
				outcome = "UNKNOWN"
				reason = "EXECUTION_OUTCOME_UNKNOWN"
			}
			decision, err := retry.Decide(retryFacts(task, attempt.ID, now, outcome, false))
			if err != nil {
				return err
			}
			next := decision.State
			var completed *time.Time
			if next.Terminal() {
				completed = &now
			}
			available := task.view.AvailableAt
			if decision.NextAt != nil {
				available = *decision.NextAt
			}
			result, err := tx.ExecContext(ctx, `UPDATE pulse_attempts SET state=?,ended_at=?,error_code=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, attemptState, now, reason, now, c.ns, attempt.ID, attempt.Revision)
			if err := exactlyOne(result, err); err != nil {
				return err
			}
			result, err = tx.ExecContext(ctx, `UPDATE pulse_tasks SET state=?,effect_uncertain=?,lease_until=NULL,available_at=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, next, decision.Uncertain, available, completed, now, c.ns, c.id, task.view.Revision)
			if err := exactlyOne(result, err); err != nil {
				return err
			}
			if err := releasePermits(ctx, tx, c.ns, attempt.ID, now); err != nil {
				return err
			}
			changed = next
			return emitEvent(ctx, tx, c.ns, c.id, task.view.Revision+1, "task", resultEvent(next), map[string]any{"attempt_id": attempt.ID, "epoch": fmt.Sprint(attempt.Epoch), "reason": reason, "node_id": s.node, "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 != nil {
			return out, err
		}
		switch changed {
		case pulse.TaskWaiting:
			out.Requeued++
		case pulse.TaskFailed:
			out.Failed++
		case pulse.TaskNeedsReview:
			out.NeedsReview++
		}
	}
	return out, nil
}