FCQRSFCQRS

Step 5: Restart during shipping

The steps so far ran in one process from start to end. This step stops the process in the middle of an order and starts it again, to show what FCQRS recovers and what it runs a second time. The workflow does not change; only the program does.

This page changes the program from step 4.

Stop the process inside a step

In Program.fs (Program.cs), replace the stand-in carrier. A call now takes two seconds, and with crash the carrier stops the whole process after it ships, before it returns:

Shared setup
module Order =
    open System
    open System.Threading.Tasks
    open FCQRS.Common
    open FCQRS.Workflows

    // What an order workflow is started with. FCQRS stores it in the journal.
    type OrderRequest = {
        Customer: string
        Item: string
        Quantity: int
        UnitPrice: decimal
    }

    // What the workflow returns. FCQRS stores it too.
    type OrderResult = {
        Status: string
        Reservation: string
        Shipment: string
        Confirmation: Guid
        PlacedAt: DateTime
    }

    // The warehouse is an external system. The key is the same for every
    // attempt of one step, so the warehouse can recognise a repeated call.
    type IWarehouse =
        abstract Reserve:
            key: string * item: string * quantity: int -> Task<string>

    // The carrier is another external system. It returns a shipment ID.
    type ICarrier =
        abstract Ship: key: string * reservation: string -> Task<string>

    // Up to five attempts, 200 ms apart.
    let reserveRetry = {
        Retry.times 5 with
            Schedule = FixedInterval(TimeSpan.FromMilliseconds 200.0)
    }

    // "order" is stored with every instance. It must not change.
    let definition (warehouse: IWarehouse) (carrier: ICarrier) =
        Workflow.define "order" (fun (request: OrderRequest) ->
            workflow {
                // A failed attempt is not stored. FCQRS tries again until an
                // attempt succeeds or the policy gives up.
                let! reservation =
                    activity "reserve-stock" reserveRetry (fun attempt ->
                        warehouse.Reserve(
                            attempt.IdempotencyKey, request.Item, request.Quantity))
                // The body must give the same answers on every replay, so it
                // reads the clock and makes GUIDs only through steps. FCQRS
                // stores their values and replays them.
                let! placedAt = now
                let! confirmation = newGuid
                // An order over 1000 waits for a person to approve it, for at
                // most three seconds in this program. With no answer by then,
                // the wait returns None.
                let total = request.UnitPrice * decimal request.Quantity
                let approvalWait = TimeSpan.FromSeconds 3.0
                let! approved =
                    workflow {
                        if total <= 1000m then
                            return true
                        else
                            let! answer =
                                waitSignalFor<bool> "approval" approvalWait
                            return defaultArg answer false
                    }
                if not approved then
                    return {
                        Status = "not approved"
                        Reservation = reservation
                        Shipment = ""
                        Confirmation = confirmation
                        PlacedAt = placedAt
                    }
                else
                    // Packing takes two seconds. The timer's due time is
                    // stored, so a restart does not start the wait again.
                    do! sleep (TimeSpan.FromSeconds 2.0)
                    let! shipment =
                        activity "ship" (Retry.times 3) (fun attempt ->
                            carrier.Ship(attempt.IdempotencyKey, reservation))
                    return {
                        Status = "shipped"
                        Reservation = reservation
                        Shipment = shipment
                        Confirmation = confirmation
                        PlacedAt = placedAt
                    }
                })
open System
open System.Collections.Generic
open System.Threading
open FCQRS.Workflows
open Order
// Unchanged from step 4.
let api = Unchecked.defaultof<FCQRS.Common.IActor>
type Warehouse() =
    interface IWarehouse with
        member _.Reserve(_, _, _) = Tasks.Task.FromResult "r-1"

// A stand-in for the carrier. A call takes two seconds. With `crash`, the
// process stops after the carrier ships and before FCQRS stores the result.
type Carrier(crash: bool) =
    let shipments = Dictionary<string, string>()
    interface ICarrier with
        member _.Ship(key, reservation) =
            task {
                Thread.Sleep 2000
                if not (shipments.ContainsKey key) then
                    shipments[key] <- $"s-{shipments.Count + 1}"
                    printfn $"  carrier shipped {reservation} (key {key})"
                if crash then
                    printfn "  the process stops before this step is stored"
                    exit 1
                return shipments[key]
            }

// A stand-in for the carrier. A call takes two seconds. With `crash`, the
// process stops after the carrier ships and before FCQRS stores the result.
public sealed class Carrier(bool crash) : ICarrier
{
    private readonly Dictionary<string, string> shipments = new();

    public Task<string> Ship(string key, string reservation)
    {
        Thread.Sleep(2000);
        if (!shipments.ContainsKey(key))
        {
            shipments[key] = $"s-{shipments.Count + 1}";
            Console.WriteLine($"  carrier shipped {reservation} (key {key})");
        }
        if (crash)
        {
            Console.WriteLine("  the process stops before this step is stored");
            Environment.Exit(1);
        }
        return Task.FromResult(shipments[key]);
    }
}

