Crash process on repeated pipeline handle exceptions?

Suppose I start and wait on a pipeline like this:

#r "nuget: Propulsion, 3.0.0"

open Propulsion

async {
  use src : Feed.Core.SourcePipeline =
    Unchecked.defaultof<_> // ...

  do! src.Await()
}
|> Async.RunSynchronously

How do I make this async crash (and therefore the entire process) when the handle function throws an exception?

See tests like https://github.com/jet/propulsion/blob/master/tests/Propulsion.Tests/SourceTests.fs
dotnet-templates repo illustrates how to coordinate a teardown in various samples such as this one (though the tools impls might be equally useful)

Bottom line is that multiple parts of the system may trigger a shutdown:

source hard faulting and/or reaching the end
sink hard faulting
external trigger (single CtrlC for example)

In the case of a hard stop, you want a controlled shutdown of sink and source (including flushing checkpoints)

In the case of reaching end of a source, the sink needs to stay up until it's projected all the inputs

There are also tests here and in dotnet-templates that illustrate how you can await processing (including recursively treiggered follow-on actions) re integration tests

The Monitor coordinates those (TL;DR it takes a start reader checkpoint and waits with a timeout until checkpointing has moved beyond that aggregate position)

The other thing that might be relevant is the health checks stuff which lets you instigate teardown from aggregated status beyond individual exceptions encountered at a single handler invocation level
can't recall how far I got with writing docs but there are tests
https://github.com/jet/propulsion/blob/master/tests/Propulsion.Tests/SinkHealthTests.fs

Note in general the core assumption is that there can be significant transient failure traffic (stemming from the original use case of feeding multiple TB of data from an event store into CosmosStore with the implied pervasive hot partitions being in the mix

There's intentionally no backoffs and/or dead streams support baked in as no useful pattern has emerged to date, so there's placeholders such as #129

If you have some more context as to the specifics of your consumption scenario I'm happy to answer here (even though in a perfect world I'd write up a doc section exploring it)

Please feel free to @ me for any question like this in future

Thanks for the links, I will dig into them 👍

Let me expand on the scenario above...

In this script we push an event to a VolatileStore every second.
The IDs increase in sequence.
Any event with ID > 3 is hard-coded to fail, simulating a non-transient error.
I would like these to crash the process so that the orchestrator sees the failure.
However, src.Await() will keep retrying, presumably forever.

#r "nuget: Serilog, 4.3.1"
#r "nuget: Serilog.Sinks.Console, 4.1.0"
#r "nuget: FsCodec.SystemTextJson, 3.1.0"
#r "nuget: Propulsion, 3.0.0"
#r "nuget: Propulsion.MemoryStore, 3.0.0"

open System
open Serilog
open Equinox.MemoryStore
open Propulsion
open FsCodec
open FsCodec.SystemTextJson

type Fresh =
  {
    ID : int
  }

type Event =
  | Fresh of Fresh
  interface TypeShape.UnionContract.IUnionContract

let codec =
  Codec.Create<Event>()
  |> FsCodec.Encoder.Uncompressed

let handle (log : ILogger) (stream : StreamName) events =
  let r = Random()
  async {
    for event in events do
      log.Information($"Stream: %A{stream}, Event: %A{event}")

      match codec.Decode event with
      | ValueSome (Fresh { ID = id }) ->
        // Simulate a failure
        if id > 3 then
          failwith $"Failed at %i{id}"
      | ValueNone ->
        log.Error $"Failed to decode event %A{event}"

    return (), Sinks.Events.next events
  }

