160 lines
8.1 KiB
Go
160 lines
8.1 KiB
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/engine"
|
||
weave "git.yuntuo.hk/k2safe/orbit/weave/model"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/recovery"
|
||
)
|
||
|
||
// processedWorkflowPollResult 只接受当前最后任务的已处理权威结果，不把人工声明当作PENDING事实。
|
||
func processedWorkflowPollResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, last core.ID, now time.Time) (*workflowResult, error) {
|
||
task, err := readTask(ctx, tx, g.view.NamespaceID, last, false)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if task.State != "SUCCEEDED" || task.EffectUncertain {
|
||
return nil, fault.Conflict
|
||
}
|
||
var id core.ID
|
||
if err := tx.QueryRowContext(ctx, `SELECT id FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND source_activation_id=? AND source_revision=? AND state='PROCESSED'`, g.view.NamespaceID, g.view.ID, last, row.view.ActivationID, task.Revision).Scan(&id); err != nil {
|
||
return nil, err
|
||
}
|
||
source, err := readWorkflowResultState(ctx, tx, g, workflowCandidate{ns: g.view.NamespaceID, workflow: g.view.ID, id: id}, now, "PROCESSED")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if source == nil || source.task != task.ID || source.activation != *row.view.ActivationID || source.step != row || source.revision != task.Revision || source.snapshot.State != "SUCCEEDED" || source.snapshot.EffectUncertain || !sameWorkflowJSON(source.snapshot.Result, task.Result) {
|
||
return nil, errors.New("reviewed poll result proof mismatch")
|
||
}
|
||
kind, err := g.program.ClassifyPoll(row.view.Key, source.snapshot.Result)
|
||
if err != nil || kind != engine.PollPending {
|
||
return nil, fault.Conflict
|
||
}
|
||
return source, nil
|
||
}
|
||
|
||
func noPendingWorkflowSignals(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, sequence uint32) error {
|
||
var pending int
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_signals WHERE namespace_id=? AND workflow_id=? AND target_step_key=? AND target_activation_sequence=? AND state='PENDING'`, g.view.NamespaceID, g.view.ID, row.view.Key, sequence).Scan(&pending); err != nil {
|
||
return err
|
||
}
|
||
if pending != 0 {
|
||
return fault.Conflict
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ContinuePoll 为已核实的PENDING恢复原等待，不重查同轮、不重置尝试，也不自动解除实例核查。
|
||
func (s *WorkflowRecoveryStore) ContinuePoll(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.ContinuePollInput) (recovery.Result, error) {
|
||
return s.mutate(ctx, t, id, key, "workflow.poll.continue", in, in.Decision, func(tx *sql.Tx, g *workflowGraph, _ core.ID, now time.Time, out *recovery.Result) (map[string]any, error) {
|
||
row, err := workflowRecoveryStep(g, in.StepDecision)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
node, ok := g.program.Step(row.view.Key)
|
||
if !ok || node.Poll == nil {
|
||
return nil, fault.Invalid("continue poll requires a POLL step")
|
||
}
|
||
if g.view.StopKind != nil || g.view.CancelRequestedAt != nil || !now.Before(g.view.DeadlineAt) {
|
||
return nil, fault.Conflict
|
||
}
|
||
if err := workflowStepSettled(ctx, tx, g, row); err != nil {
|
||
return nil, err
|
||
}
|
||
p, last, err := readWorkflowPollBudget(ctx, tx, g, row)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if p.CurrentTaskID != nil || p.UsedRounds >= p.MaxRounds || !now.Before(p.DeadlineAt) {
|
||
return nil, fault.Conflict
|
||
}
|
||
if _, err := readWorkflowPollRoundHistory(ctx, tx, g, row, p, p.UsedRounds); err != nil {
|
||
return nil, err
|
||
}
|
||
source, err := processedWorkflowPollResult(ctx, tx, g, row, last, now)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !source.occurred.Before(p.DeadlineAt) {
|
||
return nil, fault.Conflict
|
||
}
|
||
next := minimumTime(source.occurred.Add(time.Duration(p.IntervalMS)*time.Millisecond), p.DeadlineAt)
|
||
meta := map[string]any{"step_id": row.view.ID, "activation_id": row.view.ActivationID, "previous_state": row.view.State, "previous_output": json.RawMessage(row.output), "previous_resolved_at": p.ResolvedAt, "task_id": last, "source_event_id": source.event, "source_revision": source.revision, "round": p.UsedRounds, "next_poll_at": next, "deadline_at": p.DeadlineAt, "pending_confirmed": true}
|
||
if node.Poll.FinalSignal != nil {
|
||
w, err := readWorkflowSignalWait(ctx, tx, g, row)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if w.winner != nil || w.state != "WAITING" && w.state != "CANCELLED" || !now.Before(w.deadline) {
|
||
return nil, fault.Conflict
|
||
}
|
||
if err := noPendingWorkflowSignals(ctx, tx, g, row, w.sequence); err != nil {
|
||
return nil, err
|
||
}
|
||
meta["wait_id"], meta["previous_wait_state"] = w.id, w.state
|
||
result, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='WAITING',resolved_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state=? AND winner_signal_id IS NULL`, now, t.NamespaceID, w.id, w.state)
|
||
if err := exactlyOne(result, err); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
result, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET next_poll_at=?,resolved_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_task_id IS NULL`, next, now, t.NamespaceID, p.ID, p.Revision)
|
||
if err := exactlyOne(result, err); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := saveWorkflowResultStep(ctx, tx, g, row, weave.StepWaiting, nil, now); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, next); err != nil {
|
||
return nil, err
|
||
}
|
||
meta["step_revision"] = row.view.Revision
|
||
out.Step = &row.view
|
||
return meta, nil
|
||
})
|
||
}
|
||
|
||
// verifyWorkflowPollPending 保留原Inbox处置；人工接续必须有匹配当前快照的不可变审计证明。
|
||
func verifyWorkflowPollPending(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, p weave.PollState, last core.ID, now time.Time) error {
|
||
var disposition string
|
||
if err := tx.QueryRowContext(ctx, `SELECT error_code FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND source_activation_id=? AND state='PROCESSED' ORDER BY source_revision DESC LIMIT 1`, g.view.NamespaceID, g.view.ID, last, row.view.ActivationID).Scan(&disposition); err != nil {
|
||
return err
|
||
}
|
||
if disposition == "POLL_PENDING" {
|
||
return nil
|
||
}
|
||
source, err := processedWorkflowPollResult(ctx, tx, g, row, last, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var raw []byte
|
||
if err := tx.QueryRowContext(ctx, "SELECT `after` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow' AND resource_id=? AND action='workflow.poll.continue' AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.step_id'))=? AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.activation_id'))=? AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.task_id'))=? ORDER BY created_at DESC,id DESC LIMIT 1", g.view.NamespaceID, g.view.ID, string(row.view.ID), string(*row.view.ActivationID), string(last)).Scan(&raw); err != nil {
|
||
return err
|
||
}
|
||
var proof struct {
|
||
StepID core.ID `json:"step_id"`
|
||
ActivationID core.ID `json:"activation_id"`
|
||
TaskID core.ID `json:"task_id"`
|
||
SourceEventID core.ID `json:"source_event_id"`
|
||
SourceRevision uint64 `json:"source_revision"`
|
||
StepRevision uint64 `json:"step_revision"`
|
||
Round uint32 `json:"round"`
|
||
NextPollAt time.Time `json:"next_poll_at"`
|
||
DeadlineAt time.Time `json:"deadline_at"`
|
||
PendingConfirmed bool `json:"pending_confirmed"`
|
||
}
|
||
expected := minimumTime(source.occurred.Add(time.Duration(p.IntervalMS)*time.Millisecond), p.DeadlineAt)
|
||
if json.Unmarshal(raw, &proof) != nil || proof.StepID != row.view.ID || proof.ActivationID != *row.view.ActivationID || proof.TaskID != last || proof.SourceEventID != source.event || proof.SourceRevision != source.revision || proof.StepRevision != row.view.Revision || proof.Round != p.UsedRounds || !proof.PendingConfirmed || !proof.DeadlineAt.Equal(p.DeadlineAt) || p.NextPollAt == nil || !proof.NextPollAt.Equal(*p.NextPollAt) || !proof.NextPollAt.Equal(expected) {
|
||
return errors.New("poll continuation audit proof invalid")
|
||
}
|
||
return nil
|
||
}
|