diff --git a/docs/decisions/0001-pipelines-halt-on-error-by-default.md b/docs/decisions/0001-pipelines-halt-on-error-by-default.md new file mode 100644 index 0000000..5c360de --- /dev/null +++ b/docs/decisions/0001-pipelines-halt-on-error-by-default.md @@ -0,0 +1,135 @@ +# ADR-0001: Pipelines Halt Step Progression on Error by Default + +**Status:** Accepted +**Date:** 2026-05-19 + +## Context + +A composed command - or any step whose inner `PipelineBuilder.Build()` catches an +exception - sets `context.Exception` / `context.IsError` on the shared pipeline +context, but the pipeline continues executing subsequent steps on `default` data. The +real-world failure (Ringba/Ringba-v2 PR #5597): an Aerospike deserialization error +inside a composed `getAccountByIdCommand` was swallowed, the next step received +`account == null`, and the pipeline reported a misleading, data-destroying "Account +not found." validation outcome. The original exception sat unread on the context. + +An external proposal (`pipeline-error.md`) diagnosed this as `ExecuteAsync`-based +composition swallowing exceptions on an isolated inner `CommandResult.Context`, and +proposed rewriting all `CommandStatementBuilder` overloads to call `ExecuteAsync` and +re-throw via `ExceptionDispatchInfo`. Investigation showed that diagnosis is false for +this codebase: + +- `CommandStatementBuilder` composes commands via `command.PipelineFunction` bound + into the parent pipeline with the shared context. It never calls `ExecuteAsync`. + There is no isolated inner context. +- `docs/site/command-pattern.md` explicitly documents shared-context composition as + the intended design ("context flows through naturally ... shared state, middleware, + exception handling, and cancellation are all preserved"). +- Existing passing tests (`CommandStatementBuilderTests`) pin shared-context behavior; + the proposed `ExecuteAsync` rewrite would break them, contradicting its "purely + additive" claim. + +Two forces were in tension: + +1. Boundary model. The outermost pipeline returns a result and does not throw - + directly analogous to Roslyn's `CSharpCompilation.Emit()` returning an `EmitResult` + with `Diagnostics` rather than throwing on a compile error. `CommandResult + { Context }` with `Success` / `IsError` / `Exception` is the diagnostics analog. + This boundary model is considered correct and must be preserved. + +2. Internal progression. Roslyn does not run the emit phase on garbage after binding + fails; it halts phase progression and surfaces diagnostics at the boundary. This + pipeline halts internal progression on cancellation (`Binder.ProcessPipelineAsync` + then `HandleCancellationRequested`, with every binder bailing via + `if (canceled) return default;`) but has no equivalent for error. + +The validation subsystem already establishes the house pattern for halting on a +logical failure: `context.CancelAfter()` plus diagnostic-as-data in `context.Items` +plus result at the boundary (`PipelineValidationExtensions`, +`ValidationAction.CancelAfter`). Exceptions caught by `Build()` are the only failure +class that does not enter this established halt path. + +Alternatives considered: + +- Original proposal (ExecuteAsync rewrite + rethrow). Rejected: fights the correct + boundary model, breaks documented shared-context design and existing tests, and + patches one builder for a defect that lives in the shared binder/build layer. +- Strategy A - parallel `errored` flag through `ProcessPipelineAsync`. Add an error + short-circuit alongside the cancellation one, touching the base method plus ~8 + binder call sites. Rejected: non-minimal, adds a second halt mechanism, no added + correctness over Strategy B. +- Strategy B - reuse the existing cancel-after halt. Chosen (see Decision). + +## Decision + +We will make pipelines halt step progression on error by default, reusing the existing +cancellation short-circuit rather than introducing a new mechanism. + +- In `PipelineBuilder.Build()` and `BuildAsProcedure()` catch blocks, when the + halt-on-error policy is enabled, call `context.CancelAfter()` in addition to setting + `context.Exception = ex`. The existing `HandleCancellationRequested` short-circuit + in `Binder.ProcessPipelineAsync` then halts all downstream steps. No binder changes. + All 8 step/block binders + (Pipe / Call / PipeIf / CallIf / ForEach / Reduce / WaitAll / CallBlock) already + bail via the existing `if (canceled)` pattern off the single + `Binder.ProcessPipelineAsync` chokepoint. `WrapBinder` / `HookBinder` are middleware + decorators and are out of scope. `WaitAll` forks child contexts via `Clone(false)`; + parallel join semantics remain owned by its reducer (a known, documented boundary). + +- The outermost boundary model is unchanged: `Build()` still returns a result and does + not throw; `context.Exception` / `IsError` carry the diagnostic. This preserves the + Roslyn-style result-plus-diagnostics contract. + +- The behavior is governed by a halt-on-error policy carried on the context as a peer + of `Throws`, seeded by `PipelineContextFactory` from a new public `PipelineOptions` + object configured via an `Action configure` delegate on + `AddPipeline(...)` at DI wire-up. `PipelineOptions` is the single extensible seam for + future pipeline-wide policy (chosen over a bare `bool` parameter to avoid + option-parameter accretion). The default is halt-on-error = true (greenfield, IJW, + Roslyn-correct); omitting the `configure` delegate keeps it on. A manually + constructed `new PipelineContext()` also defaults to true. No per-call override is + introduced in v1 (the signature can be extended later without a breaking change if a + real need emerges). + +Backward-compatibility contract: After this change, the new halt-on-error behavior is +the default. An existing user can restore the prior run-through behavior by explicitly +configuring the halt-on-error option to false at pipeline DI wire-up. No code changes +are required to adopt the new default. + +## Consequences + +Easier: + +- `.PipeAsync(command)` and every other step behave like an ordinary `await` from the + caller's perspective: an error stops the pipeline instead of silently corrupting + downstream data. The Ringba `context.ThrowIfError()` workaround becomes unnecessary. +- One halt mechanism for all logical failures (validation, cancellation, error), + consistent with the existing validation pattern. Minimal change surface; zero binder + edits; low regression risk. + +Harder / tradeoffs accepted: + +- An errored pipeline now also reports `IsCanceled == true` (halt uses `CancelAfter`). + This is consistent with existing behavior - validation failures already set + `IsCanceled` today. `Success == !IsError && !IsCanceled` is unaffected, and + `IsError` continues to distinguish exception failures. This overlap must be + documented explicitly. +- This is a behavioral change for any existing pipeline that intentionally relies on + running steps after an error was stashed (for example inline compensation that + inspects or clears `context.Exception`). Such users must set the opt-out option at + wire-up. The change is gated precisely so this remains a one-line, code-free + migration. +- Parallel `WaitAll` branches run on forked (`Clone(false)`) contexts; halt-on-error + applies per fork, and join/aggregation semantics remain the reducer's + responsibility. This boundary is unchanged but must be documented. + +Follow-on / constrained: + +- Supersedes the `pipeline-error.md` proposal entirely; a corrected proposal documents + the implementation (`docs/proposals/0001-halt-on-error.md`). +- A future per-call or per-builder override (for example opt a single composed command + out of halt-on-error) can be added later as a non-breaking signature extension if + demand is demonstrated. Deliberately out of scope now. +- `docs/site/command-pattern.md` must be revised: the "exception handling ... + preserved" line now means "errors halt the pipeline and surface at the boundary," + and a Halt-on-Error / boundary-model section must be added. diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md new file mode 100644 index 0000000..cf34023 --- /dev/null +++ b/docs/decisions/INDEX.md @@ -0,0 +1,5 @@ +# Architecture Decision Records + +| ADR | Title | Status | Date | +| --- | ----- | ------ | ---- | +| [0001](0001-pipelines-halt-on-error-by-default.md) | Pipelines Halt Step Progression on Error by Default | Accepted | 2026-05-19 | diff --git a/docs/proposals/0001-halt-on-error.md b/docs/proposals/0001-halt-on-error.md new file mode 100644 index 0000000..fb32004 --- /dev/null +++ b/docs/proposals/0001-halt-on-error.md @@ -0,0 +1,203 @@ +# Proposal 0001: Halt-on-Error for Pipelines + +**Status:** Implemented (branch `feat/halt-on-error`) +**Date:** 2026-05-19 +**Decision:** ADR-0001 (docs/decisions/0001-pipelines-halt-on-error-by-default.md) +**Supersedes:** `pipeline-error.md` (desktop draft) - rejected; see "Why the original +draft was wrong" below. + +## Goal + +When any pipeline step fails with an exception, stop running subsequent steps and +surface the failure at the pipeline boundary as a result (not a throw). Make this the +default; allow existing users to restore the old run-through behavior with one +wire-up option. + +## Why the original draft was wrong + +The desktop draft (`pipeline-error.md`) is rejected. Its premise does not match this +codebase: + +- It claims `CommandStatementBuilder` invokes `command.ExecuteAsync(...)` and swallows + exceptions on an isolated inner `CommandResult.Context`. False. Composition binds + `command.PipelineFunction` into the parent with the shared context; `ExecuteAsync` + is never called. The draft's own footnote ("the exception lands on the outer context + too, via the shared parent") contradicts its own premise. +- Its "purely additive" claim is false: switching composition to `ExecuteAsync` would + break documented shared-context behavior (`docs/site/command-pattern.md`) and + existing passing tests (`CommandStatementBuilderTests`). +- It patches one builder for a defect that is library-wide and lives in the + build/binder layer. +- Its sample code does not compile against the real binder structure, and its test + fixtures use the wrong `CommandFunction` constructor signature. + +## Correct diagnosis + +The outermost pipeline correctly returns a result and does not throw - analogous to +Roslyn `CSharpCompilation.Emit()` returning `EmitResult` + `Diagnostics`. That +boundary model is correct and is NOT changing. + +The actual defect: internal step progression halts on cancellation but not on error. + +- `Binder.ProcessPipelineAsync` (`src/Hyperbee.Pipeline/Binders/Abstractions/Binder.cs`) + is the single chokepoint every step/block binder funnels through. It calls + `HandleCancellationRequested`; every binder bails via `if (canceled) return default;`. +- The validation subsystem already establishes the house pattern for halting on a + logical failure: `context.CancelAfter()` + diagnostic-as-data + result at boundary + (`PipelineValidationExtensions`, `ValidationAction.CancelAfter`). +- An exception caught by `PipelineBuilder.Build()` sets `context.Exception` but never + enters that halt path. That single gap is the bug. + +## Design (Strategy B - reuse the existing cancel-after halt) + +When `Build()` / `BuildAsProcedure()` catches an exception and the halt-on-error policy +is enabled, also call `context.CancelAfter()`. The existing cancellation short-circuit +then halts every downstream step with zero binder changes. Boundary behavior is +unchanged: a result is returned, `context.Exception` / `IsError` carry the diagnostic. + +### File changes (all in `src/Hyperbee.Pipeline/` unless noted) + +1. `Context/IPipelineContext.cs` - add `bool HaltOnError { get; }` (peer of `Throws`). +2. `Context/PipelineContext.cs` - add `HaltOnError` as an `init` property defaulting to + `true` (same pattern as the existing `Logger` / `ServiceProvider` init properties, + so the factory sets it via object initializer and a manual `new PipelineContext()` + gets the new default). The `(source, throws)` clone constructor copies + `HaltOnError = source.HaltOnError` so forked `WaitAll` branches inherit it. +3. `PipelineBuilder.cs` - in both `Build()` and `BuildAsProcedure()` catch blocks: + + ``` + catch ( Exception ex ) + { + context.Exception = ex; + + if ( context.HaltOnError && !context.IsCanceled ) + context.CancelAfter(); // enters the existing halt path + + if ( context.Throws ) + throw; + } + ``` + + `OperationCanceledException` already arrives with the token canceled, so the + `!context.IsCanceled` guard avoids a redundant `CancelAfter` and keeps cancellation + semantics intact. +4. `Context/PipelineOptions.cs` (new) - public options object, the single extensible + seam for pipeline-wide policy: + + ``` + public sealed class PipelineOptions + { + public bool HaltOnError { get; set; } = true; + } + ``` + +5. `Context/IPipelineContextFactory.cs` / `Context/PipelineContextFactory.cs` - the + factory carries a resolved `PipelineOptions` (default instance when none supplied) + and stamps `HaltOnError = options.HaltOnError` onto every `PipelineContext` it + creates via the object initializer. `CreateFactory(...)` gains a `PipelineOptions` + parameter; the existing single-instance behavior is preserved. +6. `Extensions/ServiceCollectionExtensions.cs` - add an + `Action configure = null` to both existing `AddPipeline` overloads + (it composes with `includeAllServices` and the `implementationFactory` overload). + The delegate mutates a default `PipelineOptions` (so omitting it keeps + halt-on-error on); the result is passed to `CreateFactory`. Wire-up shapes: + + ``` + // greenfield - halt-on-error is the default, nothing to configure + services.AddPipeline(); + + // legacy opt-out - one explicit setting, no code changes elsewhere + services.AddPipeline( o => o.HaltOnError = false ); + + // composes with the factory overload + services.AddPipeline( + ( factorySvcs, root ) => { /* ... */ }, + o => o.HaltOnError = false ); + ``` +7. `docs/site/command-pattern.md` - rewrite the "exception handling ... preserved" + sentence; add a "Halt-on-Error and the Boundary Model" section explaining + result-not-throw at the boundary, halt-between-steps internally, and the + `AddPipeline( o => o.HaltOnError = false )` opt-out. +8. Changelog - note the new default and the one-line opt-out. + +### Backward compatibility + +After this change the new halt-on-error behavior is the default. An existing user +restores the prior run-through behavior by explicitly setting the halt-on-error option +to `false` at pipeline DI wire-up. No code changes are required to adopt the new +default. A manually constructed `new PipelineContext()` also defaults to halt-on-error. + +## Tests + +New tests, reusing `CommandStatementBuilderTests` conventions (MSTest, AAA, NSubstitute +for `IPipelineContextFactory` / `ILogger`). Target ~8-10 tests, not 32. + +1. `step_throws_should_halt_pipeline_and_skip_subsequent_steps` - a `.Pipe` after a + throwing step does not run; result is `default`; `context.IsError` true; + `context.Exception` is the thrown instance. +2. `composed_command_throws_should_halt_outer_pipeline` - the Ringba shape: a composed + command throws, the follow-up validation step never runs, no misleading "not found". +3. `halt_on_error_false_preserves_legacy_run_through` - with the opt-out, subsequent + steps still run, result mirrors current behavior; `context.IsError` true. +4. `boundary_does_not_throw_on_error` - outermost `Build()` returns a result, does not + throw, when `Throws` is false (default). +5. `boundary_throws_when_Throws_true` - existing `Throws` semantics unchanged. +6. `errored_pipeline_reports_IsError_and_IsCanceled` - pins the documented overlap; + `Success` is false; `IsError` distinguishes from a plain cancellation. +7. `OperationCanceledException_is_not_double_canceled` - cancellation path unchanged; + no redundant `CancelAfter`; `CancellationValue` behavior intact. +8. `procedure_pipeline_halts_on_error` - same via `BuildAsProcedure()`. +9. `WaitAll_branch_error_is_isolated_to_fork` - a throwing parallel branch halts its + own fork; the reducer still receives per-branch results; pins the documented + parallel boundary. +10. `manual_PipelineContext_defaults_to_halt_on_error` - non-DI construction gets the + new default. + +`dotnet test` green; `dotnet build -warnaserror` clean. Existing +`CommandStatementBuilderTests` (shared-context) must remain green unchanged. + +## API surface note + +`IPipelineContext` gains a `bool HaltOnError { get; }` member. `PipelineContext` is the +only concrete implementer in the repo, so this is source-compatible internally. It is, +however, a breaking change for any external code that implements `IPipelineContext` +directly (they must add the member). This is acceptable for a minor/feature release of +a library that controls its own versioning; it is called out here so the release notes +can flag it. `CreateFactory` gained an optional trailing `PipelineOptions` parameter +(appended last) so all existing positional call sites compile unchanged. + +## Out of scope (deliberately) + +- Per-call / per-builder halt override. Omitted in v1 to avoid additive bias; the + context/option signature can be extended later without a breaking change. +- Changing the outermost boundary model (still returns a result, still does not throw). +- Auto-merging validation results between composed and parent contexts (unrelated; + composition already shares the context). +- Any `CommandStatementBuilder` / `ExecuteAsync` signature change. None is needed - + the fix is in the build/context layer. + +## Acceptance criteria + +- [x] `IPipelineContext.HaltOnError` exists; `PipelineContext` defaults it to `true` + (init property) and propagates it through the clone constructor. +- [x] `Build()` and `BuildAsProcedure()` call `context.CancelAfter()` on caught + exception when `HaltOnError` and not already canceled; `Throws` behavior + unchanged. +- [x] No binder changes; all existing binder/cancellation tests remain green. +- [x] New public `PipelineOptions { HaltOnError = true }`; both `AddPipeline` + overloads accept `Action configure = null` and compose with the + existing `includeAllServices` / `implementationFactory` overloads. +- [x] `AddPipeline( o => o.HaltOnError = false )` reproduces the prior run-through + behavior with no other code change; omitting `configure` keeps halt-on-error on. +- [x] Existing `CommandStatementBuilderTests` pass unchanged. +- [x] New test matrix implemented and green (9 tests in `HaltOnErrorTests.cs`; + consolidated the plain-step case into composed-command coverage since a single + `Build()` already short-circuits via normal exception propagation - the + composed-command path is the actual defect surface). +- [x] `docs/site/command-pattern.md` updated (ASCII only); `dependency-injection.md` + gains a Pipeline Options section. No `CHANGELOG` file exists in the repo + (versioning via nbgv / GitHub Releases) - release note deferred to the release + process; see handoff note. +- [x] `dotnet build -warnaserror` clean (full solution, net10.0). +- [x] All 244 tests green across all 6 test projects (net10.0); existing + `CommandStatementBuilderTests` unchanged and passing. diff --git a/docs/site/command-pattern.md b/docs/site/command-pattern.md index 34cef1d..fd8f0c2 100644 --- a/docs/site/command-pattern.md +++ b/docs/site/command-pattern.md @@ -176,8 +176,9 @@ command that uses the provider gets consistent middleware without any extra boil Commands expose their inner pipeline delegate via the `PipelineFunction` property. This allows one command's pipeline to directly compose another command's pipeline as a step, without calling `ExecuteAsync`. The key -benefit is that the pipeline context flows through naturally -- shared state, middleware, exception handling, -and cancellation are all preserved. +benefit is that the pipeline context flows through naturally -- shared state, middleware, and cancellation are +preserved, and an error in a composed command halts the outer pipeline (see +[Halt-on-Error and the Boundary Model](#halt-on-error-and-the-boundary-model)). ### PipeAsync with Commands @@ -240,6 +241,39 @@ Conditionally compose a command's pipeline based on a runtime condition. .CallIf( ( ctx, arg ) => arg.StartsWith( "log:" ), _logCommand ) ``` +### Halt-on-Error and the Boundary Model + +A pipeline has two distinct boundaries, and they behave differently on error. + +The outermost boundary returns a result and does not throw. Running a built pipeline +(or a command's `ExecuteAsync`) captures any failure on the context rather than +propagating it as an exception -- `CommandResult.Context` exposes `Success`, +`IsError`, and `Exception` as diagnostics. This is the same model as a compiler that +returns a result with diagnostics rather than throwing on a compile error. (Set +`context.Throws` to opt a specific run into rethrowing instead.) + +Between steps, progression halts on error. When any step -- including a composed +command -- fails with an exception, the pipeline stops: subsequent steps do not run, +and the failure surfaces at the boundary as `IsError` / `Exception`. This prevents a +swallowed error from feeding `default` data into later steps. Because the halt reuses +the cancellation short-circuit, an errored pipeline also reports `IsCanceled == true`; +`IsError` distinguishes an error from a plain cancellation. + +This halt-on-error behavior is the default. To restore the prior behavior, where +steps continue running after an error is recorded, configure it once at DI wire-up: + +```csharp +services.AddPipeline( options => options.HaltOnError = false ); +``` + +The option composes with the other `AddPipeline` overloads: + +```csharp +services.AddPipeline( + ( factoryServices, rootProvider ) => { /* register factory services */ }, + options => options.HaltOnError = false ); +``` + ### Implicit Conversion `CommandFunction` and `CommandProcedure` define implicit conversion operators to their respective delegate diff --git a/docs/site/dependency-injection.md b/docs/site/dependency-injection.md index 3eaf1e8..2f1e31c 100644 --- a/docs/site/dependency-injection.md +++ b/docs/site/dependency-injection.md @@ -46,6 +46,21 @@ services.AddPipeline( (factoryServices, rootProvider) => } ); ``` +### Pipeline Options + +Pipeline-wide behavior is configured with a `configure` delegate accepted by every +`AddPipeline` overload. Currently this exposes `HaltOnError` (default `true`), which +controls whether a step error halts the pipeline. See +[Halt-on-Error and the Boundary Model](command-pattern.md#halt-on-error-and-the-boundary-model). + +```csharp +// restore the prior run-through-on-error behavior +services.AddPipeline( options => options.HaltOnError = false ); + +// composes with the other overloads +services.AddPipeline( includeAllServices: true, configure: options => options.HaltOnError = false ); +``` + ### Example 4 Register Pipeline services manually and provide Pipeline dependencies using a specialized container. diff --git a/src/Hyperbee.Pipeline/Context/IPipelineContext.cs b/src/Hyperbee.Pipeline/Context/IPipelineContext.cs index 416cf7c..5d46e5c 100644 --- a/src/Hyperbee.Pipeline/Context/IPipelineContext.cs +++ b/src/Hyperbee.Pipeline/Context/IPipelineContext.cs @@ -11,6 +11,7 @@ public interface IPipelineContext Exception Exception { get; set; } bool Throws { get; } + bool HaltOnError { get; } bool Success { get; } bool IsError { get; } diff --git a/src/Hyperbee.Pipeline/Context/PipelineContext.cs b/src/Hyperbee.Pipeline/Context/PipelineContext.cs index 4dda5eb..95ae5b0 100644 --- a/src/Hyperbee.Pipeline/Context/PipelineContext.cs +++ b/src/Hyperbee.Pipeline/Context/PipelineContext.cs @@ -30,6 +30,7 @@ protected PipelineContext( PipelineContext source, bool throws ) Logger = source.Logger; ServiceProvider = source.ServiceProvider; Throws = throws; + HaltOnError = source.HaltOnError; } private object _cancellationValue; @@ -70,6 +71,7 @@ private set public Exception Exception { get; set; } public bool Throws { get; } + public bool HaltOnError { get; init; } = true; public bool Success => !IsError && !IsCanceled; public bool IsError => Exception != null; diff --git a/src/Hyperbee.Pipeline/Context/PipelineContextFactory.cs b/src/Hyperbee.Pipeline/Context/PipelineContextFactory.cs index fa58bbe..44a23c5 100644 --- a/src/Hyperbee.Pipeline/Context/PipelineContextFactory.cs +++ b/src/Hyperbee.Pipeline/Context/PipelineContextFactory.cs @@ -5,14 +5,16 @@ namespace Hyperbee.Pipeline.Context; public class PipelineContextFactory : IPipelineContextFactory { private readonly IServiceProvider _serviceProvider; + private readonly PipelineOptions _options; - private PipelineContextFactory( IServiceProvider serviceProvider ) + private PipelineContextFactory( IServiceProvider serviceProvider, PipelineOptions options ) { // private instantiation guarantees a single instance. // this is important so that both DI and manual (non-DI) usage // use the same instance. _serviceProvider = serviceProvider; + _options = options ?? new PipelineOptions(); } public IPipelineContext Create( ILogger logger, CancellationToken cancellation = default ) @@ -20,19 +22,20 @@ private PipelineContextFactory( IServiceProvider serviceProvider ) return new PipelineContext( cancellation ) { Logger = logger, - ServiceProvider = _serviceProvider + ServiceProvider = _serviceProvider, + HaltOnError = _options.HaltOnError }; } public static IPipelineContextFactory Instance { get; private set; } - public static IPipelineContextFactory CreateFactory( IServiceProvider serviceProvider = null, bool resetFactory = false ) + public static IPipelineContextFactory CreateFactory( IServiceProvider serviceProvider = null, bool resetFactory = false, PipelineOptions options = null ) { if ( resetFactory ) { - return Instance = new PipelineContextFactory( serviceProvider ); + return Instance = new PipelineContextFactory( serviceProvider, options ); } - return Instance ??= new PipelineContextFactory( serviceProvider ); + return Instance ??= new PipelineContextFactory( serviceProvider, options ); } } diff --git a/src/Hyperbee.Pipeline/Context/PipelineOptions.cs b/src/Hyperbee.Pipeline/Context/PipelineOptions.cs new file mode 100644 index 0000000..d60ed03 --- /dev/null +++ b/src/Hyperbee.Pipeline/Context/PipelineOptions.cs @@ -0,0 +1,11 @@ +namespace Hyperbee.Pipeline.Context; + +public sealed class PipelineOptions +{ + // When true (default), a step that fails with an exception halts pipeline + // progression: the exception is captured on the context and the pipeline + // short-circuits to the boundary instead of running subsequent steps on + // default data. Set false to restore the prior run-through behavior. + + public bool HaltOnError { get; set; } = true; +} diff --git a/src/Hyperbee.Pipeline/Extensions/ServiceCollectionExtensions.cs b/src/Hyperbee.Pipeline/Extensions/ServiceCollectionExtensions.cs index b1252a4..3f38b2b 100644 --- a/src/Hyperbee.Pipeline/Extensions/ServiceCollectionExtensions.cs +++ b/src/Hyperbee.Pipeline/Extensions/ServiceCollectionExtensions.cs @@ -6,16 +6,20 @@ namespace Hyperbee.Pipeline; public static class ServiceCollectionExtensions { - public static IServiceCollection AddPipeline( this IServiceCollection services, bool includeAllServices = false ) + public static IServiceCollection AddPipeline( this IServiceCollection services, bool includeAllServices = false, Action configure = null ) { return services.AddSingleton( serviceProvider => { var factoryServices = includeAllServices ? serviceProvider : null; // use application wide service provider, or none - return PipelineContextFactory.CreateFactory( factoryServices ); + + var options = new PipelineOptions(); + configure?.Invoke( options ); + + return PipelineContextFactory.CreateFactory( factoryServices, options: options ); } ); } - public static IServiceCollection AddPipeline( this IServiceCollection services, Action implementationFactory ) + public static IServiceCollection AddPipeline( this IServiceCollection services, Action implementationFactory, Action configure = null ) { ArgumentNullException.ThrowIfNull( implementationFactory ); @@ -24,7 +28,10 @@ public static IServiceCollection AddPipeline( this IServiceCollection services, var factoryServices = new ServiceCollection(); // use a specialized factory services container implementationFactory( factoryServices, serviceProvider ); - return PipelineContextFactory.CreateFactory( factoryServices.BuildServiceProvider() ); + var options = new PipelineOptions(); + configure?.Invoke( options ); + + return PipelineContextFactory.CreateFactory( factoryServices.BuildServiceProvider(), options: options ); } ); } diff --git a/src/Hyperbee.Pipeline/PipelineBuilder.cs b/src/Hyperbee.Pipeline/PipelineBuilder.cs index 60a6817..fdab9ad 100644 --- a/src/Hyperbee.Pipeline/PipelineBuilder.cs +++ b/src/Hyperbee.Pipeline/PipelineBuilder.cs @@ -29,6 +29,12 @@ public FunctionAsync Build() { context.Exception = ex; + // Halt step progression on error by reusing the cancellation + // short-circuit. The boundary still returns a result (no throw + // unless context.Throws); IsError/Exception carry the diagnostic. + if ( context.HaltOnError && !context.IsCanceled ) + context.CancelAfter(); + if ( context.Throws ) throw; } @@ -50,6 +56,12 @@ public ProcedureAsync BuildAsProcedure() { context.Exception = ex; + // Halt step progression on error by reusing the cancellation + // short-circuit. The boundary still returns a result (no throw + // unless context.Throws); IsError/Exception carry the diagnostic. + if ( context.HaltOnError && !context.IsCanceled ) + context.CancelAfter(); + if ( context.Throws ) throw; } diff --git a/test/Hyperbee.Pipeline.Tests/HaltOnErrorTests.cs b/test/Hyperbee.Pipeline.Tests/HaltOnErrorTests.cs new file mode 100644 index 0000000..d9e5944 --- /dev/null +++ b/test/Hyperbee.Pipeline.Tests/HaltOnErrorTests.cs @@ -0,0 +1,529 @@ +using System.Threading.Tasks; +using Hyperbee.Pipeline.Commands; +using Hyperbee.Pipeline.Context; +using Microsoft.Extensions.Logging; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using NSubstitute; + +namespace Hyperbee.Pipeline.Tests; + +[TestClass] +public class HaltOnErrorTests +{ + // A composed command whose inner pipeline throws. Composition binds the + // command's PipelineFunction onto the shared parent context, so the inner + // Build() is what captures the exception. + + private sealed class ThrowingFunctionCommand : CommandFunction, ICommandFunction + { + public ThrowingFunctionCommand() + : base( Substitute.For(), Substitute.For() ) + { + } + + protected override FunctionAsync CreatePipeline() + { + return PipelineFactory + .Start() + .Pipe( ThrowString ) + .Build(); + } + + private static string ThrowString( IPipelineContext context, string argument ) + => throw new InvalidOperationException( "boom" ); + } + + private sealed class ThrowingProcedureCommand : CommandProcedure, ICommandProcedure + { + public ThrowingProcedureCommand() + : base( Substitute.For(), Substitute.For() ) + { + } + + protected override ProcedureAsync CreatePipeline() + { + return PipelineFactory + .Start() + .Call( ThrowVoid ) + .BuildAsProcedure(); + } + + private static void ThrowVoid( IPipelineContext context, string argument ) + => throw new InvalidOperationException( "boom" ); + } + + [TestMethod] + public async Task Composed_command_error_should_halt_outer_pipeline_by_default() + { + // Arrange + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + .Pipe( ( _, arg ) => + { + followUpRan = true; + return arg; + } ) + .Build(); + + var context = new PipelineContext(); + + // Act + var result = await pipeline( context, "input" ); + + // Assert - the step after the composed command never ran (the Ringba fix) + Assert.IsFalse( followUpRan ); + Assert.IsNull( result ); + Assert.IsTrue( context.IsError ); + Assert.IsInstanceOfType( context.Exception, typeof( InvalidOperationException ) ); + Assert.AreEqual( "boom", context.Exception.Message ); + // halt reuses the cancellation short-circuit + Assert.IsTrue( context.IsCanceled ); + Assert.IsFalse( context.Success ); + } + + [TestMethod] + public async Task Composed_command_error_should_run_through_when_HaltOnError_false() + { + // Arrange - explicit legacy opt-out + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + .Pipe( ( _, arg ) => + { + followUpRan = true; + return arg ?? "ran-with-default"; + } ) + .Build(); + + var context = new PipelineContext { HaltOnError = false }; + + // Act + var result = await pipeline( context, "input" ); + + // Assert - prior behavior preserved: subsequent steps still run + Assert.IsTrue( followUpRan ); + Assert.AreEqual( "ran-with-default", result ); + Assert.IsTrue( context.IsError ); // exception is still recorded + Assert.IsFalse( context.IsCanceled ); // but the pipeline did not halt + } + + [TestMethod] + public async Task Boundary_should_not_throw_on_error_when_Throws_is_false() + { + // Arrange + var pipeline = PipelineFactory + .Start() + .PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + .Build(); + + var context = new PipelineContext(); // Throws defaults false + + // Act - boundary returns a result, does not throw + var result = await pipeline( context, "input" ); + + // Assert + Assert.IsNull( result ); + Assert.IsTrue( context.IsError ); + } + + [TestMethod] + public async Task Boundary_should_throw_on_error_when_Throws_is_true() + { + // Arrange + var pipeline = PipelineFactory + .Start() + .PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + .Build(); + + var context = new PipelineContext().Clone( throws: true ); + + // Act & Assert - existing Throws semantics unchanged + await Assert.ThrowsExactlyAsync( + () => pipeline( context, "input" ) ); + } + + [TestMethod] + public async Task Composed_procedure_error_should_halt_outer_pipeline_by_default() + { + // Arrange + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .CallAsync( (ICommandProcedure) new ThrowingProcedureCommand() ) + .Pipe( ( _, arg ) => + { + followUpRan = true; + return arg; + } ) + .Build(); + + var context = new PipelineContext(); + + // Act + var result = await pipeline( context, "input" ); + + // Assert - follow-up skipped; Call preserves input, surfaced as the + // cancellation value when the halt short-circuits at the boundary + Assert.IsFalse( followUpRan ); + Assert.AreEqual( "input", result ); + Assert.IsTrue( context.IsError ); + Assert.IsTrue( context.IsCanceled ); + } + + [TestMethod] + public void Manual_PipelineContext_should_default_to_HaltOnError() + { + Assert.IsTrue( new PipelineContext().HaltOnError ); + Assert.IsTrue( new PipelineContext().Clone().HaltOnError ); + Assert.IsFalse( new PipelineContext { HaltOnError = false }.Clone().HaltOnError ); + } + + [TestMethod] + public async Task Plain_cancellation_should_remain_distinct_from_error() + { + // Arrange - a cancel (not an error) must not set IsError + var pipeline = PipelineFactory + .Start() + .Pipe( ( _, _ ) => 1 ) + .Pipe( ( ctx, _ ) => + { + ctx.CancelAfter(); + return 2; + } ) + .Pipe( ( _, _ ) => 3 ) + .Build(); + + var context = new PipelineContext(); + + // Act + var result = await pipeline( context, 0 ); + + // Assert - existing cancellation behavior is unaffected by halt-on-error + Assert.AreEqual( 2, result ); + Assert.IsTrue( context.IsCanceled ); + Assert.IsFalse( context.IsError ); + } + + // --- hardening: composition shapes, nesting, conditionals, enumeration, parallel --- + + private sealed class ThrowingIntCommand : CommandFunction, ICommandFunction + { + public ThrowingIntCommand() + : base( Substitute.For(), Substitute.For() ) + { + } + + protected override FunctionAsync CreatePipeline() + => PipelineFactory.Start().Pipe( ThrowInt ).Build(); + + private static int ThrowInt( IPipelineContext context, int argument ) + => throw new InvalidOperationException( "boom" ); + } + + // A command whose pipeline composes another (throwing) command - command of command. + private sealed class NestingCommand : CommandFunction, ICommandFunction + { + public NestingCommand() + : base( Substitute.For(), Substitute.For() ) + { + } + + protected override FunctionAsync CreatePipeline() + => PipelineFactory + .Start() + .PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + .Build(); + } + + private sealed class SideEffectProcedureCommand : CommandProcedure, ICommandProcedure + { + public bool Ran { get; private set; } + + public SideEffectProcedureCommand() + : base( Substitute.For(), Substitute.For() ) + { + } + + protected override ProcedureAsync CreatePipeline() + => PipelineFactory + .Start() + .Call( ( _, _ ) => Ran = true ) + .BuildAsProcedure(); + } + + [TestMethod] + public async Task Nested_composed_command_error_should_halt_through_all_levels() + { + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .PipeAsync( (ICommandFunction) new NestingCommand() ) // composes ThrowingFunctionCommand + .Pipe( ( _, arg ) => { followUpRan = true; return arg; } ) + .Build(); + + var context = new PipelineContext(); + + var result = await pipeline( context, "input" ); + + Assert.IsFalse( followUpRan ); + Assert.IsNull( result ); + Assert.IsTrue( context.IsError ); + Assert.AreEqual( "boom", context.Exception.Message ); + Assert.IsTrue( context.IsCanceled ); + } + + [TestMethod] + public async Task PipeIf_true_with_throwing_command_should_halt() + { + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .PipeIf( ( _, _ ) => true, (ICommandFunction) new ThrowingFunctionCommand() ) + .Pipe( ( _, arg ) => { followUpRan = true; return arg; } ) + .Build(); + + var context = new PipelineContext(); + + await pipeline( context, "input" ); + + Assert.IsFalse( followUpRan ); + Assert.IsTrue( context.IsError ); + Assert.IsTrue( context.IsCanceled ); + } + + [TestMethod] + public async Task PipeIf_false_should_not_error_or_halt() + { + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .PipeIf( ( _, _ ) => false, (ICommandFunction) new ThrowingFunctionCommand() ) + .Pipe( ( _, arg ) => { followUpRan = true; return arg + "-ok"; } ) + .Build(); + + var context = new PipelineContext(); + + var result = await pipeline( context, "input" ); + + Assert.IsTrue( followUpRan ); + Assert.AreEqual( "input-ok", result ); + Assert.IsFalse( context.IsError ); + Assert.IsFalse( context.IsCanceled ); + } + + [TestMethod] + public async Task CallIf_true_with_throwing_command_should_halt() + { + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .CallIf( ( _, _ ) => true, (ICommandProcedure) new ThrowingProcedureCommand() ) + .Pipe( ( _, arg ) => { followUpRan = true; return arg; } ) + .Build(); + + var context = new PipelineContext(); + + await pipeline( context, "input" ); + + Assert.IsFalse( followUpRan ); + Assert.IsTrue( context.IsError ); + Assert.IsTrue( context.IsCanceled ); + } + + [TestMethod] + public async Task PipeAsync_with_selector_throwing_command_should_halt() + { + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .PipeAsync( new ThrowingIntCommand(), selector: ( _, arg ) => arg.Length ) + .Pipe( ( _, arg ) => { followUpRan = true; return arg; } ) + .Build(); + + var context = new PipelineContext(); + + await pipeline( context, "input" ); + + Assert.IsFalse( followUpRan ); + Assert.IsTrue( context.IsError ); + Assert.IsTrue( context.IsCanceled ); + } + + [TestMethod] + public async Task Error_should_prevent_subsequent_side_effect_command() + { + var sideEffect = new SideEffectProcedureCommand(); + + var pipeline = PipelineFactory + .Start() + .PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + .CallAsync( (ICommandProcedure) sideEffect ) + .Build(); + + var context = new PipelineContext(); + + await pipeline( context, "input" ); + + // the side effect (e.g. "send email") must not run after an upstream error + Assert.IsFalse( sideEffect.Ran ); + Assert.IsTrue( context.IsError ); + } + + [TestMethod] + public async Task Value_type_pipeline_error_should_return_default_and_record_error() + { + var pipeline = PipelineFactory + .Start() + .PipeAsync( (ICommandFunction) new ThrowingIntCommand() ) + .Pipe( ( _, n ) => n + 1 ) + .Build(); + + var context = new PipelineContext(); + + var result = await pipeline( context, 41 ); + + Assert.AreEqual( 0, result ); // default(int), not 42 + Assert.IsTrue( context.IsError ); + Assert.IsTrue( context.IsCanceled ); + } + + [TestMethod] + public async Task ForEach_composed_command_error_should_halt_remaining_iterations_and_outer() + { + var processed = 0; + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .Pipe( ( _, arg ) => arg.Split( ' ' ) ) + .ForEach().Type( builder => builder + .Pipe( ( _, e ) => { processed++; return e; } ) + .PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + ) + .Pipe( ( _, arg ) => { followUpRan = true; return arg; } ) + .Build(); + + var context = new PipelineContext(); + + await pipeline( context, "a b c" ); + + // only the first element is processed before the composed command halts + Assert.AreEqual( 1, processed ); + Assert.IsFalse( followUpRan ); + Assert.IsTrue( context.IsError ); + Assert.IsTrue( context.IsCanceled ); + } + + [TestMethod] + public async Task WaitAll_branch_error_should_be_isolated_to_fork_and_not_auto_halt_parent() + { + WaitAllResult[] captured = null; + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .WaitAll( + b => b.Create( + branch => branch.Pipe( ( _, arg ) => arg + "-ok" ), + branch => branch.PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + ), + ( _, input, results ) => { captured = results; return input; } + ) + .Pipe( ( _, arg ) => { followUpRan = true; return arg; } ) + .Build(); + + var context = new PipelineContext(); + + var result = await pipeline( context, "input" ); + + // the failing branch's error is contained on its forked context... + Assert.IsNotNull( captured ); + Assert.HasCount( 2, captured ); + var anyForkErrored = false; + foreach ( var r in captured ) + anyForkErrored |= r.Context.IsError; + Assert.IsTrue( anyForkErrored ); + + // ...and does NOT auto-halt the parent (join semantics are the reducer's job) + Assert.IsFalse( context.IsError ); + Assert.IsFalse( context.IsCanceled ); + Assert.IsTrue( followUpRan ); + Assert.AreEqual( "input", result ); + } + + [TestMethod] + public async Task WaitAll_reducer_can_propagate_branch_error_to_halt_parent() + { + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .WaitAll( + b => b.Create( + branch => branch.Pipe( ( _, arg ) => arg ), + branch => branch.PipeAsync( (ICommandFunction) new ThrowingFunctionCommand() ) + ), + ( ctx, input, results ) => + { + foreach ( var r in results ) + { + if ( r.Context.IsError ) + { + ctx.CancelAfter(); // recommended pattern to propagate a branch failure + break; + } + } + + return input; + } + ) + .Pipe( ( _, arg ) => { followUpRan = true; return arg; } ) + .Build(); + + var context = new PipelineContext(); + + await pipeline( context, "input" ); + + Assert.IsFalse( followUpRan ); + Assert.IsTrue( context.IsCanceled ); + } + + [TestMethod] + public async Task Programmatic_context_Exception_without_throw_should_not_halt() + { + // halt-on-error triggers only on a thrown-and-caught exception, not on a + // programmatic context.Exception assignment (consistent with the validation + // / CancelAfter pattern for deliberate short-circuiting). + var followUpRan = false; + + var pipeline = PipelineFactory + .Start() + .Pipe( ( ctx, arg ) => + { + ctx.Exception = new InvalidOperationException( "annotated, not thrown" ); + return arg; + } ) + .Pipe( ( _, arg ) => { followUpRan = true; return arg + "-ok"; } ) + .Build(); + + var context = new PipelineContext(); + + var result = await pipeline( context, "input" ); + + Assert.IsTrue( followUpRan ); + Assert.AreEqual( "input-ok", result ); + Assert.IsTrue( context.IsError ); // recorded + Assert.IsFalse( context.IsCanceled ); // but not halted + } +}