feat: recover workflow waits within original deadlines

2026-09-18 06:10:04 +08:00
parent 56b9a2ccfe
commit af1ae4cb21
25 changed files with 941 additions and 52 deletions
--- a/internal/storage/mysql/workflow_clock.go
+++ b/internal/storage/mysql/workflow_clock.go
@@ -61,10 +61,6 @@ func (r *WorkflowRuntime) advanceClock(ctx context.Context, tx *sql.Tx, g *workf
 		}
 		return false, deleteWorkflowClock(ctx, tx, g, row)
 	}
-	var activated time.Time
-	if err := tx.QueryRowContext(ctx, `SELECT created_at FROM weave_activations WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=?`, c.ns, checkpoint.ActivationID, c.workflow, step).Scan(&activated); err != nil {
-		return false, err
-	}
 	if checkpoint.Kind == "POLL" {
 		node, _ := g.program.Step(row.view.Key)
 		if node.Poll.FinalSignal != nil {
@@ -103,7 +99,11 @@ func (r *WorkflowRuntime) advanceClock(ctx context.Context, tx *sql.Tx, g *workf
 		return false, err
 	}
 	node, _ := g.program.Step(row.view.Key)
-	intended := activated.Add(time.Duration(node.Timer.DelayMS) * time.Millisecond)
+	origin, err := workflowActivationOrigin(ctx, tx, g, row)
+	if err != nil {
+		return false, err
+	}
+	intended := origin.started.Add(time.Duration(node.Timer.DelayMS) * time.Millisecond)
 	if kind != "TIMER" || state != "WAITING" || name.Valid || version.Valid || len(correlation) != 0 || !deadline.Equal(minimumTime(intended, g.view.DeadlineAt)) {
 		return false, errors.New("workflow timer wait invalid")
 	}
--- a/internal/storage/mysql/workflow_consume.go
+++ b/internal/storage/mysql/workflow_consume.go
@@ -31,6 +31,11 @@ type workflowResult struct {
 }

 func readWorkflowResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, c workflowCandidate, now time.Time) (*workflowResult, error) {
+	return readWorkflowResultState(ctx, tx, g, c, now, "PENDING")
+}
+
+// readWorkflowResultState 复用不可变来源核验；管理恢复只读取已处理快照，不改写原Inbox处置。
+func readWorkflowResultState(ctx context.Context, tx *sql.Tx, g *workflowGraph, c workflowCandidate, now time.Time, required string) (*workflowResult, error) {
 	var state string
 	var due time.Time
 	var raw []byte
@@ -42,7 +47,7 @@ func readWorkflowResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, c wor
 	if err != nil {
 		return nil, err
 	}
-	if state != "PENDING" || due.After(now) {
+	if state != required || required == "PENDING" && due.After(now) {
 		return nil, nil
 	}
 	if _, err := jsonvalue.Canonical(raw); err != nil {
--- a/internal/storage/mysql/workflow_poll_clock.go
+++ b/internal/storage/mysql/workflow_poll_clock.go
@@ -159,13 +159,9 @@ func (r *WorkflowRuntime) advancePoll(ctx context.Context, tx *sql.Tx, g *workfl
 		return false, err
 	}
 	// 只有上一轮成功查询并已分类为PENDING才可以消耗新轮次，不能用时间推进绕过失败预算。
-	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`, c.ns, c.workflow, lastTask, row.view.ActivationID).Scan(&disposition); err != nil {
+	if err := verifyWorkflowPollPending(ctx, tx, g, row, p, lastTask, now); err != nil {
 		return false, err
 	}
-	if disposition != "POLL_PENDING" {
-		return false, errors.New("poll previous round was not pending")
-	}
 	in, err := pollTaskInput(ctx, tx, g, row, policy.Task)
 	if err != nil {
 		return false, err
--- a/internal/storage/mysql/workflow_poll_continue.go
+++ b/internal/storage/mysql/workflow_poll_continue.go
@@ -0,0 +1,159 @@
+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
+}
--- a/internal/storage/mysql/workflow_recovery.go
+++ b/internal/storage/mysql/workflow_recovery.go
@@ -155,6 +155,17 @@ func (s *WorkflowRecoveryStore) ReviewTask(ctx context.Context, t auth.Token, id
 }

 func workflowRecoveryStep(g *workflowGraph, in recovery.StepDecision) (*workflowStep, error) {
+	row, err := workflowRecoveryTarget(g, in)
+	if err != nil {
+		return nil, err
+	}
+	if row.view.State != weave.StepNeedsReview && row.view.State != weave.StepFailed && row.view.State != weave.StepCancelled {
+		return nil, fault.Conflict
+	}
+	return row, nil
+}
+
+func workflowRecoveryTarget(g *workflowGraph, in recovery.StepDecision) (*workflowStep, error) {
 	row := g.steps[in.StepKey]
 	if row == nil || in.StepKey == workflowRootKey {
 		return nil, fault.NotFound
@@ -162,9 +173,6 @@ func workflowRecoveryStep(g *workflowGraph, in recovery.StepDecision) (*workflow
 	if row.view.ActivationID == nil || *row.view.ActivationID != in.ActivationID || row.view.Revision != in.ExpectedStepRevision {
 		return nil, fault.Conflict
 	}
-	if row.view.State != weave.StepNeedsReview && row.view.State != weave.StepFailed && row.view.State != weave.StepCancelled {
-		return nil, fault.Conflict
-	}
 	return row, nil
 }

@@ -217,11 +225,11 @@ func (s *WorkflowRecoveryStore) ResolveStep(ctx context.Context, t auth.Token, i
 				}
 				typ = node.Poll.Task.ResultType
 			case node.Timer != nil:
-				var activated time.Time
-				if err := tx.QueryRowContext(ctx, `SELECT created_at FROM weave_activations WHERE namespace_id=? AND id=?`, t.NamespaceID, row.view.ActivationID).Scan(&activated); err != nil {
+				origin, err := workflowActivationOrigin(ctx, tx, g, row)
+				if err != nil {
 					return nil, err
 				}
-				if now.Before(activated.Add(time.Duration(node.Timer.DelayMS) * time.Millisecond)) {
+				if now.Before(origin.started.Add(time.Duration(node.Timer.DelayMS) * time.Millisecond)) {
 					return nil, fault.Conflict
 				}
 				typ = definition.Type{Kind: "NULL"}
--- a/internal/storage/mysql/workflow_runtime_process_integration_test.go
+++ b/internal/storage/mysql/workflow_runtime_process_integration_test.go
@@ -408,4 +408,89 @@ func TestIntegrationWorkflowRuntimeProcess(t *testing.T) {
 	}
 	t.Log("POLL manual same-round recovery survived all-stop and two fresh controllers; reduced request attempts, original deadline, old activation and subsequent business-round progression preserved")

+	for _, p := range processes {
+		stop(p)
+	}
+	pending := runtimeStart(t, f, recoveryPollSpec(f.resources.QueueID, true))
+	grantWorkflowRecovery(f)
+	grantWorkflowSignal(f)
+	ref = retryStart(t, f)
+	retryFinish(t, f, ref, "UNKNOWN", false)
+	start("pending-review-a")
+	start("pending-review-b")
+	wait(func() bool { return f.count("weave_instances", "id=? AND state='NEEDS_REVIEW'", pending.ID) == 1 }, "query unknown review")
+	for _, p := range processes {
+		stop(p)
+	}
+	task := f.get(ref.TaskID)
+	_, err = recovery.New(NewWorkflowRecoveryStore(db)).ReviewTask(ctx, f.token, pending.ID, string(executionID(t)), recovery.ReviewTaskInput{Decision: recoveryDecision(t, f, pending.ID), TaskID: task.ID, ExpectedTaskRevision: task.Revision, ExecutionSettled: true, Outcome: "SUCCEEDED", Result: json.RawMessage(`"PENDING"`)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	start("pending-proof-a")
+	start("pending-proof-b")
+	wait(func() bool { return f.count("weave_inbox", "workflow_id=? AND state='PENDING'", pending.ID) == 0 }, "reviewed pending evidence consumed")
+	for _, p := range processes {
+		stop(p)
+	}
+	continueReviewedPoll(t, f, pending.ID)
+	continueWorkflow(t, f, pending.ID)
+	start("pending-continue-a")
+	start("pending-continue-b")
+	wait(func() bool {
+		return f.count("weave_poll_states", "workflow_id=? AND used_rounds=2 AND current_task_id IS NOT NULL", pending.ID) == 1
+	}, "reviewed pending resumes next round across all-stop")
+	ref = retryStart(t, f)
+	finishQueryOnly(t, f, ref, "DONE")
+	wait(func() bool { return f.count("weave_instances", "id=? AND state='SUCCEEDED'", pending.ID) == 1 }, "continued poll completion")
+	for _, kind := range []string{"SIGNAL", "TIMER"} {
+		for _, p := range processes {
+			stop(p)
+		}
+		delay := uint64(30000)
+		key := "signal"
+		if kind == "TIMER" {
+			delay = 3000
+			key = "timer"
+		}
+		waiting := runtimeStart(t, f, waitingParallelSpec(f.resources.QueueID, kind, delay))
+		grantWorkflowRecovery(f)
+		grantWorkflowSignal(f)
+		ref = retryStart(t, f)
+		retryFinish(t, f, ref, "FAILED", false)
+		start("wait-review-" + kind + "-a")
+		start("wait-review-" + kind + "-b")
+		wait(func() bool { return f.count("weave_instances", "id=? AND state='NEEDS_REVIEW'", waiting.ID) == 1 }, "parallel failure leaves waiting sibling")
+		for _, p := range processes {
+			stop(p)
+		}
+		var due time.Time
+		if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, waiting.ID).Scan(&due); err != nil {
+			t.Fatal(err)
+		}
+		rearmWorkflowWait(t, f, waiting.ID, key)
+		if f.count("weave_waits", "workflow_id=? AND deadline_at=?", waiting.ID, due) != 2 {
+			t.Fatal("wait deadline reset after all-stop")
+		}
+		resolveWorkflowStep(t, f, waiting.ID, "left", "SUCCEEDED", `{"n":9007199254740993}`)
+		continueWorkflow(t, f, waiting.ID)
+		start("wait-rearm-" + kind + "-a")
+		start("wait-rearm-" + kind + "-b")
+		if kind == "SIGNAL" {
+			input := signalInput(waiting)
+			input.TargetActivationSequence = 2
+			sendSignal(t, f, waiting, input)
+		}
+		wait(func() bool {
+			return f.count("weave_steps", "workflow_id=? AND `key`='task0' AND state='ACTIVE'", waiting.ID) == 1
+		}, "rearmed wait completed and started one successor")
+		ref = retryStart(t, f)
+		retryFinish(t, f, ref, "SUCCEEDED", false)
+		wait(func() bool { return f.count("weave_instances", "id=? AND state='SUCCEEDED'", waiting.ID) == 1 }, "rearmed wait workflow completed")
+	}
+	if f.count("pulse_tasks", "1=1") != 22 || f.used() != 0 {
+		t.Fatal("wait recovery duplicated task or leaked permits")
+	}
+	t.Log("reviewed PENDING continuation plus SIGNAL/TIMER new generations survived full controller shutdown and two fresh nodes; original intervals/deadlines and durable proof preserved, exactly 22 tasks overall")
+
 }
--- a/internal/storage/mysql/workflow_signal_runtime.go
+++ b/internal/storage/mysql/workflow_signal_runtime.go
@@ -98,13 +98,9 @@ func (r *WorkflowRuntime) applyWorkflowSignal(ctx context.Context, tx *sql.Tx, g
 			if row.view.State != weave.StepWaiting || poll.NextPollAt == nil {
 				return false, false, "", errors.New("signal poll continuation missing")
 			}
-			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 {
+			if err := verifyWorkflowPollPending(ctx, tx, g, row, poll, last, now); err != nil {
 				return false, false, "", err
 			}
-			if disposition != "POLL_PENDING" {
-				return false, false, "", errors.New("signal poll previous round not pending")
-			}
 		}
 	}
 	v, err := selectedWorkflowSignal(ctx, tx, g, row, w, now)
--- a/internal/storage/mysql/workflow_wait_rearm.go
+++ b/internal/storage/mysql/workflow_wait_rearm.go
@@ -0,0 +1,159 @@
+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"
+	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
+	"git.yuntuo.hk/k2safe/orbit/weave/recovery"
+)
+
+// RearmWait 仅对独立SIGNAL/TIMER换代，固定首次等待的预算；到期事实不能通过新建等待清除。
+func (s *WorkflowRecoveryStore) RearmWait(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.RearmWaitInput) (recovery.Result, error) {
+	return s.mutate(ctx, t, id, key, "workflow.wait.rearm", in, in.Decision, func(tx *sql.Tx, g *workflowGraph, _ core.ID, now time.Time, out *recovery.Result) (map[string]any, error) {
+		row, err := workflowRecoveryTarget(g, in.StepDecision)
+		if err != nil {
+			return nil, err
+		}
+		node, ok := g.program.Step(row.view.Key)
+		if !ok || node.Signal == nil && node.Timer == nil {
+			return nil, fault.Invalid("wait rearm requires a SIGNAL or TIMER step")
+		}
+		if row.view.State != weave.StepWaiting && row.view.State != weave.StepNeedsReview && row.view.State != weave.StepFailed && row.view.State != weave.StepCancelled {
+			return nil, fault.Conflict
+		}
+		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
+		}
+		origin, err := workflowActivationOrigin(ctx, tx, g, row)
+		if err != nil {
+			return nil, err
+		}
+		if origin.sequence >= recovery.MaxWaitActivations {
+			return nil, fault.Conflict
+		}
+		if err := verifyWorkflowWaitHistory(ctx, tx, g, row, origin); err != nil {
+			return nil, err
+		}
+		var w workflowSignalWait
+		if node.Signal != nil {
+			w, err = readWorkflowSignalWait(ctx, tx, g, row)
+			if err != nil {
+				return nil, err
+			}
+			if err := noPendingWorkflowSignals(ctx, tx, g, row, w.sequence); err != nil {
+				return nil, err
+			}
+		} else {
+			var count int
+			if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, t.NamespaceID, id, row.view.ActivationID).Scan(&count); err != nil {
+				return nil, err
+			}
+			if count != 1 {
+				return nil, errors.New("timer rearm wait set invalid")
+			}
+			var kind string
+			var name, version *string
+			var wake *core.ID
+			if err := tx.QueryRowContext(ctx, `SELECT id,kind,state,deadline_at,signal_name,payload_version,correlation_key,winner_signal_id,wake_task_id FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, t.NamespaceID, id, row.view.ActivationID).Scan(&w.id, &kind, &w.state, &w.deadline, &name, &version, &w.correlation, &w.winner, &wake); err != nil {
+				return nil, err
+			}
+			expectedInput, _ := json.Marshal(map[string]uint64{"delay_ms": node.Timer.DelayMS})
+			if kind != "TIMER" || name != nil || version != nil || w.correlation != "" || wake != nil || !sameWorkflowJSON(row.input, expectedInput) || !w.deadline.Equal(minimumTime(origin.started.Add(time.Duration(node.Timer.DelayMS)*time.Millisecond), g.view.DeadlineAt)) {
+				return nil, errors.New("timer rearm contract mismatch")
+			}
+		}
+		if w.winner != nil || w.state != "WAITING" && w.state != "CANCELLED" || !now.Before(w.deadline) {
+			return nil, fault.Conflict
+		}
+		activation, err := core.NewID()
+		if err != nil {
+			return nil, err
+		}
+		wait, err := core.NewID()
+		if err != nil {
+			return nil, err
+		}
+		beforeState := row.view.State
+		if row.view.State == weave.StepWaiting {
+			result, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state='CANCELLED',completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state='WAITING'`, now, now, t.NamespaceID, in.ActivationID, id, row.view.ID)
+			if err := exactlyOne(result, err); err != nil {
+				return nil, err
+			}
+		}
+		if _, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='CANCELLED',resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state='WAITING'`, now, now, t.NamespaceID, w.id); err != nil {
+			return nil, err
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,input,created_at,updated_at) VALUES (?,?,?,?,?,'WAITING',?,?,?)`, t.NamespaceID, activation, id, row.view.ID, origin.sequence+1, string(origin.input), now, now); err != nil {
+			return nil, err
+		}
+		var name, version *string
+		if w.contract != nil {
+			name = &w.contract.Name
+			version = &w.contract.PayloadVersion
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at,signal_name,payload_version,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, t.NamespaceID, wait, id, activation, row.view.Kind, []byte(w.correlation), w.deadline, name, version, now, now); err != nil {
+			return nil, err
+		}
+		result, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state='WAITING',current_activation_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_activation_id=?`, activation, now, t.NamespaceID, row.view.ID, row.view.Revision, in.ActivationID)
+		if err := exactlyOne(result, err); err != nil {
+			return nil, err
+		}
+		row.view.State, row.view.ActivationID = weave.StepWaiting, &activation
+		row.view.Revision++
+		row.output = nil
+		due := w.deadline
+		if node.Signal != nil {
+			due = now
+		}
+		if err := saveWorkflowClock(ctx, tx, g, row, string(row.view.Kind), row.view.Revision, due); err != nil {
+			return nil, err
+		}
+		if err := emitEvent(ctx, tx, t.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": id, "state": row.view.State, "activation_id": activation}); err != nil {
+			return nil, err
+		}
+		out.Step = &row.view
+		return map[string]any{"step_id": row.view.ID, "previous_activation_id": in.ActivationID, "activation_id": activation, "sequence": origin.sequence + 1, "previous_state": beforeState, "previous_wait_id": w.id, "previous_wait_state": w.state, "wait_id": wait, "deadline_at": w.deadline, "wait_verified": true}, nil
+	})
+}
+
+func verifyWorkflowWaitHistory(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, origin workflowActivationBase) error {
+	rows, err := tx.QueryContext(ctx, `SELECT id,sequence,state,input,task_id FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? ORDER BY sequence LIMIT 33`, g.view.NamespaceID, g.view.ID, row.view.ID)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	var count uint32
+	var last core.ID
+	for rows.Next() {
+		var id core.ID
+		var sequence uint32
+		var state string
+		var input []byte
+		var task *core.ID
+		if err := rows.Scan(&id, &sequence, &state, &input, &task); err != nil {
+			return err
+		}
+		count++
+		if sequence != count || task != nil || !sameWorkflowJSON(input, origin.input) || id != *row.view.ActivationID && state != "NEEDS_REVIEW" && state != "FAILED" && state != "CANCELLED" {
+			return errors.New("wait activation history invalid")
+		}
+		last = id
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	if count != origin.sequence || last != *row.view.ActivationID {
+		return errors.New("wait activation history incomplete")
+	}
+	return nil
+}
--- a/internal/storage/mysql/workflow_wait_recovery_integration_test.go
+++ b/internal/storage/mysql/workflow_wait_recovery_integration_test.go
@@ -0,0 +1,305 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"git.yuntuo.hk/k2safe/orbit/internal/fault"
+	core "git.yuntuo.hk/k2safe/orbit/model"
+	"git.yuntuo.hk/k2safe/orbit/weave/definition"
+	"git.yuntuo.hk/k2safe/orbit/weave/instance"
+	"git.yuntuo.hk/k2safe/orbit/weave/recovery"
+)
+
+func waitingParallelSpec(queue core.ID, kind string, delay uint64) definition.Specification {
+	spec := runtimeSpec(queue, 1)
+	tail := spec.Body.Steps[0]
+	join := runtimeParallel(spec)
+	var wait definition.Step
+	var typ definition.Type
+	if kind == "SIGNAL" {
+		wait = signalSpec(queue, false).Body.Steps[0]
+		wait.Signal.TimeoutMS = delay
+		typ = spec.InputType
+	} else {
+		wait = timerSpec(queue, delay, false).Body.Steps[0]
+		typ = definition.Type{Kind: "NULL"}
+	}
+	join.Parallel.Branches[1].Body = definition.Block{Steps: []definition.Step{wait}, OutputType: typ, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: wait.Key}}}
+	spec.Body.Steps = []definition.Step{join, tail}
+	return spec
+}
+func waitRecoveryFixture(t *testing.T, ctx context.Context, db *sql.DB, kind string, delay uint64) (*executionFixture, instance.View) {
+	t.Helper()
+	f := newExecutionFixture(t, ctx, db)
+	v := runtimeStart(t, f, waitingParallelSpec(f.resources.QueueID, kind, delay))
+	grantWorkflowRecovery(f)
+	grantWorkflowSignal(f)
+	ref := retryStart(t, f)
+	retryFinish(t, f, ref, "FAILED", false)
+	runtimeConsume(t, f, newRuntime(t, db), ref.TaskID)
+	return f, v
+}
+func rearmWorkflowWait(t *testing.T, f *executionFixture, v core.ID, key string) recovery.Result {
+	t.Helper()
+	out, err := recovery.New(NewWorkflowRecoveryStore(f.db)).RearmWait(f.ctx, f.token, v, string(executionID(t)), recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v, key), WaitVerified: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return out
+}
+func reviewedPendingPoll(t *testing.T, ctx context.Context, db *sql.DB, final bool) (*executionFixture, instance.View, core.ID) {
+	t.Helper()
+	f := newClockFixture(t, ctx, db)
+	v := runtimeStart(t, f, recoveryPollSpec(f.resources.QueueID, final))
+	grantWorkflowRecovery(f)
+	grantWorkflowSignal(f)
+	ref := retryStart(t, f)
+	retryFinish(t, f, ref, "UNKNOWN", false)
+	runtimeConsume(t, f, newRuntime(t, db), ref.TaskID)
+	task := f.get(ref.TaskID)
+	_, err := recovery.New(NewWorkflowRecoveryStore(db)).ReviewTask(ctx, f.token, v.ID, string(executionID(t)), recovery.ReviewTaskInput{Decision: recoveryDecision(t, f, v.ID), TaskID: task.ID, ExpectedTaskRevision: task.Revision, ExecutionSettled: true, Outcome: "SUCCEEDED", Result: json.RawMessage(`"PENDING"`)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	runtimeConsume(t, f, newRuntime(t, db), task.ID)
+	return f, v, task.ID
+}
+func continueReviewedPoll(t *testing.T, f *executionFixture, v core.ID) recovery.Result {
+	t.Helper()
+	out, err := recovery.New(NewWorkflowRecoveryStore(f.db)).ContinuePoll(f.ctx, f.token, v, string(executionID(t)), recovery.ContinuePollInput{StepDecision: recoveryStep(t, f, v, "query"), PendingConfirmed: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return out
+}
+func TestIntegrationWorkflowWaitRecovery(t *testing.T) {
+	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
+	if dsn == "" {
+		t.Skip("requires disposable MySQL")
+	}
+	ctx, stop := context.WithTimeout(context.Background(), 180*time.Second)
+	defer stop()
+	db, err := Open(ctx, dsn, 32, 8)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if err := Migrate(ctx, db); err != nil {
+		t.Fatal(err)
+	}
+	api := recovery.New(NewWorkflowRecoveryStore(db))
+	t.Run("reviewed_pending_keeps_proof_and_interval_then_new_business_round", func(t *testing.T) {
+		f, v, task := reviewedPendingPoll(t, ctx, db, false)
+		before := recoveryStep(t, f, v.ID, "query")
+		out := continueReviewedPoll(t, f, v.ID)
+		if out.Instance.State != "NEEDS_REVIEW" || out.Step.State != "WAITING" || *out.Step.ActivationID != before.ActivationID || f.count("pulse_tasks", "1=1") != 1 || f.count("weave_inbox", "source_task_id=? AND error_code='STEP_ALREADY_SETTLED'", task) != 1 {
+			t.Fatal("continuation restarted request or changed old source")
+		}
+		var next, occurred time.Time
+		if err := db.QueryRowContext(ctx, `SELECT p.next_poll_at,e.occurred_at FROM weave_poll_states p JOIN pulse_tasks t ON t.namespace_id=p.namespace_id AND t.id=? JOIN weave_inbox i ON i.namespace_id=t.namespace_id AND i.source_task_id=t.id AND i.source_revision=t.revision JOIN orbit_events e ON e.namespace_id=i.namespace_id AND e.id=i.event_id WHERE p.namespace_id=? AND p.workflow_id=?`, task, f.resources.NamespaceID, v.ID).Scan(&next, &occurred); err != nil || !next.Equal(occurred.Add(time.Second)) {
+			t.Fatal("original interval lost", err)
+		}
+		c := clockCandidate(t, f, v.ID, "query")
+		runClock(t, f, c)
+		if f.count("pulse_tasks", "1=1") != 1 {
+			t.Fatal("review gate or interval bypassed")
+		}
+		continueWorkflow(t, f, v.ID)
+		c = clockCandidate(t, f, v.ID, "query")
+		waitClockDue(t, f, c)
+		runClock(t, f, c)
+		ref := retryStart(t, f)
+		if f.get(ref.TaskID).Retry.MaxAttempts != 3 {
+			t.Fatal("next round inherited old request budget")
+		}
+		finishPoll(t, f, ref, "DONE")
+		runtimeView(t, f, v.ID, "SUCCEEDED")
+		if f.count("pulse_tasks", "1=1") != 2 || f.count("pulse_attempts", "state='UNKNOWN'") != 1 || f.used() != 0 {
+			t.Fatal("request duplicated or history lost")
+		}
+	})
+	t.Run("reviewed_pending_final_signal_still_wins_without_new_round", func(t *testing.T) {
+		f, v, _ := reviewedPendingPoll(t, ctx, db, true)
+		continueReviewedPoll(t, f, v.ID)
+		continueWorkflow(t, f, v.ID)
+		sendSignal(t, f, v, pollSignalInput(v))
+		runClock(t, f, clockCandidate(t, f, v.ID, "query"))
+		runtimeView(t, f, v.ID, "SUCCEEDED")
+		if f.count("pulse_tasks", "1=1") != 1 || f.count("weave_waits", "state='SATISFIED' AND winner_signal_id IS NOT NULL") != 1 {
+			t.Fatal("continued final wait lost signal")
+		}
+	})
+	t.Run("pending_continuation_audit_failure_is_atomic", func(t *testing.T) {
+		f, v, _ := reviewedPendingPoll(t, ctx, db, false)
+		in := recovery.ContinuePollInput{StepDecision: recoveryStep(t, f, v.ID, "query"), PendingConfirmed: true}
+		key := string(executionID(t))
+		trigger := "reject_poll_continue_" + string(executionID(t))
+		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled continuation audit failure'")
+		_, err := api.ContinuePoll(ctx, f.token, v.ID, key, in)
+		f.exec("DROP TRIGGER " + trigger)
+		if err == nil || f.count("weave_poll_states", "resolved_at IS NOT NULL AND next_poll_at IS NULL") != 1 || f.count("weave_pending_advances", "1=1") != 0 || recoveryDecision(t, f, v.ID).ExpectedRevision != in.ExpectedRevision {
+			t.Fatal("partial continuation committed", err)
+		}
+		first, err := api.ContinuePoll(ctx, f.token, v.ID, key, in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		again, err := api.ContinuePoll(ctx, f.token, v.ID, key, in)
+		if err != nil || !again.Deduplicated || again.AuditID != first.AuditID {
+			t.Fatal("continuation receipt changed", err)
+		}
+	})
+	t.Run("continuation_rejects_forged_pending_and_exhausted_rounds", func(t *testing.T) {
+		f, v, _ := failedPoll(t, ctx, db, false)
+		if _, err := api.ContinuePoll(ctx, f.token, v.ID, string(executionID(t)), recovery.ContinuePollInput{StepDecision: recoveryStep(t, f, v.ID, "query"), PendingConfirmed: true}); fault.Code(err) != "STATE_CONFLICT" {
+			t.Fatal("failure treated as pending", err)
+		}
+		f = newClockFixture(t, ctx, db)
+		spec := recoveryPollSpec(f.resources.QueueID, false)
+		spec.Body.Steps[0].Poll.MaxRounds = 1
+		v = runtimeStart(t, f, spec)
+		grantWorkflowRecovery(f)
+		ref := retryStart(t, f)
+		finishPoll(t, f, ref, "PENDING")
+		if _, err := api.ContinuePoll(ctx, f.token, v.ID, string(executionID(t)), recovery.ContinuePollInput{StepDecision: recoveryStep(t, f, v.ID, "query"), PendingConfirmed: true}); fault.Code(err) != "STATE_CONFLICT" || f.count("pulse_tasks", "1=1") != 1 {
+			t.Fatal("round limit reset", err)
+		}
+	})
+	t.Run("continuation_clock_rejects_changed_due_without_matching_audit", func(t *testing.T) {
+		f, v, _ := reviewedPendingPoll(t, ctx, db, false)
+		continueReviewedPoll(t, f, v.ID)
+		continueWorkflow(t, f, v.ID)
+		c := clockCandidate(t, f, v.ID, "query")
+		waitClockDue(t, f, c)
+		f.exec(`UPDATE weave_poll_states SET next_poll_at=DATE_SUB(next_poll_at,INTERVAL 1 MICROSECOND) WHERE namespace_id=?`, f.resources.NamespaceID)
+		_, err := newRuntime(t, db).process(ctx, c, false)
+		f.exec(`UPDATE weave_poll_states SET next_poll_at=DATE_ADD(next_poll_at,INTERVAL 1 MICROSECOND) WHERE namespace_id=?`, f.resources.NamespaceID)
+		if err == nil || f.count("pulse_tasks", "1=1") != 1 {
+			t.Fatal("changed checkpoint bypassed audit proof")
+		}
+	})
+	t.Run("signal_rearm_changes_generation_without_changing_deadline", func(t *testing.T) {
+		f, v := waitRecoveryFixture(t, ctx, db, "SIGNAL", 30000)
+		before := recoveryStep(t, f, v.ID, "signal")
+		var deadline time.Time
+		if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=?`, f.resources.NamespaceID).Scan(&deadline); err != nil {
+			t.Fatal(err)
+		}
+		out := rearmWorkflowWait(t, f, v.ID, "signal")
+		if *out.Step.ActivationID == before.ActivationID || f.count("weave_waits", "deadline_at=?", deadline) != 2 || f.count("weave_activations", "id=? AND state='CANCELLED'", before.ActivationID) != 1 {
+			t.Fatal("wait history or original deadline lost")
+		}
+		old := sendSignal(t, f, v, signalInput(v))
+		if old.Signal.Disposition != "STALE_ACTIVATION" {
+			t.Fatal("old wait signal accepted")
+		}
+		resolveWorkflowStep(t, f, v.ID, "left", "SUCCEEDED", `{"n":9007199254740993}`)
+		continueWorkflow(t, f, v.ID)
+		in := signalInput(v)
+		in.TargetActivationSequence = 2
+		sendSignal(t, f, v, in)
+		runClock(t, f, clockCandidate(t, f, v.ID, "signal"))
+		ref := retryStart(t, f)
+		retryFinish(t, f, ref, "SUCCEEDED", false)
+		runtimeConsume(t, f, newRuntime(t, db), ref.TaskID)
+		runtimeView(t, f, v.ID, "SUCCEEDED")
+	})
+	t.Run("timer_rearm_uses_first_activation_elapsed_time", func(t *testing.T) {
+		f, v := waitRecoveryFixture(t, ctx, db, "TIMER", 2000)
+		var original time.Time
+		if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=?`, f.resources.NamespaceID).Scan(&original); err != nil {
+			t.Fatal(err)
+		}
+		time.Sleep(300 * time.Millisecond)
+		out := rearmWorkflowWait(t, f, v.ID, "timer")
+		if f.count("weave_waits", "deadline_at=?", original) != 2 {
+			t.Fatal("timer rearm reset original delay")
+		}
+		resolveWorkflowStep(t, f, v.ID, "left", "SUCCEEDED", `{"n":9007199254740993}`)
+		continueWorkflow(t, f, v.ID)
+		c := clockCandidate(t, f, v.ID, "timer")
+		waitClockDue(t, f, c)
+		runClock(t, f, c)
+		if f.count("weave_activations", "id=? AND state='SUCCEEDED'", out.Step.ActivationID) != 1 {
+			t.Fatal("timer expected new activation delay")
+		}
+	})
+	t.Run("wait_rearm_bound_same_key_race_and_stale_target", func(t *testing.T) {
+		f, v := waitRecoveryFixture(t, ctx, db, "SIGNAL", 60000)
+		in := recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v.ID, "signal"), WaitVerified: true}
+		key := string(executionID(t))
+		var wg sync.WaitGroup
+		errs := make(chan error, 8)
+		results := make(chan recovery.Result, 8)
+		for i := 0; i < 8; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				out, err := api.RearmWait(ctx, f.token, v.ID, key, in)
+				errs <- err
+				results <- out
+			}()
+		}
+		wg.Wait()
+		close(errs)
+		close(results)
+		for err := range errs {
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+		fresh := 0
+		for out := range results {
+			if !out.Deduplicated {
+				fresh++
+			}
+		}
+		if fresh != 1 {
+			t.Fatal("duplicate wait generations")
+		}
+		stale := in
+		stale.ExpectedRevision = recoveryDecision(t, f, v.ID).ExpectedRevision
+		if _, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), stale); fault.Code(err) != "STATE_CONFLICT" {
+			t.Fatal("old activation rearmed latest wait", err)
+		}
+		for sequence := uint32(2); sequence < recovery.MaxWaitActivations; sequence++ {
+			rearmWorkflowWait(t, f, v.ID, "signal")
+		}
+		if _, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v.ID, "signal"), WaitVerified: true}); fault.Code(err) != "STATE_CONFLICT" || f.count("weave_waits", "1=1") != int(recovery.MaxWaitActivations) {
+			t.Fatal("wait rearm unbounded", err)
+		}
+	})
+	t.Run("pending_signal_cannot_be_discarded_by_rearm", func(t *testing.T) {
+		f, v := waitRecoveryFixture(t, ctx, db, "SIGNAL", 30000)
+		sendSignal(t, f, v, signalInput(v))
+		_, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v.ID, "signal"), WaitVerified: true})
+		if fault.Code(err) != "STATE_CONFLICT" || f.count("weave_signals", "state='PENDING'") != 1 || f.count("weave_waits", "1=1") != 1 {
+			t.Fatal("pending evidence discarded", err)
+		}
+	})
+	t.Run("wait_rearm_audit_failure_and_expired_deadline", func(t *testing.T) {
+		f, v := waitRecoveryFixture(t, ctx, db, "TIMER", 1500)
+		in := recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v.ID, "timer"), WaitVerified: true}
+		trigger := "reject_wait_rearm_" + string(executionID(t))
+		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled rearm audit failure'")
+		_, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), in)
+		f.exec("DROP TRIGGER " + trigger)
+		if err == nil || f.count("weave_waits", "state='WAITING'") != 1 || f.count("weave_activations", "id=? AND state='WAITING'", in.ActivationID) != 1 || recoveryDecision(t, f, v.ID).ExpectedRevision != in.ExpectedRevision {
+			t.Fatal("partial wait rearm committed", err)
+		}
+		var due time.Time
+		if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=?`, f.resources.NamespaceID).Scan(&due); err != nil {
+			t.Fatal(err)
+		}
+		time.Sleep(time.Until(due.Add(20 * time.Millisecond)))
+		if _, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), in); fault.Code(err) != "STATE_CONFLICT" || f.count("weave_waits", "1=1") != 1 {
+			t.Fatal("expired timer was restarted", err)
+		}
+	})
+}