From 3182f82a510d2a61bdcbc4460f4c2350bd9208c8 Mon Sep 17 00:00:00 2001 From: Henry Date: Tue, 8 Sep 2026 20:27:47 +0800 Subject: [PATCH] test(office): prove a dropped children-completed wake gets recovered (#3486) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- .../service/scheduler_wake_reconciler_test.go | 229 ++++++++++++++++++ 1 file changed, 229 insertions(+) diff --git a/apps/backend/internal/office/service/scheduler_wake_reconciler_test.go b/apps/backend/internal/office/service/scheduler_wake_reconciler_test.go index bf9b46cfab..be313b6f48 100644 --- a/apps/backend/internal/office/service/scheduler_wake_reconciler_test.go +++ b/apps/backend/internal/office/service/scheduler_wake_reconciler_test.go @@ -3,11 +3,13 @@ package service_test import ( "context" "fmt" + "sync" "testing" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/office/models" "github.com/kandev/kandev/internal/office/service" + "github.com/kandev/kandev/internal/workflow/engine" ) // adoptOffice satisfies HasOfficeAdoption via an office_projects row, the @@ -324,6 +326,233 @@ func TestParentWakeReconciler_UnresolvedAndPausedDoNotStarveHealthyParent(t *tes t.Fatalf("zz-healthy-parent was never dispatched across 3 ticks behind 10 sticky candidates: %#v", dispatcher.Calls()) } +// oneShotFailureDispatcher wraps a queueRunDispatcher and fails the first +// call for a chosen trigger, then delegates every other call (including +// later calls for that same trigger) to the wrapped dispatcher. It records +// every call it sees so a test can assert dispatch counts and operation ids +// on both the failed attempt and the later recovery. +type oneShotFailureDispatcher struct { + inner *queueRunDispatcher + failTrigger engine.Trigger + failArmed bool + failErr error + + mu sync.Mutex + calls []dispatcherCall +} + +func (d *oneShotFailureDispatcher) HandleTrigger( + ctx context.Context, taskID string, trigger engine.Trigger, payload any, opID string, +) error { + d.mu.Lock() + d.calls = append(d.calls, dispatcherCall{taskID, trigger, payload, opID}) + d.mu.Unlock() + if d.failArmed && trigger == d.failTrigger { + d.failArmed = false + return d.failErr + } + return d.inner.HandleTrigger(ctx, taskID, trigger, payload, opID) +} + +func (d *oneShotFailureDispatcher) Calls() []dispatcherCall { + d.mu.Lock() + defer d.mu.Unlock() + out := make([]dispatcherCall, len(d.calls)) + copy(out, d.calls) + return out +} + +// countingDispatcher wraps a queueRunDispatcher and counts every call it +// sees, so a test can assert the reconciler never even attempted a +// dispatch instead of only checking the resulting run count. +type countingDispatcher struct { + inner *queueRunDispatcher + + mu sync.Mutex + n int +} + +func (d *countingDispatcher) HandleTrigger( + ctx context.Context, taskID string, trigger engine.Trigger, payload any, opID string, +) error { + d.mu.Lock() + d.n++ + d.mu.Unlock() + return d.inner.HandleTrigger(ctx, taskID, trigger, payload, opID) +} + +func (d *countingDispatcher) Count() int { + d.mu.Lock() + defer d.mu.Unlock() + return d.n +} + +// publishChildDone publishes the task.moved event that fires +// finalizeDone -> queueChildrenCompletedRun for parentID, the same event +// shape TestWakeOperationID_UnifiedAcrossEdgeAndReconcilerPaths uses. +func publishChildDone(t *testing.T, eb bus.EventBus, parentID, workerID string) { + t.Helper() + ctx := context.Background() + moveEvent := bus.NewEvent("task.moved", "test", map[string]string{ + "task_id": parentID + "-child-0", + "workspace_id": "ws-1", + "from_step_id": "step-1", + "to_step_id": "step-done", + "to_step_name": "Done", + "from_step_name": "In Progress", + "assignee_agent_profile_id": workerID, + "parent_id": parentID, + }) + if err := eb.Publish(ctx, "task.moved", moveEvent); err != nil { + t.Fatalf("publish task.moved: %v", err) + } +} + +// TestParentWakeReconciler_RecoversFailedEdgeDispatch is the regression test +// requested on PR #3271's review (discussion r3909576685): queueChildrenCompletedRun +// can fail after AreAllChildrenTerminal succeeds, including when the workflow +// dispatcher returns an error, and finalizeDone only logs that failure at Warn +// because ParentWakeReconciler is the documented recovery path. This proves +// the recovery actually happens: the edge-triggered dispatch is made to fail +// once, then a reconciler Tick must re-deliver the same wake (same operation +// id) as a real queued run. +func TestParentWakeReconciler_RecoversFailedEdgeDispatch(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + dispatcher := &oneShotFailureDispatcher{ + inner: &queueRunDispatcher{svc: svc}, + failTrigger: engine.TriggerOnChildrenCompleted, + failArmed: true, + failErr: fmt.Errorf("injected on_children_completed dispatch failure"), + } + svc.SetWorkflowEngineDispatcher(dispatcher) + + adoptOffice(t, svc, "ws-1") + seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1") + + // Edge-triggered path: the child lands in Done, firing + // finalizeDone -> queueChildrenCompletedRun for parent-1, which fails + // on the injected dispatcher error and is swallowed at Warn. + publishChildDone(t, eb, "parent-1", "worker-1") + + calls := dispatcher.Calls() + if len(calls) != 1 { + t.Fatalf("want exactly one dispatch attempt after the failed edge publish, got %d: %#v", + len(calls), calls) + } + if calls[0].taskID != "parent-1" || calls[0].trigger != engine.TriggerOnChildrenCompleted { + t.Fatalf("failed edge dispatch = task %q, trigger %q; want task parent-1, trigger %q", + calls[0].taskID, calls[0].trigger, engine.TriggerOnChildrenCompleted) + } + edgeOpID := calls[0].opID + if edgeOpID == "" { + t.Fatal("failed edge dispatch attempt carried an empty operation id") + } + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 0 { + t.Fatalf("failed edge dispatch left a queued run: %#v", runs) + } + receipt, err := svc.GetWakeReceiptForTest(ctx, "parent-1") + if err != nil { + t.Fatalf("get wake receipt: %v", err) + } + if receipt != nil { + t.Fatalf("failed edge dispatch left a receipt: %#v", receipt) + } + + // Level-triggered path: ListStuckParents still finds parent-1 stuck + // because nothing was persisted for it, so Tick must re-deliver. + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick: %v", err) + } + + calls = dispatcher.Calls() + if len(calls) != 2 { + t.Fatalf("want a second dispatch attempt from the reconciler, got %d: %#v", + len(calls), calls) + } + if calls[1].taskID != "parent-1" || calls[1].trigger != engine.TriggerOnChildrenCompleted { + t.Fatalf("reconciler dispatch = task %q, trigger %q; want task parent-1, trigger %q", + calls[1].taskID, calls[1].trigger, engine.TriggerOnChildrenCompleted) + } + reconcilerOpID := calls[1].opID + if reconcilerOpID != edgeOpID { + t.Fatalf("reconciler recovery used a different operation id than the failed edge attempt: edge=%q reconciler=%q", + edgeOpID, reconcilerOpID) + } + + runsAfter, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs after tick: %v", err) + } + if len(runsAfter) != 1 { + t.Fatalf("want exactly one queued run after recovery, got %d: %#v", len(runsAfter), runsAfter) + } + if runsAfter[0].Reason != service.RunReasonTaskChildrenCompleted { + t.Fatalf("recovered run reason = %q, want %q", runsAfter[0].Reason, service.RunReasonTaskChildrenCompleted) + } + + receiptAfter, err := svc.GetWakeReceiptForTest(ctx, "parent-1") + if err != nil { + t.Fatalf("get wake receipt after tick: %v", err) + } + if receiptAfter == nil || receiptAfter.DeliveryOperationID != edgeOpID { + t.Fatalf("recovery receipt = %#v, want an operation id matching the failed edge attempt %q", receiptAfter, edgeOpID) + } +} + +// TestParentWakeReconciler_DoesNotDoubleQueueASuccessfulEdgeDispatch is the +// non-vacuity control for TestParentWakeReconciler_RecoversFailedEdgeDispatch: +// identical setup, but the edge dispatch is left to succeed. It must be the +// one that queues the run, and the following Tick must add nothing, proving +// the recovery assertions above discriminate on the edge path's outcome +// rather than passing regardless of it. +func TestParentWakeReconciler_DoesNotDoubleQueueASuccessfulEdgeDispatch(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + + dispatcher := &countingDispatcher{inner: &queueRunDispatcher{svc: svc}} + svc.SetWorkflowEngineDispatcher(dispatcher) + + adoptOffice(t, svc, "ws-1") + seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1") + + publishChildDone(t, eb, "parent-1", "worker-1") + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 1 { + t.Fatalf("successful edge dispatch did not queue a run: %#v", runs) + } + if dispatcher.Count() != 1 { + t.Fatalf("want exactly one dispatch attempt from the successful edge publish, got %d", dispatcher.Count()) + } + + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick: %v", err) + } + + if dispatcher.Count() != 1 { + t.Fatalf("want the reconciler to skip dispatch entirely behind a successful edge delivery, got %d attempts", dispatcher.Count()) + } + + runsAfter, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs after tick: %v", err) + } + if len(runsAfter) != 1 { + t.Fatalf("tick added a run behind a successful edge dispatch: %#v", runsAfter) + } +} + // TestWakeOperationID_UnifiedAcrossEdgeAndReconcilerPaths is the regression // test for the duplicate on_children_completed wake race: the edge-triggered // path (queueChildrenCompletedRun, fired off task.moved) and the