Document recoverability pipeline

Snippets/Core/Core_8/Audit/AddAuditData.cs

@@ -9,7 +9,7 @@ public class CustomAuditDataBehavior : Behavior<IAuditContext>
     {
         public override Task Invoke(IAuditContext context, Func<Task> next)
         {
-            context.AddAuditData("myKey", "myValue");
+            context.AuditMetadata["myKey"] = "MyValue";
             return next();
         }
     }

Snippets/Core/Core_8/Core_8.csproj

@@ -8,7 +8,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.AspNet.Mvc" Version="5.2.7" />
     <PackageReference Include="Microsoft.ServiceFabric.Services" Version="2.*" />
-    <PackageReference Include="NServiceBus" Version="8.0.0-alpha.1902" />
+    <PackageReference Include="NServiceBus" Version="8.0.0-alpha.3019" />
     <PackageReference Include="NServiceBus.Callbacks" Version="3.*" />
     <PackageReference Include="NServiceBus.Encryption.MessageProperty" Version="2.*" />
     <PackageReference Include="NServiceBus.Metrics.ServiceControl" Version="3.*" />

Snippets/Core/Core_8/Pipeline/CustomForkConnector.cs

@@ -21,7 +21,7 @@ await next()
             var auditAddress = "AuditAddress";
 
             // Fork into new pipeline
-            await fork(this.CreateAuditContext(message, auditAddress, context))
+            await fork(this.CreateAuditContext(message, auditAddress, TimeSpan.FromHours(1), context))
                 .ConfigureAwait(false);
         }
     }

Snippets/Core/Core_8/Recoverability/Pipeline.cs

@@ -0,0 +1,62 @@
+namespace Core8.Recoverability
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Threading.Tasks;
+    using NServiceBus;
+    using NServiceBus.Pipeline;
+
+    #region custom-recoverability-action
+
+    public class EnableExternalBodyStorageBehavior : Behavior<IRecoverabilityContext>
+    {
+        private readonly IExternalBodyStorage storage;
+
+        public EnableExternalBodyStorageBehavior(IExternalBodyStorage storage)
+        {
+            this.storage = storage;
+        }
+
+        public async override Task Invoke(IRecoverabilityContext context, Func<Task> next)
+        {
+            if(context.RecoverabilityAction is MoveToError errorAction)
+            {
+                var message = context.ErrorContext.Message;
+                var bodyUrl = await storage.StoreBody(message.MessageId, message.Body);
+
+                context.Metadata["body-url"] = bodyUrl;
+
+                context.RecoverabilityAction = new SkipFailedMessageBody(errorAction.ErrorQueue);
+            }
+
+            await next();
+        }
+
+        class SkipFailedMessageBody : MoveToError
+        {
+            public SkipFailedMessageBody(string errorQueue) : base(errorQueue)
+            {
+            }
+
+            public override IReadOnlyCollection<IRoutingContext> GetRoutingContexts(IRecoverabilityActionContext context)
+            {
+                var routingContexts = base.GetRoutingContexts(context);
+
+                foreach (var routingContext in routingContexts)
+                {
+                    // clear out the message body
+                    routingContext.Message.UpdateBody(ReadOnlyMemory<byte>.Empty);
+                }
+
+                return routingContexts;
+            }
+        }
+    }
+
+    #endregion
+
+    public interface IExternalBodyStorage
+    {
+        Task<string> StoreBody(string messageId, ReadOnlyMemory<byte> body);
+    }
+}

menu/menu.yaml

@@ -519,6 +519,8 @@
       Title: Custom Policy
     - Url: nservicebus/recoverability/subscribing-to-error-notifications
       Title: Error notifications
+    - Url: nservicebus/recoverability/pipeline
+      Title: Recoverability pipeline
   - Url: nservicebus/pipeline
     Title: Pipeline
     Articles:

nservicebus/pipeline/index.md

@@ -2,29 +2,24 @@
 title: Message Handling Pipeline
 summary: Overview of the message handling pipeline
 reviewed: 2020-05-07
+component: Core
 redirects:
 - nservicebus/nservicebus-pipeline-intro
 related:
 - samples/header-manipulation
 - samples/pipeline/fix-messages-using-behavior
 ---
 
-NServiceBus has the concept of a _pipeline_ which refers to the series of actions taken when an incoming message is processed and an outgoing message is sent.
+partial: intro
 
-## Customizing the pipeline
+## Pipeline customization
 
-There are several ways to customize this pipeline with varying levels of complexity.
+There are several ways to customize the pipelines with varying levels of complexity.
 
 * [Manipulate the pipeline with behaviors](/nservicebus/pipeline/manipulate-with-behaviors.md)
 * [Steps, stages and connectors](/nservicebus/pipeline/steps-stages-connectors.md)
 * [Message mutators](/nservicebus/pipeline/message-mutators.md)
 * [Abort the pipeline](/nservicebus/pipeline/aborting.md)
 * [React to pipeline events](/nservicebus/pipeline/events.md)
 