Replace the registration. The program reads --crash from its command line and gives it to the carrier:

// Run the program with --crash to stop it in the middle of shipping.
let crash = Environment.GetCommandLineArgs() |> Array.contains "--crash"

// Registering the workflow also resumes the orders that were running when
// the process last stopped.
let orders =
    Workflows.register api (definition (Warehouse()) (Carrier crash))

// Run the program with --crash to stop it in the middle of shipping.
var crash = args.Contains("--crash");

// Registering the workflow also resumes the orders that were running when
// the process last stopped.
var orders = WorkflowHost.Register(
    actor, new OrderWorkflow(new Warehouse(), new Carrier(crash)));

Replace the code that starts the orders. This step places only o-1:

// Start an order. A start with an ID that is already in the journal stores
// nothing and returns the order's status.
let start id request =
    match orders.Start id request |> Async.RunSynchronously with
    | StartResult.Started -> printfn $"Started {id}"
    | StartResult.AlreadyStarted status ->
        printfn $"{id} is already started ({status})"

// Start an order, send it an approval if given one, and wait for its end.
let place id quantity approval =
    let request = {
        Customer = "ada"
        Item = "widget"
        Quantity = quantity
        UnitPrice = 40m
    }
    start id request
    match approval with
    | Some approve ->
        Workflows.signal orders id "approval" approve |> Async.RunSynchronously
    | None -> ()
    let outcome =
        orders.Result id (TimeSpan.FromSeconds 30.0) |> Async.RunSynchronously
    match outcome with
    | WorkflowOutcome.Finished result when result.Status = "shipped" ->
        printfn $"{id} finished: shipped as {result.Shipment}"
    | WorkflowOutcome.Finished result ->
        printfn $"{id} finished: {result.Status}"
    | other -> printfn $"{id} ended: {other}"

place "o-1" 2 None

// Start an order. A start with an ID that is already in the journal stores
// nothing and returns false.
async Task Start(string id, OrderRequest request)
{
    if (await orders.StartAsync(id, request))
        Console.WriteLine($"Started {id}");
    else
        Console.WriteLine(
            $"{id} is already started ({await orders.StatusAsync(id)})");
}

// Start an order, send it an approval if given one, and wait for its end.
async Task Place(string id, int quantity, bool? approval)
{
    await Start(id, new OrderRequest("ada", "widget", quantity, 40m));
    if (approval is bool approve)
        await orders.SignalAsync(id, "approval", approve);
    var result = await orders.ResultAsync(id, TimeSpan.FromSeconds(30));
    Console.WriteLine(result.Status == "shipped"
        ? $"{id} finished: shipped as {result.Shipment}"
        : $"{id} finished: {result.Status}");
}

await Place("o-1", 2, null);

Run it with --crash

Delete bin/Debug/net11.0/orders.db, then run the program with --crash:

dotnet run -- --crash

Started o-1
  warehouse busy (key order/o-1/0)
  warehouse busy (key order/o-1/0)
  warehouse reserved 2 widget (key order/o-1/0)
  carrier shipped r-1 (key order/o-1/4)
  the process stops before this step is stored

The process exited while o-1 was inside ship. The carrier shipped, but FCQRS never received the result, so the journal has ship's StepScheduled record and no StepCompleted record for it. The same happens when a real process is killed, loses power, or is redeployed during a call.

Run it again

Keep orders.db and run the program without the flag:

dotnet run

o-1 is already started (Running ["4 ship"])
  carrier shipped r-1 (key order/o-1/4)
o-1 finished: shipped as s-1
Journal of order/default-shard/o-1:
  1 Started
  2 StepScheduled
  3 StepCompleted
  4 StepCompleted
  5 StepCompleted
  6 StepScheduled
  7 StepCompleted
  8 StepScheduled
  9 StepCompleted
  10 Finished

Registering the workflow found o-1 in the journal and resumed it before the program asked: the start found it already running, waiting on ship at position 4. To resume an order, FCQRS ran the body from the beginning and replayed the stored results:

  • reserve-stock had a stored result, so the warehouse was not called. No warehouse line appears.
  • now and newGuid returned their stored values, so the confirmation number and the order time are the ones from the first run.
  • sleep had ended before the process stopped, and its end was stored, so it did not wait again.
  • ship had begun and had no result, so FCQRS ran it again, with the same key, order/o-1/4.

What the carrier must do

The carrier was called twice for one shipment. The stand-in printed carrier shipped twice because it lost its memory of keys when the process stopped. A real carrier must store the key with the shipment and answer a repeated key with the first shipment. FCQRS cannot do this for it: no stored record can tell whether a call that was running when the process stopped reached the other system.

The same holds for every activity. FCQRS gives each step a stable key, stores each result once it arrives, and never runs a step whose result is stored. The external system must make the repeated call harmless.

Next

Step 6: Charge the customer sends a command to an account from the workflow, and releases the stock when the payment is refused.