async {
  let log = LoggerConfiguration().WriteTo.Console().CreateLogger()

  let stats =
    {
      new Streams.Stats<_>(log, TimeSpan.FromMinutes 1., TimeSpan.FromMinutes 1.) with
        member _.HandleOk _x = ()
        member _.HandleExn(log, exn) =
          log.Error(exn, exn.Message)
    }

  let store = VolatileStore()

  let cat = MemoryStoreCategory(store, "demo", codec, (fun _ events -> events), [||])
  let decider = Equinox.Decider.forStream log cat (StreamId.create "a")

  // Push an event every second
  let! pushEvents =
    async {
      for i = 0 to 1000 do
        let! _ =
          decider.Transact
            (fun _ ->
              async {
                let event = Fresh { ID = i }
                return (), [| event |]
              })

        do! Async.Sleep(TimeSpan.FromSeconds 1.)
    }
    |> Async.StartChild

  use pipeline = Propulsion.Sinks.Factory.StartConcurrent(log, 10, 1, handle log, stats)
  use src = MemoryStore.MemoryStoreSource(log, store, (fun _ -> true), pipeline).Start()

  do! src.Await() // Can we configure a crash here, eventually?
  do! pushEvents
}
|> Async.RunSynchronously

yes src.Await is waiting for the source to drain, checkpoint and stop

In order to get to to that point, the sink needs to declare the event in the stream to have been processed (and/or monitor/sink side detects problem in sink and pulls down the stack instead of waiting for that ending)

If there's a non-transient error, the default behavior (as there is no backoff pending the above linked PR and/or a similar feature) is to retry loop

Then, as the wall clock time since first fail climbs and no successes happen on that stream, it becomes Stuck state. The Stuck state combined with the retry count can be used to then have the handler and/or an external inspector take one of two high level actions:

shout stop (use CT to pull down the stack, call Stop on the Sink etc)
take a note of the stuck stream externally (add to an external deadlettered streams list), and return success from the handler so it gets taken off the stuck list

