diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a6abd72e..997c122f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,7 @@ The `Unreleased` section name is replaced by the expected version of next releas ### Added - Add handling for `-T` TCP/IP switch on EventStore args [#46](https://github.com/jet/dotnet-templates/pulls/46) +- `equinox-fc`: Fulfilment-Center inspired example utilizing Process Manager patterns with basic `Equinox.MemoryStore` and `Equinox.Cosmos` tests [#40](https://github.com/jet/dotnet-templates/pulls/40) ### Changed diff --git a/README.md b/README.md index 5d56ea78c..e01f8b0b3 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ This repo hosts the source for Jet's [`dotnet new`](https://docs.microsoft.com/e - [`eqxweb`](equinox-web/README.md) - Boilerplate for an ASP .NET Core 2 Web App, with an associated storage-independent Domain project using [Equinox](https://github.com/jet/equinox). - [`eqxwebcs`](equinox-web-csharp/README.md) - Boilerplate for an ASP .NET Core 2 Web App, with an associated storage-independent Domain project using [Equinox](https://github.com/jet/equinox), _ported to C#_. - [`eqxtestbed`](equinox-testbed/README.md) - Host that allows running back-to-back benchmarks when prototyping models using [Equinox](https://github.com/jet/equinox), using different stores and/or store configuration parameters. +- [`eqxfc`](equinox-fc/README.md) - Samples showcasing various modeling and testing techniques such as (FsCheck-based) unit tests and use of `MemoryStore` for integration tests. ## [Propulsion](https://github.com/jet/propulsion) related diff --git a/dotnet-templates.sln b/dotnet-templates.sln index 5db9a23d5..150ced066 100644 --- a/dotnet-templates.sln +++ b/dotnet-templates.sln @@ -80,6 +80,18 @@ EndProjectSection EndProject Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Equinox.Templates", "src\Equinox.Templates\Equinox.Templates.fsproj", "{8C92B728-85A5-4231-863A-E4236E46CC36}" EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "fcInventory", "fcInventory", "{4946576F-1558-49ED-A272-6F4D92FB0031}" +ProjectSection(SolutionItems) = preProject + equinox-fc\.template.config\template.json = equinox-fc\.template.config\template.json + equinox-fc\README.md = equinox-fc\README.md +EndProjectSection +EndProject +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Domain", "equinox-fc\Domain\Domain.fsproj", "{B3CFC965-6AB9-47E8-AA47-548A8D8A2E2C}" +EndProject +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Domain.Tests", "equinox-fc\Domain.Tests\Domain.Tests.fsproj", "{49890A45-D6C2-4EF6-87AD-39960E03E254}" +EndProject +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Watchdog", "equinox-fc\Watchdog\Watchdog.fsproj", "{46B8B7C9-3334-4C13-A339-57571C14F2B9}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -128,10 +140,18 @@ Global {F66A5BFE-7C81-44DC-97DE-3FD8C83B8F06}.Debug|Any CPU.Build.0 = Debug|Any CPU {F66A5BFE-7C81-44DC-97DE-3FD8C83B8F06}.Release|Any CPU.ActiveCfg = Release|Any CPU {F66A5BFE-7C81-44DC-97DE-3FD8C83B8F06}.Release|Any CPU.Build.0 = Release|Any CPU - {8C92B728-85A5-4231-863A-E4236E46CC36}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {8C92B728-85A5-4231-863A-E4236E46CC36}.Debug|Any CPU.Build.0 = Debug|Any CPU - {8C92B728-85A5-4231-863A-E4236E46CC36}.Release|Any CPU.ActiveCfg = Release|Any CPU - {8C92B728-85A5-4231-863A-E4236E46CC36}.Release|Any CPU.Build.0 = Release|Any CPU + {B3CFC965-6AB9-47E8-AA47-548A8D8A2E2C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {B3CFC965-6AB9-47E8-AA47-548A8D8A2E2C}.Debug|Any CPU.Build.0 = Debug|Any CPU + {B3CFC965-6AB9-47E8-AA47-548A8D8A2E2C}.Release|Any CPU.ActiveCfg = Release|Any CPU + {B3CFC965-6AB9-47E8-AA47-548A8D8A2E2C}.Release|Any CPU.Build.0 = Release|Any CPU + {49890A45-D6C2-4EF6-87AD-39960E03E254}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {49890A45-D6C2-4EF6-87AD-39960E03E254}.Debug|Any CPU.Build.0 = Debug|Any CPU + {49890A45-D6C2-4EF6-87AD-39960E03E254}.Release|Any CPU.ActiveCfg = Release|Any CPU + {49890A45-D6C2-4EF6-87AD-39960E03E254}.Release|Any CPU.Build.0 = Release|Any CPU + {46B8B7C9-3334-4C13-A339-57571C14F2B9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {46B8B7C9-3334-4C13-A339-57571C14F2B9}.Debug|Any CPU.Build.0 = Debug|Any CPU + {46B8B7C9-3334-4C13-A339-57571C14F2B9}.Release|Any CPU.ActiveCfg = Release|Any CPU + {46B8B7C9-3334-4C13-A339-57571C14F2B9}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(NestedProjects) = preSolution {F66A5BFE-7C81-44DC-97DE-3FD8C83B8F06} = {B72FFAAE-7801-41B2-86F5-FD90E97A30F7} @@ -145,5 +165,8 @@ Global {D7ACBDF8-5F24-420F-9657-20096CE08B49} = {818D28A6-E6AB-4416-BDA6-1577C5D54447} {B6389F9E-A8E4-4BD7-B4C0-703B1A69BEA1} = {E7434881-8655-4C22-82CD-91ADB5123A73} {36C2D70A-F292-4481-8ADA-5066A80F92B2} = {1F3C9245-F973-43A3-97C9-5E527B93060C} + {B3CFC965-6AB9-47E8-AA47-548A8D8A2E2C} = {4946576F-1558-49ED-A272-6F4D92FB0031} + {49890A45-D6C2-4EF6-87AD-39960E03E254} = {4946576F-1558-49ED-A272-6F4D92FB0031} + {46B8B7C9-3334-4C13-A339-57571C14F2B9} = {4946576F-1558-49ED-A272-6F4D92FB0031} EndGlobalSection EndGlobal diff --git a/equinox-fc/.template.config/template.json b/equinox-fc/.template.config/template.json new file mode 100644 index 000000000..b8bac62f6 --- /dev/null +++ b/equinox-fc/.template.config/template.json @@ -0,0 +1,18 @@ +{ + "$schema": "http://json.schemastore.org/template", + "author": "@jet @bartelink", + "classifications": [ + "Equinox", + "Event Sourcing", + "Fc", + "Propulsion" + ], + "tags": { + "language": "F#" + }, + "identity": "Equinox.Fc", + "name": "Equinox Fc Example", + "shortName": "fcInventory", + "sourceName": "Fc", + "preferNameDirectory": true +} \ No newline at end of file diff --git a/equinox-fc/Domain.Tests/AllocationTests.fs b/equinox-fc/Domain.Tests/AllocationTests.fs new file mode 100644 index 000000000..a8f7fd3d4 --- /dev/null +++ b/equinox-fc/Domain.Tests/AllocationTests.fs @@ -0,0 +1,10 @@ +module AllocationTests + +open Allocation +open FsCheck.Xunit +open Swensen.Unquote + +let [] ``codec can roundtrip`` event = + let ee = Events.codec.Encode(None, event) + let ie = FsCodec.Core.TimelineEvent.Create(0L, ee.EventType, ee.Data) + test <@ Some event = Events.codec.TryDecode ie @> \ No newline at end of file diff --git a/equinox-fc/Domain.Tests/AllocatorTests.fs b/equinox-fc/Domain.Tests/AllocatorTests.fs new file mode 100644 index 000000000..bca6d2fb3 --- /dev/null +++ b/equinox-fc/Domain.Tests/AllocatorTests.fs @@ -0,0 +1,59 @@ +module AllocatorTests + +open Allocator +open FsCheck.Xunit +open Swensen.Unquote +open System + +type Command = + | Commence of AllocationId * DateTimeOffset + | Complete of AllocationId * Events.Reason + +type Result = + | Accepted + | Conflict of AllocationId + +let execute cmd state = + match cmd with + | Commence (a, c) -> + match decideCommence a c state with + | CommenceResult.Accepted, es -> Accepted, es + | CommenceResult.Conflict a, es -> Conflict a, es + | Complete (a, r) -> let es = decideComplete a r state in Accepted, es + +let [] properties c1 c2 = + let res, events = execute c1 Fold.initial + let state1 = Fold.fold Fold.initial events + match c1, res, events, state1 with + | Commence (a, c), Accepted, [Events.Commenced ({ allocationId = ea; cutoff = ec } as e)], state -> + test <@ a = ea && c = ec && state = Some e @> + | Complete _, Accepted, [], None -> + () // Non-applicable Complete requests are simply ignored + | _, res, l, _ -> + test <@ List.isEmpty l && res = Accepted @> + + let res, events = execute c2 state1 + let state2 = Fold.fold state1 events + match state1, c2, res, events, state2 with + // As per above, normal commence + | None, Commence (a, c), Accepted, [Events.Commenced ({ allocationId = ea; cutoff = ec } as e)], state -> + test <@ a = ea && c = ec && state = Some e @> + // Idempotent accept if same allocationId + | Some active as s1, Commence (a, _), Accepted, [], s2 -> + test <@ s1 = s2 && active.allocationId = a @> + // Conflict reports owner allocator + | Some active as s1, Commence (a2, _), Conflict a1, [], s2 -> + test <@ s1 = s2 && a2 <> a1 && a1 = active.allocationId @> + // Correct complete for same allocator is accepted + | Some active, Complete (a, r), Accepted, [Events.Completed { allocationId = ea; reason = er }], None -> + test <@ er = r && ea = a && active.allocationId = a @> + // Completes not for the same allocator are ignored + | Some active as s1, Complete (a, _), Accepted, [], s2 -> + test <@ active.allocationId <> a && s2 = s1 @> + | _, _, res, l, _ -> + test <@ List.isEmpty l && res = Accepted @> + +let [] ``codec can roundtrip`` event = + let ee = Events.codec.Encode(None, event) + let ie = FsCodec.Core.TimelineEvent.Create(0L, ee.EventType, ee.Data) + test <@ Some event = Events.codec.TryDecode ie @> \ No newline at end of file diff --git a/equinox-fc/Domain.Tests/Domain.Tests.fsproj b/equinox-fc/Domain.Tests/Domain.Tests.fsproj new file mode 100644 index 000000000..76190995d --- /dev/null +++ b/equinox-fc/Domain.Tests/Domain.Tests.fsproj @@ -0,0 +1,35 @@ + + + + netcoreapp2.1 + 5 + false + true + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/equinox-fc/Domain.Tests/Infrastructure.fs b/equinox-fc/Domain.Tests/Infrastructure.fs new file mode 100644 index 000000000..1c6c0b920 --- /dev/null +++ b/equinox-fc/Domain.Tests/Infrastructure.fs @@ -0,0 +1,51 @@ +[] +module Fc.Infrastructure + +open Serilog +open System + +let (|Id|) (x : Guid) = x.ToString "N" |> FSharp.UMX.UMX.tag +let inline mkId () = Guid.NewGuid() |> (|Id|) +let (|Ids|) (xs : Guid[]) = xs |> Array.map (|Id|) +let (|IdsAtLeastOne|) (Id x, Ids xs) = Seq.append xs (Seq.singleton x) |> Seq.toArray + +module EnvVar = + + let tryGet k = Environment.GetEnvironmentVariable k |> Option.ofObj + +module Cosmos = + + open Equinox.Cosmos + let connect () = + match EnvVar.tryGet "EQUINOX_COSMOS_CONNECTION", EnvVar.tryGet "EQUINOX_COSMOS_DATABASE", EnvVar.tryGet "EQUINOX_COSMOS_CONTAINER" with + | Some s, Some d, Some c -> + let appName = "Domain.Tests" + let discovery = Discovery.FromConnectionString s + let connector = Connector(TimeSpan.FromSeconds 5., 5, TimeSpan.FromSeconds 5., Serilog.Log.Logger) + let connection = connector.Connect(appName, discovery) |> Async.RunSynchronously + let context = Context(connection, d, c) + let cache = Equinox.Cache (appName, 10) + context, cache + | s, d, c -> + failwithf "Connection, Database and Container EQUINOX_COSMOS_* Environment variables are required (%b,%b,%b)" + (Option.isSome s) (Option.isSome d) (Option.isSome c) + +/// Adapts the XUnit ITestOutputHelper to be a Serilog Sink +type TestOutputAdapter(testOutput : Xunit.Abstractions.ITestOutputHelper) = + let template = "{Timestamp:HH:mm:ss.fff zzz} [{Level:u3}] {Message} {Properties}{NewLine}{Exception}" + let formatter = Serilog.Formatting.Display.MessageTemplateTextFormatter(template, null); + let writeSerilogEvent logEvent = + use writer = new System.IO.StringWriter() + formatter.Format(logEvent, writer) + let messageLine = string writer + testOutput.WriteLine messageLine + System.Diagnostics.Debug.Write messageLine + interface Serilog.Core.ILogEventSink with member __.Emit logEvent = writeSerilogEvent logEvent + +/// Creates a Serilog Log chain emitting to the cited Sink (only) +let createLogger sink = + Serilog.LoggerConfiguration() +// .MinimumLevel.Debug() + .Destructure.FSharpTypes() + .WriteTo.Sink(sink) + .CreateLogger() diff --git a/equinox-fc/Domain.Tests/LocationEpochTests.fs b/equinox-fc/Domain.Tests/LocationEpochTests.fs new file mode 100644 index 000000000..4184d9d13 --- /dev/null +++ b/equinox-fc/Domain.Tests/LocationEpochTests.fs @@ -0,0 +1,57 @@ +module Fc.LocationEpochTests + +open FsCheck.Xunit +open Location.Epoch +open Swensen.Unquote + +let interpret transactionId delta _balance = + match delta with + | 0 -> (), [] + | delta when delta < 0 -> (), [Events.Removed {| delta = -delta; id = transactionId |}] + | delta -> (), [Events.Added {| delta = delta; id = transactionId |}] + +let validateAndInterpret transactionId expectedBalance delta balance = + test <@ expectedBalance = balance @> + interpret transactionId delta balance + +let verifyDeltaEvent transactionId delta events = + let dEvents = events |> List.filter (function Events.Added _ | Events.Removed _ -> true | _ -> false) + test <@ interpret transactionId delta (Unchecked.defaultof<_>) = ((), dEvents) @> + +let [] properties transactionId carriedForward delta1 closeImmediately delta2 close = + + (* Starting with an empty stream, we'll need to supply the balance carried forward, optionally we apply a delta and potentially close *) + + let initialShouldClose _state = closeImmediately + let res, events = + sync (Some carriedForward) (validateAndInterpret transactionId carriedForward delta1) initialShouldClose Fold.initial + let cfEvents events = events |> List.filter (function Events.CarriedForward _ -> true | _ -> false) + let closeEvents events = events |> List.filter (function Events.Closed -> true | _ -> false) + let state1 = Fold.fold Fold.initial events + let expectedBalance = carriedForward.initial + delta1 + // Only expect closing if it was requested + let expectImmediateClose = closeImmediately + test <@ Option.isSome res.result + && expectedBalance = res.balance @> + test <@ [Events.CarriedForward { initial = carriedForward }] = cfEvents events + && (not expectImmediateClose || 1 = Seq.length (closeEvents events)) @> + verifyDeltaEvent transactionId delta1 events + + (* After initializing, validate we don't need to supply a carriedForward, and don't produce a CarriedForward event *) + + let shouldClose _state = close + let { isOpen = isOpen; result = worked; balance = bal }, events = + sync None (validateAndInterpret transactionId expectedBalance delta2) shouldClose state1 + let expectedBalance = if expectImmediateClose then expectedBalance else expectedBalance + delta2 + test <@ [] = cfEvents events + && (expectImmediateClose || not close || 1 = Seq.length (closeEvents events)) @> + test <@ (expectImmediateClose || close || isOpen) + && expectedBalance = bal @> + if not expectImmediateClose then + test <@ Option.isSome worked @> + verifyDeltaEvent transactionId delta2 events + +let [] ``codec can roundtrip`` event = + let ee = Events.codec.Encode(None, event) + let ie = FsCodec.Core.TimelineEvent.Create(0L, ee.EventType, ee.Data) + test <@ Some event = Events.codec.TryDecode ie @> diff --git a/equinox-fc/Domain.Tests/LocationSeriesTests.fs b/equinox-fc/Domain.Tests/LocationSeriesTests.fs new file mode 100644 index 000000000..65f682545 --- /dev/null +++ b/equinox-fc/Domain.Tests/LocationSeriesTests.fs @@ -0,0 +1,44 @@ +module Fc.LocationSeriesTests + +open FsCheck.Xunit +open FSharp.UMX +open Swensen.Unquote +open Location.Series + +let [] properties c1 c2 = + let events = interpretAdvanceIngestionEpoch c1 Fold.initial + let state1 = Fold.fold Fold.initial events + let epoch0 = %0 + match c1, events, state1 with + // Started events are not written for < 0 + | n, [], activeEpoch when n < epoch0 -> + test <@ None = activeEpoch @> + // Any >=0 value should trigger a Started event, initially + | n, [Events.Started { epoch = ee }], Some activatedEpoch -> + test <@ n >= epoch0 && n = ee && n = activatedEpoch @> + // Nothing else should yield events + | _, l, _ -> + test <@ List.isEmpty l @> + + let events = interpretAdvanceIngestionEpoch c2 state1 + let state2 = Fold.fold state1 events + match state1, c2, events, state2 with + // Started events are not written for < 0 + | None, n, [], activeEpoch when n < epoch0 -> + test <@ None = activeEpoch @> + // Any >= 0 epochId should trigger a Started event if first command didnt do anything + | None, n, [Events.Started { epoch = ee }], Some activatedEpoch -> + let eEpoch = %ee + test <@ n >= epoch0 && n = eEpoch && n = activatedEpoch @> + // Any higher epochId should trigger a Started event (gaps are fine - we are only tying to reduce walks) + | Some s1, n, [Events.Started { epoch = ee }], Some activatedEpoch -> + let eEpoch = %ee + test <@ n > s1 && n = eEpoch && n > epoch0 && n = activatedEpoch @> + // Nothing else should yield events + | _, _, l, _ -> + test <@ List.isEmpty l @> + +let [] ``codec can roundtrip`` event = + let ee = Events.codec.Encode(None, event) + let ie = FsCodec.Core.TimelineEvent.Create(0L, ee.EventType, ee.Data) + test <@ Some event = Events.codec.TryDecode ie @> diff --git a/equinox-fc/Domain.Tests/LocationTests.fs b/equinox-fc/Domain.Tests/LocationTests.fs new file mode 100644 index 000000000..7e2217f7d --- /dev/null +++ b/equinox-fc/Domain.Tests/LocationTests.fs @@ -0,0 +1,71 @@ +module Fc.LocationTests + +open Fc.Location +open FsCheck.Xunit +open FSharp.UMX +open Swensen.Unquote +open System + +/// Helpers to match `module Cosmos` wrapping inside the impl +module Location = + + module MemoryStore = + + open Equinox.MemoryStore + + module Series = + + let resolver store = Resolver(store, Series.Events.codec, Series.Fold.fold, Series.Fold.initial).Resolve + + module Epoch = + + let resolver store = Resolver(store, Epoch.Events.codec, Epoch.Fold.fold, Epoch.Fold.initial).Resolve + + let createService (zeroBalance, shouldClose) store = + let maxAttempts = Int32.MaxValue + let series = Series.create (Series.resolver store) maxAttempts + let epochs = Epoch.create (Epoch.resolver store) maxAttempts + create (zeroBalance, shouldClose) (series, epochs) + +let run (service : Location.Service) (IdsAtLeastOne locations, deltas : _[], transactionId) = Async.RunSynchronously <| async { + let runId = mkId () // Need to make making state in store unique when replaying or shrinking + let locations = locations |> Array.map (fun x -> % (sprintf "%O/%O" x runId)) + + let updates = deltas |> Seq.mapi (fun i x -> locations.[i % locations.Length], x) |> Seq.cache + + (* Apply random deltas *) + + let adjust delta (bal : Epoch.Fold.Balance) = + let value = max -bal delta + if value = 0 then 0, [] + elif value < 0 then value, [Epoch.Events.Removed {| delta = -value; id = transactionId |}] + else value, [Epoch.Events.Added {| delta = value; id = transactionId |}] + let! appliedDeltas = seq { for loc, x in updates -> async { let! _, eff = service.Execute(loc, adjust x) in return loc,eff } } |> Async.Parallel + let expectedBalances = Seq.append (seq { for l in locations -> l, 0}) appliedDeltas |> Seq.groupBy fst |> Seq.map (fun (l, xs) -> l, xs |> Seq.sumBy snd) |> Set.ofSeq + + (* Verify loading yields identical state *) + + let! balances = seq { for loc in locations -> async { let! bal, () = service.Execute(loc,(fun _ -> (), [])) in return loc,bal } } |> Async.Parallel + test <@ expectedBalances = Set.ofSeq balances @> } + +let [] ``MemoryStore properties`` maxEvents args = + let store = Equinox.MemoryStore.VolatileStore() + let zeroBalance = 0 + let maxEvents = max 1 maxEvents + let shouldClose (state : Epoch.Fold.OpenState) = state.count > maxEvents + let service = Location.MemoryStore.createService (zeroBalance, shouldClose) store + run service args + +type Cosmos(testOutput) = + + let context, cache = Cosmos.connect () + + let log = testOutput |> TestOutputAdapter |> createLogger + do Serilog.Log.Logger <- log + + let [] properties maxEvents args = + let zeroBalance = 0 + let maxEvents = max 1 maxEvents + let shouldClose (state : Epoch.Fold.OpenState) = state.count > maxEvents + let service = Location.Cosmos.createService (zeroBalance, shouldClose) (context, cache, 50) + run service args diff --git a/equinox-fc/Domain.Tests/TicketListTests.fs b/equinox-fc/Domain.Tests/TicketListTests.fs new file mode 100644 index 000000000..5932f8805 --- /dev/null +++ b/equinox-fc/Domain.Tests/TicketListTests.fs @@ -0,0 +1,45 @@ +module TicketListTests + +open FsCheck.Xunit +open Swensen.Unquote +open TicketList + +let [] properties c1 c2 = + let events = interpret c1 Fold.initial + let state1 = Fold.fold Fold.initial events + match c1, events, state1 with + // Empty request -> no Event + | (_, []), [], state -> + test <@ Set.isEmpty state @> + | (a, t), [Events.Allocated { allocatorId = ea; ticketIds = et }], state -> + test <@ a = ea @> + test <@ state = set t @> + test <@ state = set et @> + | _, l, _ -> + test <@ List.isEmpty l @> + + let events = interpret c2 state1 + let state2 = Fold.fold state1 events + test <@ Fold.fold state2 [Fold.snapshot state2] = state2 @> + match state1, c2, events, state2 with + // Empty request -> no Event, same state + | s1, (_, []), [], state -> + test <@ state = s1 @> + // Redundant request -> No Event, same state + | s1, (_, t), [], _ -> + test <@ Set.isSuperset s1 (set t) @> + // Two consecutive commands should both manifest in the state + | s1, (a, t), [Events.Allocated { allocatorId = ea; ticketIds = et }], state -> + test <@ a = ea @> + let et = Set et + test <@ Set.isSuperset (set t) et @> + test <@ Set.intersect s1 et |> Set.isEmpty @> + test <@ Set.isSuperset state s1 @> + test <@ Set.isSuperset state et @> + | _, _, l, _ -> + test <@ List.isEmpty l @> + +let [] ``codec can roundtrip`` event = + let ee = Events.codec.Encode(None, event) + let ie = FsCodec.Core.TimelineEvent.Create(0L, ee.EventType, ee.Data) + test <@ Some event = Events.codec.TryDecode ie @> \ No newline at end of file diff --git a/equinox-fc/Domain.Tests/TicketTests.fs b/equinox-fc/Domain.Tests/TicketTests.fs new file mode 100644 index 000000000..ffb05d87f --- /dev/null +++ b/equinox-fc/Domain.Tests/TicketTests.fs @@ -0,0 +1,82 @@ +module TicketTests + +open FsCheck.Xunit +open Swensen.Unquote +open Ticket +open Ticket.Fold + +/// We want to generate Allocate requests with and without the same listId in some cases +let (|MaybeSameCommands|) = function + | Allocate _ as x, Allocate _, cmd3, Choice1Of2 () -> x, x, cmd3 + | cmd1, (Allocate _ as x), Allocate _, Choice1Of2 () -> cmd1, x, x + | cmd1, cmd2, cmd3, (Choice1Of2 ()|Choice2Of2 ()) -> cmd1, cmd2, cmd3 + +/// Explicitly generate sequences with the same allocator running twice or three times +let (|MaybeSameIds|) = function + | Choice1Of4 a -> a, a, a + | Choice2Of4 (a, b) -> a, a, b + | Choice3Of4 (a, b) -> a, b, b + | Choice4Of4 (a, b, c) -> a, b, c + +let (|Invariants|) = function + // Revokes always succeed iff Unallocated + | Unallocated, Revoke, true, [], Unallocated -> + () + // Everything else fails + | _, _, res, e, _ -> + test <@ not res && List.isEmpty e @> + +let (|ReservedCases|_|) allocator = function + // Reserve given unallocated + | Unallocated, Reserve, true, [Events.Reserved { allocatorId = a }], state -> + test <@ a = allocator && state = Reserved a @> + Some () + // Idempotent reserve request + | Reserved a, Reserve, true, [], _ -> + test <@ a = allocator @> + Some () + // Revokes not by the owner are reported as successful, but we force the real owner to do the real relinquish + | (Reserved by | Allocated(by, _)), Revoke, true, [], _ -> + test <@ by <> allocator @> + Some () + // Revokes succeed iff by the owner + | (Reserved by | Allocated(by, _)), Revoke, true, [Events.Revoked], Unallocated -> + test <@ by = allocator @> + Some () + // Reservations can transition to Allocations as long as it's the same Allocator requesting + | Reserved a, Allocate l, true, [Events.Allocated { allocatorId = ea; listId = el }], Allocated (sa, sl) -> + test <@ a = allocator && a = ea && a = sa && l = el && l = sl @> + Some() + | _ -> None + +let [] properties (MaybeSameIds (a1, a2, a3)) (MaybeSameCommands (c1, c2, c3)) = + let res, events = decide a1 c1 Fold.initial + let state1 = Fold.fold Fold.initial events + + match Fold.initial, c1, res, events, state1 with + | _, Reserve, true, [Events.Reserved { allocatorId = a }], Reserved sa -> + test <@ a = a1 && sa = a1 @> + | Invariants -> () + + let res, events = decide a2 c2 state1 + let state2 = Fold.fold state1 events + match state1, c2, res, events, state2 with + | ReservedCases a2 -> () + | Invariants -> () + + let res, events = decide a3 c3 state2 + let state3 = Fold.fold state2 events + match state2, c3, res, events, state3 with + // Idempotent allocate ignore + | Allocated (a, l), Allocate l3, true, [], _ -> + test <@ a = a3 && l = l3 @> + // Allocated -> Revoked + | Allocated (a, _), Revoke, true, [Events.Revoked], Unallocated -> + test <@ a = a3 @> + | ReservedCases a3 -> () + | Invariants -> () + +let [] ``codec can roundtrip`` event = + let ee = Events.codec.Encode(None, event) + let ie = FsCodec.Core.TimelineEvent.Create(0L, ee.EventType, ee.Data) + test <@ Some event = Events.codec.TryDecode ie @> \ No newline at end of file diff --git a/equinox-fc/Domain/Allocation.fs b/equinox-fc/Domain/Allocation.fs new file mode 100644 index 000000000..026eea375 --- /dev/null +++ b/equinox-fc/Domain/Allocation.fs @@ -0,0 +1,257 @@ +module Allocation + +let [] Category = "Allocation" +let streamName allocationId = FsCodec.StreamName.create Category (AllocationId.toString allocationId) + +// NOTE - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +module Events = + + let [] CategoryId = "Allocation" + let (|For|) id = FsCodec.StreamName.create CategoryId (AllocationId.toString id) + + type Commenced = { ticketIds : TicketId[] } + type Tickets = { ticketIds : TicketId[] } + type Allocated = { ticketIds : TicketId[]; listId : TicketListId } + type Assigned = { listId : TicketListId } + type Snapshotted = { ticketIds : TicketId[] } + type Event = + /// Records full set of targets (so Abort can Revoke all potential in flight Reservations) + | Commenced of Commenced + /// Tickets verified as not being attainable (Allocated, not just Reserved) + | Failed of Tickets + /// Tickets verified as having been marked Reserved + | Reserved of Tickets + /// Confirming cited tickets are to be allocated to the cited list + | Allocated of Allocated + /// Records intention to release cited tickets (while Running, not implicitly via Aborted) + | Released of Tickets + /// Transitioning to phase where (Commenced-Allocated) get Returned by performing Releases on the Tickets + | Cancelled + /// Confirming cited tickets have been assigned to the list + | Assigned of Assigned + /// Records confirmed Revokes of cited Tickets + | Revoked of Tickets + /// Allocated + Returned = Commenced ==> Open for a new Commenced to happen + | Completed + // Dummy event to make Equinox.EventStore happy (see `module EventStore`) + | Snapshotted + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = NotStarted | Running of States | Canceling of States | Completed + and States = + { unknown : Set + failed : Set + reserved : Set + assigning : Events.Allocated list + releasing : Set + stats : Stats } + and Stats = + { requested : int + denied : int + reserved : int + releasing : int + assigned : int list } + let (|Idle|Acquiring|Releasing|) = function NotStarted | Completed -> Idle | Running s -> Acquiring s | Canceling s -> Releasing s + module States = + let (|ToSet|) = set + let private withKnown xs x = { x with unknown = Set.difference x.unknown xs } + let withFailed (ToSet xs) x = { withKnown xs x with failed = x.failed |> Set.union xs } + let withReserved (ToSet xs) x = { withKnown xs x with reserved = x.reserved |> Set.union xs } + let withRevoked (ToSet xs) x = { withKnown xs x with reserved = Set.difference x.reserved xs } + let withReleasing (ToSet xs) x ={ withKnown xs x with releasing = x.releasing |> Set.union xs } // TODO + let withAssigned listId x = // TODO + let decided, remaining = x.assigning |> List.partition (fun x -> x.listId = listId) + let xs = seq { for x in decided do yield! x.ticketIds } + { withRevoked xs x with assigning = remaining } + let initial = NotStarted + let evolve state = function + | Events.Commenced e -> + match state with + | NotStarted -> Running { unknown = set e.ticketIds; failed = Set.empty; reserved = Set.empty; assigning = []; releasing = Set.empty + stats = { requested = 0; denied = 0; reserved = 0; releasing = 0; assigned = [] } } + | x -> failwithf "Can only Commence when NotStarted, not %A" x + | Events.Failed e -> + match state with + | Idle -> failwith "Cannot have Failed if Idle" + | Acquiring s -> Running (s |> States.withFailed e.ticketIds) + | Releasing s -> Canceling (s |> States.withFailed e.ticketIds) + | Events.Reserved e -> + match state with + | Idle -> failwith "Cannot have Reserved if Idle" + | Acquiring s -> Running (s |> States.withReserved e.ticketIds) + | Releasing s -> Canceling (s |> States.withReserved e.ticketIds) + | Events.Allocated e -> + match state with + | Idle -> failwith "Cannot have Allocating if Idle" + | Acquiring s -> Running { s with assigning = e :: s.assigning} + | Releasing s -> Canceling { s with assigning = e :: s.assigning} + | Events.Released e -> + match state with + | Idle -> failwith "Cannot have Releasing if Idle" + | Acquiring s -> Running (s |> States.withReleasing e.ticketIds) + | Releasing s -> Canceling (s |> States.withReleasing e.ticketIds) + | Events.Cancelled -> + match state with + | Acquiring s -> Canceling s + | x -> failwithf "Can only Abort when Running, not %A" x + | Events.Assigned e -> + match state with + | Idle -> failwith "Cannot have Allocated if Idle" + | Acquiring s -> Running (s |> States.withAssigned e.listId) + | Releasing s -> Canceling (s |> States.withAssigned e.listId) + | Events.Revoked e -> + match state with + | Idle -> failwith "Cannot have Released if Idle" + | Acquiring s -> Running (s |> States.withRevoked e.ticketIds) + | Releasing s -> Canceling (s |> States.withRevoked e.ticketIds) + | Events.Completed -> + match state with + | Acquiring s + | Releasing s when Set.isEmpty s.unknown && Set.isEmpty s.reserved && List.isEmpty s.assigning -> + Completed + | x -> failwithf "Can only Complete when reservations and unknowns resolved, not %A" x + | Events.Snapshotted -> state // Dummy event, see EventStore bindings + let fold : State -> Events.Event seq -> State = Seq.fold evolve + let isOrigin = function Events.Completed -> true | Events.Snapshotted | _ -> false + +/// Current state of the workflow based on the present state of the Aggregate +type ProcessState = + | NotStarted + | Running of reserved : TicketId list * toAssign : Events.Allocated list * toRelease : TicketId list * toReserve : TicketId list + | Idle of reserved : TicketId list + | Cancelling of toAssign : Events.Allocated list * toRelease : TicketId list + | Completed + static member FromFoldState = function + | Fold.NotStarted -> + NotStarted + | Fold.Running e -> + match Set.toList e.reserved, e.assigning, Set.toList e.releasing, Set.toList e.unknown with + | res, [], [], [] -> + Idle (reserved = res) + | res, ass, rel, tor -> + Running (reserved = res, toAssign = ass, toRelease = rel, toReserve = tor) + | Fold.Canceling e -> + Cancelling (toAssign = e.assigning, toRelease = [yield! e.reserved; yield! e.unknown; yield! e.releasing]) + | Fold.Completed -> + Completed + +/// Updates recording attained progress +type Update = + | Failed of tickets : TicketId list + | Reserved of tickets : TicketId list + | Assigned of listId : TicketListId + | Revoked of tickets : TicketId list + +let (|ToSet|) xs = set xs +let (|SetEmpty|_|) s = if Set.isEmpty s then Some () else None + +/// Map processed work to associated events that are to be recorded in the stream +let decideUpdate update state = + let owned (s : Fold.States) = Set.union s.releasing (set <| seq { yield! s.unknown; yield! s.reserved }) + match state, update with + | (Fold.Completed | Fold.NotStarted), (Failed _|Reserved _|Assigned _|Revoked _) as x -> + failwithf "Folds.Completed or NotStarted cannot handle (Failed|Revoked|Assigned) %A" x + | (Fold.Running s|Fold.Canceling s), Reserved (ToSet xs) -> + match set s.unknown |> Set.intersect xs with SetEmpty -> [] | changed -> [Events.Reserved { ticketIds = Set.toArray changed }] + | (Fold.Running s|Fold.Canceling s), Failed (ToSet xs) -> + match owned s |> Set.intersect xs with SetEmpty -> [] | changed -> [Events.Failed { ticketIds = Set.toArray changed }] + | (Fold.Running s|Fold.Canceling s), Revoked (ToSet xs) -> + match owned s |> Set.intersect xs with SetEmpty -> [] | changed -> [Events.Revoked { ticketIds = Set.toArray changed }] + | (Fold.Running s|Fold.Canceling s), Assigned listId -> + if s.assigning |> List.exists (fun x -> x.listId = listId) then [Events.Assigned { listId = listId }] else [] + +/// Holds events accumulated from a series of decisions while also evolving the presented `state` to reflect the pended events +type private Accumulator() = + let acc = ResizeArray() + member __.Ingest state : 'res * Events.Event list -> 'res * Fold.State = function + | res, [] -> res, state + | res, [e] -> acc.Add e; res, Fold.evolve state e + | res, xs -> acc.AddRange xs; res, Fold.fold state (Seq.ofList xs) + member __.Accumulated = List.ofSeq acc + +/// Impetus provided to the Aggregate Service from the Process Manager +type Command = + | Commence of tickets : TicketId list + | Apply of assign : Events.Allocated list * release : TicketId list + | Cancel + +/// Apply updates, decide whether Command is applicable, emit state reflecting work to be completed to conclude the in-progress workflow (if any) +let sync (updates : Update seq, command : Command) (state : Fold.State) : (bool*ProcessState) * Events.Event list = + let acc = Accumulator() + + (* Apply any updates *) + let mutable state = state + for x in updates do + let (), state' = acc.Ingest state ((), decideUpdate x state) + state <- state' + + (* Decide whether the Command is now acceptable *) + let accepted, state = + acc.Ingest state <| + match state, command with + (* Ignore on the basis of being idempotent in the face of retries *) + // TOCONSIDER how to represent that a request is being denied e.g. due to timeout vs due to being complete + | (Fold.Idle|Fold.Releasing _), Apply _ -> + false, [] + (* Defer; Need to allow current request to progress before it can be considered *) + | (Fold.Acquiring _|Fold.Releasing _), Commence _ -> + true, [] // TODO validate idempotent ? + (* Ok on the basis of idempotency *) + | (Fold.Idle|Fold.Releasing _), Cancel -> + true, [] + (* Ok; Currently idle, normal Commence request*) + | Fold.Idle, Commence tickets -> + true, [Events.Commenced { ticketIds = Array.ofList tickets }] + (* Ok; normal apply to distribute held tickets *) + | Fold.Acquiring s, Apply (assign, release) -> + let avail = System.Collections.Generic.HashSet s.reserved + let toAssign = [for a in assign -> { a with ticketIds = a.ticketIds |> Array.where avail.Remove }] + let toRelease = (Set.empty, release) ||> List.fold (fun s x -> if avail.Remove x then Set.add x s else s) + true, [ + for x in toAssign do if (not << Array.isEmpty) x.ticketIds then yield Events.Allocated x + match toRelease with SetEmpty -> () | toRelease -> yield Events.Released { ticketIds = Set.toArray toRelease }] + (* Ok, normal Cancel *) + | Fold.Acquiring _, Cancel -> + true, [Events.Cancelled] + + (* Yield outstanding processing requirements (if any), together with events accumulated based on the `updates` *) + (accepted, ProcessState.FromFoldState state), acc.Accumulated + +type Service internal (resolve : AllocationId -> Equinox.Stream) = + + member __.Sync(allocationId, updates, command) : Async = + let stream = resolve allocationId + stream.Transact(sync (updates, command)) + +let create resolve = + let resolve pickListId = + let stream = resolve (streamName pickListId) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = 2) + Service(resolve) + +module EventStore = + + // while there are competing writers [which might cause us to have to retry a Transact], this should be infrequent + let opt = Equinox.ResolveOption.AllowStale + let create (context, cache) = + let cacheStrategy = Equinox.EventStore.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + // We should be reaching Completed state frequently so no actual Snapshots should get written + let resolver = Equinox.EventStore.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy) + let resolve id = resolver.Resolve(id, opt) + create resolve + +module Cosmos = + + // TODO impl snapshots + let makeEmptyUnfolds events _state = events, [] + let accessStrategy = Equinox.Cosmos.AccessStrategy.Custom (Fold.isOrigin,makeEmptyUnfolds) + // while there are competing writers [which might cause us to have to retry a Transact], this should be infrequent + let opt = Equinox.ResolveOption.AllowStale + let create (context, cache) = + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolve id = resolver.Resolve(id,opt) + create resolve \ No newline at end of file diff --git a/equinox-fc/Domain/Allocator.fs b/equinox-fc/Domain/Allocator.fs new file mode 100644 index 000000000..7a7fe38cc --- /dev/null +++ b/equinox-fc/Domain/Allocator.fs @@ -0,0 +1,81 @@ +module Allocator + +open System + +let [] Category = "Allocator" +let streamName allocatorId = FsCodec.StreamName.create Category (AllocatorId.toString allocatorId) + +// NOTE - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +module Events = + + let [] CategoryId = "Allocator" + let (|For|) id = FsCodec.StreamName.create CategoryId (AllocatorId.toString id) + + type Commenced = { allocationId : AllocationId; cutoff : DateTimeOffset } + type Completed = { allocationId : AllocationId; reason : Reason } + and [)>] + Reason = Ok | TimedOut | Cancelled + type Snapshotted = { active : Commenced option } + type Event = + | Commenced of Commenced + | Completed of Completed + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = Events.Commenced option + let initial = None + let evolve _state = function + | Events.Commenced e -> Some e + | Events.Completed _ -> None + let fold : State -> Events.Event seq -> State = Seq.fold evolve + +type CommenceResult = Accepted | Conflict of AllocationId + +let decideCommence allocationId cutoff : Fold.State -> CommenceResult*Events.Event list = function + | None -> Accepted, [Events.Commenced { allocationId = allocationId; cutoff = cutoff }] + | Some { allocationId = tid } when allocationId = tid -> Accepted, [] // Accept replay idempotently + | Some curr -> Conflict curr.allocationId, [] // Reject attempts at commencing overlapping transactions + +let decideComplete allocationId reason : Fold.State -> Events.Event list = function + | Some { allocationId = tid } when allocationId = tid -> [Events.Completed { allocationId = allocationId; reason = reason }] + | Some _ | None -> [] // Assume replay; accept but don't write + +type Service internal (resolve : AllocatorId -> Equinox.Stream) = + + member __.Commence(allocatorId, allocationId, cutoff) : Async = + let stream = resolve allocatorId + stream.Transact(decideCommence allocationId cutoff) + + member __.Complete(allocatorId, allocationId, reason) : Async = + let stream = resolve allocatorId + stream.Transact(decideComplete allocationId reason) + +let create resolve = + let resolve pickListId = + let stream = resolve (streamName pickListId) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = 2) + Service(resolve) + +module EventStore = + + let accessStrategy = Equinox.EventStore.AccessStrategy.LatestKnownEvent + let create (context, cache) = + let cacheStrategy = Equinox.EventStore.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + // while there are competing writers [which might cause us to have to retry a Transact], this should be infrequent + let opt = Equinox.ResolveOption.AllowStale + let resolver = Equinox.EventStore.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolve id = resolver.Resolve(id,opt) + create resolve + +module Cosmos = + + let accessStrategy = Equinox.Cosmos.AccessStrategy.LatestKnownEvent + // while there are competing writers [which might cause us to have to retry a Transact], this should be infrequent + let opt = Equinox.ResolveOption.AllowStale + let create (context, cache) = + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolve id = resolver.Resolve(id,opt) + create resolve diff --git a/equinox-fc/Domain/Domain.fsproj b/equinox-fc/Domain/Domain.fsproj new file mode 100644 index 000000000..a443df179 --- /dev/null +++ b/equinox-fc/Domain/Domain.fsproj @@ -0,0 +1,29 @@ + + + + netstandard2.0 + 5 + true + + + + + + + + + + + + + + + + + + + + + + + diff --git a/equinox-fc/Domain/Infrastructure.fs b/equinox-fc/Domain/Infrastructure.fs new file mode 100644 index 000000000..4c263fab8 --- /dev/null +++ b/equinox-fc/Domain/Infrastructure.fs @@ -0,0 +1,64 @@ +namespace global + +open FSharp.UMX // see https://github.com/fsprojects/FSharp.UMX - % operator and ability to apply units of measure to Guids+strings + +(* Locations *) + +type LocationId = string +and [] locationId +module LocationId = + let parse (value : string) : LocationId = %value + let toString (value : LocationId) : string = %value + +type LocationEpochId = int +and [] locationEpochId +module LocationEpochId = + let parse (value : int) : LocationEpochId = %value + let next (value : LocationEpochId) : LocationEpochId = % (%value + 1) + let toString (value : LocationEpochId) : string = string %value + +type InventoryId = string +and [] inventoryId +module InventoryId = + let parse (value : string) : InventoryId = %value + let toString (value : InventoryId) : string = %value + +type InventoryEpochId = int +and [] inventoryEpochId +module InventoryEpochId = + let parse (value : int) : InventoryEpochId = %value + let next (value : InventoryEpochId) : InventoryEpochId = % (%value + 1) + let toString (value : InventoryEpochId) : string = string %value + +type InventoryTransactionId = string +and [] inventoryTransactionId +module InventoryTransactionId = + let parse (value : string) : InventoryTransactionId = %value + let (|Parse|) = parse + let toString (value : InventoryTransactionId) : string = %value + +(* Tickets *) + +type TicketId = string +and [] ticketId +module TicketId = + let parse (value : string) : TicketId = let raw = value in %raw + let toString (value : TicketId) : string = %value + +type TicketListId = string +and [] ticketListId +module TicketListId = + let parse (value : string) : TicketListId = let raw = value in %raw + let toString (value : TicketListId) : string = %value + +type AllocationId = string +and [] allocationId +module AllocationId = + let parse (value : string) : AllocationId = let raw = value in %raw + let toString (value : AllocationId) : string = %value + +type AllocatorId = string +and [] allocatorId +module AllocatorId = + let parse (value : string) : AllocatorId = let raw = value in %raw + let toString (value : AllocatorId) : string = %value \ No newline at end of file diff --git a/equinox-fc/Domain/Inventory.fs b/equinox-fc/Domain/Inventory.fs new file mode 100644 index 000000000..490d7c847 --- /dev/null +++ b/equinox-fc/Domain/Inventory.fs @@ -0,0 +1,131 @@ +namespace Fc.Inventory + +open Epoch +open Equinox.Core // we use Equinox's AsyncCacheCell helper below +open FSharp.UMX + +type internal IdsCache<'Id>() = + let all = System.Collections.Concurrent.ConcurrentDictionary<'Id, unit>() // Bounded only by relatively low number of physical pick tickets IRL + static member Create init = let x = IdsCache() in x.Add init; x + member __.Add ids = for x in ids do all.[x] <- () + member __.Contains id = all.ContainsKey id + +/// Maintains active Epoch Id in a thread-safe manner while ingesting items into the `series` of `epochs` +/// Prior to first add, reads `lookBack` epochs to seed the cache, in order to minimize the number of duplicated Ids we ingest +type Service2 internal (inventoryId, series : Series.Service, epochs : Epoch.Service, lookBack, capacity) = + + let log = Serilog.Log.ForContext() + + // Maintains what we believe to be the currently open EpochId + // Guaranteed to be set only after `previousIds.AwaitValue()` + let mutable activeEpochId = Unchecked.defaultof<_> + + // We want max one request in flight to establish the pre-existing Events from which the TransactionIds cache will be seeded + let previousEpochs = AsyncCacheCell> list> <| async { + let! startingId = series.ReadIngestionEpoch(inventoryId) + activeEpochId <- %startingId + let read epochId = async { let! r = epochs.TryIngest(inventoryId, epochId, (fun _ -> 1), Seq.empty) in return r.transactionIds } + return [ for epoch in (max 0 (%startingId - lookBack)) .. (%startingId - 1) -> AsyncCacheCell(read %epoch) ] } + + // TransactionIds cache - used to maintain a list of transactions that have already been ingested in order to avoid db round-trips + let previousIds : AsyncCacheCell> = AsyncCacheCell <| async { + let! previousEpochs = previousEpochs.AwaitValue() + let! ids = seq { for x in previousEpochs -> x.AwaitValue() } |> Async.Parallel + return IdsCache.Create(Seq.concat ids) } + + let tryIngest events = async { + let! previousIds = previousIds.AwaitValue() + let initialEpochId = %activeEpochId + + let rec aux epochId totalIngested items = async { + let SeqPartition f = Seq.toArray >> Array.partition f + let dup, fresh = items |> SeqPartition (Epoch.Events.chooseInventoryTransactionId >> Option.exists previousIds.Contains) + let fullCount = List.length items + let dropping = fullCount - Array.length fresh + if dropping <> 0 then log.Information("Ignoring {count}/{fullCount} duplicate ids: {ids} for {epochId}", dropping, fullCount, dup, epochId) + if Array.isEmpty fresh then + return totalIngested + else + let! res = epochs.TryIngest(inventoryId, epochId, capacity, fresh) + log.Information("Added {count} items to {inventoryId:l}/{epochId}", res.added, inventoryId, epochId) + // The adding is potentially redundant; we don't care + previousIds.Add res.transactionIds + // Any writer noticing we've moved to a new epoch shares the burden of marking it active + if not res.isClosed && activeEpochId < %epochId then + log.Information("Marking {inventoryId:l}/{epochId} active", inventoryId, epochId) + do! series.AdvanceIngestionEpoch(inventoryId, epochId) + System.Threading.Interlocked.CompareExchange(&activeEpochId, %epochId, activeEpochId) |> ignore + let totalIngestedTransactions = totalIngested + res.added + match res.rejected with + | [] -> return totalIngestedTransactions + | rej -> return! aux (InventoryEpochId.next epochId) totalIngestedTransactions rej } + return! aux initialEpochId 0 events + } + + /// Upon startup, we initialize the TransactionIds cache with recent epochs; we want to kick that process off before our first ingest + member __.Initialize() = previousIds.AwaitValue() |> Async.Ignore + + /// Feeds the events into the sequence of transactions. Returns the number actually added [excluding duplicates] + member __.Ingest(events : Epoch.Events.Event list) : Async = tryIngest events + +module internal Helpers = + + let create inventoryId maxTransactionsPerEpoch lookBackLimit (series, epochs) = + let remainingEpochCapacity (state: Epoch.Fold.State) = + let currentLen = state.ids.Count + max 0 (maxTransactionsPerEpoch - currentLen) + Service2(inventoryId, series, epochs, lookBack = lookBackLimit, capacity = remainingEpochCapacity) + +module Cosmos = + + let create inventoryId maxTransactionsPerEpoch lookBackLimit (context, cache) = + let series = Series.Cosmos.create (context, cache) + let epochs = Epoch.Cosmos.create (context, cache) + Helpers.create inventoryId maxTransactionsPerEpoch lookBackLimit (series, epochs) + +module Processor = + + type Service(transactions : Transaction.Service, locations : Fc.Location.Service, inventory : Service2) = + + let execute transactionId = + let f = Fc.Location.Epoch.decide transactionId + let rec aux update = async { + let! action = transactions.Apply(transactionId, update) + let aux event = aux (Some event) + match action with + | Transaction.Adjust (loc, bal) -> + match! locations.Execute(loc, f (Fc.Location.Epoch.Reset bal)) with + | Fc.Location.Epoch.Accepted _ -> return! aux Transaction.Events.Adjusted + | Fc.Location.Epoch.Denied -> return failwith "Cannot Deny Reset" + | Fc.Location.Epoch.DupFromPreviousEpoch -> return failwith "TODO walk back to previous epoch" + | Transaction.Remove (loc, delta) -> + match! locations.Execute(loc, f (Fc.Location.Epoch.Remove delta)) with + | Fc.Location.Epoch.Accepted bal -> return! aux (Transaction.Events.Removed { balance = bal }) + | Fc.Location.Epoch.Denied -> return! aux Transaction.Events.Failed + | Fc.Location.Epoch.DupFromPreviousEpoch -> return failwith "TODO walk back to previous epoch" + | Transaction.Add (loc, delta) -> + match! locations.Execute(loc, f (Fc.Location.Epoch.Add delta)) with + | Fc.Location.Epoch.Accepted bal -> return! aux (Transaction.Events.Added { balance = bal }) + | Fc.Location.Epoch.Denied -> return failwith "Cannot Deny Add" + | Fc.Location.Epoch.DupFromPreviousEpoch -> return failwith "TODO walk back to previous epoch" + | Transaction.Log (Transaction.Adjusted _) -> + let! _count = inventory.Ingest([Fc.Inventory.Epoch.Events.Adjusted { transactionId = transactionId }]) + return! aux Transaction.Events.Logged + | Transaction.Log (Transaction.Transferred _) -> + let! _count = inventory.Ingest([Fc.Inventory.Epoch.Events.Transferred { transactionId = transactionId }]) + return! aux Transaction.Events.Logged + | Transaction.Finish success -> + return success + } + aux + let run transactionId req = execute transactionId (Some req) + + member __.Adjust(transactionId, location, quantity) = + run transactionId (Fc.Inventory.Transaction.Events.AdjustmentRequested { location = location; quantity = quantity }) + + member __.TryTransfer(transactionId, source, destination, quantity) = + run transactionId (Fc.Inventory.Transaction.Events.TransferRequested { source = source; destination = destination; quantity = quantity }) + + /// Used by Watchdog to force conclusion of a transaction whose progress has stalled + member __.Push(transactionId) = async { + let! _ = execute transactionId None in () } \ No newline at end of file diff --git a/equinox-fc/Domain/InventoryEpoch.fs b/equinox-fc/Domain/InventoryEpoch.fs new file mode 100644 index 000000000..bb12310e3 --- /dev/null +++ b/equinox-fc/Domain/InventoryEpoch.fs @@ -0,0 +1,96 @@ +/// Manages the ingestion (and deduplication based on a TransactionId) of events reflecting transfers or stock adjustments +/// that have been effected across a given set of Inventory +/// See Inventory.Service for surface level API which manages the ingestion, including transitioning to a new Epoch when an epoch reaches 'full' state +module Fc.Inventory.Epoch + +let [] Category = "InventoryEpoch" +let streamName (inventoryId, epochId) = FsCodec.StreamName.compose Category [InventoryId.toString inventoryId; InventoryEpochId.toString epochId] + +// NB - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +[] +module Events = + + type TransactionRef = { transactionId : InventoryTransactionId } + type Snapshotted = { closed: bool; ids : InventoryTransactionId[] } + + type Event = + | Adjusted of TransactionRef + | Transferred of TransactionRef + | Closed + | Snapshotted of Snapshotted + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + + /// Used for deduplicating input events + let chooseInventoryTransactionId = function + | Adjusted { transactionId = id } | Transferred { transactionId = id } -> Some id + | Closed | Snapshotted _ -> None + +module Fold = + + type State = { closed : bool; ids : Set } + let initial = { closed = false; ids = Set.empty } + let evolve state = function + | (Events.Adjusted e | Events.Transferred e) -> { state with ids = Set.add e.transactionId state.ids } + | Events.Closed -> { state with closed = true } + | Events.Snapshotted e -> { closed = e.closed; ids = Set.ofArray e.ids } + let fold : State -> Events.Event seq -> State = Seq.fold evolve + let isOrigin = function Events.Snapshotted _ -> true | _ -> false + let snapshot s = Events.Snapshotted { closed = s.closed; ids = Array.ofSeq s.ids } + +type Result = + { /// Indicates whether this epoch is closed (either previously or as a side-effect this time) + isClosed : bool + /// Count of items added to this epoch. May be less than requested due to removal of duplicates and/or rejected items + added : int + /// residual items that [are not duplicates and] were not accepted into this epoch + rejected : Events.Event list + /// identifiers for all items in this epoch + transactionIds : Set } + +let decideSync capacity events (state : Fold.State) : Result * Events.Event list = + let isFresh = function + | Events.Adjusted { transactionId = id } + | Events.Transferred { transactionId = id } -> (not << state.ids.Contains) id + | Events.Closed | Events.Snapshotted _ -> false + let news = events |> Seq.filter isFresh |> List.ofSeq + let closed, allowing, markClosed, residual = + let newCount = List.length news + if state.closed then + true, 0, false, news + else + let capacityNow = capacity state + let accepting = min capacityNow newCount + let closing = accepting = capacityNow + let residual = List.skip accepting news + closing, accepting, closing, residual + let events = + [ if allowing <> 0 then yield! news + if markClosed then yield Events.Closed ] + let state' = Fold.fold state events + { isClosed = closed; added = allowing; rejected = residual; transactionIds = state'.ids }, events + +type Service internal (resolve : InventoryId * InventoryEpochId -> Equinox.Stream) = + + /// Attempt ingestion of `events` into the cited Epoch. + /// - None will be accepted if the Epoch is `closed` + /// - The `capacity` function will be passed a non-closed `state` in order to determine number of items that can be admitted prior to closing + /// - If the computed capacity result is >= the number of items being submitted (which may be 0), the Epoch will be marked Closed + /// NOTE the result may include rejected items (which the caller is expected to feed into a successor epoch) + member __.TryIngest(inventoryId, epochId, capacity, events) : Async = + let stream = resolve (inventoryId, epochId) + stream.Transact(decideSync capacity events) + +let create resolve = + let resolve ids = + let stream = resolve (streamName ids) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = 2) + Service(resolve) + +module Cosmos = + + let accessStrategy = Equinox.Cosmos.AccessStrategy.Snapshot (Fold.isOrigin, Fold.snapshot) + let create (context, cache) = + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + create resolver.Resolve diff --git a/equinox-fc/Domain/InventorySeries.fs b/equinox-fc/Domain/InventorySeries.fs new file mode 100644 index 000000000..f16d16705 --- /dev/null +++ b/equinox-fc/Domain/InventorySeries.fs @@ -0,0 +1,60 @@ +/// Manages a) the ingestion epoch id b) the current checkpointed read position for a long-running Inventory Series +/// See InventoryEpoch for the logic managing the actual events logged within a given epoch +/// See Inventory.Service for the surface API which manages the writing +module Fc.Inventory.Series + +let [] Category = "InventorySeries" +let streamName inventoryId = FsCodec.StreamName.create Category (InventoryId.toString inventoryId) + +// NB - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +[] +module Events = + + type Started = { epoch : InventoryEpochId } + type Event = + | Started of Started + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = InventoryEpochId option + let initial = None + let evolve _state = function + | Events.Started e -> Some e.epoch + let fold : State -> Events.Event seq -> State = Seq.fold evolve + +let queryActiveEpoch state = state |> Option.defaultValue (InventoryEpochId.parse 0) + +let interpretAdvanceIngestionEpoch epochId (state : Fold.State) = + if queryActiveEpoch state >= epochId then [] + else [Events.Started { epoch = epochId }] + +type Service internal (resolve : InventoryId -> Equinox.Stream) = + + member __.ReadIngestionEpoch(inventoryId) : Async = + let stream = resolve inventoryId + stream.Query queryActiveEpoch + + member __.AdvanceIngestionEpoch(inventoryId, epochId) : Async = + let stream = resolve inventoryId + stream.Transact(interpretAdvanceIngestionEpoch epochId) + +let create resolve = + let resolve locationId = + let stream = resolve (streamName locationId) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = 2) + Service(resolve) + +module Cosmos = + + // For this stream, we uniformly use stale reads as: + // a) we don't require any information from competing writers + // b) while there are competing writers [which might cause us to have to retry a Transact], this should be infrequent + let opt = Equinox.ResolveOption.AllowStale + let accessStrategy = Equinox.Cosmos.AccessStrategy.LatestKnownEvent + let create (context, cache) = + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolve id = resolver.Resolve(id, opt) + create resolve diff --git a/equinox-fc/Domain/InventoryTransaction.fs b/equinox-fc/Domain/InventoryTransaction.fs new file mode 100644 index 000000000..cbaa66aeb --- /dev/null +++ b/equinox-fc/Domain/InventoryTransaction.fs @@ -0,0 +1,204 @@ +/// Process Manager used to: +/// - Coordinate competing attempts to transfer quantities from stock; if balance is 3 one of contesting requests to remove 2 or 3 items must reach `Failed` +/// - maintain rolling balance of stock levels per Location +/// - while recording any transfers or adjustment in an overall Inventory record +/// The Process is driven by two collaborating actors: +/// 1) The 'happy path', where a given actor is executing the known steps of the command flow +/// In the normal case, such an actor will bring the flow to a terminal state (Completed or Failed) +/// 2) A watchdog-projector, which reacts to observed events in this Category by stepping in to complete in-flight requests that have stalled +/// This represents the case where a 'happy path' actor died, or experienced another impediment on the path. +module Fc.Inventory.Transaction + +let [] Category = "InventoryTransaction" +let streamName transactionId = FsCodec.StreamName.create Category (InventoryTransactionId.toString transactionId) + +// NB - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +[] +module Events = + + type AdjustmentRequested = { location : LocationId; quantity : int } + type TransferRequested = { source : LocationId; destination : LocationId; quantity : int } + type Removed = { balance : int } + type Added = { balance : int } + + type Event = + (* Stock Adjustment Flow *) + | AdjustmentRequested of AdjustmentRequested + | Adjusted + + (* Stock Transfer Flow *) + | TransferRequested of TransferRequested + | Failed // terminal + | Removed of Removed + | Added of Added + + (* Successful completion *) + | Logged + | Completed + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +type Action = + | Adjust of LocationId * int + | Remove of LocationId * int + | Add of LocationId * int + | Log of LoggingState + | Finish of success : bool +and LoggingState = + | Adjusted of Events.AdjustmentRequested + | Transferred of Added +and Added = { request : Events.TransferRequested; removed : Events.Removed; added : Events.Added } + +module Fold = + + type State = + | Initial + | Running of RunningState + | Logging of LoggingState + | Completed of TerminalState + and RunningState = + | Adjust of Events.AdjustmentRequested + | Transfer of TransferState + and TransferState = + | Requested of Events.TransferRequested + | Adding of Removed + and TerminalState = + | Adjusted of Events.AdjustmentRequested + | Transferred of Added + | TransferFailed of Events.TransferRequested + and Removed = { request : Events.TransferRequested; removed : Events.Removed } + let initial = Initial + let evolve state event = + match state, event with + (* Adjustment Process *) + | Initial, Events.AdjustmentRequested r -> + Running (Adjust r) + | Running (Adjust r), Events.Adjusted -> + Logging (LoggingState.Adjusted r) + + (* Transfer Process *) + | Initial, Events.TransferRequested e -> + Running (Transfer (Requested e)) + + | Running (Transfer (Requested s)), Events.Failed -> + Completed (TransferFailed s) + + | Running (Transfer (Requested s)), Events.Removed e -> + Running (Transfer (Adding { request = s; removed = e })) + | Running (Transfer (Adding s)), Events.Added e -> + Logging (LoggingState.Transferred { request = s.request; removed = s.removed; added = e }) + + (* Log result *) + | Logging (LoggingState.Adjusted s), Events.Logged -> + Completed (Adjusted s) + | Logging (LoggingState.Transferred s), Events.Logged -> + Completed (Transferred s) + + (* Any disallowed state changes represent gaps in the model, so we fail fast *) + | state, event -> failwithf "Unexpected %A when %A" event state + let fold : State -> Events.Event seq -> State = Seq.fold evolve + + /// Validates an event actually represents an acceptable, non-redundant state transition + let filter event state = + match state, event with + | Initial, Events.AdjustmentRequested _ + | Initial, Events.TransferRequested _ + | Running (Adjust _), Events.Adjusted + | Running (Transfer (Requested _)), Events.Failed + | Running (Transfer (Requested _)), Events.Removed _ + | Running (Transfer (Adding _)), Events.Added _ + | Logging _, Events.Logged -> + [event] + | _ -> [] + + /// Determines the next action (if any) to be carried out in this workflwo + let nextAction : State -> Action = function + | Initial -> failwith "Cannot interpret Initial state" + | Running (Adjust r) -> Action.Adjust (r.location, r.quantity) + | Running (Transfer (Requested r)) -> Action.Remove (r.source, r.quantity) + | Running (Transfer (Adding r)) -> Action.Add (r.request.destination, r.request.quantity) + | Logging s -> Action.Log s + | Completed (TransferFailed _) -> Finish false + | Completed (Transferred _ | Adjusted _) -> Finish true + +/// Given an event from the Process's timeline, yields the State, in order that it can be completed +let decide update (state : Fold.State) : Action * Events.Event list = + let events = + match update with + | None -> [] + | Some update -> Fold.filter update state + let state' = Fold.fold state events + Fold.nextAction state', events + +type Service internal (resolve : InventoryTransactionId -> Equinox.Stream) = + + member __.Apply(transactionId, update) : Async = + let stream = resolve transactionId + stream.Transact(decide update) + +let create resolve = + let resolve inventoryTransactionId = + let stream = resolve (streamName inventoryTransactionId) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = 2) + Service(resolve) + +module Cosmos = + + // in the happy path case, the event stream will typically be short, and the state cached, so snapshotting is less critical + let accessStrategy = Equinox.Cosmos.AccessStrategy.Unoptimized + // ... and there will generally be a single actor touching it at a given time, so we don't need to do a load (which would be more expensive than normal given the `accessStrategy`) before we sync + let opt = Equinox.AllowStale + let create (context, cache) = + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolve id = resolver.Resolve(id, opt) + create resolve + +/// Handles requirement to infer when a transaction is 'stuck' +/// Note we don't want to couple to the state in a deep manner; thus we track: +/// a) when the request intent is established (aka when *Requested is logged) +/// b) when a transaction reports that it has Completed +module Watchdog = + + module Events = + + // TOCONSIDER: this can be written more generically by just grabbing the time of the very first event + type Event = + | AdjustmentRequested + | TransferRequested + | Completed + interface TypeShape.UnionContract.IUnionContract + type TimestampAndEvent = System.DateTimeOffset * Event + let codec = + let up (encoded : FsCodec.ITimelineEvent<_>, message) : TimestampAndEvent = encoded.Timestamp, message + let down _union = failwith "Not Implemented" + FsCodec.NewtonsoftJson.Codec.Create(up, down) + + module Fold = + + type State = Initial | Active of startTime: System.DateTimeOffset | Completed + let initial = Initial + let evolve _state = function + | startTime, (Events.AdjustmentRequested | Events.TransferRequested ) -> Active startTime + | _, Events.Completed -> Completed + let fold : State -> Events.TimestampAndEvent seq -> State = Seq.fold evolve + + type Status = Complete | Active | Stuck + let categorize cutoffTime = function + | Fold.Initial -> Active // cover (hypothetical) corner case where we don't have any valid events yet + | Fold.Active startTime when startTime < cutoffTime -> Stuck + | Fold.Active _ -> Active + | Fold.Completed -> Complete + + let fold : Events.TimestampAndEvent seq -> Fold.State = + Fold.fold Fold.initial + + let (|FoldToWatchdogState|) events : Fold.State = + events + |> Seq.choose Events.codec.TryDecode + |> fold + + let (|Match|_|) = function + | FsCodec.StreamName.CategoryAndId (Category, InventoryTransactionId.Parse transId), FoldToWatchdogState state -> + Some (transId, state) + | _ -> None \ No newline at end of file diff --git a/equinox-fc/Domain/ListAllocation.fs b/equinox-fc/Domain/ListAllocation.fs new file mode 100644 index 000000000..0485d9077 --- /dev/null +++ b/equinox-fc/Domain/ListAllocation.fs @@ -0,0 +1,22 @@ +module ListAllocation + +open System + +type Service(maxListLen, allocators : Allocator.Service, allocations : Allocation.Service, lists : TicketList.Service, tickets : Ticket.Service) = + + member __.Commence(allocatorId, allocationId, tickets, transactionTimeout) : Async<_> = async { + let cutoff = let now = DateTimeOffset.UtcNow in now.Add transactionTimeout + let! state = allocators.Commence(allocatorId, allocationId, cutoff) + // TODO cancel timed out conflicting work + let! _, state = allocations.Sync(allocationId, Seq.empty, Allocation.Commence tickets) + return state } + + member __.Read(allocationId) : Async<_> = async { + let! _, state = allocations.Sync(allocationId, Seq.empty, Allocation.Command.Apply ([], [])) + // TODO incorporate allocator state + return state } + + member __.Cancel(allocatorId, allocationId) : Async<_> = async { + let! _, state = allocations.Sync(allocationId, Seq.empty, Allocation.Command.Cancel) + // TODO propagate to allocator with reason + return state } diff --git a/equinox-fc/Domain/Location.fs b/equinox-fc/Domain/Location.fs new file mode 100644 index 000000000..37c08479d --- /dev/null +++ b/equinox-fc/Domain/Location.fs @@ -0,0 +1,48 @@ +namespace Fc.Location + +[] +type Wip<'R> = + | Pending of decide : (Epoch.Fold.State -> 'R * Epoch.Events.Event list) + | Complete of 'R + +/// Manages Reads and Writes for a Series of Epochs, with a running total being carried forward to the next Epoch when it's marked Closed +type Service internal (zeroBalance, toBalanceCarriedForward, shouldClose, series : Series.Service, epochs : Epoch.Service) = + + let execute locationId originEpochId = + let rec aux epochId balanceToCarryForward wip = async { + let decide state = match wip with Complete r -> r, [] | Pending decide -> decide state + match! epochs.Sync(locationId, epochId, balanceToCarryForward, decide, shouldClose) with + | { result = Some res; isOpen = true } -> + if originEpochId <> epochId then + do! series.AdvanceIngestionEpoch(locationId, epochId) + return res + | { history = history; result = Some res } -> + let successorEpochId = LocationEpochId.next epochId + let cf = toBalanceCarriedForward history + return! aux successorEpochId (Some cf) (Complete res) + | { history = history } -> + let successorEpochId = LocationEpochId.next epochId + let cf = toBalanceCarriedForward history + return! aux successorEpochId (Some cf) wip } + aux + + member __.Execute(locationId, decide) = async { + let! activeEpoch = series.TryReadIngestionEpoch locationId + let originEpochId, epochId, balanceCarriedForward = + match activeEpoch with + | None -> LocationEpochId.parse -1, LocationEpochId.parse 0, Some zeroBalance + | Some activeEpochId -> activeEpochId, activeEpochId, None + return! execute locationId originEpochId epochId balanceCarriedForward (Pending decide)} + +[] +module Helpers = + + let create (zeroBalance, toBalanceCarriedForward, shouldClose) (series, epochs) = + Service(zeroBalance, toBalanceCarriedForward, shouldClose, series, epochs) + +module Cosmos = + + let createService (zeroBalance, toBalanceCarriedForward, shouldClose) (context, cache, maxAttempts) = + let series = Series.Cosmos.createService (context, cache, maxAttempts) + let epochs = Epoch.Cosmos.create (context, cache, maxAttempts) + create (zeroBalance, toBalanceCarriedForward, shouldClose) (series, epochs) diff --git a/equinox-fc/Domain/LocationEpoch.fs b/equinox-fc/Domain/LocationEpoch.fs new file mode 100644 index 000000000..b5ef12c40 --- /dev/null +++ b/equinox-fc/Domain/LocationEpoch.fs @@ -0,0 +1,148 @@ +/// Manages Stock adjustments and deltas for a given Location +/// Provides for controlled opening and closing of an epoch, carrying forward incoming balances when a given Epoch reaches a 'full' state +/// See Location.Service for the logic that allows competing readers/writers to co-operate in bringing this about +module Fc.Location.Epoch + +let [] Category = "LocationEpoch" +let streamName (locationId, epochId) = FsCodec.StreamName.compose Category [LocationId.toString locationId; LocationEpochId.toString epochId] + +// NOTE - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +[] +module Events = + + type CarriedForward = { initial : int; recentTransactions : InventoryTransactionId[] } + type Event = + | CarriedForward of CarriedForward + | Added of {| delta : int; id : InventoryTransactionId |} + | Removed of {| delta : int; id : InventoryTransactionId |} + | Reset of {| value : int; id : InventoryTransactionId |} + | Closed + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = + | Initial + | Open of Record list // reverse order, i.e. most recent first + | Closed of Record list // trimmed + and Record = + | Init of Events.CarriedForward + | Step of Step + and Step = { balance : Balance; id : InventoryTransactionId } + and Balance = int + let initial = Initial + let (|Current|) = function + | (Init { initial = bal } | Step { balance = bal }) :: _ -> bal + | [] -> failwith "Cannot transact when no CarriedForward" + let evolve state event = + match event, state with + | Events.CarriedForward e, Initial -> Open [Init e] + | Events.Added e, Open (Current cur as log) -> Open (Step { id = e.id ; balance = cur + e.delta } :: log) + | Events.Removed e, Open (Current cur as log) -> Open (Step { id = e.id ; balance = cur - e.delta } :: log) + | Events.Reset e, Open log -> Open (Step { id = e.id ; balance = e.value } :: log) + | Events.Closed, Open log -> Closed log + | Events.CarriedForward _, (Open _ | Closed _ as x) -> failwithf "CarriedForward : Unexpected %A" x + | (Events.Added _ | Events.Removed _ | Events.Reset _ | Events.Closed) as e, (Initial | Closed _ as s) -> + failwithf "Unexpected %A when %A" e s + let fold = Seq.fold evolve + +/// Holds events accumulated from a series of decisions while also evolving the presented `state` to reflect the pended events +type private Accumulator() = + let acc = ResizeArray() + member __.Ingest state : 'res * Events.Event list -> 'res * Fold.State = function + | res, [] -> res, state + | res, [e] -> acc.Add e; res, Fold.evolve state e + | res, xs -> acc.AddRange xs; res, Fold.fold state (Seq.ofList xs) + member __.Accumulated = List.ofSeq acc + +type Result<'t> = { history : Fold.Record list; result : 't option; isOpen : bool } + +let sync (carriedForward : Events.CarriedForward option) + (decide : Fold.State -> 't * Events.Event list) + shouldClose + state + : Result<'t> * Events.Event list = + + let acc = Accumulator() + // 1. Guarantee a CarriedForward event at the start of any Epoch's event stream + let (), state = + acc.Ingest state <| + match state with + | Fold.Initial -> (), [Events.CarriedForward (Option.get carriedForward )] + | Fold.Open _ | Fold.Closed _ -> (), [] + // 2. Transact (unless we determine we're in Closed state) + let result, state = + acc.Ingest state <| + match state with + | Fold.Initial -> failwith "We've just guaranteed not Initial" + | Fold.Open history -> let r, es = decide state in Some r, es + | Fold.Closed _ -> None, [] + // 3. Finally (iff we're `Open`, have run a `decide` and `shouldClose`), we generate a Closed event + let (history, isOpen), _ = + acc.Ingest state <| + match state with + | Fold.Initial -> failwith "Can't be Initial" + | Fold.Open history -> + if shouldClose history then (history, false), [Events.Closed] + else (history, true), [] + | Fold.Closed history -> (history, false), [] + { history = history; result = result; isOpen = isOpen }, acc.Accumulated + +type DupCheckResult = NotDuplicate | IdempotentInsert of Fold.Balance | DupCarriedForward +let private tryFindDup transactionId (history : Fold.Record list) = + let tryMatch : Fold.Record -> Fold.Balance option option = function + | Fold.Step { balance = bal; id = id } when id = transactionId -> Some (Some bal) + | Fold.Init { recentTransactions = prevs } when prevs |> Array.contains transactionId -> Some None + | _ -> None + match history |> Seq.tryPick tryMatch with + | None -> NotDuplicate + | Some None -> DupCarriedForward + | Some (Some bal) -> IdempotentInsert bal + +type Command = + | Reset of value : int + | Add of delta : int + | Remove of delta : int + +type Result = Denied | Accepted of Fold.Balance | DupFromPreviousEpoch + +let decide transactionId command (state: Fold.State) = + match state with + | Fold.Closed _ | Fold.Initial -> failwithf "Cannot apply in state %A" state + | Fold.Open (Fold.Current cur as history) -> + + match tryFindDup transactionId history with + | IdempotentInsert bal -> Accepted bal, [] + | DupCarriedForward -> DupFromPreviousEpoch, [] + | NotDuplicate -> + + let accepted, events = + match command with + | Reset value -> true, [Events.Reset {| value = value; id = transactionId |}] + | Add delta -> true, [Events.Added {| delta = delta; id = transactionId |}] + | Remove delta when delta > cur -> false, [] + | Remove delta -> true, [Events.Removed {| delta = delta; id = transactionId |}] + match Fold.fold state events with + | Fold.Open (Fold.Current cur) -> (if accepted then Accepted cur else Denied), events + | s -> failwithf "Unexpected state %A" s + +type Service internal (resolve : LocationId * LocationEpochId -> Equinox.Stream) = + + member __.Sync<'R>(locationId, epochId, prevEpochBalanceCarriedForward, decide, shouldClose) : Async> = + let stream = resolve (locationId, epochId) + stream.Transact(sync prevEpochBalanceCarriedForward decide shouldClose) + +let create resolve maxAttempts = + let resolve locationId = + let stream = resolve (streamName locationId) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = maxAttempts) + Service(resolve) + +module Cosmos = + + let accessStrategy = Equinox.Cosmos.AccessStrategy.Unoptimized + let create (context, cache, maxAttempts) = + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + create resolver.Resolve maxAttempts diff --git a/equinox-fc/Domain/LocationSeries.fs b/equinox-fc/Domain/LocationSeries.fs new file mode 100644 index 000000000..606c4c468 --- /dev/null +++ b/equinox-fc/Domain/LocationSeries.fs @@ -0,0 +1,54 @@ +/// Manages the active epoch for a given Location +module Fc.Location.Series + +let [] Category = "LocationSeries" +let streamName locationId = FsCodec.StreamName.create Category (LocationId.toString locationId) + +// NOTE - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +[] +module Events = + + type Started = { epoch : LocationEpochId } + type Event = + | Started of Started + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = LocationEpochId option + let initial : State = None + let private evolve _state = function + | Events.Started e -> Some e.epoch + let fold = Seq.fold evolve + +let interpretAdvanceIngestionEpoch (epochId : LocationEpochId) (state : Fold.State) = + if epochId < LocationEpochId.parse 0 then [] else + + [if state |> Option.forall (fun s -> s < epochId) then yield Events.Started { epoch = epochId }] + +type Service internal (resolve : LocationId -> Equinox.Stream) = + + member __.TryReadIngestionEpoch(locationId) : Async = + let stream = resolve locationId + stream.Query id + + member __.AdvanceIngestionEpoch(locationId, epochId) : Async = + let stream = resolve locationId + stream.Transact(interpretAdvanceIngestionEpoch epochId) + +let create resolve maxAttempts = + let resolve locationId = + let stream = resolve (streamName locationId) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = maxAttempts) + Service(resolve) + +module Cosmos = + + let accessStrategy = Equinox.Cosmos.AccessStrategy.LatestKnownEvent + let opt = Equinox.ResolveOption.AllowStale + let createService (context, cache, maxAttempts) = + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolve id = resolver.Resolve(id, opt) + create resolve maxAttempts diff --git a/equinox-fc/Domain/Ticket.fs b/equinox-fc/Domain/Ticket.fs new file mode 100644 index 000000000..69f2de7b2 --- /dev/null +++ b/equinox-fc/Domain/Ticket.fs @@ -0,0 +1,86 @@ +module Ticket + +let [] Category = "Ticket" +let streamName ticketId = FsCodec.StreamName.create Category (TicketId.toString ticketId) + +// NOTE - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +module Events = + + let [] CategoryId = "Ticket" + let (|For|) id = FsCodec.StreamName.create CategoryId (TicketId.toString id) + + type Reserved = { allocatorId : AllocatorId } + type Allocated = { allocatorId : AllocatorId; listId : TicketListId } + + type Event = + | Reserved of Reserved + | Allocated of Allocated + | Revoked + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = Unallocated | Reserved of by : AllocatorId | Allocated of by : AllocatorId * on : TicketListId + let initial = Unallocated + let private evolve _state = function + | Events.Reserved e -> Reserved e.allocatorId + | Events.Allocated e -> Allocated (e.allocatorId, e.listId) + | Events.Revoked -> Unallocated + // because each event supersedes the previous one, we only ever need to fold the last event + let fold state events = + Seq.tryLast events |> Option.fold evolve state + +type Command = + /// permitted if nobody owns it (or idempotently ok if we are the owner) + | Reserve + /// permitted if the allocator has it reserved (or idempotently ok if already on list) + | Allocate of on : TicketListId + /// must be performed by the owner; attempts by non-owner to deallocate get ignored as a new owner now has that responsibility + /// (but are not failures from an Allocator's perspective) + | Revoke + +let decide (allocator : AllocatorId) (command : Command) (state : Fold.State) : bool * Events.Event list = + match command, state with + | Reserve, Fold.Unallocated -> true, [Events.Reserved { allocatorId = allocator }] // normal case -> allow+record + | Reserve, Fold.Reserved by when by = allocator -> true, [] // idempotently permit + | Reserve, (Fold.Reserved _ | Fold.Allocated _) -> false, [] // report failure, nothing to write + | Allocate list, Fold.Allocated (by, l) when by = allocator && l = list -> true, [] // idempotent processing + | Allocate list, Fold.Reserved by when by = allocator -> true, [Events.Allocated { allocatorId = allocator; listId = list }] // normal + | Allocate _, (Fold.Allocated _ | Fold.Unallocated | Fold.Reserved _) -> false, [] // Fail if someone else has reserved or allocated, or we are jumping straight to Allocated without Reserving first + | Revoke, Fold.Unallocated -> true, [] // idempotent handling + | Revoke, (Fold.Reserved by | Fold.Allocated (by, _)) when by = allocator -> true, [Events.Revoked] // release Reservation or Allocation + | Revoke, (Fold.Reserved _ | Fold.Allocated _ ) -> true, [] // NOTE we report success of achieving the intent (but, critically, we leave it to the actual owner to manage any actual revoke) + +type Service internal (resolve : TicketId -> Equinox.Stream) = + + /// Attempts to achieve the intent represented by `command`. High level semantics as per comments on Command (see decide for lowdown) + /// `false` is returned if a competing allocator holds it (or we're attempting to jump straight to Allocated without first Reserving) + member __.Sync(pickTicketId, allocator, command : Command) : Async = + let stream = resolve pickTicketId + stream.Transact(decide allocator command) + +let create resolve = + let resolve ticketId = + let stream = resolve (streamName ticketId) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = 2) + Service(resolve) + +module EventStore = + + let accessStrategy = Equinox.EventStore.AccessStrategy.LatestKnownEvent + let create (context, cache) = + let cacheStrategy = Equinox.EventStore.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + // because we only ever need the last event, we use the Equinox.EventStore access strategy that optimizes around that + let resolver = Equinox.EventStore.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + create resolver.Resolve + +module Cosmos = + + let accessStrategy = Equinox.Cosmos.AccessStrategy.LatestKnownEvent + let create (context, cache) = + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + // because we only ever need the last event to build the state, we feed the events we are writing + // (there's always exactly one if we are writing), into the unfolds slot so a single point read with etag check gets us state in one trip + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + create resolver.Resolve diff --git a/equinox-fc/Domain/TicketList.fs b/equinox-fc/Domain/TicketList.fs new file mode 100644 index 000000000..26993a743 --- /dev/null +++ b/equinox-fc/Domain/TicketList.fs @@ -0,0 +1,70 @@ +module TicketList + +let [] Category = "TicketList" +let streamName listId = FsCodec.StreamName.create Category (TicketListId.toString listId) + +// NOTE - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +module Events = + + let [] CategoryId = "TicketList" + let (|For|) id = FsCodec.StreamName.create CategoryId (TicketListId.toString id) + + type Allocated = { allocatorId : AllocatorId; ticketIds : TicketId[] } + type Snapshotted = { ticketIds : TicketId[] } + type Event = + | Allocated of Allocated + | Snapshotted of Snapshotted + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = Set + let initial = Set.empty + let evolve state = function + | Events.Allocated e -> (state, e.ticketIds) ||> Array.fold (fun m x -> Set.add x m) + | Events.Snapshotted e -> Set.ofArray e.ticketIds + let fold : State -> Events.Event seq -> State = Seq.fold evolve + let isOrigin = function Events.Snapshotted _ -> true | Events.Allocated _ -> false + let snapshot state = Events.Snapshotted { ticketIds = Set.toArray state } + +let interpret (allocatorId : AllocatorId, allocated : TicketId list) (state : Fold.State) : Events.Event list = + match allocated |> Seq.except state |> Seq.distinct |> Seq.toArray with + | [||] -> [] + | news -> [Events.Allocated { allocatorId = allocatorId; ticketIds = news }] + +type Service internal (resolve : TicketListId -> Equinox.Stream) = + + member __.Sync(pickListId, allocatorId, assignedTickets) : Async = + let stream = resolve pickListId + stream.Transact(interpret (allocatorId, assignedTickets)) + +let create resolve = + let resolve pickListId = + let stream = resolve (streamName pickListId) + Equinox.Stream(Serilog.Log.ForContext(), stream, maxAttempts = 2) + Service(resolve) + +module EventStore = + + // we _could_ use this Access Strategy, but because we are only generally doing a single shot write, its unwarranted + // let accessStrategy = AccessStrategy.RollingSnapshots (Folds.isOrigin, Folds.snapshot) + // while there are competing writers (which might cause us to have to retry a Transact and discover it is redundant), there is never a cost to being wrong + let opt = Equinox.ResolveOption.AllowStale + let create (context, cache) = + let cacheStrategy = Equinox.EventStore.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.EventStore.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy) + let resolve id = resolver.Resolve(id,opt) + create resolve + +module Cosmos = + + // while there are competing writers (which might cause us to have to retry a Transact and discover it is redundant), there is never a cost to being wrong + let opt = Equinox.ResolveOption.AllowStale + // we want reads and writes (esp idempotent ones) to have optimal RU efficiency so we go the extra mile to do snapshotting into the Tip + let accessStrategy = Equinox.Cosmos.AccessStrategy.Snapshot (Fold.isOrigin, Fold.snapshot) + let create (context, cache)= + let cacheStrategy = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let resolver = Equinox.Cosmos.Resolver(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolve id = resolver.Resolve(id, opt) + create resolve \ No newline at end of file diff --git a/equinox-fc/Fc.sln b/equinox-fc/Fc.sln new file mode 100644 index 000000000..7496986dc --- /dev/null +++ b/equinox-fc/Fc.sln @@ -0,0 +1,67 @@ + +Microsoft Visual Studio Solution File, Format Version 12.00 +# Visual Studio 15 +VisualStudioVersion = 15.0.26124.0 +MinimumVisualStudioVersion = 15.0.26124.0 +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Domain", "Domain\Domain.fsproj", "{4722DF4D-DBE1-4030-9C55-6BA3A4147322}" +EndProject +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Domain.Tests", "Domain.Tests\Domain.Tests.fsproj", "{AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}" +EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = ".project", ".project", "{FEADBAC9-0CA2-440B-AF9B-184FED9B362E}" +ProjectSection(SolutionItems) = preProject + README.md = README.md +EndProjectSection +EndProject +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Watchdog", "Watchdog\Watchdog.fsproj", "{B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}" +EndProject +Global + GlobalSection(SolutionConfigurationPlatforms) = preSolution + Debug|Any CPU = Debug|Any CPU + Debug|x64 = Debug|x64 + Debug|x86 = Debug|x86 + Release|Any CPU = Release|Any CPU + Release|x64 = Release|x64 + Release|x86 = Release|x86 + EndGlobalSection + GlobalSection(SolutionProperties) = preSolution + HideSolutionNode = FALSE + EndGlobalSection + GlobalSection(ProjectConfigurationPlatforms) = postSolution + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Debug|Any CPU.Build.0 = Debug|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Debug|x64.ActiveCfg = Debug|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Debug|x64.Build.0 = Debug|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Debug|x86.ActiveCfg = Debug|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Debug|x86.Build.0 = Debug|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Release|Any CPU.ActiveCfg = Release|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Release|Any CPU.Build.0 = Release|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Release|x64.ActiveCfg = Release|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Release|x64.Build.0 = Release|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Release|x86.ActiveCfg = Release|Any CPU + {4722DF4D-DBE1-4030-9C55-6BA3A4147322}.Release|x86.Build.0 = Release|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Debug|Any CPU.Build.0 = Debug|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Debug|x64.ActiveCfg = Debug|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Debug|x64.Build.0 = Debug|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Debug|x86.ActiveCfg = Debug|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Debug|x86.Build.0 = Debug|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Release|Any CPU.ActiveCfg = Release|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Release|Any CPU.Build.0 = Release|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Release|x64.ActiveCfg = Release|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Release|x64.Build.0 = Release|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Release|x86.ActiveCfg = Release|Any CPU + {AA589DC7-D0B7-4F95-BC29-6B89AF4E9280}.Release|x86.Build.0 = Release|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Debug|Any CPU.Build.0 = Debug|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Debug|x64.ActiveCfg = Debug|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Debug|x64.Build.0 = Debug|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Debug|x86.ActiveCfg = Debug|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Debug|x86.Build.0 = Debug|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Release|Any CPU.ActiveCfg = Release|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Release|Any CPU.Build.0 = Release|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Release|x64.ActiveCfg = Release|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Release|x64.Build.0 = Release|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Release|x86.ActiveCfg = Release|Any CPU + {B57DFA65-DC4F-4673-A5DC-7B00A08EFFF0}.Release|x86.Build.0 = Release|Any CPU + EndGlobalSection +EndGlobal diff --git a/equinox-fc/README.md b/equinox-fc/README.md new file mode 100644 index 000000000..350598cff --- /dev/null +++ b/equinox-fc/README.md @@ -0,0 +1,26 @@ +# Equinox FC Sample + +This project was generated using: + + dotnet new -i Equinox.Templates # just once, to install in the local templates store + + dotnet new eqxfc # use --help to see options regarding storage subsystem configuration etc + +# `Location*` + +The `Location`* `module`s illustrates a way to approach the modelling of a long-running state by having writers adhere to a common protocol when writing: + +- the `LocationEpoch` category represents a span of time, which is guaranteed to have a `CarriedForward` event representing the opening balance, and/or the balance carried forward when the preceding epoch was marked `Closed` +- the `LocationSeries` category bears the verified active epoch (competing readers/writers read this optimistically on a cached basis; in the event that they're behind, they'll compete to log the successful commencement of a new epoch, which cannot happen before the `CarriedForward` event for the successor epoch has been committed) + +- `Domain.Tests` includes (`FsCheck`) Property-based unit tests, and an integration tests that demonstrates parallel writes (that trigger Optimistic Concurrency Control-based conflict resolution, including against `MemoryStore`) + +## Notes + +- Referencing an `Equinox.*` Store module from the `Domain` project is not mandatory; it's common to defer all wiring and configuration of the elements in `module Cosmos`, `module EventStore` etc. and instead maintain that alongside the Composition Root, outside of the `Domain` project + +- While using an `AccessStrategy` such as `Snapshot` may in some cases be relevant too, in the general case, using the `Equinox.Cache`, combined with having a compact Fold `State` and a sufficiently constrained maximum number/size of events means the state can be established within a predictable latency range. + +- Representing a long-running state in this fashion is no panacea; in modeling a system, the ideal is to have streams that have a naturally constrained number of events over their lifetime. + +- [Original PR](https://github.com/jet/dotnet-templates/pull/40) \ No newline at end of file diff --git a/equinox-fc/Watchdog/Handler.fs b/equinox-fc/Watchdog/Handler.fs new file mode 100644 index 000000000..b96deb7a3 --- /dev/null +++ b/equinox-fc/Watchdog/Handler.fs @@ -0,0 +1,46 @@ +module Fc.Watchdog.Handler + +open System + +[] +type Outcome = Completed | Deferred | Nudged + +type Stats(log, ?statsInterval, ?stateInterval) = + inherit Propulsion.Streams.Projector.Stats(log, defaultArg statsInterval (TimeSpan.FromMinutes 1.), defaultArg stateInterval (TimeSpan.FromMinutes 5.)) + + let mutable completed, deferred, resolved = 0, 0, 0 + + override __.HandleOk res = res |> function + | Outcome.Completed -> completed <- completed + 1 + | Outcome.Deferred -> deferred <- deferred + 1 + | Outcome.Nudged -> resolved <- resolved + 1 + + override __.DumpStats () = + if completed <> 0 || deferred <> 0 then + log.Information(" Completed {completed} Deferred {deferred} Resolved {resolved}", completed, deferred, resolved) + completed <- 0; deferred <- 0; resolved <- 0 + +open Fc.Inventory.Transaction + +let tryHandle driveTransaction (stream, span : Propulsion.Streams.StreamSpan<_>) : Async = async { + let processingStuckCutoff = let now = DateTimeOffset.UtcNow in now.AddSeconds -10. + match stream, span.events with + | Watchdog.Match (transId, state) -> + match Watchdog.categorize processingStuckCutoff state with + | Watchdog.Complete -> + return Outcome.Completed + | Watchdog.Active -> + // Visiting every second is not too expensive; we don't want to be warming the data center for no purpose + do! Async.Sleep 1000 // ms + return Outcome.Deferred + | Watchdog.Stuck -> + do! driveTransaction transId + return Outcome.Nudged + | _ -> return failwith "Not a relevant event" } + +let handleStreamEvents tryHandle (stream, span : Propulsion.Streams.StreamSpan<_>) : Async = async { + match! tryHandle (stream, span) with + // if we're deferring, we need to retain the events (all of them, as the time of the first one is salient) + | Outcome.Deferred as r -> return span.index, r + // if it's now complete (either organically, or with our help), we can release all the events pertaining to this transaction + | Outcome.Completed | Outcome.Nudged as r -> return span.index + span.events.LongLength, r } diff --git a/equinox-fc/Watchdog/Program.fs b/equinox-fc/Watchdog/Program.fs new file mode 100644 index 000000000..fe86c6514 --- /dev/null +++ b/equinox-fc/Watchdog/Program.fs @@ -0,0 +1,398 @@ +module Fc.Watchdog.Program + +open Propulsion.Cosmos +open Propulsion.EventStore +open Serilog +open System + +module EnvVar = + + let tryGet varName : string option = Environment.GetEnvironmentVariable varName |> Option.ofObj + let set varName value : unit = Environment.SetEnvironmentVariable(varName, value) + +module Settings = + + let private initEnvVar var key loadF = + if None = EnvVar.tryGet var then + printfn "Setting %s from %A" var key + EnvVar.set var (loadF key) + + let initialize () = + // e.g. initEnvVar "EQUINOX_COSMOS_COLLECTION" "CONSUL KEY" readFromConsul + () // TODO add any custom logic preprocessing commandline arguments and/or gathering custom defaults from external sources, etc + +module CmdParser = + + exception MissingArg of string + let private getEnvVarForArgumentOrThrow varName argName = + match EnvVar.tryGet varName with + | None -> raise (MissingArg(sprintf "Please provide a %s, either as an argument or via the %s environment variable" argName varName)) + | Some x -> x + let private defaultWithEnvVar varName argName = function + | None -> getEnvVarForArgumentOrThrow varName argName + | Some x -> x + open Argu + open Equinox.Cosmos + open Equinox.EventStore + [] + type Parameters = + | [] ConsumerGroupName of string + | [] MaxReadAhead of int + | [] MaxWriters of int + | [] Verbose + | [] VerboseConsole + + + | [] Es of ParseResults + | [] SrcCosmos of ParseResults + interface IArgParserTemplate with + member a.Usage = + match a with + | ConsumerGroupName _ -> "Projector consumer group name." + | MaxReadAhead _ -> "maximum number of batches to let processing get ahead of completion. Default: 16." + | MaxWriters _ -> "maximum number of concurrent streams on which to process at any time. Default: 64." + | Verbose -> "request Verbose Logging. Default: off." + | VerboseConsole -> "request Verbose Console Logging. Default: off." + | Es _ -> "specify EventStore input parameters." + | SrcCosmos _ -> "specify CosmosDB input parameters." + and Arguments(a : ParseResults) = + member __.ConsumerGroupName = a.GetResult ConsumerGroupName + member __.Verbose = a.Contains Parameters.Verbose + member __.VerboseConsole = a.Contains VerboseConsole + member __.MaxReadAhead = a.GetResult(MaxReadAhead, 16) + member __.MaxConcurrentStreams = a.GetResult(MaxWriters, 64) + member __.StatsInterval = TimeSpan.FromMinutes 1. + + member val Source : Choice = + match a.TryGetSubCommand() with + | Some (Es es) -> Choice1Of2 (EsSourceArguments es) + | Some (SrcCosmos cosmos) -> Choice2Of2 (CosmosSourceArguments cosmos) + | _ -> raise (MissingArg "Must specify one of cosmos or es for Src") + member x.SourceParams() : Choice = + match x.Source with + | Choice1Of2 srcE -> + let startPos, cosmos = srcE.StartPos, srcE.Cosmos + Log.Information("Processing Consumer Group {groupName} from {startPos} (force: {forceRestart}) in Database {db} Container {container}", + x.ConsumerGroupName, startPos, srcE.ForceRestart, cosmos.Database, cosmos.Container) + Log.Information("Ingesting in batches of [{minBatchSize}..{batchSize}], reading up to {maxReadAhead} uncommitted batches ahead", + srcE.MinBatchSize, srcE.StartingBatchSize, x.MaxReadAhead) + Choice1Of2 (srcE, cosmos, + { groupName = x.ConsumerGroupName; start = startPos; checkpointInterval = srcE.CheckpointInterval; tailInterval = srcE.TailInterval + forceRestart = srcE.ForceRestart + batchSize = srcE.StartingBatchSize; minBatchSize = srcE.MinBatchSize; gorge = srcE.Gorge; streamReaders = 0 }) + | Choice2Of2 srcC -> + let disco, auxColl = + match srcC.LeaseContainer with + | None -> srcC.Discovery, { database = srcC.Database; container = srcC.Container + "-aux" } + | Some sc -> srcC.Discovery, { database = srcC.Database; container = sc } + Log.Information("Max read backlog: {maxReadAhead}", x.MaxReadAhead) + Log.Information("Processing Lease {leaseId} in Database {db} Container {container} with maximum document count limited to {maxDocuments}", + x.ConsumerGroupName, auxColl.database, auxColl.container, srcC.MaxDocuments) + if srcC.FromTail then Log.Warning("(If new projector group) Skipping projection of all existing events.") + srcC.LagFrequency |> Option.iter (fun s -> Log.Information("Dumping lag stats at {lagS:n0}s intervals", s.TotalSeconds)) + Choice2Of2 (srcC, (disco, auxColl, x.ConsumerGroupName, srcC.FromTail, srcC.MaxDocuments, srcC.LagFrequency)) + and [] CosmosSourceParameters = + | [] FromTail + | [] MaxDocuments of int + | [] LagFreqM of float + | [] LeaseContainer of string + + | [] ConnectionMode of Equinox.Cosmos.ConnectionMode + | [] Connection of string + | [] Database of string + | [] Container of string // Actually Mandatory, but stating that is not supported + | [] Timeout of float + | [] Retries of int + | [] RetriesWaitTime of float + + | [] Cosmos of ParseResults + interface IArgParserTemplate with + member a.Usage = a |> function + | FromTail -> "(iff the Consumer Name is fresh) - force skip to present Position. Default: Never skip an event." + | MaxDocuments _ -> "maximum item count to request from feed. Default: unlimited" + | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: off" + | LeaseContainer _ -> "specify Container Name for Leases container. Default: `sourceContainer` + `-aux`." + + | ConnectionMode _ -> "override the connection mode. Default: Direct." + | Connection _ -> "specify a connection string for a Cosmos account. (optional if environment variable EQUINOX_COSMOS_CONNECTION specified)" + | Database _ -> "specify a database name for Cosmos account. (optional if environment variable EQUINOX_COSMOS_DATABASE specified)" + | Container _ -> "specify a container name within `Database`" + | Timeout _ -> "specify operation timeout in seconds. Default: 5." + | Retries _ -> "specify operation retries. Default: 1." + | RetriesWaitTime _ -> "specify max wait-time for retry when being throttled by Cosmos in seconds. Default: 5." + + | Cosmos _ -> "CosmosDb Sink parameters." + and CosmosSourceArguments(a : ParseResults) = + member __.FromTail = a.Contains CosmosSourceParameters.FromTail + member __.MaxDocuments = a.TryGetResult MaxDocuments + member __.LagFrequency = a.TryGetResult LagFreqM |> Option.map TimeSpan.FromMinutes + member __.LeaseContainer = a.TryGetResult CosmosSourceParameters.LeaseContainer + + member __.Mode = a.GetResult(CosmosSourceParameters.ConnectionMode, Equinox.Cosmos.ConnectionMode.Direct) + member __.Discovery = Discovery.FromConnectionString __.Connection + member __.Connection = a.TryGetResult CosmosSourceParameters.Connection |> defaultWithEnvVar "EQUINOX_COSMOS_CONNECTION" "Connection" + member __.Database = a.TryGetResult CosmosSourceParameters.Database |> defaultWithEnvVar "EQUINOX_COSMOS_DATABASE" "Database" + member __.Container = a.GetResult CosmosSourceParameters.Container + member __.Timeout = a.GetResult(CosmosSourceParameters.Timeout, 5.) |> TimeSpan.FromSeconds + member __.Retries = a.GetResult(CosmosSourceParameters.Retries, 1) + member __.MaxRetryWaitTime = a.GetResult(CosmosSourceParameters.RetriesWaitTime, 5.) |> TimeSpan.FromSeconds + member x.BuildConnectionDetails() = + let (Discovery.UriAndKey (endpointUri, _)) as discovery = x.Discovery + Log.Information("Source CosmosDb {mode} {endpointUri} Database {database} Container {container}", + x.Mode, endpointUri, x.Database, x.Container) + Log.Information("Source CosmosDb timeout {timeout}s; Throttling retries {retries}, max wait {maxRetryWaitTime}s", + (let t = x.Timeout in t.TotalSeconds), x.Retries, (let t = x.MaxRetryWaitTime in t.TotalSeconds)) + let c = Equinox.Cosmos.Connector(x.Timeout, x.Retries, x.MaxRetryWaitTime, Log.Logger, mode=x.Mode) + discovery, { database = x.Database; container = x.Container }, c + + member val Cosmos = + match a.TryGetSubCommand() with + | Some (CosmosSourceParameters.Cosmos cosmos) -> CosmosArguments cosmos + | _ -> raise (MissingArg "Must specify cosmos details") + and [] EsSourceParameters = + | [] FromTail + | [] Gorge of int + | [] Tail of intervalS: float + | [] ForceRestart + | [] BatchSize of int + | [] MinBatchSize of int + | [] Position of int64 + | [] Chunk of int + | [] Percent of float + + | [] Verbose + | [] Timeout of float + | [] Retries of int + | [] HeartbeatTimeout of float + | [] Tcp + | [] Host of string + | [] Port of int + | [] Username of string + | [] Password of string + + | [] Cosmos of ParseResults + interface IArgParserTemplate with + member a.Usage = a |> function + | FromTail -> "Start the processing from the Tail" + | Gorge _ -> "Request Parallel readers phase during initial catchup, running one chunk (256MB) apart. Default: off" + | Tail _ -> "attempt to read from tail at specified interval in Seconds. Default: 1" + | ForceRestart _ -> "Forget the current committed position; start from (and commit) specified position. Default: start from specified position or resume from committed." + | BatchSize _ -> "maximum item count to request from feed. Default: 4096" + | MinBatchSize _ -> "minimum item count to drop down to in reaction to read failures. Default: 512" + | Position _ -> "EventStore $all Stream Position to commence from" + | Chunk _ -> "EventStore $all Chunk to commence from" + | Percent _ -> "EventStore $all Stream Position to commence from (as a percentage of current tail position)" + + | Verbose -> "Include low level Store logging." + | Tcp -> "Request connecting direct to a TCP/IP endpoint. Default: Use Clustered mode with Gossip-driven discovery (unless environment variable EQUINOX_ES_TCP specifies 'true')." + | Host _ -> "TCP mode: specify a hostname to connect to directly. Clustered mode: use Gossip protocol against all A records returned from DNS query. (optional if environment variable EQUINOX_ES_HOST specified)" + | Port _ -> "specify a custom port. Uses value of environment variable EQUINOX_ES_PORT if specified. Defaults for Cluster and Direct TCP/IP mode are 30778 and 1113 respectively." + | Username _ -> "specify a username. (optional if environment variable EQUINOX_ES_USERNAME specified)" + | Password _ -> "specify a Password. (optional if environment variable EQUINOX_ES_PASSWORD specified)" + | Timeout _ -> "specify operation timeout in seconds. Default: 20." + | Retries _ -> "specify operation retries. Default: 3." + | HeartbeatTimeout _ -> "specify heartbeat timeout in seconds. Default: 1.5." + + | Cosmos _ -> "CosmosDB (Checkpoint) Store parameters." + and EsSourceArguments(a : ParseResults) = + member __.Gorge = a.TryGetResult Gorge + member __.TailInterval = a.GetResult(Tail, 1.) |> TimeSpan.FromSeconds + member __.ForceRestart = a.Contains ForceRestart + member __.StartingBatchSize = a.GetResult(BatchSize, 4096) + member __.MinBatchSize = a.GetResult(MinBatchSize, 512) + member __.StartPos = + match a.TryGetResult Position, a.TryGetResult Chunk, a.TryGetResult Percent, a.Contains EsSourceParameters.FromTail with + | Some p, _, _, _ -> Absolute p + | _, Some c, _, _ -> StartPos.Chunk c + | _, _, Some p, _ -> Percentage p + | None, None, None, true -> StartPos.TailOrCheckpoint + | None, None, None, _ -> StartPos.StartOrCheckpoint + + member __.Discovery = + match __.Tcp, __.Port with + | false, None -> Discovery.GossipDns __.Host + | false, Some p -> Discovery.GossipDnsCustomPort (__.Host, p) + | true, None -> Discovery.Uri (UriBuilder("tcp", __.Host).Uri) + | true, Some p -> Discovery.Uri (UriBuilder("tcp", __.Host, p).Uri) + member __.Tcp = + a.Contains EsSourceParameters.Tcp + || EnvVar.tryGet "EQUINOX_ES_TCP" |> Option.exists (fun s -> String.Equals(s, bool.TrueString, StringComparison.OrdinalIgnoreCase)) + member __.Port = match a.TryGetResult Port with Some x -> Some x | None -> EnvVar.tryGet "EQUINOX_ES_PORT" |> Option.map int + member __.Host = a.TryGetResult Host |> defaultWithEnvVar "EQUINOX_ES_HOST" "Host" + member __.User = a.TryGetResult Username |> defaultWithEnvVar "EQUINOX_ES_USERNAME" "Username" + member __.Password = a.TryGetResult Password |> defaultWithEnvVar "EQUINOX_ES_PASSWORD" "Password" + member __.Retries = a.GetResult(EsSourceParameters.Retries, 3) + member __.Timeout = a.GetResult(EsSourceParameters.Timeout, 20.) |> TimeSpan.FromSeconds + member __.Heartbeat = a.GetResult(HeartbeatTimeout, 1.5) |> TimeSpan.FromSeconds + member x.Connect(log: ILogger, storeLog: ILogger, appName, connectionStrategy) = + let s (x : TimeSpan) = x.TotalSeconds + let discovery = x.Discovery + log.ForContext("host", x.Host).ForContext("port", x.Port) + .Information("EventStore {discovery} heartbeat: {heartbeat}s Timeout: {timeout}s Retries {retries}", + discovery, s x.Heartbeat, s x.Timeout, x.Retries) + let log=if storeLog.IsEnabled Serilog.Events.LogEventLevel.Debug then Logger.SerilogVerbose storeLog else Logger.SerilogNormal storeLog + let tags=["M", Environment.MachineName; "I", Guid.NewGuid() |> string] + Connector(x.User, x.Password, x.Timeout, x.Retries, log=log, heartbeatTimeout=x.Heartbeat, tags=tags) + .Establish(appName, discovery, connectionStrategy) |> Async.RunSynchronously + + member __.CheckpointInterval = TimeSpan.FromHours 1. + member val Cosmos : CosmosArguments = + match a.TryGetSubCommand() with + | Some (EsSourceParameters.Cosmos cosmos) -> CosmosArguments cosmos + | _ -> raise (MissingArg "Must specify `cosmos` checkpoint store when source is `es`") + and [] CosmosParameters = + | [] Connection of string + | [] ConnectionMode of ConnectionMode + | [] Database of string + | [] Container of string + | [] Timeout of float + | [] Retries of int + | [] RetriesWaitTime of float + interface IArgParserTemplate with + member a.Usage = + match a with + | ConnectionMode _ -> "override the connection mode. Default: Direct." + | Connection _ -> "specify a connection string for a Cosmos account. (optional if environment variable EQUINOX_COSMOS_CONNECTION specified)" + | Database _ -> "specify a database name for Cosmos store. (optional if environment variable EQUINOX_COSMOS_DATABASE specified)" + | Container _ -> "specify a container name for Cosmos store. (optional if environment variable EQUINOX_COSMOS_CONTAINER specified)" + | Timeout _ -> "specify operation timeout in seconds. Default: 5." + | Retries _ -> "specify operation retries. Default: 1." + | RetriesWaitTime _ -> "specify max wait-time for retry when being throttled by Cosmos in seconds. Default: 5." + and CosmosArguments(a : ParseResults) = + member __.Mode = a.GetResult(CosmosParameters.ConnectionMode, Equinox.Cosmos.ConnectionMode.Direct) + member __.Connection = a.TryGetResult CosmosParameters.Connection |> defaultWithEnvVar "EQUINOX_COSMOS_CONNECTION" "Connection" + member __.Database = a.TryGetResult CosmosParameters.Database |> defaultWithEnvVar "EQUINOX_COSMOS_DATABASE" "Database" + member __.Container = a.TryGetResult CosmosParameters.Container |> defaultWithEnvVar "EQUINOX_COSMOS_CONTAINER" "Container" + member __.Timeout = a.GetResult(CosmosParameters.Timeout, 5.) |> TimeSpan.FromSeconds + member __.Retries = a.GetResult(CosmosParameters.Retries, 1) + member __.MaxRetryWaitTime = a.GetResult(CosmosParameters.RetriesWaitTime, 5.) |> TimeSpan.FromSeconds + member x.BuildConnectionDetails() = + let (Discovery.UriAndKey (endpointUri, _) as discovery) = Discovery.FromConnectionString x.Connection + Log.Information("CosmosDb {mode} {endpointUri} Database {database} Container {container}", + x.Mode, endpointUri, x.Database, x.Container) + Log.Information("CosmosDb timeout {timeout}s; Throttling retries {retries}, max wait {maxRetryWaitTime}s", + (let t = x.Timeout in t.TotalSeconds), x.Retries, (let t = x.MaxRetryWaitTime in t.TotalSeconds)) + let connector = Equinox.Cosmos.Connector(x.Timeout, x.Retries, x.MaxRetryWaitTime, Log.Logger, mode=x.Mode) + discovery, x.Database, x.Container, connector + + /// Parse the commandline; can throw exceptions in response to missing arguments and/or `-h`/`--help` args + let parse argv = + let programName = System.Reflection.Assembly.GetEntryAssembly().GetName().Name + let parser = ArgumentParser.Create(programName = programName) + parser.ParseCommandLine argv |> Arguments + +module Logging = + + let initialize verbose changeLogVerbose = + Log.Logger <- + LoggerConfiguration() + .Destructure.FSharpTypes() + .Enrich.FromLogContext() + |> fun c -> if verbose then c.MinimumLevel.Debug() else c + // LibLog writes to the global logger, so we need to control the emission + |> fun c -> let cfpl = if changeLogVerbose then Serilog.Events.LogEventLevel.Debug else Serilog.Events.LogEventLevel.Warning + c.MinimumLevel.Override("Microsoft.Azure.Documents.ChangeFeedProcessor", cfpl) + |> fun c -> let isCfp429a = Filters.Matching.FromSource("Microsoft.Azure.Documents.ChangeFeedProcessor.LeaseManagement.DocumentServiceLeaseUpdater").Invoke + let isCfp429b = Filters.Matching.FromSource("Microsoft.Azure.Documents.ChangeFeedProcessor.PartitionManagement.LeaseRenewer").Invoke + let isCfp429c = Filters.Matching.FromSource("Microsoft.Azure.Documents.ChangeFeedProcessor.PartitionManagement.PartitionLoadBalancer").Invoke + let isCfp429d = Filters.Matching.FromSource("Microsoft.Azure.Documents.ChangeFeedProcessor.FeedProcessing.PartitionProcessor").Invoke + let isCfp x = isCfp429a x || isCfp429b x || isCfp429c x || isCfp429d x + if changeLogVerbose then c else c.Filter.ByExcluding(fun x -> isCfp x) + |> fun c -> let t = "[{Timestamp:HH:mm:ss} {Level:u3}] {partitionKeyRangeId,2} {Message:lj} {NewLine}{Exception}" + c.WriteTo.Console(theme=Sinks.SystemConsole.Themes.AnsiConsoleTheme.Code, outputTemplate=t) + |> fun c -> c.CreateLogger() + +let [] AppName = "Watchdog" + +module EventStoreContext = + let cache = Equinox.Cache(AppName, sizeMb = 10) + let create connection = Equinox.EventStore.Context(connection, Equinox.EventStore.BatchingPolicy(maxBatchSize=500)) + let tryMapEvent isWhitelisted (x : EventStore.ClientAPI.ResolvedEvent) = + match x.Event with + | e when not e.IsJson || e.EventStreamId.StartsWith "$" || not (isWhitelisted e.EventStreamId) -> None + | PropulsionStreamEvent e -> Some e + +let transactionStreamPrefix = sprintf "%s-" Fc.Inventory.Transaction.Category +let isTransactionStream : string -> bool = function sn -> sn.StartsWith transactionStreamPrefix + +let build (args : CmdParser.Arguments) = + match args.SourceParams() with + | Choice1Of2 (srcE, cosmos, spec) -> + let connectEs () = srcE.Connect(Log.Logger, Log.Logger, AppName, Equinox.EventStore.ConnectionStrategy.ClusterSingle Equinox.EventStore.NodePreference.PreferSlave) + let (discovery, database, container, connector) = cosmos.BuildConnectionDetails() + + let connection = connector.Connect(AppName, discovery) |> Async.RunSynchronously + let cache = Equinox.Cache(AppName, sizeMb = 10) + let context = Equinox.Cosmos.Context(connection, database, container) + + let resolveCheckpointStream = + let codec = FsCodec.NewtonsoftJson.Codec.Create() + let caching = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, TimeSpan.FromMinutes 20.) + let access = Equinox.Cosmos.AccessStrategy.Custom (Checkpoint.Fold.isOrigin, Checkpoint.Fold.transmute) + fun target -> Equinox.Cosmos.Resolver(context, codec, Checkpoint.Fold.fold, Checkpoint.Fold.initial, caching, access).Resolve(target, Equinox.AllowStale) + let checkpoints = Checkpoint.CheckpointSeries(spec.groupName, Log.ForContext(), resolveCheckpointStream) + let handle (_stream, _span) : Async = failwith "TODO" // Handler.handleStreamEvents (Handler.tryHandle driveTransaction) + let sink = + Propulsion.Streams.StreamsProjector.Start( + Log.Logger, args.MaxReadAhead, args.MaxConcurrentStreams, handle, stats = Handler.Stats(Log.Logger)) + let connect () = let c = connectEs () in c.ReadConnection + let runPipeline = + EventStoreSource.Run( + Log.Logger, sink, checkpoints, connect, spec, EventStoreContext.tryMapEvent isTransactionStream, + args.MaxReadAhead, args.StatsInterval) + sink, runPipeline + | Choice2Of2 (srcCosmos, (auxDiscovery, aux, leaseId, startFromTail, maxDocuments, lagFrequency)) -> + let (discovery, database, container, connector) = srcCosmos.Cosmos.BuildConnectionDetails() + let connection = connector.Connect(AppName, discovery) |> Async.RunSynchronously + let cache = Equinox.Cache(AppName, sizeMb = 10) + let context = Equinox.Cosmos.Context(connection, database, container) + let inventoryService = + let inventoryId = InventoryId.parse "FC000" + let maxTransactionsPerEpoch = 100 + let lookBackLimit = 10 + Fc.Inventory.Cosmos.create inventoryId maxTransactionsPerEpoch lookBackLimit (context, cache) + let transactionService = Fc.Inventory.Transaction.Cosmos.create (context, cache) + let locations = + let zeroBalance : Fc.Location.Epoch.Events.CarriedForward = { initial = 0; recentTransactions = [||] } + let chooseTransactionIds = function + | Fc.Location.Epoch.Fold.Init { recentTransactions = ids } -> Seq.ofArray ids + | Fc.Location.Epoch.Fold.Step { id = id } -> Seq.singleton id + let toBalanceCarriedForward (Fc.Location.Epoch.Fold.Current cur as records) : Fc.Location.Epoch.Events.CarriedForward = + { initial = cur; recentTransactions = records |> Seq.collect chooseTransactionIds |> Seq.truncate 5 |> Seq.toArray } + let shouldClose x = false + let maxAttempts = 3 + Fc.Location.Cosmos.createService (zeroBalance, toBalanceCarriedForward, shouldClose) (context, cache, maxAttempts) + let processor = Fc.Inventory.Processor.Service(transactionService, locations, inventoryService) + + let handle = Handler.handleStreamEvents (Handler.tryHandle processor.Push) + let sink = + Propulsion.Streams.StreamsProjector.Start( + Log.Logger, args.MaxReadAhead, args.MaxConcurrentStreams, handle, stats = Handler.Stats(Log.Logger)) + let mapToStreamItems (docs : Microsoft.Azure.Documents.Document seq) : Propulsion.Streams.StreamEvent<_> seq = + docs + |> Seq.collect EquinoxCosmosParser.enumStreamEvents + |> Seq.filter (fun e -> e.stream |> FsCodec.StreamName.toString |> isTransactionStream) + let source = { database = database; container = container } + let createObserver () = CosmosSource.CreateObserver(Log.Logger, sink.StartIngester, mapToStreamItems) + let runPipeline = + CosmosSource.Run(Log.Logger, connector.CreateClient(AppName, discovery), source, aux, + leaseId, startFromTail, createObserver, + ?maxDocuments=maxDocuments, ?lagReportFreq=lagFrequency, auxClient=connector.CreateClient(AppName, auxDiscovery)) + sink, runPipeline + +let run argv = + try let args = CmdParser.parse argv + Logging.initialize args.Verbose args.VerboseConsole + Settings.initialize () + let projector, runSourcePipeline = build args + runSourcePipeline |> Async.Start + projector.AwaitCompletion() |> Async.RunSynchronously + if projector.RanToCompletion then 0 else 2 + with :? Argu.ArguParseException as e -> eprintfn "%s" e.Message; 1 + | CmdParser.MissingArg msg -> eprintfn "%s" msg; 1 + | e -> Log.Fatal(e, "Exiting"); 1 + +[] +let main argv = + try run argv + finally Log.CloseAndFlush() diff --git a/equinox-fc/Watchdog/Watchdog.fsproj b/equinox-fc/Watchdog/Watchdog.fsproj new file mode 100644 index 000000000..ebc91054c --- /dev/null +++ b/equinox-fc/Watchdog/Watchdog.fsproj @@ -0,0 +1,26 @@ + + + + Exe + netcoreapp3.1 + 5 + + + + + + + + + + + + + + + + + + + +