orbit/internal/storage/mysql/workflow_controls.go

package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/auth"
	"git.yuntuo.hk/k2safe/orbit/weave/control"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
)

// WorkflowControlStore 控制决定和门控同事务，不在请求中等待外部执行停止。
type WorkflowControlStore struct{ *TaskStore }

func NewWorkflowControlStore(db *sql.DB) *WorkflowControlStore {
	return &WorkflowControlStore{NewTaskStore(db)}
}

var _ control.Store = (*WorkflowControlStore)(nil)

func (s *WorkflowControlStore) Apply(ctx context.Context, t auth.Token, id core.ID, key string, in control.Input) (control.Result, error) {
	var out control.Result
	err := s.transaction(ctx, t, control.Action, func(tx *sql.Tx, p auth.Permissions) error {
		if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil {
			return err
		}
		identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "workflow.control", string(id), key, "workflow", id, in)
		if err != nil {
			return err
		}
		g, _, err := lockWorkflowGraph(ctx, tx, t.NamespaceID, id, false)
		if err != nil {
			return err
		}
		for _, q := range g.version.Analysis.Queues {
			if !p.AllowsQueue(q) {
				return fault.NotFound
			}
		}
		if cached != nil {
			if err := json.Unmarshal(cached, &out); err != nil {
				return err
			}
			if out.Instance.NamespaceID != t.NamespaceID || out.Instance.ID != id || out.Instance.DefinitionVersionID != g.view.DefinitionVersionID {
				return errors.New("workflow control receipt mismatch")
			}
			out.Deduplicated = true
			return nil
		}
		if g.view.State.Terminal() || g.view.ControlRevision != in.ExpectedControlRevision {
			return fault.Conflict
		}
		before := g.view
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		switch in.Operation {
		case "PAUSE":
			if g.view.Paused {
				return fault.Conflict
			}
			g.view.Paused = true
		case "RESUME":
			if !g.view.Paused {
				return fault.Conflict
			}
			g.view.Paused = false
		case "CANCEL":
			if g.view.CancelRequestedAt != nil {
				return fault.Conflict
			}
			g.view.CancelRequestedAt = &now
			if err := recordWorkflowStop(ctx, tx, g, "CANCEL", now); err != nil {
				return err
			}
			if g.view.State != weave.InstanceNeedsReview {
				g.view.State = weave.InstanceCancelling
			}
		default:
			return fault.Invalid("unsupported workflow control operation")
		}
		res, err := tx.ExecContext(ctx, `UPDATE weave_instances SET paused=?,state=?,cancel_requested_at=?,control_revision=control_revision+1,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND control_revision=? AND revision=?`, g.view.Paused, g.view.State, g.view.CancelRequestedAt, now, t.NamespaceID, id, in.ExpectedControlRevision, g.view.Revision)
		if err := exactlyOne(res, err); err != nil {
			return err
		}
		g.view.ControlRevision++
		g.view.Revision++
		gatePaused := g.view.Paused || g.view.StopKind != nil || g.view.State == weave.InstanceNeedsReview || g.view.State == weave.InstanceCancelling
		if _, err := tx.ExecContext(ctx, `UPDATE pulse_execution_gates SET paused=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, gatePaused, now, t.NamespaceID, g.gate); err != nil {
			return err
		}
		if in.Operation == "CANCEL" || in.Operation == "RESUME" {
			if err := saveWorkflowAdvance(ctx, tx, g, now); err != nil {
				return err
			}
		}
		meta := map[string]any{"operation": in.Operation, "paused": g.view.Paused, "state": g.view.State, "control_revision": g.view.ControlRevision, "stop_kind": g.view.StopKind, "stop_requested_at": g.view.StopRequestedAt, "cancel_requested_at": g.view.CancelRequestedAt}
		if err := emitEvent(ctx, tx, t.NamespaceID, id, g.view.Revision, "workflow", "workflow.controlled", meta); err != nil {
			return err
		}
		if err := appendAudit(ctx, tx, t, identity, id, "workflow", "workflow.control", in.Reason, map[string]any{"paused": before.Paused, "state": before.State, "control_revision": before.ControlRevision, "stop_kind": before.StopKind}, meta); err != nil {
			return err
		}
		out.Instance, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
		if err != nil {
			return err
		}
		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
	})
	return out, err
}

// recordWorkflowStop 冻结首次停止原因；后来的用户取消不能抹去已启动的失败结清。
// 调用方同时关闭门控、保存根检查点和审计，不能将此函数单独提交。
func recordWorkflowStop(ctx context.Context, tx *sql.Tx, g *workflowGraph, kind string, now time.Time) error {
	if g.view.StopKind != nil {
		return nil
	}
	if kind != "CANCEL" && kind != "FAILURE" {
		return errors.New("invalid workflow stop kind")
	}
	res, err := tx.ExecContext(ctx, `UPDATE weave_instances SET stop_kind=?,stop_requested_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND stop_kind IS NULL`, kind, now, now, g.view.NamespaceID, g.view.ID, g.view.Revision)
	if err := exactlyOne(res, err); err != nil {
		return err
	}
	g.view.StopKind = &kind
	g.view.StopRequestedAt = &now
	g.view.Revision++
	return nil
}