diff --git a/backend/src/CodeSpace.Core/Handlers/CommandHandlers/Workflows/ReconcileStuckRunsCommandHandler.cs b/backend/src/CodeSpace.Core/Handlers/CommandHandlers/Workflows/ReconcileStuckRunsCommandHandler.cs index 8ee056bdf..2d40c9df9 100644 --- a/backend/src/CodeSpace.Core/Handlers/CommandHandlers/Workflows/ReconcileStuckRunsCommandHandler.cs +++ b/backend/src/CodeSpace.Core/Handlers/CommandHandlers/Workflows/ReconcileStuckRunsCommandHandler.cs @@ -28,6 +28,7 @@ public async Task Handle(ReconcileStuckRunsCommand r RecoveredStrandedTimerWait = summary.RecoveredStrandedTimerWait, RecoveredStrandedSupervisorInfraParkWait = summary.RecoveredStrandedSupervisorInfraParkWait, RecoveredStrandedSubworkflowParent = summary.RecoveredStrandedSubworkflowParent, + CancelledUnwaitedChildRun = summary.CancelledUnwaitedChildRun, }; } } diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/IWorkflowRunDispatcher.cs b/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/IWorkflowRunDispatcher.cs index 829b0044c..c5e1dd669 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/IWorkflowRunDispatcher.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/IWorkflowRunDispatcher.cs @@ -28,6 +28,9 @@ public interface IWorkflowRunDispatcher /// false if the row was not in Pending state (another caller won /// the race, OR the run is already terminal). This is NOT an error; the loser /// silently returns. Reconciler relies on this idempotence. + /// false, too, for a sub-workflow child with a finished ancestor (reached through + /// sub-workflow children only — a rerun's lineage never counts). Nothing waits for it; the + /// row is left as it is, for the stuck-run reconciler to cancel. /// /// /// Throws when the background-job client itself fails to enqueue (e.g. Hangfire storage diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/RunAncestry.cs b/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/RunAncestry.cs new file mode 100644 index 000000000..532177771 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/RunAncestry.cs @@ -0,0 +1,41 @@ +using CodeSpace.Core.Services.Workflows.Engine; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Enums; + +namespace CodeSpace.Core.Services.Workflows.Dispatch; + +/// +/// The sub-workflow ancestry walk the dispatcher refuses a run by, the same walk ExecutionAuthorityService denies an +/// agent's authority by: up the parent chain while each run is a sub-workflow child, refusing at the first ancestor that has +/// finished. Only a child's parent can be waiting for it; a rerun's or replay's parent is its lineage, finished by definition, +/// so the walk ends at the first run that is not a child. +/// +internal static class RunAncestry +{ + /// One run on the walk: what kind of run it is, the run it hangs under, and how far it has got. + internal sealed record Hop(string SourceType, Guid? ParentRunId, WorkflowRunStatus Status); + + /// The most child hops a walk takes. Nesting is capped at , so a longer chain of children is not one the engine staged. + internal const int MaxChildHops = SubworkflowService.MaxDepth; + + /// + /// Whether some ancestor of , reached through sub-workflow children only, has finished — or is missing, + /// loops back, or lies deeper than nesting allows, none of which the engine stages. reads one run, null when absent. + /// + internal static async Task HasFinishedAncestorAsync(Guid runId, Func> load) + { + var lineage = new HashSet { runId }; + var run = await load(runId).ConfigureAwait(false); + + for (var hops = 0; run is { SourceType: WorkflowRunSourceTypes.ChildWorkflow, ParentRunId: { } parentId }; hops++) + { + if (hops == MaxChildHops || !lineage.Add(parentId)) return true; + + run = await load(parentId).ConfigureAwait(false); + + if (run is null || WorkflowRunState.IsTerminal(run.Status)) return true; + } + + return false; + } +} diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/WorkflowRunDispatcher.cs b/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/WorkflowRunDispatcher.cs index d88e93209..dced2071d 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/WorkflowRunDispatcher.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Dispatch/WorkflowRunDispatcher.cs @@ -10,7 +10,7 @@ namespace CodeSpace.Core.Services.Workflows.Dispatch; /// /// Atomic CAS + revert-on-throw isolated from the three call sites (manual run, replay run, -/// webhook dispatcher). +/// webhook dispatcher), behind a refusal of any sub-workflow child whose parent run has finished. /// public sealed class WorkflowRunDispatcher : IWorkflowRunDispatcher, IScopedDependency { @@ -27,6 +27,8 @@ public WorkflowRunDispatcher(CodeSpaceDbContext db, ICodeSpaceBackgroundJobClien public async Task DispatchAsync(Guid runId, CancellationToken cancellationToken) { + if (await HasFinishedAncestorAsync(runId, cancellationToken).ConfigureAwait(false)) return RefuseUnderFinishedAncestor(runId); + // Atomic CAS — Pending → Enqueued. // ExecuteUpdateAsync emits a single UPDATE ... WHERE Status = Pending statement. // Two concurrent callers (e.g. RunManuallyAsync + a reconciler that found the same @@ -89,4 +91,22 @@ await _db.WorkflowRun throw; } } + + /// + /// Whether the run is a sub-workflow child whose ancestry has finished. Every resume, revive and reconciler re-dispatch comes + /// through here, so this is the last door before a worker walks the run: a child whose parent has ended — a timer still firing + /// after the parent failed, say — would run toward a wait nothing holds open any more, spending for no one. + /// + private async Task HasFinishedAncestorAsync(Guid runId, CancellationToken cancellationToken) => + await RunAncestry.HasFinishedAncestorAsync(runId, id => LoadHopAsync(id, cancellationToken)).ConfigureAwait(false); + + private async Task LoadHopAsync(Guid runId, CancellationToken cancellationToken) => + await _db.WorkflowRun.AsNoTracking().Where(r => r.Id == runId).Select(r => new RunAncestry.Hop(r.SourceType, r.ParentRunId, r.Status)).SingleOrDefaultAsync(cancellationToken).ConfigureAwait(false); + + /// Leave the run as it is — Pending, if a resume flipped it — for the stuck-run reconciler, which cancels a sub-workflow child no live parent waits for. + private bool RefuseUnderFinishedAncestor(Guid runId) + { + _logger.LogInformation("WorkflowRunDispatcher: run {RunId} not dispatched — it is a sub-workflow whose parent run has finished, so nothing is waiting for it", runId); + return false; + } } diff --git a/backend/src/CodeSpace.Core/Services/Workflows/IWorkflowService.cs b/backend/src/CodeSpace.Core/Services/Workflows/IWorkflowService.cs index 8909bb7c0..939919797 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/IWorkflowService.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/IWorkflowService.cs @@ -151,8 +151,8 @@ public interface IWorkflowService /// teardown kills its agents and cancels its own children) — and emit a run.cancelled /// ledger record. The teardown is best-effort, so one agent's failed kill never strands the other agents, and one /// child's failed cancel never strands its siblings or the run's remaining waits. The agent-run reconciler's - /// parent-run-terminal guard re-cleans an agent run the teardown missed, and the stuck-run reconciler never dispatches - /// a child still staged under a finished parent; nothing stops a child that had already started. + /// parent-run-terminal guard re-cleans an agent run the teardown missed, and a child the teardown left live — staged or + /// started — is no longer waited for: the dispatcher will not queue it and the stuck-run reconciler cancels it. /// The teardown runs after the terminal commits and does not stop when the request that asked for it goes away — and /// the response waits for it across the whole sub-tree: every descendant's agents, children and waits are torn down /// before this returns. Returning at the commit instead is a follow-up, not built: a TearDownStoppedRunAsync @@ -171,17 +171,19 @@ public interface IWorkflowService /// Cancelled run, whose halting / interrupted steps re-run where they stopped. CAS → Pending then post-commit /// dispatch. TEAM-SCOPED: a foreign run throws (404). Returns true if this /// call drove it; false when the run is Success / Running, still parked on a pending wait (use - /// /resume), has nothing to re-run in place, or the CAS lost to a concurrent continue. Idempotent + race-safe + /// /resume), has nothing to re-run in place, is a sub-workflow child with a finished ancestor (continue the parent + /// instead; nothing waits for the child), or the CAS lost to a concurrent continue. Idempotent + race-safe /// (the dispatcher's Pending → Enqueued CAS is the double-dispatch guard). /// /// Reviving a terminal run starts a new run generation: a walk still winding down from the stop or failure, /// and a stop's teardown still in flight, carry the old one and touch nothing of the revived run. The revive also /// closes, in the same transaction, what the ended attempt left pending — its waits Discarded, its queued and - /// running agents cancelled, its staged child runs cancelled — so the revived run never parks on, reuses or folds - /// as its own work a late teardown then ends. A running agent gets only its row flipped there; its process kill, - /// credential revoke and spend settlement run once that commit is visible, and a stop's teardown landing later - /// loses every CAS on what the revive already ended. A Continue after a Failure, which has no teardown, gets its - /// kills from the revive alone. + /// running agents cancelled — so the revived run never parks on, reuses or folds as its own work a late teardown + /// then ends. A running agent gets only its row flipped there; its process kill, credential revoke and spend + /// settlement run once that commit is visible, and so does the full cancel of every sub-workflow child the ended + /// attempt left live, staged or started: the revived walk stages its own, so nothing waits for the old one. A stop's + /// teardown landing later loses every CAS on what the revive already ended. A Continue after a Failure, which has no + /// teardown, gets its kills from the revive alone. /// /// A step the continue re-runs starts over, because a wait the stop or failure closed carries no answer. So a /// flow.sleep parks a fresh timer for its FULL delay — the time it already slept is not credited — and a diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/IStuckRunReconcilerService.cs b/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/IStuckRunReconcilerService.cs index c8afafae4..50e91eb20 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/IStuckRunReconcilerService.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/IStuckRunReconcilerService.cs @@ -52,6 +52,12 @@ namespace CodeSpace.Core.Services.Workflows.Reconciliation; /// this wait kind had NO backstop until now — a lost deadline would strand the run on the park ladder /// FOREVER, even once the model plane recovered. Re-fire the same deadline resume; idempotent against the /// real (late) job via the resume's still-Pending gate. +/// Unwaited sub-workflow child (a live workflow.child run past a 2-min staging grace that no live +/// run holds a Pending Subworkflow wait for) — its parent ended without reaching it (the engine's failure +/// cleanup cancels only a child still Pending or Enqueued; a stop's cancel of a child can fail), or moved on without +/// it (a Continue's re-walk stages a fresh child). Nothing will ever read its result, yet a timer or an agent can +/// keep it spending. Cancelled through the full run cancel, FIRST in the tick, so no sweep after it revives one; +/// every wake sweep also leaves out a child whose parent has finished, and the dispatcher refuses one. /// /// /// Idempotent + safe to call concurrently from multiple replicas because every state @@ -88,5 +94,8 @@ public sealed record StuckRunReconcileSummary /// Stranded Subworkflow parents re-fired because the child's inline on-completion resume was lost (a crash between the child's terminal commit and the resume). 0 when no parent is parked on a terminal child. public int RecoveredStrandedSubworkflowParent { get; init; } - public int Total => RedispatchedFromPending + RevertedFromEnqueued + MarkedAbandonedFromRunning + RedispatchedFromStrandedSuspended + RecoveredSupervisorAdvance + RecoveredAbandonedSupervisorRun + ReleasedRerunLeases + RecoveredStrandedTimerWait + RecoveredStrandedSupervisorInfraParkWait + RecoveredStrandedSubworkflowParent; + /// Sub-workflow children cancelled because no live run holds a Pending Subworkflow wait naming them — a parent that ended without reaching them, or moved on without them. 0 when every live child is waited for. + public int CancelledUnwaitedChildRun { get; init; } + + public int Total => RedispatchedFromPending + RevertedFromEnqueued + MarkedAbandonedFromRunning + RedispatchedFromStrandedSuspended + RecoveredSupervisorAdvance + RecoveredAbandonedSupervisorRun + ReleasedRerunLeases + RecoveredStrandedTimerWait + RecoveredStrandedSupervisorInfraParkWait + RecoveredStrandedSubworkflowParent + CancelledUnwaitedChildRun; } diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/StuckRunReconcilerService.UnwaitedChildren.cs b/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/StuckRunReconcilerService.UnwaitedChildren.cs new file mode 100644 index 000000000..0462aa189 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/StuckRunReconcilerService.UnwaitedChildren.cs @@ -0,0 +1,120 @@ +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Enums; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; + +namespace CodeSpace.Core.Services.Workflows.Reconciliation; + +/// +/// Sub-workflow children no parent is waiting for: the sweep that cancels them, and the exclusion that keeps every wake sweep +/// off a child whose parent has finished in the meantime. +/// +public sealed partial class StuckRunReconcilerService +{ + /// + /// How old a sub-workflow child must be before the sweep may judge it unwaited. The engine commits a staged child a moment + /// before it parks the parent's Subworkflow wait naming it, so for that moment a wanted child has no wait. Matches + /// , so a child staged by a walk that died before parking is cancelled on the tick that + /// would otherwise re-dispatch it. Committed, not configured. + /// + public static readonly TimeSpan UnwaitedChildStagingGrace = TimeSpan.FromMinutes(2); + + /// + /// Cancel the sub-workflow children that no live run is waiting for — least recently attempted first, at most a tick, + /// each through the full so its own teardown kills its agents and cancels the + /// children it waits for in turn. A child is left this way when its parent ends without reaching it — the engine's failure + /// cleanup cancels only a child still Pending or Enqueued, a stop's cancel of a child can fail — or when its parent moves + /// on without it: a Continue closes the ended attempt's waits and its re-walk stages a fresh child. Each cancel stands + /// alone: a failure is logged and retried next tick, and the rest of the batch still goes. An attempt that fails or cancels + /// nothing stamps the child's LastModifiedDate, so it queues behind the rest instead of holding a batch slot forever. A level below a cancelled child + /// that its teardown did not reach is unwaited from then on, and a later tick takes it. + /// + internal async Task CancelUnwaitedChildRunsAsync(CancellationToken cancellationToken) + { + try + { + var children = await FindUnwaitedChildRunsAsync(cancellationToken).ConfigureAwait(false); + + var cancelled = 0; + + foreach (var child in children) + if (await TryCancelUnwaitedChildAsync(child, cancellationToken).ConfigureAwait(false)) cancelled++; + + return cancelled; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + // A failed find stops this sweep only; the sweeps before it have already run and the next tick tries again. + _logger.LogWarning(ex, "StuckRunReconciler: the sweep for sub-workflow children no parent is waiting for failed; will retry next tick"); + return 0; + } + } + + /// + /// The live sub-workflow children past that no live run holds a Pending + /// Subworkflow wait for. The match is sound because only SubworkflowService.StageChildRunAsync writes this + /// source type, and only the engine's park writes a Subworkflow wait, its token the id of the child it just staged — + /// so every child some live run will resume is named by such a wait. A rerun or replay fork carries its own source type + /// and is never a candidate; a child a live fork waits for is named by the fork's wait like any other. + /// + private async Task> FindUnwaitedChildRunsAsync(CancellationToken cancellationToken) + { + var threshold = DateTimeOffset.UtcNow - UnwaitedChildStagingGrace; + + return await _db.WorkflowRun.AsNoTracking() + .Where(r => r.SourceType == WorkflowRunSourceTypes.ChildWorkflow && NonTerminalRunStatuses.Contains(r.Status) && r.CreatedDate < threshold) + .Where(r => !_db.WorkflowRunWait.Any(w => w.Token == r.Id.ToString() && w.WaitKind == WorkflowWaitKinds.Subworkflow && w.Status == WorkflowWaitStatuses.Pending && _db.WorkflowRun.Any(p => p.Id == w.RunId && NonTerminalRunStatuses.Contains(p.Status)))) + .OrderBy(r => r.LastModifiedDate).ThenBy(r => r.CreatedDate) + .Take(BatchSize) + .Select(r => new UnwaitedChild(r.Id, r.TeamId)) + .ToListAsync(cancellationToken).ConfigureAwait(false); + } + + private async Task TryCancelUnwaitedChildAsync(UnwaitedChild child, CancellationToken cancellationToken) + { + try + { + var outcome = await _workflowService.CancelRunAsync(child.Id, child.TeamId, cancellationToken).ConfigureAwait(false); + + if (outcome?.Cancelled == true) return true; + + await RotateToBackAsync(child.Id, cancellationToken).ConfigureAwait(false); + return false; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + // Per-row resilience: one child's failed cancel doesn't abort the sweep. + _logger.LogWarning(ex, "StuckRunReconciler: cancelling sub-workflow child run {RunId}, which no parent is waiting for, failed; will retry next tick", child.Id); + await RotateToBackAsync(child.Id, cancellationToken).ConfigureAwait(false); + return false; + } + } + + /// Stamps a live child's LastModifiedDate (ExecuteUpdateAsync skips the audit hook) so the oldest-attempted-first batch moves past it. Best effort. + private async Task RotateToBackAsync(Guid runId, CancellationToken cancellationToken) + { + try + { + var now = DateTimeOffset.UtcNow; + + await _db.WorkflowRun.Where(r => r.Id == runId && NonTerminalRunStatuses.Contains(r.Status)).ExecuteUpdateAsync(s => s.SetProperty(r => r.LastModifiedDate, now), cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogWarning(ex, "StuckRunReconciler: moving sub-workflow child run {RunId} to the back of the unwaited-child batch failed", runId); + } + } + + /// + /// Every run a wake or re-dispatch sweep may revive: all but a sub-workflow child whose parent has finished. No one waits for + /// such a child — its parent's waits closed when it ended — so reviving it would spend on work nothing will read; + /// cancels it instead, and until that lands the dispatcher refuses it. The guard is + /// by source type because ParentRunId is also a rerun's or replay's lineage, finished by definition. + /// + private IQueryable RunsWithNoFinishedParent => _db.WorkflowRun.Where(r => !(r.SourceType == WorkflowRunSourceTypes.ChildWorkflow && _db.WorkflowRun.Any(p => p.Id == r.ParentRunId && TerminalRunStatuses.Contains(p.Status)))); + + private static WorkflowRunStatus[] NonTerminalRunStatuses => WorkflowRunState.NonTerminal; + + private sealed record UnwaitedChild(Guid Id, Guid TeamId); +} diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/StuckRunReconcilerService.cs b/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/StuckRunReconcilerService.cs index 6f4a5efa2..03f9ab859 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/StuckRunReconcilerService.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Reconciliation/StuckRunReconcilerService.cs @@ -16,8 +16,9 @@ namespace CodeSpace.Core.Services.Workflows.Reconciliation; /// /// Four independent sweeps, one per stuck-state class. Order matters less than you'd think /// — each sweep's CAS protects against racing a normal flow or another reconciler tick. +/// The one that must lead is : no sweep after it revives what it cancels. /// -public sealed class StuckRunReconcilerService : IStuckRunReconcilerService, IScopedDependency +public sealed partial class StuckRunReconcilerService : IStuckRunReconcilerService, IScopedDependency { /// Threshold for "Pending too long" — past this, re-dispatch. public static readonly TimeSpan PendingStuckAfter = TimeSpan.FromMinutes(2); @@ -99,15 +100,17 @@ public sealed class StuckRunReconcilerService : IStuckRunReconcilerService, ISco private readonly IWorkflowResumeService _resumeService; private readonly ISubworkflowService _subworkflowService; private readonly IRunRecordLogger _recordLogger; + private readonly IWorkflowService _workflowService; private readonly ILogger _logger; - public StuckRunReconcilerService(CodeSpaceDbContext db, IWorkflowRunDispatcher dispatcher, IWorkflowResumeService resumeService, ISubworkflowService subworkflowService, IRunRecordLogger recordLogger, ILogger logger) + public StuckRunReconcilerService(CodeSpaceDbContext db, IWorkflowRunDispatcher dispatcher, IWorkflowResumeService resumeService, ISubworkflowService subworkflowService, IRunRecordLogger recordLogger, IWorkflowService workflowService, ILogger logger) { _db = db; _dispatcher = dispatcher; _resumeService = resumeService; _subworkflowService = subworkflowService; _recordLogger = recordLogger; + _workflowService = workflowService; _logger = logger; } @@ -137,6 +140,10 @@ public async Task ReconcileAsync(CancellationToken can var recoveredSubworkflowParents = await RecoverStrandedSubworkflowParentsAsync(cancellationToken).ConfigureAwait(false); + // LAST: every cancel is a full teardown (agent kill wave) — slow — and the wake sweeps above already skip, and the dispatcher + // already refuses, a child whose parent has finished, so the cheap sweeps never wait on this one. + var cancelledUnwaitedChildren = await CancelUnwaitedChildRunsAsync(cancellationToken).ConfigureAwait(false); + var summary = new StuckRunReconcileSummary { RedispatchedFromPending = redispatched, @@ -149,12 +156,13 @@ public async Task ReconcileAsync(CancellationToken can RecoveredStrandedTimerWait = recoveredTimerWaits, RecoveredStrandedSupervisorInfraParkWait = recoveredInfraParkWaits, RecoveredStrandedSubworkflowParent = recoveredSubworkflowParents, + CancelledUnwaitedChildRun = cancelledUnwaitedChildren, }; if (summary.Total > 0) _logger.LogInformation( - "StuckRunReconciler: redispatched={Redispatched}, reverted={Reverted}, abandoned={Abandoned}, unstranded={Unstranded}, supervisorAdvances={SupervisorAdvances}, supervisorRunsRecovered={SupervisorRunsRecovered}, rerunLeasesReleased={RerunLeasesReleased}, timerWaitsRecovered={TimerWaitsRecovered}, infraParkWaitsRecovered={InfraParkWaitsRecovered}, subworkflowParentsRecovered={SubworkflowParentsRecovered}", - summary.RedispatchedFromPending, summary.RevertedFromEnqueued, summary.MarkedAbandonedFromRunning, summary.RedispatchedFromStrandedSuspended, summary.RecoveredSupervisorAdvance, summary.RecoveredAbandonedSupervisorRun, summary.ReleasedRerunLeases, summary.RecoveredStrandedTimerWait, summary.RecoveredStrandedSupervisorInfraParkWait, summary.RecoveredStrandedSubworkflowParent); + "StuckRunReconciler: redispatched={Redispatched}, reverted={Reverted}, abandoned={Abandoned}, unstranded={Unstranded}, supervisorAdvances={SupervisorAdvances}, supervisorRunsRecovered={SupervisorRunsRecovered}, rerunLeasesReleased={RerunLeasesReleased}, timerWaitsRecovered={TimerWaitsRecovered}, infraParkWaitsRecovered={InfraParkWaitsRecovered}, subworkflowParentsRecovered={SubworkflowParentsRecovered}, unwaitedChildrenCancelled={UnwaitedChildrenCancelled}", + summary.RedispatchedFromPending, summary.RevertedFromEnqueued, summary.MarkedAbandonedFromRunning, summary.RedispatchedFromStrandedSuspended, summary.RecoveredSupervisorAdvance, summary.RecoveredAbandonedSupervisorRun, summary.ReleasedRerunLeases, summary.RecoveredStrandedTimerWait, summary.RecoveredStrandedSupervisorInfraParkWait, summary.RecoveredStrandedSubworkflowParent, summary.CancelledUnwaitedChildRun); return summary; } @@ -163,17 +171,15 @@ public async Task ReconcileAsync(CancellationToken can /// Pending older than threshold: call DispatchAsync. The dispatcher's own CAS prevents /// double-dispatch if a normal flow is racing us; we just hand the id back into the /// queue and Hangfire takes it from there. - /// Never a sub-workflow child whose parent has finished — one whose cancel failed during its parent's stop, say. - /// Dispatched, it would run under the finished parent toward a wait that parent no longer holds open; no one is waiting - /// for it. The guard is by source type because ParentRunId is also a rerun's lineage, finished by definition. + /// Never a sub-workflow child whose parent has finished () — one whose cancel + /// failed during its parent's stop, say. No one is waiting for it; cancels it. /// private async Task RedispatchStuckPendingAsync(CancellationToken cancellationToken) { var threshold = DateTimeOffset.UtcNow - PendingStuckAfter; - var stuckIds = await _db.WorkflowRun.AsNoTracking() - .Where(r => r.Status == WorkflowRunStatus.Pending && r.CreatedDate < threshold - && !(r.SourceType == WorkflowRunSourceTypes.ChildWorkflow && _db.WorkflowRun.Any(p => p.Id == r.ParentRunId && TerminalRunStatuses.Contains(p.Status)))) + var stuckIds = await RunsWithNoFinishedParent.AsNoTracking() + .Where(r => r.Status == WorkflowRunStatus.Pending && r.CreatedDate < threshold) .OrderBy(r => r.CreatedDate) .Take(BatchSize) .Select(r => r.Id) @@ -236,7 +242,7 @@ private async Task MarkAbandonedRunningAsync(CancellationToken cancellation // (or has no ledger at all). The outer join via a sub-select keeps this to a single // round-trip; the resulting IDs are the candidates we mark. var candidates = await ( - from run in _db.WorkflowRun.AsNoTracking() + from run in RunsWithNoFinishedParent.AsNoTracking() // an orphaned child is cancelled with its teardown by CancelUnwaitedChildRunsAsync, not failed bare where run.Status == WorkflowRunStatus.Running && run.StartedAt < runningThreshold let mostRecentLedger = _db.WorkflowRunRecord.AsNoTracking() .Where(rec => rec.RunId == run.Id) @@ -308,10 +314,9 @@ select run.Id /// concurrent resume that already drove the run (its flip won) leaves us with 0 rows → we skip, /// no double-dispatch. The resolved waits rehydrate as the suspended nodes' ResumePayloads on the /// re-walk, so the run continues from where it stranded. - /// Never a sub-workflow child whose parent has finished — one whose cancel failed during its parent's stop and whose - /// waits were then closed, say. Revived, it would run under the finished parent toward a wait that parent no longer holds - /// open; no one is waiting for it. It is left as it is and counted in the log, so it does not linger unseen. The guard is - /// by source type, as the Pending sweep's is, because ParentRunId is also a rerun's lineage, finished by definition. + /// Never a sub-workflow child whose parent has finished () — one whose cancel + /// failed during its parent's stop and whose waits were then closed, say. No one is waiting for it; + /// cancels it. /// private async Task RedispatchStrandedSuspendedAsync(CancellationToken cancellationToken) { @@ -347,28 +352,22 @@ private async Task RedispatchStrandedSuspendedAsync(CancellationToken cance /// /// The stranded Suspended runs to revive this tick, oldest first, at most — leaving out each - /// sub-workflow child whose parent has finished, and saying in the log how many it left. The exclusion is in the query, - /// not the loop: a child left alone stays Suspended, so one that took a place in the batch would take it every tick and - /// crowd out the runs that are wanted. + /// sub-workflow child whose parent has finished. The exclusion is in the query, not the loop: a child left out that took a + /// place in the batch would take it every tick its cancel failed and crowd out the runs that are wanted. /// private async Task> FindStrandedSuspendedIdsAsync(CancellationToken cancellationToken) { var threshold = DateTimeOffset.UtcNow - SuspendedStrandedAfter; - var stranded = _db.WorkflowRun.AsNoTracking() + return await RunsWithNoFinishedParent.AsNoTracking() .Where(r => r.Status == WorkflowRunStatus.Suspended && r.LastModifiedDate < threshold && r.CompletionParkedAt == null && !_db.WorkflowRunWait.Any(w => w.RunId == r.Id && w.Status == WorkflowWaitStatuses.Pending)) - .Select(r => new { r.Id, r.LastModifiedDate, ParentFinished = r.SourceType == WorkflowRunSourceTypes.ChildWorkflow && _db.WorkflowRun.Any(p => p.Id == r.ParentRunId && TerminalRunStatuses.Contains(p.Status)) }); - - var strandedIds = await stranded.Where(r => !r.ParentFinished).OrderBy(r => r.LastModifiedDate).Take(BatchSize).Select(r => r.Id).ToListAsync(cancellationToken).ConfigureAwait(false); - - var leftAlone = await stranded.CountAsync(r => r.ParentFinished, cancellationToken).ConfigureAwait(false); - - if (leftAlone > 0) _logger.LogInformation("StuckRunReconciler: left {Count} stranded Suspended sub-workflow(s) alone because their parent has finished", leftAlone); - - return strandedIds; + .OrderBy(r => r.LastModifiedDate) + .Take(BatchSize) + .Select(r => r.Id) + .ToListAsync(cancellationToken).ConfigureAwait(false); } /// @@ -407,7 +406,7 @@ private async Task RecoverSupervisorAdvancesAsync(CancellationToken cancell .Where(w => w.WaitKind == WorkflowWaitKinds.SupervisorDecision && w.Status == WorkflowWaitStatuses.Pending && w.CreatedAt < threshold - && _db.WorkflowRun.Any(r => r.Id == w.RunId && r.Status == WorkflowRunStatus.Suspended)) + && RunsWithNoFinishedParent.Any(r => r.Id == w.RunId && r.Status == WorkflowRunStatus.Suspended)) .OrderBy(w => w.CreatedAt) .Take(BatchSize) .Select(w => new { w.RunId, w.Id }) @@ -452,7 +451,7 @@ private async Task RecoverStrandedTimerWaitsAsync(CancellationToken cancell && w.Status == WorkflowWaitStatuses.Pending && w.WakeAt != null && w.WakeAt < threshold - && _db.WorkflowRun.Any(r => r.Id == w.RunId && r.Status == WorkflowRunStatus.Suspended)) + && RunsWithNoFinishedParent.Any(r => r.Id == w.RunId && r.Status == WorkflowRunStatus.Suspended)) .OrderBy(w => w.WakeAt) .Take(BatchSize) .Select(w => new { w.RunId, w.Id }) @@ -503,7 +502,7 @@ private async Task RecoverStrandedSupervisorInfraParkWaitsAsync(Cancellatio && w.WakeAt != null && w.WakeAt < threshold && w.PayloadJson != null - && _db.WorkflowRun.Any(r => r.Id == w.RunId && r.Status == WorkflowRunStatus.Suspended)) + && RunsWithNoFinishedParent.Any(r => r.Id == w.RunId && r.Status == WorkflowRunStatus.Suspended)) .OrderBy(w => w.WakeAt) .Take(BatchSize) .Select(w => new { w.Id, w.PayloadJson }) @@ -639,7 +638,7 @@ private async Task> FindRecoverableSupervisorRunsAsync(CancellationTo var livenessThreshold = DateTimeOffset.UtcNow - LedgerLivenessWindow; return await ( - from run in _db.WorkflowRun.AsNoTracking() + from run in RunsWithNoFinishedParent.AsNoTracking() where run.Status == WorkflowRunStatus.Running && run.StartedAt < runningThreshold let mostRecentLedger = _db.WorkflowRunRecord.AsNoTracking().Where(rec => rec.RunId == run.Id).Max(rec => (DateTimeOffset?)rec.OccurredAt) where mostRecentLedger == null || mostRecentLedger < livenessThreshold @@ -708,7 +707,7 @@ private async Task> FindPreClaimSupervisorCrashCandidatesAsync(Cancel var livenessThreshold = DateTimeOffset.UtcNow - LedgerLivenessWindow; var candidates = await ( - from run in _db.WorkflowRun.AsNoTracking() + from run in RunsWithNoFinishedParent.AsNoTracking() where run.Status == WorkflowRunStatus.Running && run.StartedAt < runningThreshold let mostRecentLedger = _db.WorkflowRunRecord.AsNoTracking().Where(rec => rec.RunId == run.Id).Max(rec => (DateTimeOffset?)rec.OccurredAt) where mostRecentLedger == null || mostRecentLedger < livenessThreshold diff --git a/backend/src/CodeSpace.Core/Services/Workflows/WorkflowRunState.cs b/backend/src/CodeSpace.Core/Services/Workflows/WorkflowRunState.cs index bc28752bd..51d3aee4b 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/WorkflowRunState.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/WorkflowRunState.cs @@ -14,6 +14,9 @@ public static class WorkflowRunState public static bool IsTerminal(WorkflowRunStatus status) => status is WorkflowRunStatus.Success or WorkflowRunStatus.Failure or WorkflowRunStatus.Cancelled; + /// The non-terminal statuses, derived once from so a query cannot drift from it. + internal static readonly WorkflowRunStatus[] NonTerminal = Enum.GetValues().Where(s => !IsTerminal(s)).ToArray(); + /// True when an operator cancel may flip to Cancelled — every non-terminal state (Pending/Enqueued/Running/Suspended). public static bool IsCancellable(WorkflowRunStatus status) => !IsTerminal(status); } diff --git a/backend/src/CodeSpace.Core/Services/Workflows/WorkflowService.cs b/backend/src/CodeSpace.Core/Services/Workflows/WorkflowService.cs index 5a9c7275a..927344148 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/WorkflowService.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/WorkflowService.cs @@ -907,6 +907,8 @@ public async Task ContinueRunAsync(Guid runId, Guid teamId, CancellationTo if (status is null) throw new KeyNotFoundException($"WorkflowRun {runId} not found in team {teamId}."); + if (await HasFinishedAncestorAsync(runId, cancellationToken).ConfigureAwait(false)) return RefuseContinueUnderFinishedAncestor(runId); + // Three IN-PLACE continues, all same-run-id revivals (never a fork, unlike replay / rerun-from-node): a STRANDED // Suspended run (the reconciler twin), a terminal FAILURE run (re-run the node[s] that halted it, where they // died), and a terminal CANCELLED run the operator stopped mid-flight (re-run the frontier the stop interrupted). @@ -920,6 +922,19 @@ public async Task ContinueRunAsync(Guid runId, Guid teamId, CancellationTo }; } + /// + /// A sub-workflow child whose ancestor has finished is not revived: no one waits for it, the dispatcher would refuse its + /// re-dispatch, and the reconciler would cancel it — a true here would report a continue that never happens. + /// + private async Task HasFinishedAncestorAsync(Guid runId, CancellationToken cancellationToken) => + await RunAncestry.HasFinishedAncestorAsync(runId, async id => await _db.WorkflowRun.AsNoTracking().Where(r => r.Id == id).Select(r => new RunAncestry.Hop(r.SourceType, r.ParentRunId, r.Status)).SingleOrDefaultAsync(cancellationToken).ConfigureAwait(false)).ConfigureAwait(false); + + private bool RefuseContinueUnderFinishedAncestor(Guid runId) + { + _logger.LogWarning("Workflow run {RunId} not continued: it is a sub-workflow child whose parent has finished, so nothing waits for it. Continue the parent instead", runId); + return false; + } + /// /// The user-triggered, single-run twin of the reconciler's stranded-Suspended re-dispatch /// (StuckRunReconcilerService.RedispatchStrandedSuspendedAsync) — it drives the EXACT continuation a stranded @@ -1056,22 +1071,24 @@ private async Task ContinueCancelledRunAsync(Guid runId, Guid teamId, Canc /// Then what the ended attempt left pending is closed (). The bump comes /// first because it takes the run row's lock: a park, a supervisor decision or spawn wave of the old walk either /// committed before it, and the close sees what it staged, or is refused at its own fence on the old generation. - /// The cancelled agents' side effects — process kills, credential revokes, spend settlement — run once that commit - /// is visible (), ahead of the revived walk's dispatch. + /// The cancelled agents' side effects — process kills, credential revokes, spend settlement — and the cancels of the + /// ended attempt's children run once that commit is visible (, + /// ), ahead of the revived walk's dispatch. /// private async Task ReviveTerminalRunAsync(Guid runId, Guid teamId, WorkflowRunStatus terminal, CancellationToken cancellationToken) { - IReadOnlyList runningCancelled; + EndedAttempt ended; await using (var transaction = await ScopedTransaction.OwnOrJoinAsync(_db.Database, cancellationToken).ConfigureAwait(false)) { if (!await BumpToNewGenerationAsync(runId, teamId, terminal, cancellationToken).ConfigureAwait(false)) return false; - runningCancelled = await CloseEndedAttemptAsync(runId, cancellationToken).ConfigureAwait(false); + ended = await CloseEndedAttemptAsync(runId, cancellationToken).ConfigureAwait(false); await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); } - await FinishRunningCancelsAsync(runningCancelled, cancellationToken).ConfigureAwait(false); + await FinishRunningCancelsAsync(ended.RunningAgents, cancellationToken).ConfigureAwait(false); + await CancelEndedAttemptChildrenAsync(ended.Children, teamId, cancellationToken).ConfigureAwait(false); return true; } @@ -1091,43 +1108,55 @@ await _db.WorkflowRun /// failure cleanup — leaves when it runs first, whether it has run yet or not, and never from less. Every still-Pending /// wait goes Discarded: closed unanswered, so its step parks a fresh one instead of the revived walk adopting it (a /// re-entered map branch) or re-parking on it (a supervisor's agent barrier). Every agent still Queued or Running is - /// cancelled, and every staged child run still Pending or Enqueued: nothing the ended attempt started can be picked - /// up by a worker, reclaimed by the revived supervisor, or folded onto its tape as still running. A Running agent - /// gets only its row flipped here (); returned, so its side effects run after - /// commit. A stop's teardown landing afterwards finds nothing of the revived run's: its close is keyed by the stopped - /// attempt's own agents and children, its run-wide close is fenced on the stopped generation, and every CAS it takes - /// on what was already cancelled here simply loses — which also makes this the only kill a Continue after a Failure - /// gets, since that path has no teardown. + /// cancelled: nothing the ended attempt started can be picked up by a worker, reclaimed by the revived supervisor, or + /// folded onto its tape as still running. A Running agent gets only its row flipped here + /// (), and every sub-workflow child still live is read here and cancelled after + /// commit — whether staged, walking or parked, since the revived walk stages its own and nothing waits for the old one + /// any more; both are returned for that. A stop's teardown landing afterwards finds nothing of the revived run's: its + /// close is keyed by the stopped attempt's own agents and children, its run-wide close is fenced on the stopped + /// generation, and every CAS it takes on what was already cancelled here simply loses — which also makes this the only + /// kill a Continue after a Failure gets, since that path has no teardown. /// - private async Task> CloseEndedAttemptAsync(Guid runId, CancellationToken cancellationToken) + private async Task CloseEndedAttemptAsync(Guid runId, CancellationToken cancellationToken) { - var stagedChildren = await StagedChildRunIdsAsync(runId, cancellationToken).ConfigureAwait(false); + var children = await LiveChildRunIdsAsync(runId, cancellationToken).ConfigureAwait(false); var waitsDiscarded = await DiscardPendingWaitsAsync(runId, cancellationToken).ConfigureAwait(false); var queuedCancelled = await CancelQueuedAgentsAsync(runId, cancellationToken).ConfigureAwait(false); var runningCancelled = await CancelRunningAgentRowsAsync(runId, cancellationToken).ConfigureAwait(false); - await CancelStagedChildRunsAsync(stagedChildren, cancellationToken).ConfigureAwait(false); + if (waitsDiscarded + queuedCancelled + runningCancelled.Count + children.Count > 0) + _logger.LogInformation("Continue closed what the ended attempt left. RunId={RunId} WaitsDiscarded={WaitsDiscarded} QueuedAgentsCancelled={QueuedAgentsCancelled} RunningAgentsCancelled={RunningAgentsCancelled} ChildrenToCancel={ChildrenToCancel}", runId, waitsDiscarded, queuedCancelled, runningCancelled.Count, children.Count); - if (waitsDiscarded + queuedCancelled + runningCancelled.Count + stagedChildren.Count > 0) - _logger.LogInformation("Continue closed what the ended attempt left. RunId={RunId} WaitsDiscarded={WaitsDiscarded} QueuedAgentsCancelled={QueuedAgentsCancelled} RunningAgentsCancelled={RunningAgentsCancelled} StagedChildren={StagedChildren}", runId, waitsDiscarded, queuedCancelled, runningCancelled.Count, stagedChildren.Count); - - return runningCancelled; + return new EndedAttempt(runningCancelled, children); } + /// What the revive leaves for after its commit: the Running agents it flipped, and the live sub-workflow children it read. + private sealed record EndedAttempt(IReadOnlyList RunningAgents, IReadOnlyList Children); + /// - /// Cancel the stopped attempt's staged child runs still Pending or Enqueued, in the revive's transaction, so no worker - /// claims one between the Continue and the stop's teardown — the same status-guarded CAS the engine's own terminal - /// cleanup takes for a staged child, which has started nothing of its own yet. One already walking is left to that - /// teardown, which cancels every child it captured through the full cancel. + /// The run's sub-workflow children not yet finished, read inside the revive's transaction — before the revived walk is + /// dispatched, so none of them is a child it staged. By parent, not by the run's waits: the engine's failure cleanup or a + /// stop's teardown may already have closed the wait that named a child still walking. /// - private async Task CancelStagedChildRunsAsync(IReadOnlyList ids, CancellationToken cancellationToken) + private async Task> LiveChildRunIdsAsync(Guid runId, CancellationToken cancellationToken) => + await _db.WorkflowRun.AsNoTracking() + .Where(r => r.ParentRunId == runId && r.SourceType == WorkflowRunSourceTypes.ChildWorkflow && WorkflowRunState.NonTerminal.Contains(r.Status)) + .OrderBy(r => r.CreatedDate) + .Select(r => r.Id) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + /// + /// Cancel the ended attempt's children through the full cancel once the revive's commit is visible, so each child's own + /// teardown kills its agents and cancels its own children in turn; one child's failure strands neither its siblings nor + /// the Continue. On , as the running agents' finishes: owed to a revive that has + /// committed. One whose cancel fails is left to the stuck-run reconciler, which cancels a child no live run waits for. + /// + private async Task CancelEndedAttemptChildrenAsync(IReadOnlyList children, Guid teamId, CancellationToken cancellationToken) { - if (ids.Count == 0) return; + if (children.Count == 0) return; - await _db.WorkflowRun - .Where(r => ids.Contains(r.Id) && (r.Status == WorkflowRunStatus.Pending || r.Status == WorkflowRunStatus.Enqueued)) - .ExecuteUpdateAsync(s => s.SetProperty(r => r.Status, WorkflowRunStatus.Cancelled).SetProperty(r => r.CompletedAt, (DateTimeOffset?)DateTimeOffset.UtcNow), cancellationToken).ConfigureAwait(false); + await _postCommit.RunAfterCommitAsync(_ => CancelChildRunsAsync(children, teamId, CancellationToken.None), cancellationToken).ConfigureAwait(false); } /// Close the run's still-Pending waits as Discarded, which no replay reader treats as an answer; returns how many. @@ -1335,10 +1364,6 @@ private async Task> StagedChildRunIdsAsync(Guid runId, int g return ChildRunIds(tokens); } - /// The child runs the run's still-Pending Subworkflow waits staged — read by the revive, inside its own transaction, before it discards those waits. - private async Task> StagedChildRunIdsAsync(Guid runId, CancellationToken cancellationToken) => - ChildRunIds(await _db.WorkflowRunWait.AsNoTracking().Where(w => w.RunId == runId && w.WaitKind == WorkflowWaitKinds.Subworkflow && w.Status == WorkflowWaitStatuses.Pending).Select(w => w.Token).ToListAsync(cancellationToken).ConfigureAwait(false)); - /// A Subworkflow wait's token is its child run id; one that does not parse names no child. private static IReadOnlyList ChildRunIds(IEnumerable tokens) => tokens.Select(t => Guid.TryParse(t, out var id) ? id : (Guid?)null).Where(id => id.HasValue).Select(id => id!.Value).ToList(); @@ -1374,8 +1399,8 @@ private async Task ReReadTerminalOutcomeAsync(Guid runId, Guid /// Each step is best-effort on its own (), as is each agent's kill and each /// child's cancel inside its step: the run is already Cancelled, so one failure never aborts the cancel, and never /// costs what comes after it — a failed read or wait close still leaves the kill-wave to run. The reconciler's - /// parent-run-terminal guard re-cleans an agent run a failure missed, and the stuck-run reconciler never dispatches a - /// child still staged under a finished parent, but nothing stops a child that had already started. + /// parent-run-terminal guard re-cleans an agent run a failure missed, and the stuck-run reconciler cancels a child a + /// failure left live — staged or started — since no live run waits for it any more. /// /// Neither wait close waits on a row lock: both skip a wait another transaction holds. That is how a Continue /// landing mid-teardown is met — its revive discards every Pending wait of the run in its own transaction and holds @@ -1398,7 +1423,7 @@ private async Task TearDownCancelledRunAsync(Guid runId, Guid teamId, int g return agentRunsCancelled; } - /// One best-effort teardown step: a failure is logged and the teardown goes on with . The reconciler's parent-run-terminal guard re-cleans an agent run a failed step missed while the run stays stopped. + /// One best-effort teardown step: a failure is logged and the teardown goes on with . The reconciler's parent-run-terminal guard re-cleans an agent run a failed step missed while the run stays stopped, and the stuck-run reconciler cancels a sub-workflow child one left live. private async Task TearDownStepAsync(Func> step, T fallback, string stepName, Guid runId) { try @@ -1407,7 +1432,7 @@ private async Task TearDownStepAsync(Func> step, T fallback, strin } catch (Exception ex) { - _logger.LogWarning(ex, "Run {RunId} was cancelled but the teardown step to {Step} failed; the rest of the teardown still runs — the reconciler catches an orphaned agent run, not an unfinished sub-workflow", runId, stepName); + _logger.LogWarning(ex, "Run {RunId} was cancelled but the teardown step to {Step} failed; the rest of the teardown still runs — the reconcilers catch an orphaned agent run or sub-workflow", runId, stepName); return fallback; } } @@ -1555,7 +1580,7 @@ private async Task CancelChildRunQuietlyAsync(Guid childRunId, Guid teamId, Canc } catch (Exception ex) { - _logger.LogWarning(ex, "Cancelling sub-workflow child run {ChildRunId} failed; its siblings and its parent's remaining waits are still torn down. A staged child is never dispatched under the finished parent, but nothing stops one that had already started", childRunId); + _logger.LogWarning(ex, "Cancelling sub-workflow child run {ChildRunId} failed; its siblings and its parent's remaining waits are still torn down. Nothing waits for it now, so the dispatcher will not queue it and the stuck-run reconciler cancels it", childRunId); } } diff --git a/backend/src/CodeSpace.Messages/Commands/Workflows/ReconcileStuckRunsCommand.cs b/backend/src/CodeSpace.Messages/Commands/Workflows/ReconcileStuckRunsCommand.cs index dc27c8e2c..d7ffde324 100644 --- a/backend/src/CodeSpace.Messages/Commands/Workflows/ReconcileStuckRunsCommand.cs +++ b/backend/src/CodeSpace.Messages/Commands/Workflows/ReconcileStuckRunsCommand.cs @@ -39,4 +39,7 @@ public sealed record ReconcileStuckRunsResponse /// Stranded Subworkflow parents re-fired because the child's inline on-completion resume was lost. 0 when no parent is parked on a terminal child. public required int RecoveredStrandedSubworkflowParent { get; init; } + + /// Sub-workflow children cancelled because no live run was waiting for them. 0 when every live child is waited for. + public required int CancelledUnwaitedChildRun { get; init; } } diff --git a/backend/tests/CodeSpace.E2ETests/Workflows/UnwaitedChildRunE2ETests.cs b/backend/tests/CodeSpace.E2ETests/Workflows/UnwaitedChildRunE2ETests.cs new file mode 100644 index 000000000..e6c3c6ee0 --- /dev/null +++ b/backend/tests/CodeSpace.E2ETests/Workflows/UnwaitedChildRunE2ETests.cs @@ -0,0 +1,326 @@ +using System.Data.Common; +using Autofac; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Services.Workflows; +using CodeSpace.Core.Services.Workflows.Engine; +using CodeSpace.Core.Services.Workflows.Reconciliation; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.IntegrationTests.Infrastructure.Jobs; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Messages.Authorization; +using CodeSpace.Messages.Commands.Workflows; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Dtos.Workflows; +using CodeSpace.Messages.Enums; +using MediatR; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Shouldly; + +namespace CodeSpace.E2ETests.Workflows; + +/// +/// 🟢 E2E (high fidelity, Rule 12): a sub-workflow child that no parent waits for any more stops spending, across the real +/// pipeline — the REAL engine walks parent and child, the REAL approval resume re-walks the parent into a failing sibling, +/// the REAL Continue and stop revive or end the parent, and the REAL stuck-run reconciler, sent through the mediator as +/// the recurring job sends it, cancels what is left. Only Hangfire's transport is the in-memory client, and a parked +/// timer's wake is fired the way its scheduled job fires it. +/// +/// The parent: start → { sub (a child that sleeps an hour), two approval gates, ready → boom (fails once) } → end. +/// The child parks on its timer; approvals re-walk the parent, boom fails it, and the engine's failure cleanup +/// cancels only children still Pending or Enqueued — so the parked child is left with nothing waiting for it. +/// +[Collection(PostgresCollection.Name)] +[Trait("Category", "E2E")] +[Trait("Surface", "Engine")] +public sealed class UnwaitedChildRunE2ETests +{ + private const int MaxTicks = 10; + + private readonly PostgresFixture _fixture; + + public UnwaitedChildRunE2ETests(PostgresFixture fixture) { _fixture = fixture; } + + [Fact] + public async Task A_timer_parked_child_whose_parent_a_sibling_failed_is_cancelled_by_the_reconciler_and_never_walks_again() + { + var world = await FailParentAroundAParkedChildAsync(); + await AgeAsync(world.ChildId); + + await ReconcileUntilCancelledAsync(world.ChildId); + + var walkedBefore = await NodeStartsAsync(world.ChildId); + (await FireTimerAsync(world.ChildId, world.TimerWaitId)).ShouldBeFalse("the cancel closed the child's timer wait, so its scheduled wake resolves nothing"); + await DrainAsync(); + + (await StatusAsync(world.ChildId)).ShouldBe(WorkflowRunStatus.Cancelled); + (await NodeStartsAsync(world.ChildId)).ShouldBe(walkedBefore, "the child never walks again — no node of it starts after the cancel"); + } + + [Fact] + public async Task Continue_after_engine_failure_leaves_exactly_one_live_child() + { + // No reconcile tick anywhere in this test: the Continue alone must end the child its failed attempt left behind. + var world = await FailParentAroundAParkedChildAsync(); + + (await ContinueAsync(world.ParentId, world.TeamId)).ShouldBeTrue("precondition: the failed parent continues in place"); + await DrainAsync(); + + var live = await LiveChildrenAsync(world.ParentId); + var survivor = live.ShouldHaveSingleItem("one sub-workflow node, one live child: the revived walk's"); + survivor.ShouldNotBe(world.ChildId, "the survivor is the child the revived walk staged"); + (await StatusAsync(world.ChildId)).ShouldBe(WorkflowRunStatus.Cancelled, "the failed attempt's parked child is cancelled by the Continue itself"); + (await PendingSubworkflowTokenAsync(world.ParentId)).ShouldBe(survivor.ToString(), "and the parent waits for exactly that child"); + } + + [Fact] + public async Task A_child_the_stop_failed_to_cancel_is_cancelled_by_the_next_reconcile() + { + var teamId = await SeedTeamAsync(); + var (parentId, childId) = await ParkParentOnASleepingChildAsync(teamId, SingleSubParentAsync); + + var childCancelFails = new FailFirstReadOf(childId); + (await CancelThroughAsync(parentId, teamId, childCancelFails))!.Cancelled.ShouldBeTrue("precondition: the parent was stopped"); + + childCancelFails.Failed.ShouldBe(childId, "precondition: the stop's teardown tried the child and that cancel failed"); + (await StatusAsync(childId)).ShouldBe(WorkflowRunStatus.Suspended, "precondition: the child outlived its parent's stop"); + + await AgeAsync(childId); + await ReconcileUntilCancelledAsync(childId); + } + + // ─── Worlds ───────────────────────────────────────────────────────────────── + + private sealed record FailedParentWorld(Guid TeamId, Guid ParentId, Guid ChildId, Guid TimerWaitId); + + /// Walk the parent until its child sleeps on a timer, approve the gate, and let boom fail the parent. + private async Task FailParentAroundAParkedChildAsync() + { + var teamId = await SeedTeamAsync(); + var (parentId, childId) = await ParkParentOnASleepingChildAsync(teamId, FailingSiblingParentAsync); + + await ApproveGateAsync(parentId, teamId); // wake 1: boom runs beside the still-parked sub and fails + await DrainAsync(); + + if (await StatusAsync(parentId) == WorkflowRunStatus.Suspended) + { + await ApproveGateAsync(parentId, teamId); // wake 2: the re-walk re-reads boom's failure — the park settled first in wake 1's wave + await DrainAsync(); + } + + (await StatusAsync(parentId)).ShouldBe(WorkflowRunStatus.Failure, "precondition: a re-walk an approval drove failed the parent at boom"); + (await StatusAsync(childId)).ShouldBe(WorkflowRunStatus.Suspended, "precondition: the engine's failure cleanup leaves a parked child as it is"); + + return new FailedParentWorld(teamId, parentId, childId, await PendingTimerWaitAsync(childId)); + } + + /// Run the parent built by until its child has walked into its hour-long sleep. + private async Task<(Guid ParentId, Guid ChildId)> ParkParentOnASleepingChildAsync(Guid teamId, Func> createParent) + { + var userId = await OwnerAsync(teamId); + var childWorkflowId = await CreateWorkflowAsync(teamId, userId, SleepingChildDefinition()); + var parentRunId = await WorkflowsTestSeed.SeedManualRunAsync(_fixture, await createParent(teamId, childWorkflowId), teamId); + + await RunEngineAsync(parentRunId); + await DrainAsync(); + + var childId = (await LiveChildrenAsync(parentRunId)).ShouldHaveSingleItem("precondition: the parent staged one child"); + (await StatusAsync(childId)).ShouldBe(WorkflowRunStatus.Suspended, "precondition: the child walked into its sleep"); + + return (parentRunId, childId); + } + + // ─── Driving ──────────────────────────────────────────────────────────────── + + /// Reconcile until the child is Cancelled, at most times: a tick takes the oldest candidates of the whole database, so a row this test owns can wait behind strangers. + private async Task ReconcileUntilCancelledAsync(Guid childId) + { + (await StatusAsync(childId)).ShouldNotBe(WorkflowRunStatus.Cancelled, "precondition: the child is live before the sweep runs"); + + for (var tick = 1; tick <= MaxTicks; tick++) + { + using (var scope = _fixture.BeginScope()) + await scope.Resolve().Send(new ReconcileStuckRunsCommand()); + + if (await StatusAsync(childId) == WorkflowRunStatus.Cancelled) return; + } + + throw new ShouldAssertException($"child {childId} was not cancelled within {MaxTicks} reconcile ticks (last status {await StatusAsync(childId)}). Each tick cancels the oldest {StuckRunReconcilerService.BatchSize} sub-workflow children no live parent waits for — check `SELECT p.status, w.status FROM workflow_run c JOIN workflow_run p ON p.id = c.parent_run_id LEFT JOIN workflow_run_wait w ON w.token = c.id::text WHERE c.id = '{childId}'`."); + } + + /// Move the child's staging past the reconciler's grace — the wall-clock time a real orphan lives through before a tick. + private async Task AgeAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + var past = DateTimeOffset.UtcNow - StuckRunReconcilerService.UnwaitedChildStagingGrace - TimeSpan.FromMinutes(1); + await scope.Resolve().WorkflowRun.Where(r => r.Id == runId).ExecuteUpdateAsync(s => s.SetProperty(r => r.CreatedDate, past)); + } + + private async Task RunEngineAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + await scope.Resolve().ExecuteRunAsync(runId, CancellationToken.None); + } + + private async Task DrainAsync() + { + using var scope = _fixture.BeginScope(); + await scope.Resolve().WaitForPendingAsync(); + } + + private async Task ApproveGateAsync(Guid runId, Guid teamId) + { + using var scope = _fixture.BeginScope(); + (await scope.Resolve().ApproveRunAsync(runId, teamId, await OwnerAsync(teamId), approved: true, comment: null, CancellationToken.None)).ShouldBeTrue("precondition: the gate's approval resolved"); + } + + /// The child's parked timer firing — the call the engine scheduled with Hangfire when it parked. + private async Task FireTimerAsync(Guid runId, Guid waitId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().ResumeWaitAsync(runId, waitId, null, CancellationToken.None); + } + + private async Task ContinueAsync(Guid runId, Guid teamId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().ContinueRunAsync(runId, teamId, CancellationToken.None); + } + + /// Stop the run through a scope whose database commands pass . + private async Task CancelThroughAsync(Guid runId, Guid teamId, IInterceptor interceptor) + { + DbContextOptions production; + using (var probe = _fixture.BeginScope()) + production = probe.Resolve>(); + + var options = new DbContextOptionsBuilder(production).AddInterceptors(interceptor).Options; + + using var scope = _fixture.BeginScope(b => b.RegisterInstance(options).As>().SingleInstance()); + return await scope.Resolve().CancelRunAsync(runId, teamId, CancellationToken.None); + } + + /// Fails, once, the first read that carries runId — a cancel's first read of its target. + private sealed class FailFirstReadOf : DbCommandInterceptor + { + private readonly Guid _runId; + private int _fired; + + public FailFirstReadOf(Guid runId) { _runId = runId; } + + public Guid? Failed { get; private set; } + + public override ValueTask> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + var carried = command.Parameters.Cast().Any(p => p.Value is Guid id && id == _runId); + + if (!carried || Interlocked.Exchange(ref _fired, 1) == 1) return ValueTask.FromResult(result); + + Failed = _runId; + throw new InvalidOperationException($"reading {_runId} failed"); + } + } + + // ─── Reading ──────────────────────────────────────────────────────────────── + + private async Task StatusAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRun.AsNoTracking().Where(r => r.Id == runId).Select(r => r.Status).SingleAsync(); + } + + private async Task> LiveChildrenAsync(Guid parentRunId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRun.AsNoTracking() + .Where(r => r.ParentRunId == parentRunId && r.SourceType == WorkflowRunSourceTypes.ChildWorkflow && r.Status != WorkflowRunStatus.Success && r.Status != WorkflowRunStatus.Failure && r.Status != WorkflowRunStatus.Cancelled) + .Select(r => r.Id).ToListAsync(); + } + + private async Task PendingTimerWaitAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRunWait.AsNoTracking().Where(w => w.RunId == runId && w.WaitKind == WorkflowWaitKinds.Timer && w.Status == WorkflowWaitStatuses.Pending).Select(w => w.Id).SingleAsync(); + } + + private async Task PendingSubworkflowTokenAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRunWait.AsNoTracking().Where(w => w.RunId == runId && w.WaitKind == WorkflowWaitKinds.Subworkflow && w.Status == WorkflowWaitStatuses.Pending).Select(w => w.Token).SingleOrDefaultAsync(); + } + + private async Task NodeStartsAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRunRecord.AsNoTracking().CountAsync(r => r.RunId == runId && r.RecordType == WorkflowRunRecordTypes.NodeStarted); + } + + // ─── Workflows ────────────────────────────────────────────────────────────── + + private async Task SeedTeamAsync() => (await WorkflowsTestSeed.SeedTeamAsync(_fixture)).TeamId; + + private async Task OwnerAsync(Guid teamId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().TeamMembership.Where(m => m.TeamId == teamId && m.Role == TeamRole.Owner).Select(m => m.UserId).SingleAsync(); + } + + private async Task CreateWorkflowAsync(Guid teamId, Guid userId, WorkflowDefinition definition) + { + using var scope = _fixture.BeginScopeAs(userId, teamId, Roles.Admin); + return await scope.Resolve().Send(new CreateWorkflowCommand { Name = "unwaited-" + Guid.NewGuid().ToString("N")[..8], Description = null, Definition = definition, Activations = new List(), Enabled = true }); + } + + // start → nap (an hour's flow.sleep) → end + private static WorkflowDefinition SleepingChildDefinition() => new() + { + SchemaVersion = 1, + Nodes = new List + { + Node("start", "trigger.manual"), + new() { Id = "nap", TypeKey = "flow.sleep", Config = WorkflowsTestSeed.Json("""{"seconds":3600}"""), Inputs = WorkflowsTestSeed.EmptyJson() }, + Node("end", "builtin.terminal"), + }, + Edges = new List { new() { From = "start", To = "nap" }, new() { From = "nap", To = "end" } }, + }; + + // start → { sub (the child), gate, gate2 (approvals), ready → boom (fails once) } → end. The first walk parks on its first + // wave — sub and the gates suspend, ready completes — so boom first runs on the re-walk an approval drives, beside the + // still-parked sub. That wave settles in order and the park comes first, so the failure lands on the next wake: the + // other gate's approval re-walks the run, which re-reads boom's failure and fails the parent. + private async Task FailingSiblingParentAsync(Guid teamId, Guid childWorkflowId) => await CreateWorkflowAsync(teamId, await OwnerAsync(teamId), new WorkflowDefinition + { + SchemaVersion = 1, + Nodes = new List + { + Node("start", "trigger.manual"), + SubNode(childWorkflowId), + Gate("gate"), + Gate("gate2"), + Flaky("ready", failTimes: 0), + Flaky("boom", failTimes: 1), + Node("end", "builtin.terminal"), + }, + Edges = new List + { + new() { From = "start", To = "sub" }, new() { From = "start", To = "gate" }, new() { From = "start", To = "gate2" }, new() { From = "start", To = "ready" }, new() { From = "ready", To = "boom" }, + new() { From = "sub", To = "end" }, new() { From = "gate", To = "end" }, new() { From = "gate2", To = "end" }, new() { From = "boom", To = "end" }, + }, + }); + + private static NodeDefinition Gate(string id) => new() { Id = id, TypeKey = "flow.wait_approval", Config = WorkflowsTestSeed.Json("""{"prompt":"Go on?"}"""), Inputs = WorkflowsTestSeed.EmptyJson() }; + + private static NodeDefinition Flaky(string id, int failTimes) => new() { Id = id, TypeKey = FlakyTestNode.Key, Config = WorkflowsTestSeed.Json($$"""{"key":"unwaited-{{id}}-{{Guid.NewGuid():N}}","failTimes":{{failTimes}}}"""), Inputs = WorkflowsTestSeed.EmptyJson() }; + + // start → sub (the child) → end + private async Task SingleSubParentAsync(Guid teamId, Guid childWorkflowId) => await CreateWorkflowAsync(teamId, await OwnerAsync(teamId), new WorkflowDefinition + { + SchemaVersion = 1, + Nodes = new List { Node("start", "trigger.manual"), SubNode(childWorkflowId), Node("end", "builtin.terminal") }, + Edges = new List { new() { From = "start", To = "sub" }, new() { From = "sub", To = "end" } }, + }); + + private static NodeDefinition SubNode(Guid childWorkflowId) => new() { Id = "sub", TypeKey = "flow.subworkflow", Config = WorkflowsTestSeed.Json($$"""{"workflowId":"{{childWorkflowId}}"}"""), Inputs = WorkflowsTestSeed.Json("""{"inputs":{}}""") }; + + private static NodeDefinition Node(string id, string typeKey) => new() { Id = id, TypeKey = typeKey, Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }; +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/DispatcherFinishedAncestorFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/DispatcherFinishedAncestorFlowTests.cs new file mode 100644 index 000000000..b8dd722ff --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/DispatcherFinishedAncestorFlowTests.cs @@ -0,0 +1,165 @@ +using Autofac; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Workflows; +using CodeSpace.Core.Services.Workflows.Dispatch; +using CodeSpace.Core.Services.Workflows.Engine; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.IntegrationTests.Infrastructure.Jobs; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Enums; +using Microsoft.EntityFrameworkCore; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Workflows; + +/// +/// 🟢 Integration (high fidelity, Rule 12): the REAL over real Postgres refuses to queue a +/// sub-workflow child whose ancestry has finished — the last door before a worker walks it, which every resume, revive and +/// reconciler re-dispatch passes through — while a rerun or replay, whose parent is only its lineage, still goes. The job +/// client records the enqueue without running the engine, so the dispatch itself is what is observed. +/// +[Collection(PostgresCollection.Name)] +[Trait("Category", "Integration")] +public sealed class DispatcherFinishedAncestorFlowTests +{ + private readonly PostgresFixture _fixture; + + public DispatcherFinishedAncestorFlowTests(PostgresFixture fixture) { _fixture = fixture; } + + [Theory] + [InlineData(WorkflowRunStatus.Success)] + [InlineData(WorkflowRunStatus.Failure)] + [InlineData(WorkflowRunStatus.Cancelled)] + public async Task A_child_of_a_finished_parent_is_not_queued(WorkflowRunStatus finished) + { + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, finished, null, WorkflowRunSourceTypes.Manual); // WorkflowEngine terminal / WorkflowService.CancelRunAsync + var childId = await SeedRunAsync(teamId, WorkflowRunStatus.Pending, parentId, WorkflowRunSourceTypes.ChildWorkflow); // SubworkflowService.StageChildRunAsync + var versionBefore = await RowVersionAsync(childId); + + (await DispatchAsync(childId)).ShouldBeFalse($"nothing waits for a child of a {finished} parent"); + + (await RowVersionAsync(childId)).ShouldBe(versionBefore, "the refusal comes before the Pending → Enqueued CAS, so the row is not written"); + } + + [Fact] + public async Task A_grandchild_is_not_queued_once_the_root_has_finished_though_its_parent_lives() + { + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var rootId = await SeedRunAsync(teamId, WorkflowRunStatus.Failure, null, WorkflowRunSourceTypes.Manual); // WorkflowEngine terminal + var childId = await SeedRunAsync(teamId, WorkflowRunStatus.Suspended, rootId, WorkflowRunSourceTypes.ChildWorkflow); // SubworkflowService.StageChildRunAsync, then its own park + var grandchildId = await SeedRunAsync(teamId, WorkflowRunStatus.Pending, childId, WorkflowRunSourceTypes.ChildWorkflow); // SubworkflowService.StageChildRunAsync + + (await DispatchAsync(grandchildId)).ShouldBeFalse("the root's finish ends every level under it"); + (await StatusAsync(grandchildId)).ShouldBe(WorkflowRunStatus.Pending); + } + + [Fact] + public async Task A_child_of_a_live_parent_is_queued() + { + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, WorkflowRunStatus.Suspended, null, WorkflowRunSourceTypes.Manual); // WorkflowEngine park + var childId = await SeedRunAsync(teamId, WorkflowRunStatus.Pending, parentId, WorkflowRunSourceTypes.ChildWorkflow); // SubworkflowService.StageChildRunAsync + + (await DispatchAsync(childId)).ShouldBeTrue(); + (await StatusAsync(childId)).ShouldBe(WorkflowRunStatus.Enqueued); + } + + [Fact] + public async Task A_child_of_a_live_rerun_fork_is_queued_though_the_forks_original_has_finished() + { + // The walk climbs only through sub-workflow children: the fork is where it stops, never the fork's lineage. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var originalId = await SeedRunAsync(teamId, WorkflowRunStatus.Failure, null, WorkflowRunSourceTypes.Manual); // WorkflowEngine terminal + var forkId = await SeedRunAsync(teamId, WorkflowRunStatus.Suspended, originalId, WorkflowRunSourceTypes.Rerun); // WorkflowService.StageForkedRunAsync, then the engine park + var childId = await SeedRunAsync(teamId, WorkflowRunStatus.Pending, forkId, WorkflowRunSourceTypes.ChildWorkflow); // SubworkflowService.StageChildRunAsync + + (await DispatchAsync(childId)).ShouldBeTrue("the live fork waits for its child"); + (await StatusAsync(childId)).ShouldBe(WorkflowRunStatus.Enqueued); + } + + [Theory] + [InlineData(WorkflowRunSourceTypes.Rerun)] + [InlineData(WorkflowRunSourceTypes.Replay)] + public async Task A_fork_of_a_finished_run_is_queued(string forkSource) + { + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var originalId = await SeedRunAsync(teamId, WorkflowRunStatus.Failure, null, WorkflowRunSourceTypes.Manual); // WorkflowEngine terminal + var forkId = await SeedRunAsync(teamId, WorkflowRunStatus.Pending, originalId, forkSource); // WorkflowService.StageForkedRunAsync + + (await DispatchAsync(forkId)).ShouldBeTrue($"a {forkSource}'s parent is its lineage, finished by definition — never a reason to hold it back"); + (await StatusAsync(forkId)).ShouldBe(WorkflowRunStatus.Enqueued); + } + + [Fact] + public async Task A_timer_firing_on_a_child_of_a_finished_parent_does_not_queue_it() + { + // The engine schedules a timer's wake with Hangfire when it parks, and nothing unschedules it when the parent ends. + // The wake resolves the child's wait and flips it to Pending; the dispatch it then asks for is refused. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, WorkflowRunStatus.Failure, null, WorkflowRunSourceTypes.Manual); // WorkflowEngine terminal + var childId = await SeedRunAsync(teamId, WorkflowRunStatus.Suspended, parentId, WorkflowRunSourceTypes.ChildWorkflow); // SubworkflowService.StageChildRunAsync, then its own park + var timerWaitId = await SeedTimerWaitAsync(childId); // WorkflowEngine.TryParkAsync + + using (var manual = JobClient().ManualExecution()) + using (var scope = _fixture.BeginScope()) + (await scope.Resolve().ResumeWaitAsync(childId, timerWaitId, null, CancellationToken.None)).ShouldBeTrue("precondition: the scheduled wake resolved the child's timer"); + + (await StatusAsync(childId)).ShouldBe(WorkflowRunStatus.Pending, "the wake flipped the child, but the dispatcher would not queue it — no worker walks it"); + } + + private async Task DispatchAsync(Guid runId) + { + using var manual = JobClient().ManualExecution(); // record the enqueue; never run the engine behind it + using var scope = _fixture.BeginScope(); + return await scope.Resolve().DispatchAsync(runId, CancellationToken.None); + } + + private InMemoryBackgroundJobClient JobClient() + { + using var scope = _fixture.BeginScope(); + return scope.Resolve(); + } + + private async Task SeedRunAsync(Guid teamId, WorkflowRunStatus status, Guid? parentRunId, string sourceType) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + var now = DateTimeOffset.UtcNow; + var requestId = Guid.NewGuid(); + var runId = Guid.NewGuid(); + + db.WorkflowRunRequest.Add(new WorkflowRunRequest { Id = requestId, TeamId = teamId, SourceType = sourceType, ActorType = "user", ActorId = SystemUsers.SeederId, NormalizedPayloadJson = "{}", Status = WorkflowRunRequestStatus.Consumed, ReceivedAt = now, VerifiedAt = now, NormalizedAt = now }); + db.WorkflowRun.Add(new WorkflowRun { Id = runId, TeamId = teamId, RunRequestId = requestId, SourceType = sourceType, ParentRunId = parentRunId, Status = status, StartedAt = status == WorkflowRunStatus.Pending ? null : now, CompletedAt = WorkflowRunState.IsTerminal(status) ? now : null, CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId }); + + await db.SaveChangesAsync(); + return runId; + } + + private async Task SeedTimerWaitAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + var waitId = Guid.NewGuid(); + db.WorkflowRunWait.Add(new WorkflowRunWait { Id = waitId, RunId = runId, NodeId = "delay", IterationKey = string.Empty, WaitKind = WorkflowWaitKinds.Timer, Token = Guid.NewGuid().ToString("N"), WakeAt = DateTimeOffset.UtcNow, Status = WorkflowWaitStatuses.Pending, CreatedAt = DateTimeOffset.UtcNow }); + + await db.SaveChangesAsync(); + return waitId; + } + + private async Task StatusAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRun.AsNoTracking().Where(r => r.Id == runId).Select(r => r.Status).SingleAsync(); + } + + private async Task RowVersionAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRun.AsNoTracking().Where(r => r.Id == runId).Select(r => r.Xmin).SingleAsync(); + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/StuckRunReconcilerFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/StuckRunReconcilerFlowTests.cs index ed476301d..31b8273b6 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Workflows/StuckRunReconcilerFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/StuckRunReconcilerFlowTests.cs @@ -11,7 +11,6 @@ using CodeSpace.Messages.Enums; using MediatR; using Microsoft.EntityFrameworkCore; -using Microsoft.Extensions.Logging; using Shouldly; namespace CodeSpace.IntegrationTests.Workflows; @@ -100,28 +99,6 @@ public async Task Recent_pending_is_NOT_redispatched() "the young Pending row must remain Pending"); } - [Theory] - [InlineData(WorkflowRunStatus.Cancelled)] - [InlineData(WorkflowRunStatus.Failure)] - [InlineData(WorkflowRunStatus.Success)] - public async Task A_stuck_pending_sub_workflow_under_a_finished_parent_is_never_dispatched(WorkflowRunStatus finished) - { - // A staged child whose cancel failed during its parent's stop sits Pending with a stale CreatedDate — exactly what - // this sweep re-dispatches. Dispatched, it ran under a finished parent toward a wait that parent had closed. The - // guard agent runs already have: a finished parent's staged child is not work anyone is waiting for. - var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); - var workflowId = await CreateWorkflowAsync(teamId, userId); - var parentId = await StageStuckRunAsync(workflowId, teamId, finished, createdAgo: TimeSpan.FromHours(1)); - var childId = await StageStuckPendingUnderAsync(workflowId, teamId, parentId, WorkflowRunSourceTypes.ChildWorkflow); - - var versionBefore = await ReadRowVersionAsync(childId); - - await ReconcileAsync(); - - (await ReadRowVersionAsync(childId)).ShouldBe(versionBefore, $"a staged sub-workflow under a {finished} parent must not be dispatched — nor touched at all"); - (await ReadStatusAsync(childId)).ShouldBe(WorkflowRunStatus.Pending); - } - [Fact] public async Task A_stuck_pending_rerun_of_a_finished_run_is_still_dispatched() { @@ -627,30 +604,7 @@ public async Task Suspended_with_zero_pending_waits_but_within_grace_window_is_N } [Theory] - [InlineData(WorkflowRunStatus.Cancelled)] - [InlineData(WorkflowRunStatus.Failure)] - [InlineData(WorkflowRunStatus.Success)] - public async Task A_stranded_suspended_sub_workflow_under_a_finished_parent_is_never_redispatched(WorkflowRunStatus finished) - { - // A child whose cancel failed during its parent's stop, and whose waits were then closed, sits Suspended with no - // pending wait — exactly what this sweep revives. Revived, it ran under a finished parent toward a wait that parent - // no longer holds open, spending on work no one is waiting for. The guard the Pending sweep has: a finished parent's - // child is not work anyone wants done. - var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); - var workflowId = await CreateWorkflowAsync(teamId, userId); - var parentId = await StageStuckRunAsync(workflowId, teamId, finished, createdAgo: TimeSpan.FromHours(1)); - var childId = await StageStrandedSuspendedUnderAsync(workflowId, teamId, parentId, WorkflowRunSourceTypes.ChildWorkflow); - - var versionBefore = await ReadRowVersionAsync(childId); - - await ReconcileAsync(); - - (await ReadRowVersionAsync(childId)).ShouldBe(versionBefore, $"a stranded sub-workflow under a {finished} parent must not be revived — nor touched at all"); - (await ReadStatusAsync(childId)).ShouldBe(WorkflowRunStatus.Suspended); - } - - [Theory] - [InlineData(WorkflowRunStatus.Running)] // the parent walking on while its child is stranded + [InlineData(WorkflowRunStatus.Running)] // the parent re-walking, its Subworkflow wait on the child still Pending [InlineData(WorkflowRunStatus.Suspended)] // the parent parked on its child's Subworkflow wait — how production holds a live parent public async Task A_stranded_suspended_sub_workflow_under_a_live_parent_is_still_redispatched(WorkflowRunStatus live) { @@ -660,8 +614,7 @@ public async Task A_stranded_suspended_sub_workflow_under_a_live_parent_is_still var workflowId = await CreateWorkflowAsync(teamId, userId); var parentId = await StageStuckRunAsync(workflowId, teamId, live, createdAgo: TimeSpan.FromMinutes(1), startedAtAgo: TimeSpan.FromMinutes(1)); var childId = await StageStrandedSuspendedUnderAsync(workflowId, teamId, parentId, WorkflowRunSourceTypes.ChildWorkflow); - - if (live == WorkflowRunStatus.Suspended) await SeedSubworkflowWaitAsync(parentId, childId); + await SeedSubworkflowWaitAsync(parentId, childId); await ReconcileAsync(); @@ -683,25 +636,6 @@ public async Task A_stranded_suspended_rerun_of_a_finished_run_is_still_redispat (await ReadStatusAsync(rerunId)).ShouldBe(WorkflowRunStatus.Enqueued, "a stranded rerun of a finished run is revived like any stranded run"); } - [Fact] - public async Task The_sweep_says_how_many_stranded_sub_workflows_it_left_alone() - { - // The children it leaves Suspended stay Suspended, so without a line they would linger unseen. - var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); - var workflowId = await CreateWorkflowAsync(teamId, userId); - var parentId = await StageStuckRunAsync(workflowId, teamId, WorkflowRunStatus.Cancelled, createdAgo: TimeSpan.FromHours(1)); - await StageStrandedSuspendedUnderAsync(workflowId, teamId, parentId, WorkflowRunSourceTypes.ChildWorkflow); - - var log = new RecordingLogger(); - - await ReconcileAsync(log); - - var entry = log.Entries.Where(e => e.Level == LogLevel.Information && e.Message.Contains("parent has finished")).ShouldHaveSingleItem("one line per sweep says so"); - - // >= not == : the count is deployment-wide (see the class note); the child this test staged is one of them. - ((int)entry.Properties["Count"]!).ShouldBeGreaterThanOrEqualTo(1, "the line carries the count as a structured property, not interpolated text"); - } - // ─── Helpers ────────────────────────────────────────────────────────────────── private async Task RunEngineAsync(Guid runId) @@ -1317,13 +1251,6 @@ private async Task ReconcileAsync() return await mediator.Send(new ReconcileStuckRunsCommand()); } - /// As , in a scope whose reconciler logs to . - private async Task ReconcileAsync(ILogger log) - { - using var scope = _fixture.BeginScope(b => b.RegisterInstance(log).As>()); - return await scope.Resolve().Send(new ReconcileStuckRunsCommand()); - } - /// The Postgres row version. Changes on any write, so it answers "was this row touched" rather than "what does it say now". private async Task ReadRowVersionAsync(Guid runId) { @@ -1358,17 +1285,4 @@ private async Task ContinueAsync(Guid runId, Guid teamId) using var scope = _fixture.BeginScope(); return await scope.Resolve().ContinueRunAsync(runId, teamId, CancellationToken.None); } - - /// Keeps each entry's level, formatted message and structured properties — the named template values an interpolated message would not carry. - private sealed class RecordingLogger : ILogger - { - public List<(LogLevel Level, string Message, IReadOnlyDictionary Properties)> Entries { get; } = new(); - - public IDisposable? BeginScope(TState state) where TState : notnull => null; - - public bool IsEnabled(LogLevel logLevel) => true; - - public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) => - Entries.Add((logLevel, formatter(state, exception), (state as IEnumerable> ?? []).ToDictionary(p => p.Key, p => p.Value))); - } } diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/SubworkflowCancelFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/SubworkflowCancelFlowTests.cs index 0904c5ef4..0b2c10e05 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Workflows/SubworkflowCancelFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/SubworkflowCancelFlowTests.cs @@ -85,7 +85,7 @@ public async Task A_child_whose_cancel_fails_strands_neither_its_siblings_nor_th (await RunStatusAsync(sibling)).ShouldBe(WorkflowRunStatus.Cancelled, "one child's failed cancel must not strand its sibling — the single UPDATE this replaced cancelled every staged child at once"); (await WaitStatusAsync(approvalWaitId)).ShouldBe(WorkflowWaitStatuses.Discarded, "nor the parent's remaining waits"); (await RunStatusAsync(parentId)).ShouldBe(WorkflowRunStatus.Cancelled); - (await RunStatusAsync(failed)).ShouldBe(started[failed], "the child whose cancel failed is left as it was: the stuck-run reconciler never dispatches a staged child under a finished parent, and nothing sweeps a started one"); + (await RunStatusAsync(failed)).ShouldBe(started[failed], "the child whose cancel failed is left as it was by the stop — the stuck-run reconciler's next tick cancels it (UnwaitedChildRunFlowTests)"); } [Fact] diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/UnwaitedChildRunFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/UnwaitedChildRunFlowTests.cs new file mode 100644 index 000000000..7fa13399b --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/UnwaitedChildRunFlowTests.cs @@ -0,0 +1,794 @@ +using System.Collections.Concurrent; +using System.Data.Common; +using System.Diagnostics; +using System.Reflection; +using System.Text.Json; +using Autofac; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Agents; +using CodeSpace.Core.Services.Agents.Sandbox; +using CodeSpace.Core.Services.Agents.Sandbox.Runners; +using CodeSpace.Core.Services.Workflows; +using CodeSpace.Core.Services.Workflows.Lifecycle; +using CodeSpace.Core.Services.Workflows.Reconciliation; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.IntegrationTests.Infrastructure.Jobs; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Messages.Agents; +using CodeSpace.Messages.Authorization; +using CodeSpace.Messages.Commands.Workflows; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Dtos.Workflows; +using CodeSpace.Messages.Enums; +using MediatR; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Workflows; + +/// +/// 🟢 Integration (high fidelity, Rule 12): the stuck-run reconciler cancels a sub-workflow child that no live parent is +/// waiting for, and no wake sweep revives one in the meantime. Drives the REAL +/// through the mediator, the REAL it cancels through, and the REAL +/// Continue, over real Postgres; a REAL detached sleeper under the local durable runner stands in for a child's agent +/// process, which the child's cancel must reap. +/// +/// Every seeded row is a state production writes; the writer is named beside it. Every sweep here is +/// deployment-wide and bounded, so each test waits on the row it owns (), never on a +/// tally; and a "left alone" claim is only made once a control row — which the sweep would have reached no earlier than +/// the owned one — shows the tick got that far. +/// +[Collection(PostgresCollection.Name)] +[Trait("Category", "Integration")] +public sealed class UnwaitedChildRunFlowTests : IDisposable +{ + private const int MaxTicks = 10; + + /// A child this old is past the staging grace; the extra minute keeps clock drift out of it. + private static readonly TimeSpan PastGrace = StuckRunReconcilerService.UnwaitedChildStagingGrace + TimeSpan.FromMinutes(1); + + private readonly PostgresFixture _fixture; + private readonly List _spoolDirs = new(); + private readonly List _launchedPids = new(); + + public UnwaitedChildRunFlowTests(PostgresFixture fixture) { _fixture = fixture; } + + // ─── Cancelled: no live parent waits for the child ─────────────────────────── + + [Theory] + [InlineData(WorkflowRunStatus.Failure, WorkflowRunStatus.Pending)] + [InlineData(WorkflowRunStatus.Failure, WorkflowRunStatus.Enqueued)] + [InlineData(WorkflowRunStatus.Failure, WorkflowRunStatus.Running)] + [InlineData(WorkflowRunStatus.Failure, WorkflowRunStatus.Suspended)] + [InlineData(WorkflowRunStatus.Cancelled, WorkflowRunStatus.Pending)] + [InlineData(WorkflowRunStatus.Cancelled, WorkflowRunStatus.Enqueued)] + [InlineData(WorkflowRunStatus.Cancelled, WorkflowRunStatus.Running)] + [InlineData(WorkflowRunStatus.Cancelled, WorkflowRunStatus.Suspended)] + [InlineData(WorkflowRunStatus.Success, WorkflowRunStatus.Pending)] + [InlineData(WorkflowRunStatus.Success, WorkflowRunStatus.Enqueued)] + [InlineData(WorkflowRunStatus.Success, WorkflowRunStatus.Running)] + [InlineData(WorkflowRunStatus.Success, WorkflowRunStatus.Suspended)] + public async Task A_child_no_parent_waits_for_is_cancelled_and_its_agent_killed(WorkflowRunStatus parentStatus, WorkflowRunStatus childStatus) + { + if (OperatingSystem.IsWindows()) return; // the agent process is a /bin/sh sleeper under the local durable runner + + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, new RunSeed(parentStatus)); // WorkflowEngine terminal / WorkflowService.CancelRunAsync + var childId = await SeedChildAsync(teamId, parentId, childStatus, PastGrace); + await SeedEndedParentWaitAsync(parentId, childId, parentStatus); + var agentId = await SeedRunningAgentAsync(teamId, childId); + var pid = await LaunchAgentProcessAsync(agentId); + + ProcessAlive(pid).ShouldBeTrue($"precondition: the child's agent process runs before the sweep — {ProcessLiveness.Describe(pid)}"); + + await ReconcileUntilAsync(() => HasStatusAsync(childId, WorkflowRunStatus.Cancelled), $"the {childStatus} child {childId} of a {parentStatus} parent reaching Cancelled"); + + (await AgentStatusAsync(agentId)).ShouldBe(AgentRunStatus.Cancelled, "the child is cancelled through the full cancel, whose teardown runs the kill-wave over the child's agents"); + (await WaitForProcessGoneAsync(pid)).ShouldBeTrue($"the child's agent process must be reaped by the child's teardown — check `ps -p {pid}`; {ProcessLiveness.Describe(pid)}"); + } + + [Fact] + public async Task A_pending_wait_under_a_finished_parent_does_not_keep_its_child() + { + // The engine's terminal cleanup is best-effort and swallows its failure, so a Failure parent can keep its wait Pending. + // A parent that has finished resumes nothing, whatever its waits say. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure)); // WorkflowEngine terminal + var childId = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Suspended, PastGrace); + await SeedWaitAsync(parentId, WorkflowWaitKinds.Subworkflow, childId.ToString(), WorkflowWaitStatuses.Pending); // WorkflowEngine.TryParkAsync; CancelPendingWaitsAndChildrenAsync then failed + + await ReconcileUntilAsync(() => HasStatusAsync(childId, WorkflowRunStatus.Cancelled), $"child {childId} of a Failure parent that still holds its Pending wait reaching Cancelled"); + } + + [Fact] + public async Task A_continued_parents_prior_attempt_child_is_cancelled() + { + // Continue revives the parent in place (live again) and closes the ended attempt's waits; the walk it dispatches stages + // a fresh child. The old child is then under a LIVE parent that no longer waits for it. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Pending) { Generation = 1 }); // WorkflowService.BumpToNewGenerationAsync + var childId = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Running, PastGrace); + await SeedWaitAsync(parentId, WorkflowWaitKinds.Subworkflow, childId.ToString(), WorkflowWaitStatuses.Discarded); // WorkflowService.DiscardPendingWaitsAsync + + await ReconcileUntilAsync(() => HasStatusAsync(childId, WorkflowRunStatus.Cancelled), $"the prior attempt's child {childId} of a continued parent reaching Cancelled"); + } + + [Fact] + public async Task A_grandchild_whose_cancel_failed_in_its_parents_teardown_is_cancelled_on_a_later_tick() + { + // One tick cancels the child; the child's own teardown cancels the grandchild it waits for. When that inner cancel fails, + // the grandchild is left under a parent that has now finished — and the next tick takes it. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var rootId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure)); // WorkflowEngine terminal + var childId = await SeedChildAsync(teamId, rootId, WorkflowRunStatus.Suspended, PastGrace); + await SeedWaitAsync(rootId, WorkflowWaitKinds.Subworkflow, childId.ToString(), WorkflowWaitStatuses.Discarded); // WorkflowEngine.CancelPendingWaitsAndChildrenAsync + var grandchildId = await SeedChildAsync(teamId, childId, WorkflowRunStatus.Running, PastGrace); + await SeedWaitAsync(childId, WorkflowWaitKinds.Subworkflow, grandchildId.ToString(), WorkflowWaitStatuses.Pending); // WorkflowEngine.TryParkAsync + + var grandchildCancelFails = new FailFirstReadOf(grandchildId); + await ReconcileUntilAsync(() => HasStatusAsync(childId, WorkflowRunStatus.Cancelled), $"child {childId} reaching Cancelled", null, grandchildCancelFails); + + grandchildCancelFails.Failed.ShouldBe(grandchildId, "precondition: the child's teardown tried the grandchild and that cancel failed"); + (await StatusAsync(grandchildId)).ShouldBe(WorkflowRunStatus.Running, "precondition: the grandchild outlived the tick that cancelled its parent"); + + await ReconcileUntilAsync(() => HasStatusAsync(grandchildId, WorkflowRunStatus.Cancelled), $"grandchild {grandchildId}, under a parent that has now finished, reaching Cancelled"); + } + + [Fact] + public async Task One_failing_cancel_does_not_stop_the_rest() + { + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure)); // WorkflowEngine terminal + var failing = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Running, PastGrace + TimeSpan.FromMinutes(1)); // older: the sweep reaches it first + var sibling = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Running, PastGrace); + + var faults = new FailingCancels(failing); + await ReconcileUntilAsync(() => HasStatusAsync(sibling, WorkflowRunStatus.Cancelled), $"child {sibling}, the one after a child whose cancel throws, reaching Cancelled", faults); + + faults.Attempted.ShouldContain(failing, "precondition: the sweep tried the older child first and that cancel threw"); + (await StatusAsync(failing)).ShouldBe(WorkflowRunStatus.Running, "the child whose cancel failed waits for the next tick"); + } + + // ─── Left alone ───────────────────────────────────────────────────────────── + + [Theory] + [InlineData(WorkflowRunStatus.Pending)] // a sibling's wake flipped it to re-walk, its child's wait still Pending + [InlineData(WorkflowRunStatus.Enqueued)] + [InlineData(WorkflowRunStatus.Running)] // re-walking + [InlineData(WorkflowRunStatus.Suspended)] // parked on the child — how production holds a live parent + public async Task A_child_its_live_parent_waits_for_is_left_alone(WorkflowRunStatus parentStatus) + { + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, new RunSeed(parentStatus)); // WorkflowEngine park / WorkflowResumeService flip / dispatcher / engine claim + var childId = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Running, PastGrace + TimeSpan.FromMinutes(5)); + await SeedWaitAsync(parentId, WorkflowWaitKinds.Subworkflow, childId.ToString(), WorkflowWaitStatuses.Pending); // WorkflowEngine.TryParkAsync + + await AssertLeftAloneAsync(teamId, childId, controlAge: PastGrace, $"the child of a {parentStatus} parent that waits for it"); + } + + [Fact] + public async Task A_child_a_live_rerun_fork_waits_for_is_left_alone() + { + // The fork's own parent is its lineage, finished by definition; the child's parent is the fork, which lives and waits. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var originalId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure)); // WorkflowEngine terminal + var forkId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Suspended) { SourceType = WorkflowRunSourceTypes.Rerun, ParentRunId = originalId }); // WorkflowService.StageForkedRunAsync, then the engine park + var childId = await SeedChildAsync(teamId, forkId, WorkflowRunStatus.Running, PastGrace + TimeSpan.FromMinutes(5)); + await SeedWaitAsync(forkId, WorkflowWaitKinds.Subworkflow, childId.ToString(), WorkflowWaitStatuses.Pending); // WorkflowEngine.TryParkAsync + + await AssertLeftAloneAsync(teamId, childId, controlAge: PastGrace, "the child a live rerun fork waits for"); + } + + [Theory] + [InlineData(WorkflowRunSourceTypes.Rerun)] + [InlineData(WorkflowRunSourceTypes.Replay)] + public async Task A_fork_of_a_finished_run_is_left_alone(string forkSource) + { + // ParentRunId is also a fork's lineage. Only a sub-workflow child is ever cancelled for its parent. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var originalId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure)); // WorkflowEngine terminal + var forkId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Suspended) { SourceType = forkSource, ParentRunId = originalId, Age = PastGrace + TimeSpan.FromMinutes(5) }); // WorkflowService.StageForkedRunAsync, then the engine park + await SeedWaitAsync(forkId, WorkflowWaitKinds.Approval, Guid.NewGuid().ToString("N"), WorkflowWaitStatuses.Pending); // WorkflowEngine.TryParkAsync + + await AssertLeftAloneAsync(teamId, forkId, controlAge: PastGrace, $"a {forkSource} fork of a finished run"); + } + + [Fact] + public async Task A_child_inside_its_staging_grace_is_left_alone() + { + // The engine commits a staged child a moment before it parks the wait that names it. Inside that moment the child has + // no wait, and the grace is what keeps the sweep off it. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Running)); // WorkflowEngine claim — walking, about to park + var childId = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Pending, TimeSpan.Zero); // staged, its wait not yet parked + + // The control is older than the owned child (a fresh child cannot sort ahead of one past the grace), so one more tick + // after the control falls gives the owned child its turn had the grace not kept it out. + await AssertLeftAloneAsync(teamId, childId, controlAge: PastGrace, "a child still inside its staging grace", extraTicks: 2); + } + + // ─── Not woken while its cancel keeps failing ─────────────────────────────── + + public enum WakePath { StrandedSuspended, Timer, SupervisorSelfAdvance, SupervisorInfraPark, AbandonedSupervisorAfterClaim, AbandonedSupervisorBeforeClaim } + + [Theory] + [InlineData(WakePath.StrandedSuspended)] + [InlineData(WakePath.Timer)] + [InlineData(WakePath.SupervisorSelfAdvance)] + [InlineData(WakePath.SupervisorInfraPark)] + [InlineData(WakePath.AbandonedSupervisorAfterClaim)] + [InlineData(WakePath.AbandonedSupervisorBeforeClaim)] + public async Task An_orphan_whose_cancel_keeps_failing_is_not_woken(WakePath path) + { + // Until its cancel lands, an orphan must not be re-walked by any of the sweeps that wake a stuck run. Its cancel is made + // to fail so the orphan stays put; a top-level control in the same stuck shape, due after the orphan, proves the tick + // reached that sweep's batch past the orphan's place in it. + var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var workflowId = path is WakePath.AbandonedSupervisorBeforeClaim ? await CreateSupervisorWorkflowAsync(teamId, userId) : (Guid?)null; + var parentId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure)); // WorkflowEngine terminal + var orphanId = await SeedStuckAsync(path, teamId, workflowId, parentId, overdueBy: TimeSpan.FromMinutes(20)); + await SeedWaitAsync(parentId, WorkflowWaitKinds.Subworkflow, orphanId.ToString(), WorkflowWaitStatuses.Discarded); // WorkflowEngine.CancelPendingWaitsAndChildrenAsync + var controlId = await SeedStuckAsync(path, teamId, workflowId, parentRunId: null, overdueBy: TimeSpan.FromMinutes(10)); + + var faults = new FailingCancels(orphanId); + await ReconcileUntilAsync(() => WokenAsync(path, controlId), $"the {path} sweep waking control run {controlId}", faults); + + faults.Attempted.ShouldContain(orphanId, "precondition: the sweep tried to cancel the orphan and that cancel failed"); + (await WokenAsync(path, orphanId)).ShouldBeFalse($"the {path} sweep must not wake a sub-workflow child whose parent has finished — nothing is waiting for it"); + } + + // ─── Continue ─────────────────────────────────────────────────────────────── + + [Theory] + [InlineData(WorkflowRunStatus.Failure)] + [InlineData(WorkflowRunStatus.Cancelled)] + public async Task Continue_cancels_the_ended_attempts_started_child_through_the_full_cancel(WorkflowRunStatus ended) + { + if (OperatingSystem.IsWindows()) return; + + // The engine's failure cleanup cancels only a child still Pending or Enqueued, and a stop whose child cancel failed + // leaves one started. Continue revives the parent and its re-walk stages a fresh child, so the old one is cancelled — + // with its agents — on the Continue itself, no reconcile tick in between. + var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var workflowId = await CreateParentWorkflowAsync(teamId, userId); + var parentId = await SeedRunAsync(teamId, new RunSeed(ended) { WorkflowId = workflowId }); // WorkflowEngine terminal / WorkflowService.CancelRunAsync + await RecordEndedAttemptAsync(parentId, ended); + var childId = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Suspended, TimeSpan.Zero); + await SeedWaitAsync(parentId, WorkflowWaitKinds.Subworkflow, childId.ToString(), WorkflowWaitStatuses.Discarded); // CancelPendingWaitsAndChildrenAsync / the stop's CancelPendingWaitsAsync + var agentId = await SeedRunningAgentAsync(teamId, childId); + var pid = await LaunchAgentProcessAsync(agentId); + + (await ContinueAsync(parentId, teamId)).ShouldBeTrue("precondition: the ended parent continued in place"); + + (await StatusAsync(childId)).ShouldBe(WorkflowRunStatus.Cancelled, $"the {ended} attempt's started child is cancelled by the Continue — nothing waits for it once the revived walk stages its own"); + (await AgentStatusAsync(agentId)).ShouldBe(AgentRunStatus.Cancelled, "through the full cancel, whose teardown kills the child's agents"); + (await WaitForProcessGoneAsync(pid)).ShouldBeTrue($"the child's agent process must be reaped — check `ps -p {pid}`; {ProcessLiveness.Describe(pid)}"); + } + + [Theory] + [InlineData(WorkflowRunStatus.Failure, false)] + [InlineData(WorkflowRunStatus.Cancelled, false)] + [InlineData(WorkflowRunStatus.Success, false)] + [InlineData(WorkflowRunStatus.Suspended, true)] // control: a live parent keeps its child continuable + public async Task Continue_refuses_a_failed_child_whose_parent_has_finished(WorkflowRunStatus parentStatus, bool continues) + { + // Continuing it would report success and bump its generation, and then the dispatcher would refuse the run and the + // reconciler cancel it: nothing waits for a child once its parent has finished. + var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var workflowId = await CreateParentWorkflowAsync(teamId, userId); + var parentId = await SeedRunAsync(teamId, new RunSeed(parentStatus)); // WorkflowEngine terminal / park + var childId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure) { SourceType = WorkflowRunSourceTypes.ChildWorkflow, ParentRunId = parentId, WorkflowId = workflowId }); // SubworkflowService.StageChildRunAsync, then the engine's failure + await RecordEndedAttemptAsync(childId, WorkflowRunStatus.Failure); + + (await ContinueAsync(childId, teamId)).ShouldBe(continues); + + var status = await StatusAsync(childId); + + if (continues) status.ShouldNotBe(WorkflowRunStatus.Failure, "a live parent's child revives"); + else status.ShouldBe(WorkflowRunStatus.Failure, "a refused continue leaves the child as it ended"); + } + + // ─── Isolation and rotation ───────────────────────────────────────────────── + + [Fact] + public async Task A_failing_find_does_not_stop_the_other_sweeps() + { + // The unwaited-child sweep is one sweep among several; its failure is logged and the tick still completes. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var controlId = await SeedStuckAsync(WakePath.StrandedSuspended, teamId, null, parentRunId: null, overdueBy: TimeSpan.FromMinutes(10)); + var brokenFind = new ThrowOnUnwaitedChildFind(); + + await ReconcileUntilAsync(() => WokenAsync(WakePath.StrandedSuspended, controlId), $"the stranded-Suspended sweep waking control run {controlId}", null, brokenFind); + + brokenFind.Fired.ShouldBeTrue("precondition: the unwaited-child find ran and threw"); + } + + [Fact] + public async Task A_child_whose_cancel_keeps_failing_does_not_hold_the_batch_against_a_newer_one() + { + // A full batch of children whose cancel fails sorts ahead of the one after them; a failed attempt moves a child to the + // back, so the next tick reaches the newer child. + var (teamId, _) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var parentId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure)); // WorkflowEngine terminal + var stuck = new List(); + + for (var i = 0; i < StuckRunReconcilerService.BatchSize; i++) + stuck.Add(await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Running, PastGrace + TimeSpan.FromMinutes(5))); + + var newer = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Running, PastGrace); + + var faults = new FailingCancels(stuck.ToArray()); + await ReconcileUntilAsync(() => HasStatusAsync(newer, WorkflowRunStatus.Cancelled), $"child {newer}, behind {stuck.Count} children whose cancel throws, reaching Cancelled", faults); + + (await StatusAsync(stuck[0])).ShouldBe(WorkflowRunStatus.Running, "precondition: the children whose cancel threw are still live"); + } + + // ─── Waiting ──────────────────────────────────────────────────────────────── + + /// + /// Reconcile until holds for the row the test owns, at most times. The predicate + /// must start false, or the wait waits for nothing. Each tick runs through the mediator as the recurring job sends it, with + /// ' cancels throwing and on every database command. + /// + private async Task ReconcileUntilAsync(Func> done, string signal, FailingCancels? faults = null, params IInterceptor[] interceptors) + { + (await done()).ShouldBeFalse($"precondition: {signal} must not have happened before the sweep runs"); + + for (var tick = 1; tick <= MaxTicks; tick++) + { + await ReconcileAsync(faults, interceptors); + + if (await done()) return; + } + + throw new ShouldAssertException($"{signal}: not observed within {MaxTicks} reconcile ticks. Each tick takes the oldest {StuckRunReconcilerService.BatchSize} candidates of each sweep across the whole shared database, so look for rows that keep failing ahead of this one — `SELECT id, status, created_date FROM workflow_run WHERE source_type = 'workflow.child' AND status IN ('Pending','Enqueued','Running','Suspended') ORDER BY created_date`."); + } + + /// + /// The row is never touched while the sweep demonstrably reaches its place: an unwaited control child, due no earlier than the + /// owned row, is cancelled first, then more ticks run. + /// + private async Task AssertLeftAloneAsync(Guid teamId, Guid runId, TimeSpan controlAge, string what, int extraTicks = 1) + { + var parentId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Failure)); // WorkflowEngine terminal + var controlId = await SeedChildAsync(teamId, parentId, WorkflowRunStatus.Running, controlAge); + var statusBefore = await StatusAsync(runId); + var versionBefore = await RowVersionAsync(runId); + + await ReconcileUntilAsync(() => HasStatusAsync(controlId, WorkflowRunStatus.Cancelled), $"control child {controlId} reaching Cancelled"); + + for (var tick = 0; tick < extraTicks; tick++) await ReconcileAsync(null); + + (await RowVersionAsync(runId)).ShouldBe(versionBefore, $"{what} must not be written at all"); + (await StatusAsync(runId)).ShouldBe(statusBefore); + } + + private async Task ReconcileAsync(FailingCancels? faults, params IInterceptor[] interceptors) + { + using var manual = JobClient().ManualExecution(); // record dispatches; no engine walk runs behind the sweep + using var scope = _fixture.BeginScope(b => ConfigureTick(b, faults, interceptors)); + + await scope.Resolve().Send(new ReconcileStuckRunsCommand()); + } + + private void ConfigureTick(ContainerBuilder builder, FailingCancels? faults, IInterceptor[] interceptors) + { + if (faults != null) builder.Register(c => faults.Wrap(c.Resolve())).As().InstancePerLifetimeScope(); + + if (interceptors.Length == 0) return; + + DbContextOptions production; + using (var probe = _fixture.BeginScope()) + production = probe.Resolve>(); + + builder.RegisterInstance(new DbContextOptionsBuilder(production).AddInterceptors(interceptors).Options).As>().SingleInstance(); + } + + private InMemoryBackgroundJobClient JobClient() + { + using var scope = _fixture.BeginScope(); + return scope.Resolve(); + } + + // ─── Wake paths ───────────────────────────────────────────────────────────── + + /// A run stuck the way 's sweep wakes, past that sweep's threshold — a sub-workflow child of , or a top-level run when it is null. + private async Task SeedStuckAsync(WakePath path, Guid teamId, Guid? workflowId, Guid? parentRunId, TimeSpan overdueBy) + { + var source = parentRunId is null ? WorkflowRunSourceTypes.Manual : WorkflowRunSourceTypes.ChildWorkflow; + var now = DateTimeOffset.UtcNow; + + return path switch + { + WakePath.StrandedSuspended => await SeedStrandedAsync(teamId, source, parentRunId, StuckRunReconcilerService.SuspendedStrandedAfter + overdueBy), + WakePath.Timer => await SeedParkedAsync(teamId, source, parentRunId, w => { w.WaitKind = WorkflowWaitKinds.Timer; w.WakeAt = now - StuckRunReconcilerService.TimerWakeLostAfter - overdueBy; }), + WakePath.SupervisorSelfAdvance => await SeedParkedAsync(teamId, source, parentRunId, w => { w.WaitKind = WorkflowWaitKinds.SupervisorDecision; w.NodeId = "sup"; w.IterationKey = "sup#turn1"; w.CreatedAt = now - StuckRunReconcilerService.SupervisorAdvanceLostAfter - overdueBy; }), + WakePath.SupervisorInfraPark => await SeedParkedAsync(teamId, source, parentRunId, w => { w.WaitKind = WorkflowWaitKinds.SupervisorInfraPark; w.NodeId = "sup"; w.IterationKey = "sup#infra1"; w.WakeAt = now - StuckRunReconcilerService.SupervisorInfraParkWakeLostAfter - overdueBy; w.PayloadJson = InfraParkMarker(); }), + WakePath.AbandonedSupervisorAfterClaim => await SeedAbandonedSupervisorAsync(teamId, new RunSeed(WorkflowRunStatus.Running) { SourceType = source, ParentRunId = parentRunId, StartedAgo = StuckRunReconcilerService.RunningStuckAfter + overdueBy }, claimed: true), + _ => await SeedAbandonedSupervisorAsync(teamId, new RunSeed(WorkflowRunStatus.Running) { SourceType = source, ParentRunId = parentRunId, WorkflowId = workflowId, StartedAgo = StuckRunReconcilerService.RunningStuckAfter + overdueBy }, claimed: false), + }; + } + + /// Suspended with no Pending wait — its one wait closed — and untouched for . + private async Task SeedStrandedAsync(Guid teamId, string source, Guid? parentRunId, TimeSpan idle) + { + var runId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Suspended) { SourceType = source, ParentRunId = parentRunId, Age = idle, LastModifiedAgo = idle }); // the engine's park, then a resume that resolved the wait without the flip + await SeedWaitAsync(runId, WorkflowWaitKinds.Approval, Guid.NewGuid().ToString("N"), WorkflowWaitStatuses.Resolved); // WorkflowResumeService resolve + return runId; + } + + /// Suspended on one Pending wait shaped by . + private async Task SeedParkedAsync(Guid teamId, string source, Guid? parentRunId, Action shape) + { + var runId = await SeedRunAsync(teamId, new RunSeed(WorkflowRunStatus.Suspended) { SourceType = source, ParentRunId = parentRunId, Age = TimeSpan.FromHours(1) }); // WorkflowEngine park + + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + var wait = new WorkflowRunWait { Id = Guid.NewGuid(), RunId = runId, NodeId = "delay", IterationKey = string.Empty, WaitKind = WorkflowWaitKinds.Timer, Token = Guid.NewGuid().ToString("N"), Status = WorkflowWaitStatuses.Pending, CreatedAt = DateTimeOffset.UtcNow }; + shape(wait); + db.WorkflowRunWait.Add(wait); // WorkflowEngine.TryParkAsync (a Timer / SupervisorInfraPark wait carries its deadline in wake_at) + + await db.SaveChangesAsync(); + return runId; + } + + /// A supervisor run whose worker died mid-turn: Running, started long ago, no ledger activity since — with a claimed, unfinished decision, or, before any claim, its supervisor node started. + private async Task SeedAbandonedSupervisorAsync(Guid teamId, RunSeed seed, bool claimed) + { + var runId = await SeedRunAsync(teamId, seed with { Age = seed.StartedAgo!.Value }); // the engine's Enqueued → Running claim + + if (claimed) await SeedRunningDecisionAsync(teamId, runId); + else await RecordStaleNodeStartedAsync(runId, "sup", seed.StartedAgo!.Value); + + return runId; + } + + private async Task SeedRunningDecisionAsync(Guid teamId, Guid runId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + db.SupervisorDecisionRecord.Add(new SupervisorDecisionRecord { Id = Guid.NewGuid(), TeamId = teamId, SupervisorRunId = runId, DecisionKind = SupervisorDecisionKinds.Stop, IdempotencyKey = $"stop:{Guid.NewGuid():N}", InputHash = Guid.NewGuid().ToString("N"), PayloadJson = "{}", Status = SupervisorDecisionStatus.Running }); // SupervisorDecisionLog claim + + await db.SaveChangesAsync(); + } + + /// The engine's node.started record for , as old as the dead worker that wrote it. The ledger is immutable, so the backdate lifts its trigger for the one statement. + private async Task RecordStaleNodeStartedAsync(Guid runId, string nodeId, TimeSpan ago) + { + using var scope = _fixture.BeginScope(); + var empty = new Dictionary(); + await scope.Resolve().NodeStartedAsync(runId, nodeId, string.Empty, empty, empty, CancellationToken.None); // WorkflowEngine.ExecuteNodeAsync + + var db = scope.Resolve(); + await db.Database.ExecuteSqlRawAsync("ALTER TABLE workflow_run_record DISABLE TRIGGER workflow_run_record_enforce_immutability"); + try + { + await db.Database.ExecuteSqlRawAsync("UPDATE workflow_run_record SET occurred_at = {0} WHERE run_id = {1}", DateTimeOffset.UtcNow - ago, runId); + } + finally + { + await db.Database.ExecuteSqlRawAsync("ALTER TABLE workflow_run_record ENABLE TRIGGER workflow_run_record_enforce_immutability"); + } + } + + /// Whether 's sweep has acted on the run: revived it, resolved its overdue wait, or recorded a recovery of it. + private async Task WokenAsync(WakePath path, Guid runId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + return path switch + { + WakePath.StrandedSuspended => await db.WorkflowRun.AsNoTracking().AnyAsync(r => r.Id == runId && r.Status != WorkflowRunStatus.Suspended), + WakePath.AbandonedSupervisorAfterClaim or WakePath.AbandonedSupervisorBeforeClaim => await db.WorkflowRunRecord.AsNoTracking().AnyAsync(r => r.RunId == runId && r.RecordType == WorkflowRunRecordTypes.SupervisorRunRecovered), + _ => await db.WorkflowRunWait.AsNoTracking().AnyAsync(w => w.RunId == runId && w.Status == WorkflowWaitStatuses.Resolved), + }; + } + + /// The ladder marker AgentSupervisorNode.ParkForInfraOrStopAsync stores as both the park's payload and its deadline payload. + private static string InfraParkMarker() => JsonSerializer.Serialize(new Dictionary { ["infraPark"] = true, ["parks"] = 1, ["firstParkedAtUtc"] = DateTimeOffset.UtcNow.AddMinutes(-30).ToString("o"), ["error"] = "gateway 429 (seeded)" }); + + // ─── Seeding ──────────────────────────────────────────────────────────────── + + /// A workflow_run row. Unset fields are what the run starter writes for a fresh top-level run. + private sealed record RunSeed(WorkflowRunStatus Status) + { + public string SourceType { get; init; } = WorkflowRunSourceTypes.Manual; + public Guid? ParentRunId { get; init; } + public Guid? WorkflowId { get; init; } + public TimeSpan Age { get; init; } + public TimeSpan? StartedAgo { get; init; } + public TimeSpan? LastModifiedAgo { get; init; } + public int Generation { get; init; } + } + + private async Task SeedRunAsync(Guid teamId, RunSeed seed) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + var now = DateTimeOffset.UtcNow; + var requestId = Guid.NewGuid(); + var runId = Guid.NewGuid(); + + db.WorkflowRunRequest.Add(new WorkflowRunRequest + { + Id = requestId, TeamId = teamId, WorkflowId = seed.WorkflowId, SourceType = seed.SourceType, ActorType = "user", ActorId = SystemUsers.SeederId, + NormalizedPayloadJson = "{}", Status = WorkflowRunRequestStatus.Consumed, ReceivedAt = now, VerifiedAt = now, NormalizedAt = now, + }); + db.WorkflowRun.Add(new WorkflowRun + { + Id = runId, TeamId = teamId, RunRequestId = requestId, WorkflowId = seed.WorkflowId, WorkflowVersion = seed.WorkflowId is null ? null : 1, + SourceType = seed.SourceType, ParentRunId = seed.ParentRunId, Status = seed.Status, Generation = seed.Generation, + CreatedDate = now - seed.Age, LastModifiedDate = now - (seed.LastModifiedAgo ?? TimeSpan.Zero), + EnqueuedAt = seed.Status == WorkflowRunStatus.Enqueued ? now - seed.Age : null, + StartedAt = seed.StartedAgo is { } started ? now - started : null, + CompletedAt = WorkflowRunState.IsTerminal(seed.Status) ? now : null, + CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId, + }); + + await db.SaveChangesAsync(); + return runId; + } + + /// A sub-workflow child of staged ago (SubworkflowService.StageChildRunAsync), carried on to by the dispatcher and the engine. + private Task SeedChildAsync(Guid teamId, Guid parentRunId, WorkflowRunStatus status, TimeSpan age) => + SeedRunAsync(teamId, new RunSeed(status) { SourceType = WorkflowRunSourceTypes.ChildWorkflow, ParentRunId = parentRunId, Age = age, StartedAgo = status is WorkflowRunStatus.Running or WorkflowRunStatus.Suspended ? age : null }); + + /// + /// The parent's wait on the child, as the parent's end left it: closed Discarded by a Failure's cleanup or a stop's teardown, + /// and gone under a Success — which needs the wait dropped before the end, as the engine's re-stage of a still-parked cell + /// (ReplaceCellWaitAsync) drops it. + /// + private async Task SeedEndedParentWaitAsync(Guid parentRunId, Guid childRunId, WorkflowRunStatus parentStatus) + { + if (parentStatus == WorkflowRunStatus.Success) return; + + await SeedWaitAsync(parentRunId, WorkflowWaitKinds.Subworkflow, childRunId.ToString(), WorkflowWaitStatuses.Discarded); // WorkflowEngine.CancelPendingWaitsAndChildrenAsync / WorkflowService.CancelPendingWaitsAsync + } + + private async Task SeedWaitAsync(Guid runId, string waitKind, string token, string status) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + var waitId = Guid.NewGuid(); + var closed = status != WorkflowWaitStatuses.Pending; + db.WorkflowRunWait.Add(new WorkflowRunWait { Id = waitId, RunId = runId, NodeId = $"{waitKind}-{token}", IterationKey = string.Empty, WaitKind = waitKind, Token = token, Status = status, PayloadJson = closed ? "{}" : null, CreatedAt = DateTimeOffset.UtcNow, ResolvedAt = closed ? DateTimeOffset.UtcNow : null }); + + await db.SaveChangesAsync(); + return waitId; + } + + /// A Running agent run of , held by no worker (AgentRunExecutor's claim), and the run's pending AgentRun wait on it (WorkflowEngine.TryParkAsync). + private async Task SeedRunningAgentAsync(Guid teamId, Guid workflowRunId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + var agentId = Guid.NewGuid(); + var now = DateTimeOffset.UtcNow; + + db.AgentRun.Add(new AgentRun { Id = agentId, TeamId = teamId, WorkflowRunId = workflowRunId, NodeId = "agent", Harness = "codex-cli", Status = AgentRunStatus.Running, StartedAt = now, HeartbeatAt = now, LeaseExpiresAt = now + AgentRunLiveness.Window, FenceEpoch = 1 }); + db.WorkflowRunWait.Add(new WorkflowRunWait { Id = Guid.NewGuid(), RunId = workflowRunId, NodeId = "agent", IterationKey = string.Empty, WaitKind = WorkflowWaitKinds.AgentRun, Token = agentId.ToString(), Status = WorkflowWaitStatuses.Pending, CreatedAt = now }); + + await db.SaveChangesAsync(); + return agentId; + } + + /// The records the parent's walk left: its sub-workflow step parked, and — when the run failed — the sibling step that failed it. + private async Task RecordEndedAttemptAsync(Guid runId, WorkflowRunStatus ended) + { + using var scope = _fixture.BeginScope(); + var records = scope.Resolve(); + var empty = new Dictionary(); + + await records.NodeStartedAsync(runId, "sub", string.Empty, empty, empty, CancellationToken.None); // WorkflowEngine.ExecuteNodeAsync + await records.NodeSuspendedAsync(runId, "sub", string.Empty, WorkflowWaitKinds.Subworkflow, null, CancellationToken.None); // WorkflowEngine.TryParkAsync + + if (ended != WorkflowRunStatus.Failure) return; + + await records.NodeStartedAsync(runId, "boom", string.Empty, empty, empty, CancellationToken.None); // WorkflowEngine.ExecuteNodeAsync + await records.NodeFailedAsync(runId, "boom", string.Empty, "boom", TimeSpan.FromMilliseconds(1), CancellationToken.None); // WorkflowEngine node failure + } + + // start → { sub (a sub-workflow), boom } → end + private async Task CreateParentWorkflowAsync(Guid teamId, Guid userId) + { + var childWorkflowId = await CreateWorkflowAsync(teamId, userId, WorkflowsTestSeed.MinimalDefinition()); + + return await CreateWorkflowAsync(teamId, userId, new WorkflowDefinition + { + SchemaVersion = 1, + Nodes = new List + { + new() { Id = "start", TypeKey = "trigger.manual", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + new() { Id = "sub", TypeKey = "flow.subworkflow", Config = WorkflowsTestSeed.Json($$"""{"workflowId":"{{childWorkflowId}}"}"""), Inputs = WorkflowsTestSeed.Json("""{"inputs":{}}""") }, + new() { Id = "boom", TypeKey = FlakyTestNode.Key, Config = WorkflowsTestSeed.Json($$"""{"key":"{{Guid.NewGuid():N}}","failTimes":1}"""), Inputs = WorkflowsTestSeed.EmptyJson() }, + new() { Id = "end", TypeKey = "builtin.terminal", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + }, + Edges = new List { new() { From = "start", To = "sub" }, new() { From = "start", To = "boom" }, new() { From = "sub", To = "end" }, new() { From = "boom", To = "end" } }, + }); + } + + // start → sup (agent.supervisor) → end + private Task CreateSupervisorWorkflowAsync(Guid teamId, Guid userId) => CreateWorkflowAsync(teamId, userId, new WorkflowDefinition + { + SchemaVersion = 1, + CompletionMode = WorkflowDefinition.CompletionModeShadow, + Nodes = new List + { + new() { Id = "start", TypeKey = "trigger.manual", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + new() { Id = "sup", TypeKey = "agent.supervisor", Config = WorkflowsTestSeed.Json("""{"goal":"ship the feature"}"""), Inputs = WorkflowsTestSeed.EmptyJson() }, + new() { Id = "end", TypeKey = "builtin.terminal", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + }, + Edges = new List { new() { From = "start", To = "sup" }, new() { From = "sup", To = "end" } }, + }); + + private async Task CreateWorkflowAsync(Guid teamId, Guid userId, WorkflowDefinition definition) + { + using var scope = _fixture.BeginScopeAs(userId, teamId, Roles.Admin); + return await scope.Resolve().Send(new CreateWorkflowCommand { Name = "unwaited-" + Guid.NewGuid().ToString("N")[..8], Description = null, Definition = definition, Activations = new List(), Enabled = true }); + } + + /// Launch a REAL detached sleeper under the local durable runner and record its handle on the agent run, so the agent's cancel has a live process tree to terminate. Returns the supervised pid. + private async Task LaunchAgentProcessAsync(Guid agentId) + { + var workDir = Path.Combine(Path.GetTempPath(), "cs-unwaited-test-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(workDir); + _spoolDirs.Add(workDir); + + var spec = new SandboxSpec { Command = "/bin/sh", Args = new[] { "-c", "sleep 300" }, WorkingDirectory = workDir, TimeoutSeconds = 300 }; + + SandboxHandle handle; + using (var scope = _fixture.BeginScope()) + handle = await ((ISandboxDurableRunner)scope.Resolve().Resolve(LocalProcessRunner.LocalKind)).LaunchAsync(spec, agentId.ToString("N"), CancellationToken.None); + + _spoolDirs.Add(handle.SpoolDirectory); + _launchedPids.Add(handle.ProcessId); + + using (var scope = _fixture.BeginScope()) + await scope.Resolve().SetRunnerHandleAsync(agentId, JsonSerializer.Serialize(handle, AgentJson.Options), CancellationToken.None); + + return handle.ProcessId; + } + + // ─── Reading ──────────────────────────────────────────────────────────────── + + private async Task ContinueAsync(Guid runId, Guid teamId) + { + using var manual = JobClient().ManualExecution(); // the revived walk's dispatch is recorded, not run + using var scope = _fixture.BeginScope(); + return await scope.Resolve().ContinueRunAsync(runId, teamId, CancellationToken.None); + } + + private async Task StatusAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRun.AsNoTracking().Where(r => r.Id == runId).Select(r => r.Status).SingleAsync(); + } + + private async Task HasStatusAsync(Guid runId, WorkflowRunStatus status) => await StatusAsync(runId) == status; + + /// The Postgres row version. Changes on any write, so it answers "was this row touched" rather than "what does it say now". + private async Task RowVersionAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRun.AsNoTracking().Where(r => r.Id == runId).Select(r => r.Xmin).SingleAsync(); + } + + private async Task AgentStatusAsync(Guid agentId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().AgentRun.AsNoTracking().Where(r => r.Id == agentId).Select(r => r.Status).SingleAsync(); + } + + private static bool ProcessAlive(int pid) => ProcessLiveness.IsAliveLikeTheProduct(pid); + + /// Bounded wait (5s) for the named signal: the supervised pid is gone. + private static async Task WaitForProcessGoneAsync(int pid) + { + for (var i = 0; i < 50 && ProcessAlive(pid); i++) await Task.Delay(100); + return !ProcessAlive(pid); + } + + // ─── Faults ───────────────────────────────────────────────────────────────── + + /// Cancels of these runs throw, through the workflow service the reconciler cancels with; every attempt is recorded. + public sealed class FailingCancels + { + private readonly HashSet _failing; + + public FailingCancels(params Guid[] failing) { _failing = failing.ToHashSet(); } + + public ConcurrentQueue Attempted { get; } = new(); + + public IWorkflowService Wrap(IWorkflowService inner) + { + var proxy = DispatchProxy.Create(); + ((FailingCancelProxy)(object)proxy).Bind(inner, this); + return proxy; + } + + internal bool Fails(Guid runId) + { + Attempted.Enqueue(runId); + return _failing.Contains(runId); + } + } + + public class FailingCancelProxy : DispatchProxy + { + private IWorkflowService _inner = default!; + private FailingCancels _faults = default!; + + internal void Bind(IWorkflowService inner, FailingCancels faults) + { + _inner = inner; + _faults = faults; + } + + protected override object? Invoke(MethodInfo? targetMethod, object?[]? args) + { + if (targetMethod!.Name == nameof(IWorkflowService.CancelRunAsync) && _faults.Fails((Guid)args![0]!)) + return Task.FromException(new InvalidOperationException($"cancelling run {args[0]} failed")); + + try + { + return targetMethod.Invoke(_inner, args); + } + catch (TargetInvocationException ex) when (ex.InnerException != null) + { + throw ex.InnerException; + } + } + } + + /// Throws on every read the unwaited-child find issues (the one query that matches a wait token against a run id under NOT EXISTS). + private sealed class ThrowOnUnwaitedChildFind : DbCommandInterceptor + { + public bool Fired { get; private set; } + + public override ValueTask> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + var sql = command.CommandText; + + if (!sql.Contains("NOT EXISTS", StringComparison.OrdinalIgnoreCase) || !sql.Contains("token", StringComparison.OrdinalIgnoreCase)) return ValueTask.FromResult(result); + + Fired = true; + throw new InvalidOperationException("the unwaited-child find failed"); + } + } + + /// Fails, once, the first read that carries runId — a cancel's first read of its target. + private sealed class FailFirstReadOf : DbCommandInterceptor + { + private readonly Guid _runId; + private int _fired; + + public FailFirstReadOf(Guid runId) { _runId = runId; } + + public Guid? Failed { get; private set; } + + public override ValueTask> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + var carried = command.Parameters.Cast().Any(p => p.Value is Guid id && id == _runId); + + if (!carried || Interlocked.Exchange(ref _fired, 1) == 1) return ValueTask.FromResult(result); + + Failed = _runId; + throw new InvalidOperationException($"reading {_runId} failed"); + } + } + + public void Dispose() + { + foreach (var pid in _launchedPids) + try { using var p = Process.GetProcessById(pid); if (!p.HasExited) p.Kill(entireProcessTree: true); } catch { /* best-effort */ } + + foreach (var dir in _spoolDirs) + try { Directory.Delete(dir, recursive: true); } catch { /* best-effort */ } + } +} diff --git a/backend/tests/CodeSpace.UnitTests/Handlers/Workflows/ReconcileStuckRunsCommandHandlerTests.cs b/backend/tests/CodeSpace.UnitTests/Handlers/Workflows/ReconcileStuckRunsCommandHandlerTests.cs new file mode 100644 index 000000000..689d69350 --- /dev/null +++ b/backend/tests/CodeSpace.UnitTests/Handlers/Workflows/ReconcileStuckRunsCommandHandlerTests.cs @@ -0,0 +1,48 @@ +using CodeSpace.Core.Handlers.CommandHandlers.Workflows; +using CodeSpace.Core.Services.Workflows.Reconciliation; +using CodeSpace.Messages.Commands.Workflows; +using Shouldly; + +namespace CodeSpace.UnitTests.Handlers.Workflows; + +/// +/// Unit-proves carries every sweep's count onto the response unchanged. Each +/// count gets a different value, so a field mapped from its neighbour, or left at its default, shows here and not as a +/// recurring-job log line that quietly under-reports one sweep. +/// +[Trait("Category", "Unit")] +public class ReconcileStuckRunsCommandHandlerTests +{ + [Fact] + public async Task Every_sweep_count_reaches_the_response() + { + var summary = new StuckRunReconcileSummary + { + RedispatchedFromPending = 1, RevertedFromEnqueued = 2, MarkedAbandonedFromRunning = 3, RedispatchedFromStrandedSuspended = 4, + RecoveredSupervisorAdvance = 5, RecoveredAbandonedSupervisorRun = 6, RecoveredStrandedTimerWait = 7, + RecoveredStrandedSupervisorInfraParkWait = 8, RecoveredStrandedSubworkflowParent = 9, CancelledUnwaitedChildRun = 10, + }; + + var response = await new ReconcileStuckRunsCommandHandler(new FixedReconciler(summary)).Handle(new ReconcileStuckRunsCommand(), CancellationToken.None); + + response.RedispatchedFromPending.ShouldBe(1); + response.RevertedFromEnqueued.ShouldBe(2); + response.MarkedAbandonedFromRunning.ShouldBe(3); + response.RedispatchedFromStrandedSuspended.ShouldBe(4); + response.RecoveredSupervisorAdvance.ShouldBe(5); + response.RecoveredAbandonedSupervisorRun.ShouldBe(6); + response.RecoveredStrandedTimerWait.ShouldBe(7); + response.RecoveredStrandedSupervisorInfraParkWait.ShouldBe(8); + response.RecoveredStrandedSubworkflowParent.ShouldBe(9); + response.CancelledUnwaitedChildRun.ShouldBe(10, "the children the tick cancelled because no parent was waiting for them"); + } + + private sealed class FixedReconciler : IStuckRunReconcilerService + { + private readonly StuckRunReconcileSummary _summary; + + public FixedReconciler(StuckRunReconcileSummary summary) { _summary = summary; } + + public Task ReconcileAsync(CancellationToken cancellationToken) => Task.FromResult(_summary); + } +} diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/RunAncestryTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/RunAncestryTests.cs new file mode 100644 index 000000000..d93e9312f --- /dev/null +++ b/backend/tests/CodeSpace.UnitTests/Workflows/RunAncestryTests.cs @@ -0,0 +1,154 @@ +using CodeSpace.Core.Services.Workflows.Dispatch; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Enums; +using Shouldly; + +namespace CodeSpace.UnitTests.Workflows; + +/// +/// The ancestry walk the dispatcher refuses a run by, over an in-memory chain of runs. It climbs only while each hop is a +/// sub-workflow child, so a rerun's or replay's parent — its lineage, finished by definition — never counts; it refuses on +/// the first finished or missing ancestor; and it ends on a cycle or a chain longer than nesting allows instead of looping. +/// +[Trait("Category", "Unit")] +public class RunAncestryTests +{ + [Fact] + public async Task A_run_that_is_not_a_child_has_no_ancestor_to_refuse_it() + { + var chain = new Chain(); + var original = chain.Add(WorkflowRunSourceTypes.Manual, null, WorkflowRunStatus.Failure); + var rerun = chain.Add(WorkflowRunSourceTypes.Rerun, original, WorkflowRunStatus.Pending); + + (await chain.HasFinishedAncestorAsync(rerun)).ShouldBeFalse("a rerun's parent is its lineage, not a run waiting for it"); + chain.Loaded.ShouldBe(new List { rerun }, "the walk reads the run itself and stops"); + } + + [Theory] + [InlineData(WorkflowRunStatus.Pending)] + [InlineData(WorkflowRunStatus.Enqueued)] + [InlineData(WorkflowRunStatus.Running)] + [InlineData(WorkflowRunStatus.Suspended)] + public async Task A_child_under_a_live_parent_is_not_refused(WorkflowRunStatus live) + { + var chain = new Chain(); + var parent = chain.Add(WorkflowRunSourceTypes.Manual, null, live); + var child = chain.Add(WorkflowRunSourceTypes.ChildWorkflow, parent, WorkflowRunStatus.Pending); + + (await chain.HasFinishedAncestorAsync(child)).ShouldBeFalse(); + } + + [Theory] + [InlineData(WorkflowRunStatus.Success)] + [InlineData(WorkflowRunStatus.Failure)] + [InlineData(WorkflowRunStatus.Cancelled)] + public async Task A_child_under_a_finished_parent_is_refused(WorkflowRunStatus finished) + { + var chain = new Chain(); + var parent = chain.Add(WorkflowRunSourceTypes.Manual, null, finished); + var child = chain.Add(WorkflowRunSourceTypes.ChildWorkflow, parent, WorkflowRunStatus.Pending); + + (await chain.HasFinishedAncestorAsync(child)).ShouldBeTrue($"nothing waits for a child of a {finished} parent"); + } + + [Fact] + public async Task A_grandchild_is_refused_when_its_root_has_finished_though_its_parent_lives() + { + var chain = new Chain(); + var root = chain.Add(WorkflowRunSourceTypes.Manual, null, WorkflowRunStatus.Failure); + var child = chain.Add(WorkflowRunSourceTypes.ChildWorkflow, root, WorkflowRunStatus.Suspended); + var grandchild = chain.Add(WorkflowRunSourceTypes.ChildWorkflow, child, WorkflowRunStatus.Pending); + + (await chain.HasFinishedAncestorAsync(grandchild)).ShouldBeTrue("the root's finish ends every level below it, not only the level right under it"); + } + + [Fact] + public async Task The_walk_stops_at_a_rerun_hop_and_never_reads_its_lineage() + { + var chain = new Chain(); + var original = chain.Add(WorkflowRunSourceTypes.Manual, null, WorkflowRunStatus.Failure); + var fork = chain.Add(WorkflowRunSourceTypes.Rerun, original, WorkflowRunStatus.Suspended); + var child = chain.Add(WorkflowRunSourceTypes.ChildWorkflow, fork, WorkflowRunStatus.Pending); + + (await chain.HasFinishedAncestorAsync(child)).ShouldBeFalse("the live fork is waiting for its child; the fork's own parent is lineage"); + chain.Loaded.ShouldNotContain(original, "the walk ends at the first hop that is not a sub-workflow child"); + } + + [Fact] + public async Task A_child_whose_parent_row_is_missing_is_refused() + { + var chain = new Chain(); + var child = chain.Add(WorkflowRunSourceTypes.ChildWorkflow, Guid.NewGuid(), WorkflowRunStatus.Pending); + + (await chain.HasFinishedAncestorAsync(child)).ShouldBeTrue("a parent that is gone holds no wait for the child"); + } + + [Fact] + public async Task A_cycle_of_children_is_refused_and_the_walk_ends() + { + var chain = new Chain(); + var a = Guid.NewGuid(); + var b = chain.Add(WorkflowRunSourceTypes.ChildWorkflow, a, WorkflowRunStatus.Running); + chain.Add(WorkflowRunSourceTypes.ChildWorkflow, b, WorkflowRunStatus.Running, a); + + (await chain.HasFinishedAncestorAsync(a)).ShouldBeTrue("a chain that loops back is not one the engine staged"); + chain.Loaded.Count.ShouldBeLessThanOrEqualTo(3, "the walk notices the repeat instead of going round"); + } + + [Fact] + public async Task A_chain_as_deep_as_nesting_allows_is_not_refused() + { + var (chain, leaf) = LiveChain(RunAncestry.MaxChildHops); + + (await chain.HasFinishedAncestorAsync(leaf)).ShouldBeFalse(); + } + + [Fact] + public async Task A_chain_deeper_than_nesting_allows_is_refused() + { + var (chain, leaf) = LiveChain(RunAncestry.MaxChildHops + 1); + + (await chain.HasFinishedAncestorAsync(leaf)).ShouldBeTrue("the walk is bounded — a longer chain is not one the engine staged"); + chain.Loaded.Count.ShouldBe(RunAncestry.MaxChildHops + 1, "the walk stops at its bound"); + } + + [Fact] + public async Task An_unknown_run_is_not_refused_here() + { + // The dispatcher's own Pending CAS finds no row and declines; the walk has nothing to say about it. + (await new Chain().HasFinishedAncestorAsync(Guid.NewGuid())).ShouldBeFalse(); + } + + /// A live root with sub-workflow children stacked under it; returns the deepest. + private static (Chain Chain, Guid Leaf) LiveChain(int childHops) + { + var chain = new Chain(); + var current = chain.Add(WorkflowRunSourceTypes.Manual, null, WorkflowRunStatus.Suspended); + + for (var i = 0; i < childHops; i++) current = chain.Add(WorkflowRunSourceTypes.ChildWorkflow, current, WorkflowRunStatus.Suspended); + + return (chain, current); + } + + private sealed class Chain + { + private readonly Dictionary _runs = new(); + + public List Loaded { get; } = new(); + + public Guid Add(string sourceType, Guid? parentRunId, WorkflowRunStatus status, Guid? id = null) + { + var runId = id ?? Guid.NewGuid(); + _runs[runId] = new RunAncestry.Hop(sourceType, parentRunId, status); + return runId; + } + + public Task HasFinishedAncestorAsync(Guid runId) => RunAncestry.HasFinishedAncestorAsync(runId, Load); + + private Task Load(Guid runId) + { + Loaded.Add(runId); + return Task.FromResult(_runs.GetValueOrDefault(runId)); + } + } +} diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/StuckRunReconcilerSummaryTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/StuckRunReconcilerSummaryTests.cs index bc42f4425..c99d3560d 100644 --- a/backend/tests/CodeSpace.UnitTests/Workflows/StuckRunReconcilerSummaryTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Workflows/StuckRunReconcilerSummaryTests.cs @@ -62,6 +62,24 @@ public void Total_counts_only_the_abandoned_supervisor_run_field_when_it_is_the_ summary.Total.ShouldBe(7); } + [Fact] + public void Total_counts_the_unwaited_child_cancels() + { + // A tick that only cancelled sub-workflow children no parent waits for still did work — the recurring job logs on Total > 0. + var summary = new StuckRunReconcileSummary { CancelledUnwaitedChildRun = 4 }; + + summary.Total.ShouldBe(4); + } + + [Fact] + public void UnwaitedChildStagingGrace_is_two_minutes() + { + // Pinned: a staged child has no wait naming it until its parent parks a moment later, and a child younger than this is + // never judged unwaited. Shortening it risks cancelling a child mid-staging; lengthening it lets an orphan staged by a + // walk that died before parking be re-dispatched by the Pending sweep (also two minutes) before this sweep cancels it. + StuckRunReconcilerService.UnwaitedChildStagingGrace.ShouldBe(TimeSpan.FromMinutes(2)); + } + [Fact] public void MaxSupervisorRunRecoveries_is_three() { diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/UnwaitedChildSweepBatchTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/UnwaitedChildSweepBatchTests.cs new file mode 100644 index 000000000..07e4526ca --- /dev/null +++ b/backend/tests/CodeSpace.UnitTests/Workflows/UnwaitedChildSweepBatchTests.cs @@ -0,0 +1,74 @@ +using System.Reflection; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Workflows; +using CodeSpace.Core.Services.Workflows.Reconciliation; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Dtos.Workflows; +using CodeSpace.Messages.Enums; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; + +namespace CodeSpace.UnitTests.Workflows; + +/// +/// The unwaited-child sweep's bound, on a database of its own: one tick cancels at most +/// children, oldest first, and a child some live parent is waiting for takes no place in that batch. Kept off the shared +/// integration database, where any other test's leftover children compete for the same slots. +/// +[Trait("Category", "Unit")] +public class UnwaitedChildSweepBatchTests +{ + private static readonly Guid TeamId = Guid.NewGuid(); + + [Fact] + public async Task One_tick_cancels_at_most_a_batch_of_unwaited_children_oldest_first() + { + await using var db = new CodeSpaceDbContext(new DbContextOptionsBuilder().UseInMemoryDatabase(Guid.NewGuid().ToString()).Options); + var now = DateTimeOffset.UtcNow; + var pastGrace = now - StuckRunReconcilerService.UnwaitedChildStagingGrace - TimeSpan.FromMinutes(1); + + var finishedParent = AddRun(db, WorkflowRunSourceTypes.Manual, null, WorkflowRunStatus.Failure, now - TimeSpan.FromHours(2)); // WorkflowEngine terminal + var liveParent = AddRun(db, WorkflowRunSourceTypes.Manual, null, WorkflowRunStatus.Suspended, now - TimeSpan.FromHours(2)); // WorkflowEngine park + var waited = AddRun(db, WorkflowRunSourceTypes.ChildWorkflow, liveParent, WorkflowRunStatus.Running, now - TimeSpan.FromHours(1)); // SubworkflowService.StageChildRunAsync — older than every orphan + AddSubworkflowWait(db, liveParent, waited); // WorkflowEngine.TryParkAsync + + var orphans = Enumerable.Range(1, StuckRunReconcilerService.BatchSize + 3).Select(i => AddRun(db, WorkflowRunSourceTypes.ChildWorkflow, finishedParent, WorkflowRunStatus.Suspended, pastGrace - TimeSpan.FromMinutes(i))).Reverse().ToList(); // SubworkflowService.StageChildRunAsync, oldest first + await db.SaveChangesAsync(); + + var cancels = CancelRecorder.Create(); + var reconciler = new StuckRunReconcilerService(db, null!, null!, null!, null!, cancels, NullLogger.Instance); // the sweep reads only the database and cancels through the workflow service + + var cancelled = await reconciler.CancelUnwaitedChildRunsAsync(CancellationToken.None); + + cancelled.ShouldBe(StuckRunReconcilerService.BatchSize, "one tick is bounded by the batch size"); + ((CancelRecorder)(object)cancels).Cancelled.ShouldBe(orphans.Take(StuckRunReconcilerService.BatchSize).ToList(), "oldest first, and the waited child takes no slot"); + } + + private static Guid AddRun(CodeSpaceDbContext db, string sourceType, Guid? parentRunId, WorkflowRunStatus status, DateTimeOffset createdDate) + { + var id = Guid.NewGuid(); + db.WorkflowRun.Add(new WorkflowRun { Id = id, TeamId = TeamId, RunRequestId = Guid.NewGuid(), SourceType = sourceType, ParentRunId = parentRunId, Status = status, CreatedDate = createdDate }); + return id; + } + + private static void AddSubworkflowWait(CodeSpaceDbContext db, Guid parentRunId, Guid childRunId) => + db.WorkflowRunWait.Add(new WorkflowRunWait { Id = Guid.NewGuid(), RunId = parentRunId, NodeId = "sub", IterationKey = string.Empty, WaitKind = WorkflowWaitKinds.Subworkflow, Token = childRunId.ToString(), Status = WorkflowWaitStatuses.Pending, CreatedAt = DateTimeOffset.UtcNow }); + + /// An that records each cancel and reports it done; every other member throws. + public class CancelRecorder : DispatchProxy + { + public List Cancelled { get; } = new(); + + public static IWorkflowService Create() => Create(); + + protected override object? Invoke(MethodInfo? targetMethod, object?[]? args) + { + if (targetMethod?.Name != nameof(IWorkflowService.CancelRunAsync)) throw new NotSupportedException(targetMethod?.Name); + + Cancelled.Add((Guid)args![0]!); + return Task.FromResult(new CancelRunOutcome { Cancelled = true, Status = WorkflowRunStatus.Cancelled, AgentRunsCancelled = 0 }); + } + } +}