diff --git a/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs b/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs new file mode 100644 index 00000000..0df15053 --- /dev/null +++ b/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs @@ -0,0 +1,62 @@ +using Halibut; + +namespace Squid.Core.Halibut.Resilience; + +/// +/// The single definition of "is this a TRANSIENT infrastructure failure?" — a +/// blip that should PAUSE a deployment (resumable, re-attach to the still-running +/// script) rather than fail it terminally. Referenced across layers (the pipeline +/// runner's pause-classification, the per-target TargetCatchClassifier, the +/// per-action catch in ExecuteStepsPhase, and the Halibut execution +/// strategy's re-attach probe), so the definition lives in ONE place. +/// +/// Transient = a single round-trip to the agent could not be completed, so +/// the script may still be running and a resume should re-attach to it: +/// +/// — the liveness probe gave up +/// mid-script (the script was dispatched and may still be running). +/// A base/transport — connection +/// reset, EOF, TLS handshake failure, timeout, etc. +/// +/// +/// NOT transient — these are sustained or permanent conditions, so pausing +/// for resume would only pause-loop; they must fail the deployment so an operator +/// investigates: +/// +/// The Halibut protocol/invocation subtypes — the request reached the +/// agent and was PERMANENTLY rejected (version mismatch → no matching +/// service/method; the agent's service itself threw). +/// — the per-machine breaker only opens +/// after the failure threshold (3+ consecutive failures), so it signals a +/// SUSTAINED agent problem, not a one-off blip. It is also raised BEFORE +/// any script is dispatched (fail-fast), so there is no in-flight script to +/// re-attach to. Pausing on it would loop on a genuinely dead agent and +/// flip the deliberate fail-fast-on-open-breaker contract +/// (HalibutCircuitBreakerE2ETests.BreakerForcedOpen_* → Failed). +/// +/// +public static class TransientFailureClassifier +{ + public static bool IsTransient(Exception ex) + { + if (ex is AggregateException aggregate) + return aggregate.InnerExceptions.Count > 0 && aggregate.InnerExceptions.All(IsTransient); + + if (ex is AgentUnreachableException) + return true; + + // Halibut transport failures are transient EXCEPT the protocol/invocation + // subtypes (the request reached the agent and was permanently rejected). + if (ex is HalibutClientException) + return !IsPermanentHalibutFailure(ex); + + return false; + } + + private static bool IsPermanentHalibutFailure(Exception ex) + => ex is global::Halibut.Exceptions.ServiceInvocationHalibutClientException + or global::Halibut.Exceptions.NoMatchingServiceOrMethodHalibutClientException + or global::Halibut.Exceptions.MethodNotFoundHalibutClientException + or global::Halibut.Exceptions.ServiceNotFoundHalibutClientException + or global::Halibut.Exceptions.AmbiguousMethodMatchHalibutClientException; +} diff --git a/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentCompletionHandler.cs b/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentCompletionHandler.cs index 35b38db3..4c19d51d 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentCompletionHandler.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentCompletionHandler.cs @@ -94,6 +94,31 @@ await genericDataProvider.ExecuteInTransactionAsync(async cancellationToken => }, ct).ConfigureAwait(false); } + /// + /// A transient infrastructure failure (a Halibut RPC drop that outlived the + /// library's own retries, or an agent that went unreachable mid-script) pauses + /// the deployment rather than failing it: the task transitions to + /// with its checkpoint AND in-flight script + /// pointer preserved, so a resume re-attaches to the still-running script + /// instead of re-dispatching a duplicate. Like we + /// deliberately do NOT delete the checkpoint and do NOT write a + /// DeploymentCompletion record (the deployment has not completed). The + /// historical fail-fast behaviour remains available via the + /// SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE escape hatch, which routes + /// transient failures back through . + /// + public async Task OnTransientPauseAsync(DeploymentTaskContext ctx, Exception ex, CancellationToken ct) + { + Log.Warning(ex, "[Deploy] Task {TaskId} hit a transient infrastructure failure; pausing for resume, checkpoint preserved", ctx.ServerTaskId); + + var fromState = await ResolveCurrentActiveStateAsync(ctx.ServerTaskId, ct).ConfigureAwait(false); + + await genericDataProvider.ExecuteInTransactionAsync(async cancellationToken => + { + await serverTaskService.TransitionStateAsync(ctx.ServerTaskId, fromState, TaskState.Paused, cancellationToken).ConfigureAwait(false); + }, ct).ConfigureAwait(false); + } + private async Task ResolveCurrentActiveStateAsync(int serverTaskId, CancellationToken ct) { var task = await serverTaskService.GetTaskAsync(serverTaskId, ct).ConfigureAwait(false); diff --git a/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs b/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs index 50ad1e33..0fd810dc 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs @@ -1,5 +1,6 @@ using Squid.Core.Services.DeploymentExecution.Exceptions; using Squid.Core.Services.DeploymentExecution.Lifecycle; +using Squid.Core.Services.DeploymentExecution.Pipeline.Phases; using Squid.Core.Services.DeploymentExecution.Script; using Squid.Core.Services.Deployments.ServerTask; @@ -114,6 +115,26 @@ public async Task ProcessAsync(int serverTaskId, CancellationToken ct) { await SafeCompleteAsync(ctx, () => completion.OnCancelledAsync(ctx, CancellationToken.None), new DeploymentCancelledEvent(new DeploymentEventContext())); } + catch (Exception ex) when (!timeoutCts.IsCancellationRequested && !registryCts.IsCancellationRequested && !ct.IsCancellationRequested + && TargetCatchClassifier.IsTransientInfraFailure(ex)) + { + // A transient infra failure (Halibut RPC drop after the library's + // retries, an unreachable agent, or an open breaker) pauses the + // deployment instead of failing it: the in-flight script pointer is + // preserved (the strategy clears it only on a definitive observation), + // so a resume re-attaches to the still-running script rather than + // re-dispatching a duplicate. We do NOT rethrow — Paused is a clean, + // expected outcome. This is unconditional (no opt-out): failing fast on + // a transient blip would just discard already-completed progress and + // risk a duplicate run, which has no legitimate use case. + // + // The cancel/timeout guard preserves the established precedence: a + // user-cancel (registryCts/ct) or a wall-clock timeout (timeoutCts) + // that races a transient RPC drop must NOT be reclassified as a + // transient pause — cancel/timeout win, so the exception falls through + // to their handlers (or the generic failure path). + await SafeCompleteAsync(ctx, () => completion.OnTransientPauseAsync(ctx, ex, CancellationToken.None), new DeploymentPausedEvent(new DeploymentEventContext())); + } catch (Exception ex) { await SafeCompleteAsync(ctx, () => completion.OnFailureAsync(ctx, ex, CancellationToken.None), new DeploymentFailedEvent(new DeploymentEventContext { Exception = ex })); diff --git a/src/Squid.Core/Services/DeploymentExecution/Pipeline/IDeploymentPipelinePhase.cs b/src/Squid.Core/Services/DeploymentExecution/Pipeline/IDeploymentPipelinePhase.cs index 05dc362c..b4d1471f 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/IDeploymentPipelinePhase.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/IDeploymentPipelinePhase.cs @@ -13,4 +13,11 @@ public interface IDeploymentCompletionHandler : IScopedDependency Task OnCancelledAsync(DeploymentTaskContext ctx, CancellationToken ct); Task OnPausedAsync(DeploymentTaskContext ctx, CancellationToken ct); Task OnTimedOutAsync(DeploymentTaskContext ctx, Exception ex, CancellationToken ct); + + // A transient infrastructure failure (Halibut RPC drop after the library's + // retries, or an unreachable agent) pauses the deployment for resume rather + // than failing it terminally — the still-running script is re-attached to on + // resume. Mirrors OnTimedOutAsync (Paused + checkpoint preserved); distinct + // method so the audit/log reason stays honest. + Task OnTransientPauseAsync(DeploymentTaskContext ctx, Exception ex, CancellationToken ct); } diff --git a/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/6_ExecuteStepsPhase.Execute.cs b/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/6_ExecuteStepsPhase.Execute.cs index 784ac1e1..2e9d6032 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/6_ExecuteStepsPhase.Execute.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/6_ExecuteStepsPhase.Execute.cs @@ -1,3 +1,4 @@ +using Squid.Core.Halibut.Resilience; using Squid.Core.Observability; using Squid.Core.Persistence.Entities.Deployments; using Squid.Core.Services.DeploymentExecution.Exceptions; @@ -397,6 +398,20 @@ private async Task ExecuteSingleActionAsync(PreparedAction prepared, DeploymentS { throw; } + catch (Exception ex) when (TransientFailureClassifier.IsTransient(ex)) + { + // A transient infra failure (Halibut RPC drop after the library's + // retries, an unreachable agent, or an open breaker) must PROPAGATE so + // the runner pauses the deployment for resume — regardless of whether + // the step is required. Recording it as a per-target/terminal failure + // (the path below) would route to OnFailureAsync, which deletes the + // checkpoint AND the in-flight pointer the strategy deliberately + // preserved — orphaning the still-running script and defeating resume. + // The per-target catch (TargetCatchClassifier) leaves the target + // in-flight; here, for a non-required step (which has no per-target + // catch), we still must not swallow it. + throw; + } catch (Exception ex) { result.Failed = true; diff --git a/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs b/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs index f031ffe2..7a3da284 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs @@ -1,3 +1,5 @@ +using Squid.Core.Halibut.Resilience; + namespace Squid.Core.Services.DeploymentExecution.Pipeline.Phases; /// @@ -44,10 +46,32 @@ public static Classification Classify(System.Exception ex, bool failFastCancelle if (ex is System.OperationCanceledException && failFastCancelled && !parentCtCancelled) return new Classification(MarkFailed: false, TriggerFailFast: false); + // The transient-infra case: a Halibut RPC drop (after the library's own + // retries) or an unreachable agent. The script may still be running on the + // agent, so we do NOT mark the target terminal — a resume re-attaches to it + // via its preserved in-flight pointer. We also do NOT fail-fast the peers: + // healthy targets finish their work, then the runner pauses the deployment + // (Task.WhenAll waits for all targets, so this target's exception surfaces + // after the peers complete). Guarded by !parentCtCancelled so a transient + // drop racing a real cancel/timeout still terminates the deployment. + if (IsTransientInfraFailure(ex) && !parentCtCancelled) + return new Classification(MarkFailed: false, TriggerFailFast: false); + // Everything else (genuine exception, OCE from user cancel, OCE // unrelated to our CTs): treat as failure. Cancel the failFast // cascade so peers stop work; the re-throw upstream surfaces the // actual exception. return new Classification(MarkFailed: true, TriggerFailFast: true); } + + /// + /// Whether is a transient infrastructure failure that + /// should pause (resumable) rather than fail the deployment. Delegates to the + /// single source of truth, + /// (which excludes the permanent Halibut protocol/invocation subtypes and + /// includes CircuitOpenException), kept here as the name the pipeline + /// callers already reference. + /// + public static bool IsTransientInfraFailure(System.Exception ex) + => TransientFailureClassifier.IsTransient(ex); } diff --git a/src/Squid.Core/Services/DeploymentExecution/Targets/Tentacle/Transport/HalibutMachineExecutionStrategy.cs b/src/Squid.Core/Services/DeploymentExecution/Targets/Tentacle/Transport/HalibutMachineExecutionStrategy.cs index c68468f6..3b404039 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Targets/Tentacle/Transport/HalibutMachineExecutionStrategy.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Targets/Tentacle/Transport/HalibutMachineExecutionStrategy.cs @@ -253,15 +253,24 @@ private async Task DispatchOrReattachAsync( Log.Information("[Deploy] Dispatching script to agent {MachineName} with ticket {Ticket}", request.Machine.Name, scriptTicket); - try - { - return await _observer.ObserveAndCompleteAsync(request.Machine, scriptClient, scriptTicket, scriptTimeout, ct, request.Masker, startResponse, endpoint, request.OutputSink).ConfigureAwait(false); - } - finally - { - if (_inFlightStore != null) - await _inFlightStore.ClearAsync(request.ServerTaskId, slot, ct).ConfigureAwait(false); - } + var result = await _observer.ObserveAndCompleteAsync(request.Machine, scriptClient, scriptTicket, scriptTimeout, ct, request.Masker, startResponse, endpoint, request.OutputSink).ConfigureAwait(false); + + // Clear the in-flight pointer ONLY on a definitive observation — i.e. the + // observer RETURNED (script completed, reported a non-zero exit, or hit the + // per-script timeout and was cancelled). On a THROW (a transient RPC drop + // that outlived Halibut's retries, an unreachable agent, or cancellation) + // the script may still be running on the agent, so we PRESERVE the pointer: + // the deployment pauses and a resumed run re-attaches to the still-running + // script instead of dispatching a duplicate. + await ClearInFlightAsync(request.ServerTaskId, slot).ConfigureAwait(false); + + return result; + } + + private async Task ClearInFlightAsync(int serverTaskId, DispatchSlot slot) + { + if (_inFlightStore != null) + await _inFlightStore.ClearAsync(serverTaskId, slot, CancellationToken.None).ConfigureAwait(false); } /// @@ -295,6 +304,19 @@ private async Task TryReattachAsync( } catch (Exception ex) { + // A TRANSIENT probe failure (the agent is still unreachable on resume) + // must NOT clear the pointer or dispatch fresh — the recorded script may + // still be running, so a fresh dispatch would duplicate it. Preserve the + // pointer and propagate so the deployment pauses again; a later resume + // re-probes. Only a non-transient probe error falls back to a fresh + // dispatch (today's behaviour). + if (TransientFailureClassifier.IsTransient(ex)) + { + Log.Information(ex, "[Deploy] Re-attach probe hit a transient failure for agent {MachineName} (ticket {Ticket}); preserving in-flight pointer for a later resume.", + request.Machine.Name, existingTicketId); + throw; + } + Log.Information(ex, "[Deploy] Re-attach probe failed for agent {MachineName} (ticket {Ticket}); dispatching fresh.", request.Machine.Name, existingTicketId); await _inFlightStore.ClearAsync(request.ServerTaskId, slot, ct).ConfigureAwait(false); @@ -310,14 +332,13 @@ private async Task TryReattachAsync( Log.Information("[Deploy] Re-attaching to in-flight script on agent {MachineName} (ticket {Ticket}, state {State}) after resume — skipping duplicate dispatch.", request.Machine.Name, existingTicketId, probe.State); - try - { - return await _observer.ObserveAndCompleteAsync(request.Machine, scriptClient, existingTicket, scriptTimeout, ct, request.Masker, probe, endpoint, request.OutputSink).ConfigureAwait(false); - } - finally - { - await _inFlightStore.ClearAsync(request.ServerTaskId, slot, ct).ConfigureAwait(false); - } + var result = await _observer.ObserveAndCompleteAsync(request.Machine, scriptClient, existingTicket, scriptTimeout, ct, request.Masker, probe, endpoint, request.OutputSink).ConfigureAwait(false); + + // Clear only on a definitive observation; preserve on a throw so a further + // resume re-attaches again (same rationale as DispatchOrReattachAsync). + await ClearInFlightAsync(request.ServerTaskId, slot).ConfigureAwait(false); + + return result; } /// diff --git a/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs b/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs index e1ac703e..896e60b6 100644 --- a/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs +++ b/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs @@ -252,6 +252,66 @@ await Run>( customMessage: "Both dispatch slots must be cleared once observed to completion."); } + [Fact] + public async Task ObserveThrowsTransient_PreservesInFlightPointer_ForResumeReattach() + { + // P1b: a transient infra failure during the observe loop (a Halibut RPC drop + // that outlived the library's retries) must PRESERVE the in-flight pointer — + // the script may still be running on the agent, so a resumed run re-attaches + // to it instead of dispatching a duplicate. The strategy clears the pointer + // ONLY on a definitive observation (return), never on a throw. + const int taskId = 810007, machineId = 17; + + await EnsureRowAsync(taskId).ConfigureAwait(false); + + // Fresh dispatch: StartScript is accepted, then the very first GetStatus + // throws — simulating the transient drop after the agent launched the script. + var agent = new RecordingScriptService( + getStatusScript: new[] { (ProcessState.Complete, 0) }, + completeResult: (ProcessState.Complete, 0)) + { + GetStatusThrows = new HalibutClientException("transient connection drop") + }; + + await Should.ThrowAsync(() => ExecuteWithAgentAsync(taskId, machineId, agent)).ConfigureAwait(false); + + agent.StartScriptCalls.ShouldBe(1, customMessage: "The fresh script must have been dispatched before the transient drop."); + + var preserved = await GetTicketAsync(taskId, machineId).ConfigureAwait(false); + preserved.ShouldBe(agent.StartedTickets.Single(), + customMessage: "A transient observe-loop failure must PRESERVE the in-flight pointer (the strategy clears it only on a " + + "definitive observation, never in a finally), or a resumed run re-dispatches a duplicate of the still-running script."); + } + + [Fact] + public async Task ReattachProbe_TransientFailure_PreservesPointer_DoesNotDispatchFresh() + { + // P1b (review HIGH): on resume the reattach probe (GetStatus on the recorded + // ticket) can itself hit a transient failure (the agent is still down). That + // must NOT clear the pointer or dispatch fresh — the recorded script may + // still be running, so a fresh dispatch would DUPLICATE it. The pointer is + // preserved and the failure propagates so the deployment pauses again. + const int taskId = 810008, machineId = 18; + const string ticket = "reattach-ticket-transient-probe"; + + await EnsureRowAsync(taskId).ConfigureAwait(false); + await RecordTicketAsync(taskId, machineId, ticket).ConfigureAwait(false); + + var agent = new RecordingScriptService( + getStatusScript: new[] { (ProcessState.Complete, 0) }, + completeResult: (ProcessState.Complete, 0)) + { + GetStatusThrows = new HalibutClientException("agent still unreachable on resume") + }; + + await Should.ThrowAsync(() => ExecuteWithAgentAsync(taskId, machineId, agent)).ConfigureAwait(false); + + agent.StartScriptCalls.ShouldBe(0, + customMessage: "A transient reattach-probe failure must NOT dispatch fresh — the recorded script may still be running (duplicate-execution risk)."); + (await GetTicketAsync(taskId, machineId).ConfigureAwait(false)).ShouldBe(ticket, + customMessage: "A transient reattach-probe failure must PRESERVE the recorded pointer for a later resume, not clear it."); + } + // ── Drive the real strategy with a fake agent (per-scope IHalibutClientFactory override) ── private async Task ExecuteWithAgentAsync(int taskId, int machineId, RecordingScriptService agent) @@ -341,9 +401,17 @@ public async Task StartScriptAsync(StartScriptCommand comm return Resp(command.ScriptTicket, ProcessState.Running, 0); } + /// When set, GetStatus throws this (simulating a transient RPC drop + /// that outlived Halibut's own retries) instead of returning a status. + public Exception GetStatusThrows { get; set; } + public Task GetStatusAsync(ScriptStatusRequest request) { ProbedTickets.Add(request.Ticket.TaskId); + + if (GetStatusThrows != null) + throw GetStatusThrows; + // Last scripted entry repeats so the observer's poll loop always terminates. var (state, exit) = _getStatusScript.Count > 1 ? _getStatusScript.Dequeue() : _getStatusScript.Peek(); return Task.FromResult(Resp(request.Ticket, state, exit)); diff --git a/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs b/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs new file mode 100644 index 00000000..55e7fc34 --- /dev/null +++ b/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs @@ -0,0 +1,82 @@ +using System; +using Halibut; +using Halibut.Exceptions; +using Shouldly; +using Squid.Core.Halibut.Resilience; +using Xunit; + +namespace Squid.UnitTests.Halibut.Resilience; + +/// +/// Pins — the single +/// definition of "transient infra failure that should PAUSE a deployment +/// (resumable, re-attach) rather than fail it terminally". +/// +/// The dangerous edge this guards: classifying a PERMANENT Halibut +/// protocol/invocation failure (version mismatch, missing service/method, the +/// agent's service threw) as transient would pause-loop forever — the deployment +/// pauses, resumes, hits the same permanent error, pauses again. So those +/// subtypes MUST be excluded even though they derive from +/// . +/// +public sealed class TransientFailureClassifierTests +{ + [Fact] + public void BaseHalibutClientException_IsTransient() + => TransientFailureClassifier.IsTransient(new HalibutClientException("connection reset by peer")).ShouldBeTrue(); + + [Fact] + public void AgentUnreachable_IsTransient() + => TransientFailureClassifier.IsTransient(new AgentUnreachableException("agent-1", 3)).ShouldBeTrue(); + + [Fact] + public void CircuitOpen_IsNotTransient() + // The breaker only opens after the failure threshold (3+ consecutive + // failures) — a SUSTAINED agent problem, not a one-off blip — and is raised + // BEFORE any script is dispatched, so there is nothing to re-attach to. + // Pausing on it would loop on a dead agent and break the deliberate + // fail-fast-on-open-breaker → Failed contract. + => TransientFailureClassifier.IsTransient(new CircuitOpenException(7, DateTimeOffset.UtcNow.AddSeconds(60))).ShouldBeFalse(); + + [Fact] + public void GenericException_IsNotTransient() + => TransientFailureClassifier.IsTransient(new InvalidOperationException("a real failure")).ShouldBeFalse(); + + [Theory] + [MemberData(nameof(PermanentHalibutFailures))] + public void PermanentHalibutSubtypes_AreNotTransient(Exception permanent) + { + // These reached the agent and were PERMANENTLY rejected (protocol/invocation) + // — retrying forever is pointless; they must fail, not pause-loop. + TransientFailureClassifier.IsTransient(permanent).ShouldBeFalse( + customMessage: $"{permanent.GetType().Name} is a permanent protocol/invocation failure — it must NOT be classified transient (would pause-loop forever)."); + } + + public static TheoryData PermanentHalibutFailures() => new() + { + new ServiceInvocationHalibutClientException("the agent's service threw"), + new NoMatchingServiceOrMethodHalibutClientException("no matching service/method"), + new MethodNotFoundHalibutClientException("method not found"), + new ServiceNotFoundHalibutClientException("service not found"), + new AmbiguousMethodMatchHalibutClientException("ambiguous method"), + }; + + [Fact] + public void Aggregate_AllTransient_IsTransient() + => TransientFailureClassifier.IsTransient( + new AggregateException(new HalibutClientException("a"), new AgentUnreachableException("b", 1))).ShouldBeTrue(); + + [Fact] + public void Aggregate_WithOnePermanentHalibut_IsNotTransient() + => TransientFailureClassifier.IsTransient( + new AggregateException(new HalibutClientException("a"), new NoMatchingServiceOrMethodHalibutClientException("perm"))).ShouldBeFalse(); + + [Fact] + public void Aggregate_WithOneGenuineFailure_IsNotTransient() + => TransientFailureClassifier.IsTransient( + new AggregateException(new HalibutClientException("a"), new InvalidOperationException("real"))).ShouldBeFalse(); + + [Fact] + public void Aggregate_Empty_IsNotTransient() + => TransientFailureClassifier.IsTransient(new AggregateException()).ShouldBeFalse(); +} diff --git a/tests/Squid.UnitTests/Services/Deployments/Execution/TargetCatchClassifierTests.cs b/tests/Squid.UnitTests/Services/Deployments/Execution/TargetCatchClassifierTests.cs index d71cde05..3b425345 100644 --- a/tests/Squid.UnitTests/Services/Deployments/Execution/TargetCatchClassifierTests.cs +++ b/tests/Squid.UnitTests/Services/Deployments/Execution/TargetCatchClassifierTests.cs @@ -1,5 +1,7 @@ using System; +using Halibut; using Shouldly; +using Squid.Core.Halibut.Resilience; using Squid.Core.Services.DeploymentExecution.Pipeline.Phases; using Xunit; @@ -105,4 +107,58 @@ public void NonOceException_RegardlessOfCancellationFlags_MarksFailed() classification.MarkFailed.ShouldBeTrue(); classification.TriggerFailFast.ShouldBeTrue(); } + + // ── Transient infra (P1b): the script may still be running on the agent ── + + [Theory] + [InlineData("halibut")] // HalibutClientException — RPC drop after library retries + [InlineData("agent")] // AgentUnreachableException — liveness probe gave up + public void TransientInfra_DoesNotMarkFailed_AndDoesNotFailFastPeers(string kind) + { + // The transient target's script may still be running, so it must NOT be + // marked terminal — a resume re-attaches to it via its preserved in-flight + // pointer. Healthy peers are NOT aborted (no fail-fast): they finish, then + // the deployment pauses. + var classification = TargetCatchClassifier.Classify( + Transient(kind), + failFastCancelled: false, + parentCtCancelled: false); + + classification.MarkFailed.ShouldBeFalse( + customMessage: "A transient infra failure must leave the target in-flight (not terminal) so resume re-attaches."); + classification.TriggerFailFast.ShouldBeFalse( + customMessage: "A transient infra failure must NOT abort healthy peer targets — they finish, then the deployment pauses."); + } + + [Fact] + public void TransientInfra_DuringParentCancel_MarksFailed() + { + // A transient drop racing a real user-cancel / deploy-timeout (parent CT + // cancelled) is NOT a resumable pause — the deployment is terminating, so + // treat it as a failure (don't preserve as in-flight). + var classification = TargetCatchClassifier.Classify( + new HalibutClientException("drop"), + failFastCancelled: true, + parentCtCancelled: true); + + classification.MarkFailed.ShouldBeTrue(); + } + + [Fact] + public void IsTransientInfraFailure_ClassifiesEachShape() + { + TargetCatchClassifier.IsTransientInfraFailure(new HalibutClientException("x")).ShouldBeTrue(); + TargetCatchClassifier.IsTransientInfraFailure(new AgentUnreachableException("a", 2)).ShouldBeTrue(); + TargetCatchClassifier.IsTransientInfraFailure(new InvalidOperationException("real")).ShouldBeFalse(); + + // Aggregate qualifies only when EVERY inner failure is transient. + TargetCatchClassifier.IsTransientInfraFailure( + new AggregateException(new HalibutClientException("a"), new AgentUnreachableException("b", 1))).ShouldBeTrue(); + TargetCatchClassifier.IsTransientInfraFailure( + new AggregateException(new HalibutClientException("a"), new InvalidOperationException("real"))).ShouldBeFalse(); + TargetCatchClassifier.IsTransientInfraFailure(new AggregateException()).ShouldBeFalse(); + } + + private static Exception Transient(string kind) + => kind == "agent" ? new AgentUnreachableException("agent-1", 3) : new HalibutClientException("connection reset by peer"); } diff --git a/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentExecutionLoggingTests.cs b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentExecutionLoggingTests.cs index d80855fa..4109ddc8 100644 --- a/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentExecutionLoggingTests.cs +++ b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentExecutionLoggingTests.cs @@ -2,6 +2,7 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; +using Halibut; using Squid.Core.Persistence.Entities.Deployments; using Squid.Core.Services.Common; using Squid.Core.Services.DeploymentExecution; @@ -332,6 +333,37 @@ public async Task StepFailed_DoesNotLogCompleted() completedLog.ShouldBeNull("Should NOT log step completed when step failed"); } + // ========== Transient infra failure PROPAGATES (P1b) ========== + + [Theory] + [InlineData(true)] // required step — the per-target catch leaves it in-flight + rethrows + [InlineData(false)] // NON-required step — must NOT be swallowed into FailureEncountered + public async Task TransientStrategyFailure_PropagatesOutOfPhase_RegardlessOfRequired(bool isRequired) + { + // P1b (review HIGH): a transient infra failure (Halibut RPC drop) must + // PROPAGATE out of ExecuteStepsPhase so the runner pauses for resume — even + // on a NON-required step. Pre-fix, a non-required step's transient was + // swallowed by `catch (Exception) { ...; if (step.IsRequired) throw; }`, + // which routed the deployment to OnFailureAsync (Failed + checkpoint AND + // in-flight pointer deleted), orphaning the still-running script. Contrast: + // a non-required GENUINE failure (Success=false result) is still swallowed — + // pinned by StepFailed_DoesNotLogCompleted — so only TRANSIENT propagates. + var (phase, ctx, _, _) = CreateTestHarness(); + + var transport = new TestTransport(new ThrowingStrategy(new HalibutClientException("agent dropped mid-script"))); + ctx.AllTargetsContext = new List { MakeTarget("target-1", "web", transport) }; + + var step = MakeStep("Deploy web", 1, "web"); + step.IsRequired = isRequired; + step.Actions = new List { MakeAction("RunScript") }; + ctx.Steps = new List { step }; + + await Should.ThrowAsync(() => phase.ExecuteAsync(ctx, CancellationToken.None)); + + ctx.FailureEncountered.ShouldBeFalse( + customMessage: "A transient infra failure must NOT be recorded as a terminal step failure — it propagates so the runner pauses (resumable), not fails."); + } + // ========== No Matching Targets — Plural/Singular ========== [Fact] @@ -809,6 +841,19 @@ public Task ExecuteScriptAsync(ScriptExecutionRequest req } } + // Throws a transient infra failure (vs FailingStrategy which returns a Failed + // RESULT). Used to prove the phase PROPAGATES a transient — even on a + // non-required step — instead of swallowing it. + private sealed class ThrowingStrategy : IExecutionStrategy + { + private readonly Exception _ex; + + public ThrowingStrategy(Exception ex) => _ex = ex; + + public Task ExecuteScriptAsync(ScriptExecutionRequest request, CancellationToken ct) + => throw _ex; + } + private sealed class SimpleRunScriptHandler : IActionHandler { public string ActionType => "Squid.Script"; diff --git a/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs new file mode 100644 index 00000000..fcd4adbd --- /dev/null +++ b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs @@ -0,0 +1,139 @@ +using System; +using Halibut; +using Squid.Core.Halibut.Resilience; +using Squid.Core.Services.DeploymentExecution; +using Squid.Core.Services.DeploymentExecution.Lifecycle; +using Squid.Core.Services.DeploymentExecution.Pipeline; +using Squid.Core.Services.DeploymentExecution.Script; +using Squid.Core.Services.Deployments.ServerTask; + +namespace Squid.UnitTests.Services.Deployments.Pipeline; + +/// +/// Pins how the runner classifies a TRANSIENT INFRASTRUCTURE failure (a Halibut +/// RPC drop that outlived the library's own retries, or an unreachable agent): +/// by default it pauses the deployment for resume (OnTransientPauseAsync — +/// Paused + checkpoint preserved) rather than failing it terminally, so the +/// still-running script can be re-attached to. The historical fail-fast +/// behaviour remains behind the SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE escape hatch. +/// +/// Sibling to DeploymentPipelineRunnerTimeoutTests, which pins the +/// analogous wall-clock-timeout classification. A genuine (non-transient) +/// exception must still route through OnFailureAsync + rethrow — pinned here so a +/// future regression that over-broadens the transient predicate is caught. +/// +public class DeploymentPipelineRunnerTransientPauseTests +{ + private readonly Mock _lifecycle = new(); + private readonly Mock _completion = new(); + private readonly TaskCancellationRegistry _registry = new(); + private readonly Mock _taskDataProvider = new(); + + [Theory] + [InlineData("halibut")] // HalibutClientException — RPC drop after library retries + [InlineData("agent")] // AgentUnreachableException — liveness probe gave up + public async Task TransientFailure_Default_CallsOnTransientPause_NotOnFailure(string kind) + { + var runner = CreateRunner(CreateThrowingPhase(TransientOf(kind))); + + await runner.ProcessAsync(1, CancellationToken.None); + + _completion.Verify(c => c.OnTransientPauseAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once); + _completion.Verify(c => c.OnFailureAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + _completion.Verify(c => c.OnCancelledAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task TransientFailure_Default_EmitsPausedEvent_AndDoesNotRethrow() + { + var runner = CreateRunner(CreateThrowingPhase(new HalibutClientException("connection reset by peer"))); + + await Should.NotThrowAsync(() => runner.ProcessAsync(1, CancellationToken.None)); + + _lifecycle.Verify(l => l.EmitAsync(It.IsAny(), It.IsAny()), Times.Once); + _lifecycle.Verify(l => l.EmitAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task GenuineFailure_RoutesThroughOnFailure_NotTransientPause_AndRethrows() + { + // A non-transient exception (script/RBAC/logic failure) must NOT be mistaken + // for a transient blip — it fails terminally and rethrows. + var runner = CreateRunner(CreateThrowingPhase(new InvalidOperationException("a real failure"))); + + await Should.ThrowAsync(() => runner.ProcessAsync(1, CancellationToken.None)); + + _completion.Verify(c => c.OnFailureAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once); + _completion.Verify(c => c.OnTransientPauseAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task AggregateOfTransients_PausesForResume() + { + // A parallel batch surfaces multiple target failures as an AggregateException. + // When EVERY inner failure is transient, the whole deployment pauses. + var aggregate = new AggregateException(new HalibutClientException("drop A"), new HalibutClientException("drop B")); + var runner = CreateRunner(CreateThrowingPhase(aggregate)); + + await runner.ProcessAsync(1, CancellationToken.None); + + _completion.Verify(c => c.OnTransientPauseAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once); + _completion.Verify(c => c.OnFailureAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task AggregateWithOneGenuineFailure_FailsNotPauses() + { + // A mix that includes ANY real failure is a true failure, not a pausable blip. + var aggregate = new AggregateException(new HalibutClientException("drop"), new InvalidOperationException("real")); + var runner = CreateRunner(CreateThrowingPhase(aggregate)); + + await Should.ThrowAsync(() => runner.ProcessAsync(1, CancellationToken.None)); + + _completion.Verify(c => c.OnFailureAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once); + _completion.Verify(c => c.OnTransientPauseAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task TransientFailure_DuringUserCancel_DoesNotPause() + { + // A user-cancel that races a transient RPC drop (a raw HalibutClientException, + // NOT an OperationCanceledException) must NOT be reclassified as a transient + // pause — the cancel/timeout guard makes cancel win, so the transient catch is + // skipped and the exception falls through to the failure/cancel handling. + var phase = new Mock(); + phase.Setup(p => p.Order).Returns(1); + phase.Setup(p => p.ExecuteAsync(It.IsAny(), It.IsAny())) + .Returns((_, __) => + { + _registry.TryCancel(1); + throw new HalibutClientException("connection reset while cancelling"); + }); + var runner = CreateRunner(phase.Object); + + await Should.ThrowAsync(() => runner.ProcessAsync(1, CancellationToken.None)); + + _completion.Verify(c => c.OnTransientPauseAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Never, + failMessage: "A transient drop racing a user-cancel must NOT pause — cancel/timeout win over the transient classification."); + } + + private static Exception TransientOf(string kind) + => kind == "agent" ? new AgentUnreachableException("agent-1", 3) : new HalibutClientException("connection reset by peer"); + + private IDeploymentPipelinePhase CreateThrowingPhase(Exception ex) + { + var phase = new Mock(); + phase.Setup(p => p.Order).Returns(1); + phase.Setup(p => p.ExecuteAsync(It.IsAny(), It.IsAny())) + .ThrowsAsync(ex); + + return phase.Object; + } + + // Long timeout so the wall-clock timer never fires before the phase throws. + private DeploymentPipelineRunner CreateRunner(params IDeploymentPipelinePhase[] phases) + => new(phases, _lifecycle.Object, _completion.Object, _registry, _taskDataProvider.Object) + { + DeploymentTimeout = TimeSpan.FromMinutes(5) + }; +}