The monitor has default durations before a stuck stream triggers the teardown - if you set that time to zero it will probably do what you are asking (i.e. the exn will make it failing, the time will make the continual fail be stuck, and the max stuck will trigger teardown

However as alluded to above, the chances are that to solve your whole problem, you might want to have a phase of retries even on all-but nontrasient exceptions bubble up in that way.

(This is off the top of my head - I havent looked in the code or discussed the scheme with anyone for a while so if you throw me more context I may be able to give a better answer - especially if you have further concrete details as to the actual scenario you are attempting to address)

This variant will stop eventually:

#r "nuget: Serilog, 4.3.1"
#r "nuget: Serilog.Sinks.Console, 4.1.0"
#r "nuget: FsCodec.SystemTextJson, 3.1.0"
#r "nuget: Propulsion, 3.0.0"
#r "nuget: Propulsion.MemoryStore, 3.0.0"

open System
open Serilog
open Equinox.MemoryStore
open Propulsion
open FsCodec
open FsCodec.SystemTextJson

type Fresh =
  {
    ID : int
  }

type Event =
  | Fresh of Fresh
  interface TypeShape.UnionContract.IUnionContract

let codec =
  Codec.Create<Event>()
  |> FsCodec.Encoder.Uncompressed

let handle (log : ILogger) (stream : StreamName) events =
  async {
    for event in events do
      log.Information($"Stream: %A{stream}, Event: %A{event}")

      match codec.Decode event with
      | ValueSome (Fresh { ID = id }) ->
        // Simulate a failure
        if id > 3 then
          do! Async.Sleep(TimeSpan.FromMilliseconds 100.)
          failwith $"Failed at %i{id}"
      | ValueNone ->
        log.Error $"Failed to decode event %A{event}"

    return (), Sinks.Events.next events
  }

async {
  let log = LoggerConfiguration().WriteTo.Console().CreateLogger()

  let stats =
    {
      new Streams.Stats<_>(
        log,
        statsInterval = TimeSpan.FromSeconds 1.,
        statesInterval = TimeSpan.FromSeconds 1.,
        abendThreshold = TimeSpan.FromSeconds 5.) with
        member _.HandleOk _x = ()
        member _.HandleExn(log, exn) =
          log.Error(exn, exn.Message)
    }

  let store = VolatileStore()

  let cat = MemoryStoreCategory(store, "demo", codec, (fun _ events -> events), [||])
  let decider = Equinox.Decider.forStream log cat (StreamId.create "a")

  use pipeline = Propulsion.Sinks.Factory.StartConcurrent(log, 10, 1, handle log, stats)
  use src = MemoryStore.MemoryStoreSource(log, store, (fun _ -> true), pipeline).Start()

  // Push an event every second
  let! pushEvents =
    async {
      for i = 0 to 1000 do
        let! _ =
          decider.Transact
            (fun _ ->
              async {
                let event = Fresh { ID = i }
                return (), [| event |]
              })

        do! Async.Sleep(TimeSpan.FromSeconds 1.)
    }
    |> Async.StartChild

  do! pipeline.Await()
  do! pushEvents
}
|> Async.RunSynchronously

The key changes being:

Wait on pipeline, not src
Set abendThreshold on the stats

Are there any more notes on this?

Looks good

Are there any more notes on this?

Depends - are you happy with it, or do you need tighter control?

Stats incorporates the default abend based on time/category threholds logic, but its customizable

(Dont have an IDE to hand to look closer and/or review in full detail but will do so later / in the am)

Sorry, finally looked at it (and ran it) now. I guess I did think it was too good to be true. Sometimes people attempt to build supervision trees a la actor frameworks using Async.StartChild and friends and then rely on a single exception to tear down the stack. I saw how messy that gets IRL in Jet and any sane person would run a mile. Nonetheless my eyeballing didn't scream "that'll never work", as it should...

The problem is that a source won't stop until someone tells it to (via the internal CT, driven by theStop() method)

So, while the Sink stops with your example, the Source continues on because it's got not reason not to (the max read ahead will limit how much data accumulates in the pipeline's [internal] Ingester chain, but it's on the controller to say "no point pumping anymore, nobody's ever going to pull the buffered data")

(I'll post an adjusted/corrected version shortly)

(Sorry getting pulled in lots of directions so pasting for now...)

#r "nuget: Serilog, 4.3.1"
#r "nuget: Serilog.Sinks.Console, 4.1.0"
#r "nuget: FsCodec.SystemTextJson, 3.1.0"
#r "nuget: Propulsion, 3.0.0"
#r "nuget: Propulsion.MemoryStore, 3.0.0"

open System
open Serilog
open Equinox.MemoryStore
open Propulsion
open Propulsion.Sinks
open FsCodec
open FsCodec.SystemTextJson

type Fresh =
  {
    ID : int
  }

type Event =
  | Fresh of Fresh
  interface TypeShape.UnionContract.IUnionContract

let codec =
  Codec.Create<Event>()
  |> FsCodec.Encoder.Uncompressed

let handle (log : ILogger) (stream : StreamName) events =
  async {
    for event in events do
      log.Information($"Stream: %A{stream}, Event: %A{event}")

      match codec.Decode event with
      | ValueSome (Fresh { ID = id }) ->
        // Simulate a failure
        if id > 3 then
          do! Async.Sleep(TimeSpan.FromMilliseconds 100.)
          failwith $"Failed at %i{id}"
      | ValueNone ->
        log.Error $"Failed to decode event %A{event}"

    return (), Sinks.Events.next events
  }

async {
  let log = LoggerConfiguration().WriteTo.Console().CreateLogger()

  let stats =
    {
      new Streams.Stats<_>(
        log,
        statsInterval = TimeSpan.FromSeconds 1.,
        statesInterval = TimeSpan.FromSeconds 1.,
        abendThreshold = TimeSpan.FromSeconds 5.) with
        member _.HandleOk _x = ()
        member _.HandleExn(log, exn) =
          log.Error(exn, exn.Message)
    }

  let store = VolatileStore()

  let cat = MemoryStoreCategory(store, "demo", codec, (fun _ events -> events), [||])
  let decider = Equinox.Decider.forStream log cat (StreamId.create "a")

  use pipeline = Propulsion.Sinks.Factory.StartConcurrent(log, 10, 1, handle log, stats)
  use src = MemoryStore.MemoryStoreSource(log, store, (fun _ -> true), pipeline).Start()

  // Push an event every second
  let pushEvents =
    async {
      for i = 0 to 1000 do
        let! _ =
          decider.Transact
            (fun _ ->
              async {
                let event = Fresh { ID = i }
                return (), [| event |]
              })

        do! Async.Sleep(TimeSpan.FromSeconds 1.)
    }
  // pipeline.
  return!
    Async.Parallel [|
      pushEvents
      src.AwaitWithStopOnCancellation()
      pipeline.AwaitWithStopOnCancellation() |]
}
|> Async.RunSynchronously

That code wires up three tasks, all based off the same cancellationtoken (the Async ambient one), and an exception from the sink triggers cancellation of other parties to the Parallel.
The fun bit is that (99% sure but have to run so cant confirm yet but will later) the MemoryStoreSource does not wire the Ingester it starts here into the Dispose chain for the Source, which is why that continues logging for eternity (it's not doing anything heavy, but obviously it should be stopping). The others all log correct shutdown notifications now (Sink, Submitter, Source etc). There's a small chance of there being clones of this issue, but I doubt it (e.g. CosmosStoreSource maintains a list of Observers, each with their own Ingester, but those are wired in for stopping)

(StateInterval is normally 1m in a real scenario, which in your test is 1s, that's the default for the Ingester log interval internally: ingesterStateInterval = defaultArg ingesterStateInterval stats.StateInterval.Period - that's likely why it never came ot my or anyone else's attention)

This is the test that covers it - see how it does the same parallel start wiring as we have above

Note it has a 10s state log interval, and the log is wired to the xunit Test Output, so if the process stays alive, the Memory Store's Ingester instance is logging into a disposed output sink

(afk for a few hours now; will add test coverage and correct the bug - though I think this should not have a functional effect on anything. But bottom line its good to remove the bug either way - clean shutdown is obviously a core feature on which trust is built!)))

Added a test which might be useful in terms of trimming your sample/repro in 9af8fe6

Though in general you only want to have explicit stops and sequenced shutdowns to prove individual parts are working correctly in a low level test.

The way to think about the high level consumption scenario is very much that you have N loops, and the source, the host and the sink equally get a vote in tearing down the tree. When that happens, the CT's get flipped, the Awaits coordinate controlled shutdown (i.e. flushing checkpoints), and the Async.Parallel propagates the cancellation to the others.

(3.0.2 packages pushed for Propulsion (higher FSharp.Core) and Propulsion.MemoryStore (the fix))

Thanks for this.

I think a helper like this serves my use-case:

let runPipeline (srcPipeline : Pipeline) (sinkPipeline : SinkPipeline) : Async<unit> =
  Async.Parallel [|
    srcPipeline.AwaitWithStopOnCancellation()
    sinkPipeline.AwaitWithStopOnCancellation() |]
  |> Async.Ignore<unit array>

Yeah, you're not wrong; in the general case that's all 95% of apps need. If you can find a place to mention that in the README, I'm feeling any doc PR can't make it worse than the current state 😦

The problem is that invariably there are other things that feed into the mix, i.e ASP.NET Hosting shutdown CTs and/or controlled Ctrl-C shutdown wiring, or more esoteric things.

Unfortunately giving general advice that would probably help >95% of people can be a tough call as it's very easy to oversimplify and have people assuming more complex cases are not catered for.

Yes that's all excuses of course. But another case of the perfect being the enemy of the good is that some sink wiring has not been fully unified yet (it used to be much works in V2 times - each store nearly had its own wiring whereas there's good consistency now across all sources ... except for on the Kafka side as noted in the TODO list here)

Navigation Menu

Crash process on repeated pipeline `handle` exceptions? #276

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Crash process on repeated pipeline handle exceptions? #276

Description

Activity

bartelink commented on Jun 8, 2026

bartelink commented on Jun 8, 2026

njlr commented on Jun 8, 2026

bartelink commented on Jun 8, 2026

njlr commented on Jun 8, 2026

bartelink commented on Jun 8, 2026

bartelink commented on Jun 9, 2026

bartelink commented on Jun 9, 2026

bartelink commented on Jun 9, 2026

bartelink commented on Jun 9, 2026

bartelink commented on Jun 9, 2026

njlr commented on Jun 9, 2026

bartelink commented on Jun 9, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Crash process on repeated pipeline `handle` exceptions? #276