- Unit testing of custom extensions is supported by [the `NServiceBus.Testing` library](/nservicebus/testing/#testing-a-behavior).
-
-## Features built using the pipeline
-
-* [Data Bus](/nservicebus/messaging/databus/)
-* [Message Property Encryption](/nservicebus/security/property-encryption.md)
-* [Recoverability](/nservicebus/recoverability/)
+Unit testing of custom extensions is supported by [the `NServiceBus.Testing` library](/nservicebus/testing/#testing-a-behavior).

nservicebus/pipeline/index_intro_core_[6,8).partial.md

@@ -0,0 +1 @@
+NServiceBus uses the concept of _pipelines_. A pipeline refers to the series of actions taken as a result of the triggering action, e.g. an incoming message triggers the incoming pipeline, or an outgoing message triggers the outgoing pipeline.

nservicebus/pipeline/index_intro_core_[8,).partial.md

@@ -0,0 +1,7 @@
+NServiceBus uses the concept of _pipelines_. A pipeline refers to the series of actions taken as a result of the triggering action.
+
+For example:
+
+* The incoming pipeline is triggered due to an incoming message
+* The outgoing pipeline is triggered due to an outgoing message
+* The [recoverability](/nservicebus/recoverability/) pipeline is triggered due to failure during processing

nservicebus/pipeline/steps-stages-connectors.md

@@ -7,10 +7,19 @@ related:
 - nservicebus/pipeline/manipulate-with-behaviors
 ---
 
-NServiceBus has the concept of a pipeline execution order that is executed when a message is received or dispatched. A *pipeline* refers to the series of actions taken when an incoming message is processed or an outgoing message is sent. This allows users to take full control of the incoming and outgoing message processing.
+partial: intro
 
-There are two explicit pipelines: one for the outgoing messages and one for the incoming messages.
+partial: stages
 
-Each pipeline is composed of *steps*. A step is an identifiable value in the pipeline used to programmatically define order of execution. Each step represents a behavior which will be executed at the given place within the pipeline. To add additional behavior to the pipeline by registering a new step or replace the behavior of an existing step with the custom implementation.
+partial: incoming
+
+partial: outgoing
+
+partial: recoverability
+
+partial: optional
+
+partial: extensionbag
+
+partial: connectors
 
-partial: extra

nservicebus/pipeline/steps-stages-connectors_connectors_core_[6.0,).partial.md

@@ -0,0 +1,39 @@
+## Stage connectors
+
+Stage connectors connect from the current stage (i.e. `IOutgoingLogicalMessageContext`) to another stage (i.e. `IOutgoingPhysicalMessageContext`). In order to override an existing stage, inherit from `StageConnector<TFromContext, TToContext>` and then replace an existing stage connector. Most pipeline extensions can be done by inheriting from `Behavior<TContext>`. It is rarely necessary to replace existing stage connectors. When implementing a stage connector, ensure that all required data is passed along for the next stage.
+
+snippet: CustomStageConnector
+
+## Fork Connectors
+
+```mermaid
+graph LR
+subgraph Root pipeline
+    A[TFromContext] --- B{Fork Connector}
+    B --- C[TFromContext]
+end
+subgraph Forked pipeline
+   B --> D[TForkContext]
+end
+```
+
+Fork connectors fork from a current stage (i.e. `IIncomingPhysicalMessageContext`) to another independent pipeline (i.e. `IAuditContext`). In order to override an existing fork connector inherit from `ForkConnector<TFromContext, TForkContext>` and then replace an existing fork connector.
+
+snippet: CustomForkConnector
+
+## Stage Fork Connector
+
+```mermaid
+graph LR
+subgraph Root stage
+    A[TFromContext] --- B{StageFork<br/>Connector}
+    B --- C[TToContext]
+end
+subgraph Forked pipeline
+   B --> D[TForkContext]
+end
+```
+
+Stage fork connectors are essentially a combination of a stage connector and a fork connector. They have the ability to connect from the current stage (i.e. `ITransportReceiveContext`) to another stage (i.e. `IIncomingPhysicalMessageContext`) and fork to another independent pipeline (i.e. `IBatchedDispatchContext`). In order to override an existing stage fork connector inherit from `StageForkConnector<TFromContext, TToContext, TForkContext` and then replace an existing stage fork connector.
+
+snippet: CustomStageForkConnector

nservicebus/pipeline/steps-stages-connectors_extensionbag_core_[6.0,).partial.md

@@ -0,0 +1,7 @@
+## Extension bag
+
+Pipeline contexts have an extension bag which can be used to used to create, read, update or delete custom state with a key identifier. For example, this can be used to *set* metadata- or pipeline-specific state in an incoming behavior that can be used in later pipeline stages if needed. State stored via the extension bag will be unavailable once the extension bag runs out of scope at the end of the pipeline.
+
+State set during a *forked* pipeline will not be available to the *forking* pipeline. For example, state changes during the *outgoing* pipeline will not be available in the *incoming* pipeline. If state has to be propagated, have the *forking* pipeline set a context object that the *forked* pipeline later can get and modify as needed.
+
+snippet: SetContextBetweenIncomingAndOutgoing