From 58dece1649ed4f28297ad824abc778cb9a668680 Mon Sep 17 00:00:00 2001 From: "Mars.P" Date: Thu, 11 Jun 2026 15:09:38 +0800 Subject: [PATCH 1/4] Pause (resumable) on a transient infra failure instead of failing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Halibut RPC drop that outlives the library's own retries, or an agent that goes unreachable mid-script, used to fail the deployment terminally AND — via the strategy's finally that cleared the in-flight pointer on any exit — destroy the reattach pointer, so a resume re-dispatched a duplicate of the still-running script. Classify a transient infrastructure failure (HalibutClientException / AgentUnreachableException, or an AggregateException of only those) as a PAUSE: the task transitions to Paused with its checkpoint AND in-flight pointer preserved, and a resume re-attaches to the still-running script. Reuses the #428 pause/resume machinery; env-gated by SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE (default true) so operators can opt back into fail-fast. - HalibutMachineExecutionStrategy: clear the in-flight pointer ONLY on a definitive observation (return), preserve it on a throw. - TargetCatchClassifier: a transient infra failure leaves the target in-flight (not terminal) and does NOT fail-fast healthy peers — they finish, then the deployment pauses. Owns the shared IsTransientInfraFailure. - DeploymentPipelineRunner: a transient failure routes to OnTransientPauseAsync (Paused + checkpoint preserved), not OnFailureAsync; does not rethrow. - DeploymentCompletionHandler: OnTransientPauseAsync (mirrors OnTimedOutAsync). A genuine (non-transient) exception, and a transient drop racing a real cancel/timeout, still fail terminally. Tests: - Unit: TargetCatchClassifier transient cases + IsTransientInfraFailure matrix; DeploymentPipelineRunner transient->pause vs fail-fast vs genuine failure, aggregate-of-transients, env-var pin + parse matrix - Integration: HalibutResumeReattach observe-throws-transient preserves the in-flight pointer for resume --- .../Pipeline/DeploymentCompletionHandler.cs | 25 +++ .../Pipeline/DeploymentPipelineRunner.cs | 52 +++++ .../Pipeline/IDeploymentPipelinePhase.cs | 7 + .../Pipeline/Phases/TargetCatchClassifier.cs | 35 ++++ .../HalibutMachineExecutionStrategy.cs | 42 ++-- .../Checkpoints/HalibutResumeReattachTests.cs | 39 ++++ .../Execution/TargetCatchClassifierTests.cs | 56 +++++ ...oymentPipelineRunnerTransientPauseTests.cs | 191 ++++++++++++++++++ 8 files changed, 430 insertions(+), 17 deletions(-) create mode 100644 tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs 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..1243addb 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; @@ -41,8 +42,25 @@ public sealed class DeploymentPipelineRunner(IEnumerable + /// Operator escape hatch (Rule 8): controls what happens when a deployment + /// hits a transient infrastructure failure — a Halibut RPC drop that outlived + /// the library's own retries, or an agent that went unreachable mid-script. + /// The safe default (unset / blank / unrecognised) is true — the task + /// is paused and its checkpoint + in-flight script pointer preserved so it can + /// be resumed (re-attaching to the still-running script) instead of failing + /// and re-dispatching a duplicate. Set this to a falsey value + /// (false/0/no/off, case-insensitive) to restore the + /// historical fail-fast behaviour: a transient-failed deployment transitions to + /// Failed and its checkpoint is deleted. + /// + public const string DeploymentTransientResumableEnvVar = "SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE"; + + internal const bool DefaultDeploymentTransientResumable = true; + internal TimeSpan DeploymentTimeout { get; init; } = ResolveDeploymentTimeout(); internal bool TimeoutResumable { get; init; } = ResolveTimeoutResumable(); + internal bool TransientResumable { get; init; } = ResolveTransientResumable(); internal TimeSpan ConcurrencyMaxWait { get; init; } = DefaultConcurrencyMaxWait; internal TimeSpan ConcurrencyPollInterval { get; init; } = DefaultConcurrencyPollInterval; @@ -114,6 +132,16 @@ public async Task ProcessAsync(int serverTaskId, CancellationToken ct) { await SafeCompleteAsync(ctx, () => completion.OnCancelledAsync(ctx, CancellationToken.None), new DeploymentCancelledEvent(new DeploymentEventContext())); } + catch (Exception ex) when (TransientResumable && TargetCatchClassifier.IsTransientInfraFailure(ex)) + { + // A transient infra failure (Halibut RPC drop after the library's + // retries, or an unreachable agent) 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. + 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 })); @@ -228,4 +256,28 @@ internal static bool ParseTimeoutResumable(string raw) private static bool ResolveTimeoutResumable() => ParseTimeoutResumable(Environment.GetEnvironmentVariable(DeploymentTimeoutResumableEnvVar)); + + /// + /// Parses the operator-supplied transient-resumable flag. Identical contract to + /// : safe default true (pause + + /// preserve checkpoint), only an explicit falsey token + /// (false/0/no/off, case-insensitive, surrounding + /// whitespace tolerated) opts back into fail-fast; anything unrecognised falls + /// back to the safe default so a typo can never silently discard progress. + /// + internal static bool ParseTransientResumable(string raw) + { + if (string.IsNullOrWhiteSpace(raw)) return DefaultDeploymentTransientResumable; + + var value = raw.Trim(); + + if (value.Equals("false", StringComparison.OrdinalIgnoreCase) || value == "0" + || value.Equals("no", StringComparison.OrdinalIgnoreCase) || value.Equals("off", StringComparison.OrdinalIgnoreCase)) + return false; + + return DefaultDeploymentTransientResumable; + } + + private static bool ResolveTransientResumable() + => ParseTransientResumable(Environment.GetEnvironmentVariable(DeploymentTransientResumableEnvVar)); } 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/TargetCatchClassifier.cs b/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs index f031ffe2..28469e97 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs @@ -1,3 +1,7 @@ +using System.Linq; +using Halibut; +using Squid.Core.Halibut.Resilience; + namespace Squid.Core.Services.DeploymentExecution.Pipeline.Phases; /// @@ -44,10 +48,41 @@ 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: a + /// (an RPC drop that outlived the Halibut + /// library's own retries) or an (the + /// liveness probe gave up on the agent). A parallel batch's + /// qualifies only when EVERY inner + /// failure is itself transient — a mix that includes a real script/RBAC failure + /// is a true failure, not a pausable blip. Shared by the runner's + /// pause-classification and this per-target classifier so the definition of + /// "transient" lives in one place. + /// + public static bool IsTransientInfraFailure(System.Exception ex) + { + if (ex is System.AggregateException aggregate) + return aggregate.InnerExceptions.Count > 0 && aggregate.InnerExceptions.All(IsTransientInfraFailure); + + return ex is HalibutClientException || ex is AgentUnreachableException; + } } 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..198b3231 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); } /// @@ -310,14 +319,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..8cb86e34 100644 --- a/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs +++ b/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs @@ -252,6 +252,37 @@ 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."); + } + // ── Drive the real strategy with a fake agent (per-scope IHalibutClientFactory override) ── private async Task ExecuteWithAgentAsync(int taskId, int machineId, RecordingScriptService agent) @@ -341,9 +372,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/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/DeploymentPipelineRunnerTransientPauseTests.cs b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs new file mode 100644 index 00000000..d26c8674 --- /dev/null +++ b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs @@ -0,0 +1,191 @@ +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 TransientFailure_FailFast_CallsOnFailure_NotOnTransientPause_AndRethrows() + { + // Escape hatch (SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE=false): restore the + // historical fail-fast behaviour — a transient drop routes through + // OnFailureAsync (Failed + checkpoint deleted) and the runner rethrows. + var runner = CreateFailFastRunner(CreateThrowingPhase(new HalibutClientException("connection reset by peer"))); + + 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 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, even when + // transient-resumable is on (the default). + 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); + } + + // ── Resumable-transient flag (Rule 8 env-var escape hatch) ──────────────── + + [Fact] + public void DeploymentTransientResumableEnvVar_ConstantNamePinned() + { + // Operators set this in Helm overrides / container env. Renaming it + // silently flips every opted-out tenant back to resumable pausing. + DeploymentPipelineRunner.DeploymentTransientResumableEnvVar + .ShouldBe("SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE"); + } + + [Fact] + public void DefaultDeploymentTransientResumable_IsTrue() + { + DeploymentPipelineRunner.DefaultDeploymentTransientResumable.ShouldBeTrue(); + } + + [Theory] + [InlineData(null, true)] + [InlineData("", true)] + [InlineData(" ", true)] + [InlineData("false", false)] + [InlineData("False", false)] + [InlineData(" off ", false)] + [InlineData("0", false)] + [InlineData("no", false)] + [InlineData("true", true)] + [InlineData("1", true)] + [InlineData("garbage", true)] // unrecognised → safe resumable default + public void ParseTransientResumable_HandlesAllInputs(string raw, bool expected) + { + DeploymentPipelineRunner.ParseTransientResumable(raw).ShouldBe(expected); + } + + [Fact] + public void TransientResumableEnvVar_SetFalse_DrivesProperty() + { + var original = Environment.GetEnvironmentVariable(DeploymentPipelineRunner.DeploymentTransientResumableEnvVar); + Environment.SetEnvironmentVariable(DeploymentPipelineRunner.DeploymentTransientResumableEnvVar, "false"); + + try + { + var runner = new DeploymentPipelineRunner(Array.Empty(), _lifecycle.Object, _completion.Object, _registry, _taskDataProvider.Object); + + runner.TransientResumable.ShouldBeFalse(); + } + finally + { + Environment.SetEnvironmentVariable(DeploymentPipelineRunner.DeploymentTransientResumableEnvVar, original); + } + } + + 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), + TransientResumable = true + }; + + private DeploymentPipelineRunner CreateFailFastRunner(params IDeploymentPipelinePhase[] phases) + => new(phases, _lifecycle.Object, _completion.Object, _registry, _taskDataProvider.Object) + { + DeploymentTimeout = TimeSpan.FromMinutes(5), + TransientResumable = false + }; +} From f9dc84321196ecbb51e24c9bcab5453d49c02b91 Mon Sep 17 00:00:00 2001 From: "Mars.P" Date: Thu, 11 Jun 2026 15:36:43 +0800 Subject: [PATCH 2/4] Harden transient-pause classification (adversarial review fixes) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of the transient-pause change found five reachable paths that defeated the resumable-pause guarantee. Fixes: - Centralise the transient definition in TransientFailureClassifier (Squid.Core.Halibut.Resilience) so the pipeline runner, per-target classifier, per-action catch, and the re-attach probe all agree. - NARROW the predicate: a permanent Halibut protocol/invocation failure (ServiceInvocation / NoMatchingServiceOrMethod / MethodNotFound / ServiceNotFound / AmbiguousMethodMatch) is NOT transient — classifying it so would pause-loop forever. Base/transport HalibutClientException, AgentUnreachableException, and CircuitOpenException ARE transient. - Non-required step: a transient now propagates (rethrow before the per-target/terminal failure recording) regardless of step.IsRequired — previously it was swallowed into FailureEncountered -> OnFailureAsync, which deleted the checkpoint AND the preserved in-flight pointer, orphaning the still-running script. - Re-attach probe: a TRANSIENT probe failure on resume now preserves the pointer and propagates (was: type-blind catch cleared it and dispatched fresh -> duplicate-execution risk when the agent recovers). - Runner transient catch is guarded by !timeoutCts && !registryCts && !ct so a user-cancel / wall-clock timeout racing a transient RPC drop is not reclassified as a pause (cancel/timeout win). - CircuitOpenException is treated as transient (its own contract) so a tripped breaker on resume pauses + preserves the pointer instead of forcing a terminal Failed. Tests: - Unit: TransientFailureClassifier matrix (permanent subtypes excluded, CircuitOpen/AgentUnreachable/base included, aggregate all/mixed/empty); runner transient-during-user-cancel does NOT pause - Integration: reattach-probe transient failure preserves the pointer and does not dispatch fresh --- .../Resilience/TransientFailureClassifier.cs | 53 +++++++++++++ .../Pipeline/DeploymentPipelineRunner.cs | 10 ++- .../Phases/6_ExecuteStepsPhase.Execute.cs | 15 ++++ .../Pipeline/Phases/TargetCatchClassifier.cs | 23 ++---- .../HalibutMachineExecutionStrategy.cs | 13 +++ .../Checkpoints/HalibutResumeReattachTests.cs | 29 +++++++ .../TransientFailureClassifierTests.cs | 79 +++++++++++++++++++ ...oymentPipelineRunnerTransientPauseTests.cs | 23 ++++++ 8 files changed, 227 insertions(+), 18 deletions(-) create mode 100644 src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs create mode 100644 tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs diff --git a/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs b/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs new file mode 100644 index 00000000..ef8625b6 --- /dev/null +++ b/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs @@ -0,0 +1,53 @@ +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 = the round-trip to the agent could not be completed, or the +/// agent is fail-fast-rejected by an open breaker: +/// +/// — the breaker is open; its own +/// contract says treat as transient and retry after the open window. +/// — the liveness probe gave up. +/// A base/transport — connection +/// reset, EOF, TLS handshake failure, timeout, etc. +/// +/// +/// NOT transient — the request reached the agent and was PERMANENTLY +/// rejected, so retrying forever is pointless (and would pause-loop): the Halibut +/// protocol/invocation subtypes (version mismatch → no matching service/method; +/// the agent's service itself threw → service-invocation). These are real +/// failures and must fail the deployment. +/// +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 CircuitOpenException or 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/DeploymentPipelineRunner.cs b/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs index 1243addb..155685b5 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs @@ -132,7 +132,9 @@ public async Task ProcessAsync(int serverTaskId, CancellationToken ct) { await SafeCompleteAsync(ctx, () => completion.OnCancelledAsync(ctx, CancellationToken.None), new DeploymentCancelledEvent(new DeploymentEventContext())); } - catch (Exception ex) when (TransientResumable && TargetCatchClassifier.IsTransientInfraFailure(ex)) + catch (Exception ex) when (TransientResumable + && !timeoutCts.IsCancellationRequested && !registryCts.IsCancellationRequested && !ct.IsCancellationRequested + && TargetCatchClassifier.IsTransientInfraFailure(ex)) { // A transient infra failure (Halibut RPC drop after the library's // retries, or an unreachable agent) pauses the deployment instead of @@ -140,6 +142,12 @@ public async Task ProcessAsync(int serverTaskId, CancellationToken ct) // 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. + // + // 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) 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 28469e97..7a3da284 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs @@ -1,5 +1,3 @@ -using System.Linq; -using Halibut; using Squid.Core.Halibut.Resilience; namespace Squid.Core.Services.DeploymentExecution.Pipeline.Phases; @@ -68,21 +66,12 @@ public static Classification Classify(System.Exception ex, bool failFastCancelle /// /// Whether is a transient infrastructure failure that - /// should pause (resumable) rather than fail the deployment: a - /// (an RPC drop that outlived the Halibut - /// library's own retries) or an (the - /// liveness probe gave up on the agent). A parallel batch's - /// qualifies only when EVERY inner - /// failure is itself transient — a mix that includes a real script/RBAC failure - /// is a true failure, not a pausable blip. Shared by the runner's - /// pause-classification and this per-target classifier so the definition of - /// "transient" lives in one place. + /// 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) - { - if (ex is System.AggregateException aggregate) - return aggregate.InnerExceptions.Count > 0 && aggregate.InnerExceptions.All(IsTransientInfraFailure); - - return ex is HalibutClientException || ex is AgentUnreachableException; - } + => 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 198b3231..3b404039 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Targets/Tentacle/Transport/HalibutMachineExecutionStrategy.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Targets/Tentacle/Transport/HalibutMachineExecutionStrategy.cs @@ -304,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); diff --git a/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs b/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs index 8cb86e34..896e60b6 100644 --- a/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs +++ b/tests/Squid.IntegrationTests/Services/Deployments/Checkpoints/HalibutResumeReattachTests.cs @@ -283,6 +283,35 @@ public async Task ObserveThrowsTransient_PreservesInFlightPointer_ForResumeReatt "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) diff --git a/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs b/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs new file mode 100644 index 00000000..54b9184d --- /dev/null +++ b/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs @@ -0,0 +1,79 @@ +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_IsTransient() + // The breaker's own contract says treat as transient + retry after the open + // window. A tripped breaker on resume must pause (preserve pointer), not fail. + => TransientFailureClassifier.IsTransient(new CircuitOpenException(7, DateTimeOffset.UtcNow.AddSeconds(60))).ShouldBeTrue(); + + [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/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs index d26c8674..5471972b 100644 --- a/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs +++ b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs @@ -109,6 +109,29 @@ public async Task AggregateWithOneGenuineFailure_FailsNotPauses() _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."); + } + // ── Resumable-transient flag (Rule 8 env-var escape hatch) ──────────────── [Fact] From 0e2a098b0cf9cc16f02c89fa81c461653b92dc5f Mon Sep 17 00:00:00 2001 From: "Mars.P" Date: Thu, 11 Jun 2026 15:56:44 +0800 Subject: [PATCH 3/4] Make transient-pause unconditional (drop env-var toggle) + real-phase tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove the SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE escape hatch added earlier in this PR: pausing-resumable on a transient infra blip is the only correct behaviour (failing fast would discard already-completed progress and risk a duplicate run — no legitimate use case), so it needs no opt-out. The env var never shipped, so removing it is non-breaking. The wall-clock-timeout toggle (SQUID_DEPLOYMENT_TIMEOUT_RESUMABLE, shipped in 1.8.18) is unaffected. The runner's transient catch keeps its cancel/timeout guard (!timeoutCts && !registryCts && !ct) so cancel/timeout still win. Tests: - Drop the env-var pin / parse-matrix / fail-fast tests (no longer apply) - Add real-ExecuteStepsPhase propagation tests (DeploymentExecutionLoggingTests.TransientStrategyFailure_PropagatesOutOfPhase_RegardlessOfRequired): a throwing strategy on a step proves a transient PROPAGATES out of the phase for BOTH required AND non-required steps (closing the review's non-required-swallow gap end-to-end), and FailureEncountered stays false; the existing StepFailed_DoesNotLogCompleted pins the contrast that a non-required GENUINE failure is still swallowed --- .../Pipeline/DeploymentPipelineRunner.cs | 57 +++---------- .../DeploymentExecutionLoggingTests.cs | 45 +++++++++++ ...oymentPipelineRunnerTransientPauseTests.cs | 79 +------------------ 3 files changed, 56 insertions(+), 125 deletions(-) diff --git a/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs b/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs index 155685b5..0fd810dc 100644 --- a/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs +++ b/src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs @@ -42,25 +42,8 @@ public sealed class DeploymentPipelineRunner(IEnumerable - /// Operator escape hatch (Rule 8): controls what happens when a deployment - /// hits a transient infrastructure failure — a Halibut RPC drop that outlived - /// the library's own retries, or an agent that went unreachable mid-script. - /// The safe default (unset / blank / unrecognised) is true — the task - /// is paused and its checkpoint + in-flight script pointer preserved so it can - /// be resumed (re-attaching to the still-running script) instead of failing - /// and re-dispatching a duplicate. Set this to a falsey value - /// (false/0/no/off, case-insensitive) to restore the - /// historical fail-fast behaviour: a transient-failed deployment transitions to - /// Failed and its checkpoint is deleted. - /// - public const string DeploymentTransientResumableEnvVar = "SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE"; - - internal const bool DefaultDeploymentTransientResumable = true; - internal TimeSpan DeploymentTimeout { get; init; } = ResolveDeploymentTimeout(); internal bool TimeoutResumable { get; init; } = ResolveTimeoutResumable(); - internal bool TransientResumable { get; init; } = ResolveTransientResumable(); internal TimeSpan ConcurrencyMaxWait { get; init; } = DefaultConcurrencyMaxWait; internal TimeSpan ConcurrencyPollInterval { get; init; } = DefaultConcurrencyPollInterval; @@ -132,16 +115,18 @@ public async Task ProcessAsync(int serverTaskId, CancellationToken ct) { await SafeCompleteAsync(ctx, () => completion.OnCancelledAsync(ctx, CancellationToken.None), new DeploymentCancelledEvent(new DeploymentEventContext())); } - catch (Exception ex) when (TransientResumable - && !timeoutCts.IsCancellationRequested && !registryCts.IsCancellationRequested && !ct.IsCancellationRequested + 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, or an unreachable agent) 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. + // 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) @@ -264,28 +249,4 @@ internal static bool ParseTimeoutResumable(string raw) private static bool ResolveTimeoutResumable() => ParseTimeoutResumable(Environment.GetEnvironmentVariable(DeploymentTimeoutResumableEnvVar)); - - /// - /// Parses the operator-supplied transient-resumable flag. Identical contract to - /// : safe default true (pause + - /// preserve checkpoint), only an explicit falsey token - /// (false/0/no/off, case-insensitive, surrounding - /// whitespace tolerated) opts back into fail-fast; anything unrecognised falls - /// back to the safe default so a typo can never silently discard progress. - /// - internal static bool ParseTransientResumable(string raw) - { - if (string.IsNullOrWhiteSpace(raw)) return DefaultDeploymentTransientResumable; - - var value = raw.Trim(); - - if (value.Equals("false", StringComparison.OrdinalIgnoreCase) || value == "0" - || value.Equals("no", StringComparison.OrdinalIgnoreCase) || value.Equals("off", StringComparison.OrdinalIgnoreCase)) - return false; - - return DefaultDeploymentTransientResumable; - } - - private static bool ResolveTransientResumable() - => ParseTransientResumable(Environment.GetEnvironmentVariable(DeploymentTransientResumableEnvVar)); } 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 index 5471972b..fcd4adbd 100644 --- a/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs +++ b/tests/Squid.UnitTests/Services/Deployments/Pipeline/DeploymentPipelineRunnerTransientPauseTests.cs @@ -54,26 +54,11 @@ public async Task TransientFailure_Default_EmitsPausedEvent_AndDoesNotRethrow() _lifecycle.Verify(l => l.EmitAsync(It.IsAny(), It.IsAny()), Times.Never); } - [Fact] - public async Task TransientFailure_FailFast_CallsOnFailure_NotOnTransientPause_AndRethrows() - { - // Escape hatch (SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE=false): restore the - // historical fail-fast behaviour — a transient drop routes through - // OnFailureAsync (Failed + checkpoint deleted) and the runner rethrows. - var runner = CreateFailFastRunner(CreateThrowingPhase(new HalibutClientException("connection reset by peer"))); - - 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 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, even when - // transient-resumable is on (the default). + // 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)); @@ -132,58 +117,6 @@ public async Task TransientFailure_DuringUserCancel_DoesNotPause() failMessage: "A transient drop racing a user-cancel must NOT pause — cancel/timeout win over the transient classification."); } - // ── Resumable-transient flag (Rule 8 env-var escape hatch) ──────────────── - - [Fact] - public void DeploymentTransientResumableEnvVar_ConstantNamePinned() - { - // Operators set this in Helm overrides / container env. Renaming it - // silently flips every opted-out tenant back to resumable pausing. - DeploymentPipelineRunner.DeploymentTransientResumableEnvVar - .ShouldBe("SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE"); - } - - [Fact] - public void DefaultDeploymentTransientResumable_IsTrue() - { - DeploymentPipelineRunner.DefaultDeploymentTransientResumable.ShouldBeTrue(); - } - - [Theory] - [InlineData(null, true)] - [InlineData("", true)] - [InlineData(" ", true)] - [InlineData("false", false)] - [InlineData("False", false)] - [InlineData(" off ", false)] - [InlineData("0", false)] - [InlineData("no", false)] - [InlineData("true", true)] - [InlineData("1", true)] - [InlineData("garbage", true)] // unrecognised → safe resumable default - public void ParseTransientResumable_HandlesAllInputs(string raw, bool expected) - { - DeploymentPipelineRunner.ParseTransientResumable(raw).ShouldBe(expected); - } - - [Fact] - public void TransientResumableEnvVar_SetFalse_DrivesProperty() - { - var original = Environment.GetEnvironmentVariable(DeploymentPipelineRunner.DeploymentTransientResumableEnvVar); - Environment.SetEnvironmentVariable(DeploymentPipelineRunner.DeploymentTransientResumableEnvVar, "false"); - - try - { - var runner = new DeploymentPipelineRunner(Array.Empty(), _lifecycle.Object, _completion.Object, _registry, _taskDataProvider.Object); - - runner.TransientResumable.ShouldBeFalse(); - } - finally - { - Environment.SetEnvironmentVariable(DeploymentPipelineRunner.DeploymentTransientResumableEnvVar, original); - } - } - private static Exception TransientOf(string kind) => kind == "agent" ? new AgentUnreachableException("agent-1", 3) : new HalibutClientException("connection reset by peer"); @@ -201,14 +134,6 @@ private IDeploymentPipelinePhase CreateThrowingPhase(Exception ex) private DeploymentPipelineRunner CreateRunner(params IDeploymentPipelinePhase[] phases) => new(phases, _lifecycle.Object, _completion.Object, _registry, _taskDataProvider.Object) { - DeploymentTimeout = TimeSpan.FromMinutes(5), - TransientResumable = true - }; - - private DeploymentPipelineRunner CreateFailFastRunner(params IDeploymentPipelinePhase[] phases) - => new(phases, _lifecycle.Object, _completion.Object, _registry, _taskDataProvider.Object) - { - DeploymentTimeout = TimeSpan.FromMinutes(5), - TransientResumable = false + DeploymentTimeout = TimeSpan.FromMinutes(5) }; } From 85cc38b6a71d150f9c5f3bc52bf773ee85f4cf25 Mon Sep 17 00:00:00 2001 From: "Mars.P" Date: Thu, 11 Jun 2026 16:12:04 +0800 Subject: [PATCH 4/4] Exclude CircuitOpenException from transient (keep fail-fast-on-open-breaker) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The K8s E2E HalibutCircuitBreakerE2ETests.BreakerForcedOpen_* asserts an open breaker fails the deployment fast (Failed) — a deliberate production-safety invariant. Classifying CircuitOpenException as transient (from the review's HIGH #5 suggestion) flipped that to Paused and broke it. Excluding it is the more correct design, by the same principle as the permanent-Halibut-subtype exclusion: the per-machine breaker only opens after 3+ consecutive failures — a SUSTAINED agent problem, not a one-off blip — and is 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 dead agent. The narrow resume-with-open-breaker orphan #5 described is better addressed by probe-before-breaker (a separate focused change), not by this blunt reclassification. The other four review HIGH fixes (non-required propagation, narrowed predicate, reattach-probe preserve, cancel/timeout guard) are unchanged. Test flipped: CircuitOpen_IsNotTransient. --- .../Resilience/TransientFailureClassifier.cs | 31 ++++++++++++------- .../TransientFailureClassifierTests.cs | 11 ++++--- 2 files changed, 27 insertions(+), 15 deletions(-) diff --git a/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs b/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs index ef8625b6..0df15053 100644 --- a/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs +++ b/src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs @@ -10,21 +10,30 @@ namespace Squid.Core.Halibut.Resilience; /// per-action catch in ExecuteStepsPhase, and the Halibut execution /// strategy's re-attach probe), so the definition lives in ONE place. /// -/// Transient = the round-trip to the agent could not be completed, or the -/// agent is fail-fast-rejected by an open breaker: +/// 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 breaker is open; its own -/// contract says treat as transient and retry after the open window. -/// — the liveness probe gave up. +/// — 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 — the request reached the agent and was PERMANENTLY -/// rejected, so retrying forever is pointless (and would pause-loop): the Halibut -/// protocol/invocation subtypes (version mismatch → no matching service/method; -/// the agent's service itself threw → service-invocation). These are real -/// failures and must fail the deployment. +/// 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 { @@ -33,7 +42,7 @@ public static bool IsTransient(Exception ex) if (ex is AggregateException aggregate) return aggregate.InnerExceptions.Count > 0 && aggregate.InnerExceptions.All(IsTransient); - if (ex is CircuitOpenException or AgentUnreachableException) + if (ex is AgentUnreachableException) return true; // Halibut transport failures are transient EXCEPT the protocol/invocation diff --git a/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs b/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs index 54b9184d..55e7fc34 100644 --- a/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs +++ b/tests/Squid.UnitTests/Halibut/Resilience/TransientFailureClassifierTests.cs @@ -30,10 +30,13 @@ public void AgentUnreachable_IsTransient() => TransientFailureClassifier.IsTransient(new AgentUnreachableException("agent-1", 3)).ShouldBeTrue(); [Fact] - public void CircuitOpen_IsTransient() - // The breaker's own contract says treat as transient + retry after the open - // window. A tripped breaker on resume must pause (preserve pointer), not fail. - => TransientFailureClassifier.IsTransient(new CircuitOpenException(7, DateTimeOffset.UtcNow.AddSeconds(60))).ShouldBeTrue(); + 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()