diff --git a/CHANGELOG.md b/CHANGELOG.md index 546cb6fa1..384bce275 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,8 @@ The `Unreleased` section name is replaced by the expected version of next releas ### Added +- `eqxPatterns`: `Period`: Skeleton Deciders+Tests for `Period` with Rolling Balance [#89](https://github.com/jet/dotnet-templates/pull/89) +- `eqxPatterns`: `Series`+`Epoch`: Skeleton Deciders+Tests for deduplicated ingestion of items [#89](https://github.com/jet/dotnet-templates/pull/89) - `eqxProjector --source cosmos --kafka --synthesizeSequence`: Sample code for custom parsing of document changes [#84](https://github.com/jet/dotnet-templates/pull/84) ### Changed diff --git a/README.md b/README.md index 365f0bdb3..dde272ace 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,9 @@ These templates focus solely on Consistent Processing using Equinox Stores: - [`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. +- [`eqxPatterns`](equinox-patterns/README.md) - Equinox Skeleton Deciders and Tests implementing various event sourcing patterns: + - [x] Periods with Rolling Balance carried forward + - [x] Epochs/Series/Ingester with deduplication ## [Propulsion](https://github.com/jet/propulsion) related diff --git a/dotnet-templates.sln b/dotnet-templates.sln index e20dd46f0..07fed446a 100644 --- a/dotnet-templates.sln +++ b/dotnet-templates.sln @@ -111,6 +111,16 @@ EndProjectSection EndProject Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Pruner", "propulsion-pruner\Pruner.fsproj", "{A0FB44F5-15E5-47C8-81E5-1991269849CB}" EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "eqxPatterns", "eqxPatterns", "{76721F1E-851C-4970-A276-DF61FCE3DA23}" +ProjectSection(SolutionItems) = preProject + equinox-patterns\.template.config\template.json = equinox-patterns\.template.config\template.json + equinox-patterns\README.md = equinox-patterns\README.md +EndProjectSection +EndProject +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Domain", "equinox-patterns\Domain\Domain.fsproj", "{8D9867A9-1B5D-4AD3-A890-ACC81D011C00}" +EndProject +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Domain.Tests", "equinox-patterns\Domain.Tests\Domain.Tests.fsproj", "{C899EB07-FEC8-41F8-95DC-203DA80F5A32}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -191,6 +201,14 @@ Global {A0FB44F5-15E5-47C8-81E5-1991269849CB}.Debug|Any CPU.Build.0 = Debug|Any CPU {A0FB44F5-15E5-47C8-81E5-1991269849CB}.Release|Any CPU.ActiveCfg = Release|Any CPU {A0FB44F5-15E5-47C8-81E5-1991269849CB}.Release|Any CPU.Build.0 = Release|Any CPU + {8D9867A9-1B5D-4AD3-A890-ACC81D011C00}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {8D9867A9-1B5D-4AD3-A890-ACC81D011C00}.Debug|Any CPU.Build.0 = Debug|Any CPU + {8D9867A9-1B5D-4AD3-A890-ACC81D011C00}.Release|Any CPU.ActiveCfg = Release|Any CPU + {8D9867A9-1B5D-4AD3-A890-ACC81D011C00}.Release|Any CPU.Build.0 = Release|Any CPU + {C899EB07-FEC8-41F8-95DC-203DA80F5A32}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {C899EB07-FEC8-41F8-95DC-203DA80F5A32}.Debug|Any CPU.Build.0 = Debug|Any CPU + {C899EB07-FEC8-41F8-95DC-203DA80F5A32}.Release|Any CPU.ActiveCfg = Release|Any CPU + {C899EB07-FEC8-41F8-95DC-203DA80F5A32}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(NestedProjects) = preSolution {F66A5BFE-7C81-44DC-97DE-3FD8C83B8F06} = {B72FFAAE-7801-41B2-86F5-FD90E97A30F7} @@ -210,5 +228,7 @@ Global {83BA87C3-6288-40F4-BC4F-EC3A54586CDF} = {DAE9E2B9-EDA2-4064-B0CE-FD5294549374} {B9976751-C3A6-4F8B-BEF4-278382D8EAA6} = {BBD6F425-C2F4-4857-BD68-2B13D0B4EDDE} {A0FB44F5-15E5-47C8-81E5-1991269849CB} = {7F80222A-2687-4E91-8F5B-4717DD13DEF5} + {8D9867A9-1B5D-4AD3-A890-ACC81D011C00} = {76721F1E-851C-4970-A276-DF61FCE3DA23} + {C899EB07-FEC8-41F8-95DC-203DA80F5A32} = {76721F1E-851C-4970-A276-DF61FCE3DA23} EndGlobalSection EndGlobal diff --git a/equinox-patterns/.template.config/template.json b/equinox-patterns/.template.config/template.json new file mode 100644 index 000000000..cc83d7f40 --- /dev/null +++ b/equinox-patterns/.template.config/template.json @@ -0,0 +1,21 @@ +{ + "$schema": "http://json.schemastore.org/template", + "author": "@jet @bartelink", + "classifications": [ + "Equinox", + "Event Sourcing", + "Closing Books", + "Epoch", + "Series", + "Period" + ], + "tags": { + "language": "F#", + "type": "project" + }, + "identity": "Equinox.Patterns", + "name": "Equinox Patterns Sample", + "shortName": "eqxPatterns", + "sourceName": "Patterns", + "preferNameDirectory": true +} \ No newline at end of file diff --git a/equinox-patterns/Domain.Tests/Domain.Tests.fsproj b/equinox-patterns/Domain.Tests/Domain.Tests.fsproj new file mode 100644 index 000000000..873b1ab64 --- /dev/null +++ b/equinox-patterns/Domain.Tests/Domain.Tests.fsproj @@ -0,0 +1,29 @@ + + + + netcoreapp3.1 + 5 + false + Library + + + + + + + + + + + + + + + + + + + + + + diff --git a/equinox-patterns/Domain.Tests/Infrastructure.fs b/equinox-patterns/Domain.Tests/Infrastructure.fs new file mode 100644 index 000000000..e5070c2f9 --- /dev/null +++ b/equinox-patterns/Domain.Tests/Infrastructure.fs @@ -0,0 +1,34 @@ +[] +module Patterns.Domain.Tests.Infrastructure + +open FSharp.UMX +open System +open FsCheck +open Patterns.Domain + +(* Generic FsCheck helpers *) + +let (|Id|) (x : Guid) = x.ToString "N" |> UMX.tag +let inline mkId () = Guid.NewGuid() |> (|Id|) +let (|Ids|) (xs : Guid[]) = xs |> Array.map (|Id|) + +type DomainArbs() = + + static member Item : Arbitrary = Arb.fromGen <| gen { + let! r = Arb.Default.Derive() |> Arb.toGen + let id = mkId () // TODO why doesnt `let (Id id) = Arb.generate` generate fresh every time? + return { r with id = id } + } + +type DomainProperty() = inherit FsCheck.Xunit.PropertyAttribute(Arbitrary=[|typeof|], QuietOnSuccess=true) + +/// Inspired by AutoFixture.XUnit's AutoDataAttribute - generating test data without the full Property Based Tests experience +/// By using this instead of Property, the developer has +/// a) asserted by using this property instead of [] +/// b) indirectly validated by running the tests frequently locally in DEBUG mode +/// that running the test multiple times is not a useful thing to do +#if !DEBUG +type AutoDataAttribute() = inherit FsCheck.Xunit.PropertyAttribute(Arbitrary=[|typeof|], MaxTest=1, QuietOnSuccess=true) +#else +type AutoDataAttribute() = inherit FsCheck.Xunit.PropertyAttribute(Arbitrary=[|typeof|], MaxTest=5, QuietOnSuccess=true) +#endif \ No newline at end of file diff --git a/equinox-patterns/Domain.Tests/ItemIngesterTests.fs b/equinox-patterns/Domain.Tests/ItemIngesterTests.fs new file mode 100644 index 000000000..ea6698f49 --- /dev/null +++ b/equinox-patterns/Domain.Tests/ItemIngesterTests.fs @@ -0,0 +1,67 @@ +module Patterns.Domain.Tests.ItemIngesterTests + +open Patterns.Domain +open Patterns.Domain.ItemIngester +open FsCheck.Xunit +open FSharp.UMX +open Swensen.Unquote + +let linger, lookBackLimit, maxPickTicketsPerBatch = System.TimeSpan.FromMilliseconds 1., 2, 5 + +let createSut store trancheId = + // While we use ~ 200ms when hitting Cosmos, there's no value in doing so in the context of these property based tests + let service = MemoryStore.Create(store, linger=linger, maxItemsPerEpoch=maxPickTicketsPerBatch, lookBackLimit=lookBackLimit) + service.ForTranche trancheId + +let [] properties shouldInitialize shouldUseSameSut (Id trancheId) initialItems items = + let store = Equinox.MemoryStore.VolatileStore() + Async.RunSynchronously <| async { + // Initialize with some items + let initialSut = createSut store trancheId + if shouldInitialize then do! initialSut.Initialize() + let! initialResult = initialSut.IngestMany(initialItems) + let initialExpected = initialItems |> Seq.map ItemEpoch.itemId |> Array.ofSeq + test <@ set initialExpected = set initialResult @> + + // Add some extra + let sut = if shouldUseSameSut then initialSut else createSut store trancheId + if shouldInitialize then do! sut.Initialize() + let! result = sut.IngestMany items + let expected = items |> Seq.map ItemEpoch.itemId |> Seq.except initialExpected |> Seq.distinct + test <@ set expected = set result @> + + // Add the same stuff for a different tranche; the data should be completely independent from an ingestion perspective + let differentTranche = %(sprintf "%s2" %trancheId) + let differentSutSameStore = createSut store differentTranche + let! independentResult = differentSutSameStore.IngestMany(Array.append initialItems items) + test <@ set initialResult + set result = set independentResult @> + } + +let [] ``lookBack is limited`` (Id trancheId) genItem = + let store = Equinox.MemoryStore.VolatileStore() + Async.RunSynchronously <| async { + // Initialize with more items than the lookBack accommodates + let initialSut = createSut store trancheId + let itemCount = + // Fill up lookBackLimit batches, and another one as batch 0 that we will not look include in the load + (lookBackLimit+1) * maxPickTicketsPerBatch + // Add one more so we end up with an active batchId = lookBackLimit + + 1 + let items = Array.init itemCount (fun _ -> genItem () ) + test <@ Array.distinct items = items @> + let batch0 = Array.take maxPickTicketsPerBatch items + let batchesInLookBack = Array.skip maxPickTicketsPerBatch items + let! b0Added = initialSut.IngestMany batch0 + let b0Added = Array.ofSeq b0Added + test <@ maxPickTicketsPerBatch = Array.length b0Added @> + let! batchesInLookBackAdded = initialSut.IngestMany batchesInLookBack + test <@ itemCount = Set.count (set b0Added + set batchesInLookBackAdded) @> + + // Now try to add the same items - the first batch worth should not be deduplicated + let sut = createSut store trancheId + let! result = sut.IngestMany items + let result = Array.ofSeq result + test <@ itemCount = itemCount + && result.Length = maxPickTicketsPerBatch + && set result = set b0Added @> + } diff --git a/equinox-patterns/Domain.Tests/PeriodsCarryingForward.fs b/equinox-patterns/Domain.Tests/PeriodsCarryingForward.fs new file mode 100644 index 000000000..2921bbe26 --- /dev/null +++ b/equinox-patterns/Domain.Tests/PeriodsCarryingForward.fs @@ -0,0 +1,33 @@ +/// Integration suite for `Period` +module Patterns.Domain.Tests.PeriodsCarryingForward + +open Patterns.Domain +open Patterns.Domain.Period +open FSharp.UMX +open Swensen.Unquote +open Xunit + +[] +let ``Happy path`` () = + let store = Equinox.MemoryStore.VolatileStore() + let service = MemoryStore.create store + let decide items _state = + let apply = Array.truncate 2 items + let overflow = Array.skip apply.Length items + (match overflow with [||] -> None | xs -> Some xs), // Apply max of two events + (), // result + [Events.Added {items = apply }] + let add period events = service.Transact(PeriodId.parse period, decide, events) |> Async.RunSynchronously + let read period = service.Read(PeriodId.parse period) |> Async.RunSynchronously + add 0 [| %"a"; %"b" |] + test <@ Fold.Open [| %"a"; %"b"|] = read 0 @> + add 1 [| %"c"; %"d" |] + test <@ Fold.Closed ([| %"a"; %"b"|], [| %"a"; %"b"|]) = read 0 @> + test <@ Fold.Open [| %"a"; %"b"; %"c"; %"d" |] = read 1 @> + let items period = read period |> Fold.(|Items|) + add 1 [| %"e"; %"f"; %"g" |] // >2 items, therefore triggers an overflow + test <@ [| %"a"; %"b"; %"c"; %"d"; %"e"; %"f" |] = items 1 @> + test <@ [| %"a"; %"b"; %"c"; %"d"; %"e"; %"f"; %"g" |] = items 2 @> + test <@ Fold.Initial = read 3 @> + add 3 [| %"h" |] + test <@ Fold.Open [| %"a"; %"b"; %"c"; %"d"; %"e"; %"f"; %"g"; %"h" |] = read 3 @> diff --git a/equinox-patterns/Domain/Domain.fsproj b/equinox-patterns/Domain/Domain.fsproj new file mode 100644 index 000000000..ab72c52a5 --- /dev/null +++ b/equinox-patterns/Domain/Domain.fsproj @@ -0,0 +1,23 @@ + + + + + netcoreapp3.1 + + + + + + + + + + + + + + + + + + diff --git a/equinox-patterns/Domain/Infrastructure.fs b/equinox-patterns/Domain/Infrastructure.fs new file mode 100644 index 000000000..3ef7d2da2 --- /dev/null +++ b/equinox-patterns/Domain/Infrastructure.fs @@ -0,0 +1,43 @@ +[] +module Patterns.Domain.Infrastructure + +module Equinox = + + let createDecider stream = + Equinox.Decider(Serilog.Log.Logger, stream, maxAttempts = 3) + +/// Buffers events accumulated from a series of decisions while evolving the presented `state` to reflect said proposed `Events` +type Accumulator<'event, 'state>(originState, fold : 'state -> 'event seq -> 'state) = + let pendingEvents = ResizeArray() + let mutable state = originState + + let apply (events : 'event seq) = + pendingEvents.AddRange events + state <- fold state events + + /// Run a decision function, buffering and applying any Events yielded + member _.Transact decide = + let r, events = decide state + apply events + r + + /// Run a decision function that does not yield a result +// member x.Transact decide = +// x.Transact(fun state -> (), decide state) + + /// Run an Async decision function, buffering and applying any Events yielded + member _.TransactAsync decide = async { + let! r, events = decide state + apply events + return r } + + /// Run an Async decision function that does not yield a result + member x.TransactAsync decide = async { + let! events = decide state + apply events } + +// /// Projects from the present state including accumulated events +// member _.Query f = f state + + /// Accumulated events based on the Decisions applied to date + member _.Events = List.ofSeq pendingEvents diff --git a/equinox-patterns/Domain/ItemEpoch.fs b/equinox-patterns/Domain/ItemEpoch.fs new file mode 100644 index 000000000..1d8ab6b1f --- /dev/null +++ b/equinox-patterns/Domain/ItemEpoch.fs @@ -0,0 +1,146 @@ +module Patterns.Domain.ItemEpoch + +let [] Category = "ItemEpoch" +let streamName (trancheId : ItemTrancheId, epochId : ItemEpochId) = FsCodec.StreamName.compose Category [ItemTrancheId.toString trancheId; ItemEpochId.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 Ingested = { items : Item[] } + and Item = { id : ItemId; payload : string } + type Snapshotted = { ids : ItemId[]; closed : bool } + type Event = + | Ingested of Ingested + | Closed + | Snapshotted of Snapshotted + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +let itemId (x : Events.Item) : ItemId = x.id +let (|ItemIds|) : Events.Item[] -> ItemId[] = Array.map itemId + +module Fold = + + type State = ItemId[] * bool + let initial = [||], false + let evolve (ids, closed) = function + | Events.Ingested { items = ItemIds ingestedIds } -> (Array.append ids ingestedIds, closed) + | Events.Closed -> (ids, true) + | Events.Snapshotted e -> (e.ids, e.closed) + let fold : State -> Events.Event seq -> State = Seq.fold evolve + + let isOrigin = function Events.Snapshotted _ -> true | _ -> false + let toSnapshot (ids, closed) = Events.Snapshotted { ids = ids; closed = closed } + +let notAlreadyIn (ids : ItemId seq) = + let ids = System.Collections.Generic.HashSet ids + fun (x : Events.Item) -> (not << ids.Contains) x.id + +type Result = { accepted : ItemId[]; residual : Events.Item[]; content : ItemId[]; closed : bool } + +let decide capacity candidates (currentIds, closed as state) = + match closed, candidates |> Array.filter (notAlreadyIn currentIds) with + | true, freshCandidates -> { accepted = [||]; residual = freshCandidates; content = currentIds; closed = closed }, [] + | false, [||] -> { accepted = [||]; residual = [||]; content = currentIds; closed = closed }, [] + | false, freshItems -> + // NOTE we in some cases end up triggering splitting of a request (or set of requests coalesced in the AsyncBatchingGate) + // In some cases it might be better to be a little tolerant and not be rigid about limiting things as + // - snapshots should compress well (no major incremental cost for a few more items) + // - its always good to avoid a second store roundtrip + // - splitting a batched write into multiple writes with multiple events misrepresents the facts i.e. we did + // not have 10 items 2s ago and 3 just now - we had 13 2s ago + let capacityNow = capacity freshItems currentIds + let acceptingCount = min capacityNow freshItems.Length + let closing = acceptingCount = capacityNow + let ItemIds addedItemIds as itemsIngested, residualItems = Array.splitAt acceptingCount freshItems + let events = + [ if (not << Array.isEmpty) itemsIngested then yield Events.Ingested { items = itemsIngested } + if closing then yield Events.Closed ] + let currentIds, closed = Fold.fold state events + { accepted = addedItemIds; residual = residualItems; content = currentIds; closed = closed }, events + +/// Used by the Ingester to manages ingestion of items into the epoch, i.e. the Write side +type IngestionService internal + ( capacity : Events.Item[] -> ItemId[] -> int, // let outer layers decide how many candidate items to accept given supplied Epoch state + resolve : ItemTrancheId * ItemEpochId -> Equinox.Decider) = + + /// Obtains a complete list of all the items in the specified trancheId/epochId + /// NOTE this exposes stale cached state - which is OK for the ingestion use case it needs to deal with such stale + /// state as part of handling normal race conditions, but should not be passed to anyone that's going to infer the + /// current state or version of the Epoch + member _.ReadIds(trancheId, epochId) : Async = + let decider = resolve (trancheId, epochId) + decider.Query fst + + /// Ingest the supplied items. Yields relevant elements of the post-state to enable generation of stats + /// and facilitate deduplication of incoming items in order to avoid null store round-trips where possible + member _.Ingest(trancheId, epochId, items) : Async = + let decider = resolve (trancheId, epochId) + /// NOTE decider which will initially transact against potentially stale cached state, which will trigger a + /// resync if another writer has gotten in before us. This is a conscious decision in this instance; the bulk + /// of writes are presumed to be coming from within this same process + decider.Transact(decide capacity items) + +let private create capacity resolveStream = + let resolve = streamName >> resolveStream (Some Equinox.AllowStale) >> Equinox.createDecider + IngestionService(capacity, resolve) + +module MemoryStore = + + let create capacity store = + let cat = Equinox.MemoryStore.MemoryStoreCategory(store, Events.codec, Fold.fold, Fold.initial) + let resolveStream opt sn = cat.Resolve(sn, ?option = opt) + create capacity resolveStream + +module Cosmos = + + open Equinox.CosmosStore + + let accessStrategy = AccessStrategy.Snapshot (Fold.isOrigin, Fold.toSnapshot) + let create capacity (context, cache) = + let cacheStrategy = CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let cat = CosmosStoreCategory(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolveStream opt sn = cat.Resolve(sn, ?option = opt) + create capacity resolveStream + +/// Custom Fold and caching logic compared to the IngesterService +/// - When reading, we want the full Items +/// - Caching only for one minute +/// - There's no value in using the snapshot as it does not have the full state +module Reader = + + type ReadState = Events.Item[] * bool + let initial = [||], false + let evolve (es, closed as state) = function + | Events.Ingested e -> Array.append es e.items, closed + | Events.Closed -> (es, true) + | Events.Snapshotted _ -> state // there's nothing useful in the snapshot for us to take + let fold : ReadState -> Events.Event seq -> ReadState = Seq.fold evolve + + type Service internal (resolve : ItemTrancheId * ItemEpochId -> Equinox.Decider) = + + /// Returns all the items currently held in the stream + member _.Read(trancheId, epochId) : Async = + let decider = resolve (trancheId, epochId) + decider.Query id + + let private create resolveStream = + let resolve = streamName >> resolveStream >> Equinox.createDecider + Service resolve + + module MemoryStore = + + let create store = + let cat = Equinox.MemoryStore.MemoryStoreCategory(store, Events.codec, fold, initial) + create cat.Resolve + + module Cosmos = + + open Equinox.CosmosStore + + let accessStrategy = AccessStrategy.Unoptimized + let create (context, cache) = + let cacheStrategy = CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 1.) + let cat = CosmosStoreCategory(context, Events.codec, fold, initial, cacheStrategy, accessStrategy) + create cat.Resolve diff --git a/equinox-patterns/Domain/ItemIngester.fs b/equinox-patterns/Domain/ItemIngester.fs new file mode 100644 index 000000000..bafd04941 --- /dev/null +++ b/equinox-patterns/Domain/ItemIngester.fs @@ -0,0 +1,148 @@ +/// Application Service that controls the ingestion of Items into a chain of `Epoch` streams per Tranche +/// - the `Series` aggregate maintains a pointer to the current Epoch for each Tranche +/// - as `Epoch`s complete (have `Closed` events logged), we update the `active` Epoch in the Series to reference the new one +module Patterns.Domain.ItemIngester + +open Equinox.Core +open FSharp.UMX + +type internal IdsCache() = + // NOTE: Bounded only by relatively low number of physical items IRL + let all = System.Collections.Generic.HashSet() + static member Create init = let x = IdsCache() in x.Add init; x + member _.Add ids = all.UnionWith ids + member _.Contains id = all.Contains id + +/// Maintains active EpochId in a thread-safe manner while ingesting items into the chain of `epochs` indexed by the `series` +/// Prior to first add, reads `lookBack` batches to seed the cache, in order to minimize the number of duplicated items we ingest +type ServiceForTranche internal (log : Serilog.ILogger, trancheId, epochs : ItemEpoch.IngestionService, series : ItemSeries.Service, lookBack, linger) = + + // Maintains what we believe to be the currently open EpochId + // NOTE not valid/initialized until invocation of `previousIds.AwaitValue()` has completed + let uninitializedSentinel = %ItemEpochId.unknown + let mutable activeEpochId_ = uninitializedSentinel + // NOTE see above - must not be called prior to previousIds.AwaitValue() + let effectiveEpochId () = if activeEpochId_ = uninitializedSentinel then ItemEpochId.initial else %activeEpochId_ + + // establish the pre-existing items from which the previousIds cache will be seeded + let loadPreviousEpochs loadDop : Async = async { + match! series.TryReadIngestionEpochId trancheId with + | None -> + log.Information("No starting epoch registered for {trancheId}", trancheId) + return Array.empty + | Some startingId -> + log.Information("Walking back from {trancheId}/{epochId}", trancheId, startingId) + activeEpochId_ <- %startingId + let readEpoch epochId = + log.Information("Reading {trancheId}/{epochId}", trancheId, epochId) + epochs.ReadIds(trancheId, epochId) + return! Async.Parallel(seq { for epochId in (max 0 (%startingId - lookBack)) .. (%startingId - 1) -> readEpoch %epochId }, loadDop) } + + // ItemIds cache - used to maintain a list of items that have already been ingested in order to avoid db round-trips + let previousIds : AsyncCacheCell = AsyncCacheCell <| async { + let! batches = loadPreviousEpochs 4 + return IdsCache.Create(Seq.concat batches) } + + let tryIngest items = async { + let! previousIds = previousIds.AwaitValue() + let firstEpochId = effectiveEpochId () + + let rec aux epochId ingestedItems items = async { + let dup, freshItems = items |> Array.partition (ItemEpoch.itemId >> previousIds.Contains) + let fullCount = Array.length items + let dropping = fullCount - Array.length freshItems + if dropping <> 0 then + log.Information("Ignoring {count}/{fullCount} duplicate ids: {ids} for {trancheId}/{epochId}", dropping, fullCount, dup, trancheId, epochId) + if Array.isEmpty freshItems then + return ingestedItems + else + let! res = epochs.Ingest(trancheId, epochId, freshItems) + let ingestedItemIds = Array.append ingestedItems res.accepted + if (not << Array.isEmpty) res.accepted then + log.Information("Added {count} items to {trancheId}/{epochId}", res.accepted.Length, trancheId, epochId) + // The adding is potentially redundant; we don't care + previousIds.Add res.content + // Any writer noticing we've moved to a new Epoch shares the burden of marking it active in the Series + if not res.closed && activeEpochId_ < ItemEpochId.value epochId then + log.Information("Marking {trancheId}/{epochId} active", trancheId, epochId) + do! series.MarkIngestionEpochId(trancheId, epochId) + System.Threading.Interlocked.CompareExchange(&activeEpochId_, %epochId, activeEpochId_) |> ignore + match res.residual with + | [||] -> return ingestedItemIds + | remaining -> return! aux (ItemEpochId.next epochId) ingestedItemIds remaining } + return! aux firstEpochId [||] (Array.concat items) + } + + /// Within the processing for a given Tranche, we have a Scheduler running N streams concurrently + /// If each thread works in isolation, they'll conflict with each other as they feed the ticket into the batch in epochs.Ingest + /// Instead, we enable concurrent requests to coalesce by having requests converge in this AsyncBatchingGate + /// This has the following critical effects: + /// - Traffic to CosmosDB is naturally constrained to a single flight in progress + /// (BatchingGate does not admit next batch until current has succeeded or throws) + /// - RU consumption for writing to the batch is optimized (1 write inserting 1 event document vs N writers writing N) + /// - Peak throughput is more consistent as latency is not impacted by the combination of having to: + /// a) back-off, re-read and retry if there's a concurrent write Optimistic Concurrency Check failure when writing the stream + /// b) enter a prolonged period of retries if multiple concurrent writes trigger rate limiting and 429s from CosmosDB + /// c) readers will less frequently encounter sustained 429s on the batch + let batchedIngest = AsyncBatchingGate(tryIngest, linger) + + /// Optional API enabling one to preemptively initialize the cache based on the walk back of previous epochs to load + /// ItemIds for deduplication (which would otherwise add to the latency of the first incoming request(s)). It's safe + /// to ignore the result of this call (i.e., trigger it via Async.Start) as the usual semantics of + /// AsyncBatchingGate pertain:- everyone gets to see the exception from the batch being processed, including + /// that initialization, and a failed attempt is never cached) + member _.Initialize() = previousIds.AwaitValue() |> Async.Ignore + + /// Attempts to feed the items into the sequence of epochs. + /// Returns the subset that actually got fed in this time around. + member _.IngestMany(items : ItemEpoch.Events.Item[]) : Async = async { + let! results = batchedIngest.Execute items + return System.Linq.Enumerable.Intersect(Seq.map ItemEpoch.itemId items, results) + } + + /// Attempts to feed the item into the sequence of batches. + /// Returns true if the item actually got included into an Epoch this time around. + member _.TryIngest(item : ItemEpoch.Events.Item) : Async = async { + let! result = batchedIngest.Execute(Array.singleton item) + return result |> Array.contains (ItemEpoch.itemId item) + } + +let private createServiceForTranche (epochs, lookBackLimit) series linger trancheId = + let log = Serilog.Log.ForContext() + ServiceForTranche(log, trancheId, epochs, series, lookBack=lookBackLimit, linger=linger) + +/// Each ServiceForTranche maintains significant state (set of itemIds looking back through e.g. 100 epochs), which we obv need to cache +type Service internal (createForTranche : ItemTrancheId -> ServiceForTranche) = + + // Its important we don't risk >1 instance https://andrewlock.net/making-getoradd-on-concurrentdictionary-thread-safe-using-lazy/ + // while it would be safe, there would be a risk of incurring the cost of multiple initialization loops + let forTranche = System.Collections.Concurrent.ConcurrentDictionary>() + let build trancheId = lazy createForTranche trancheId + + member _.ForTranche trancheId : ServiceForTranche = + forTranche.GetOrAdd(trancheId, build).Value + +let private maxItemsPerEpoch = 10_000 +let private linger, lookBackLimit = System.TimeSpan.FromMilliseconds 200., 100 + +type MemoryStore() = + + static member Create(store, linger, maxItemsPerEpoch, lookBackLimit) = + let remainingBatchCapacity _candidateItems currentItems = + let l = Array.length currentItems + max 0 (maxItemsPerEpoch - l) + let epochs = ItemEpoch.MemoryStore.create remainingBatchCapacity store + let series = ItemSeries.MemoryStore.create store + let createForTranche = createServiceForTranche (epochs, lookBackLimit) series linger + Service createForTranche + +module Cosmos = + + let create (context, cache) = + let remainingBatchCapacity _candidateItems currentItems = + let l = Array.length currentItems + max 0 (maxItemsPerEpoch - l) + let epochs = ItemEpoch.Cosmos.create remainingBatchCapacity (context, cache) + let series = ItemSeries.Cosmos.create (context, cache) + let createForTranche = createServiceForTranche (epochs, lookBackLimit) series linger + Service createForTranche diff --git a/equinox-patterns/Domain/ItemSeries.fs b/equinox-patterns/Domain/ItemSeries.fs new file mode 100644 index 000000000..a14ee861f --- /dev/null +++ b/equinox-patterns/Domain/ItemSeries.fs @@ -0,0 +1,84 @@ +/// Maintains a pointer into the Epoch chain for each Tranche +/// Allows an Ingester to quickly determine the current Epoch into which it should commence writing +/// As an Epoch is marked `Closed`, the Ingester will mark a new Epoch `Started` on this aggregate +module Patterns.Domain.ItemSeries + +let [] Category = "ItemSeries" +let streamName seriesId = FsCodec.StreamName.create Category (ItemSeriesId.toString seriesId) + +// NB - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +[] +module Events = + + type Event = + | Started of {| trancheId : ItemTrancheId; epochId : ItemEpochId |} + | Snapshotted of {| active : Map |} + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = Map + let initial = Map.empty + let evolve state = function + | Events.Started e -> state |> Map.add e.trancheId e.epochId + | Events.Snapshotted e -> e.active + let fold : State -> Events.Event seq -> State = Seq.fold evolve + + let isOrigin = function Events.Snapshotted _ -> true | _ -> false + let toSnapshot s = Events.Snapshotted {| active = s |} + +let tryFindEpochOfTranche : ItemTrancheId -> Fold.State -> ItemEpochId option = Map.tryFind + +let interpret trancheId epochId (state : Fold.State) = + [if state |> tryFindEpochOfTranche trancheId |> Option.forall (fun cur -> cur < epochId) && epochId >= ItemEpochId.initial then + yield Events.Started {| trancheId = trancheId; epochId = epochId |}] + +type Service internal (resolve_ : Equinox.ResolveOption option -> ItemSeriesId -> Equinox.Decider, ?seriesId) = + + let resolve = resolve_ None + let resolveStale = resolve_ (Some Equinox.AllowStale) + + // For now we have a single global sequence. This provides us an extension point should we ever need to reprocess + // NOTE we use a custom id in order to isolate data for acceptance tests + let seriesId = defaultArg seriesId ItemSeriesId.wellKnownId + + /// Exposes the set of tranches for which data is held + /// Never yields a cached value, in order to ensure a reader can traverse all tranches + member _.Read() : Async = + let decider = resolve seriesId + decider.Query id + + /// Determines the current active epoch for the specified `trancheId` + /// Uses cached values as epoch transitions are rare, and caller needs to deal with the inherent race condition in any case + member _.TryReadIngestionEpochId trancheId : Async = + let decider = resolveStale seriesId + decider.Query(tryFindEpochOfTranche trancheId) + + /// Mark specified `epochId` as live for the purposes of ingesting + /// Writers are expected to react to having writes to an epoch denied (due to it being Closed) by anointing a successor via this + member _.MarkIngestionEpochId(trancheId, epochId) : Async = + let decider = resolveStale seriesId + decider.Transact(interpret trancheId epochId) + +let private create seriesOverride resolveStream = + let resolve opt = streamName >> resolveStream opt >> Equinox.createDecider + Service(resolve, ?seriesId = seriesOverride) + +module MemoryStore = + + let create store = + let cat = Equinox.MemoryStore.MemoryStoreCategory(store, Events.codec, Fold.fold, Fold.initial) + let resolveStream opt sn = cat.Resolve(sn, ?option = opt) + create None resolveStream + +module Cosmos = + + open Equinox.CosmosStore + + let accessStrategy = AccessStrategy.Snapshot (Fold.isOrigin, Fold.toSnapshot) + let create (context, cache) = + let cacheStrategy = CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let cat = CosmosStoreCategory(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolveStream opt sn = cat.Resolve(sn, ?option = opt) + create None resolveStream diff --git a/equinox-patterns/Domain/Period.fs b/equinox-patterns/Domain/Period.fs new file mode 100644 index 000000000..fba227132 --- /dev/null +++ b/equinox-patterns/Domain/Period.fs @@ -0,0 +1,163 @@ +/// Illustrates a high level approach to how one might manage a chained series of periods which can be logically Closed +/// All write attempts adhere to a common protocol to effect this semantic:- +/// a) all preceding periods must be closed, idempotently i) computing and persisting a balance OR ii) honoring a previously recorded one +/// b) the decision is processed within the target period (which may be either Open, or being opened as part of this flow) +/// c) if appropriate, the target period may be closed as part of the same decision flow if `decideCarryForward` yields Some +module Patterns.Domain.Period + +let [] Category = "Period" +let streamName periodId = FsCodec.StreamName.create Category (PeriodId.toString periodId) + +// NOTE - these types and the union case names reflect the actual storage formats and hence need to be versioned with care +module Events = + + type ItemIds = { items : ItemId[] } + type Balance = ItemIds + type Event = + | BroughtForward of Balance + | Added of ItemIds + | CarriedForward of Balance + interface TypeShape.UnionContract.IUnionContract + let codec = FsCodec.NewtonsoftJson.Codec.Create() + +module Fold = + + type State = + | Initial + | Open of items : OpenState + | Closed of items : ItemId[] * carryingForward : ItemId[] + and OpenState = ItemId[] + let initial : State = Initial + let (|Items|) = function Initial -> [||] | Open i | Closed (i, _) -> i + open Events + let evolve (Items items) = function + | BroughtForward e + | Added e -> Open (Array.append items e.items) + | CarriedForward e -> Closed (items, e.items) + let fold = Seq.fold evolve + + /// Handles one-time opening of the Period, if applicable + let maybeOpen (getIncomingBalance : unit -> Async) state = async { + match state with + | Initial -> let! balance = getIncomingBalance () + return [BroughtForward balance] + | Open _ + | Closed _ -> return [] } + + /// Handles attempting to apply the request to the stream (assuming it's not already closed) + /// The `decide` function can signal a need to close and/or split the request by emitting it as the residual + let tryIngest (decide : 'req -> State -> 'req * 'result * Event list) req = function + | Initial -> failwith "Invalid tryIngest; stream not Open" + | Open _ as s -> let residual, result, events = decide req s + (residual, Some result), events + | Closed _ -> (req, None), [] + + /// Yields or computes the Balance to be Carried forward and/or application of the event representing that decision + let maybeClose (decideCarryForward : 'residual -> OpenState -> Async) residual state = async { + match state with + | Initial -> return failwith "Invalid maybeClose; stream not Open" + | Open s -> let! cf = decideCarryForward residual s + let events = cf |> Option.map CarriedForward |> Option.toList + return (residual, cf), events + | Closed (_, b) -> return (residual, Some { items = b }), [] } + +[] +type Rules<'request, 'result> = + { getIncomingBalance: unit -> Async + decideIngestion : 'request -> Fold.State -> 'request * 'result * Events.Event list + decideCarryForward: 'request -> Fold.OpenState -> Async } + +/// The result of the overall ingestion, consisting of +type Result<'request, 'result> = + { /// residual of the request, in the event where it was not possible to ingest it completely + residual : 'request + /// The result of the decision (assuming processing took place) + result : 'result option + /// balance being carried forward in the event that the successor period has yet to have the BroughtForward event generated + carryForward : Events.Balance option } + +/// Decision function ensuring the high level rules of an Period are adhered to viz. +/// 1. Streams must open with a BroughtForward event (obtained via Rules.getIncomingBalance if this is an uninitialized Period) +/// 2. (If the Period has not closed) Rules.decide gets to map the request to events and a residual +/// 3. Rules.decideCarryForward may trigger the closing of the Period based on the residual and/or the State by emitting Some balance +let decideIngestWithCarryForward rules req s : Async * Events.Event list> = async { + let acc = Accumulator(s, Fold.fold) + do! acc.TransactAsync(Fold.maybeOpen rules.getIncomingBalance) + let residual, result = acc.Transact(Fold.tryIngest rules.decideIngestion req) + let! residual, carryForward = acc.TransactAsync(Fold.maybeClose rules.decideCarryForward residual) + return { residual = residual; result = result; carryForward = carryForward }, acc.Events +} + +/// Manages Application of Requests to the Period's stream, including closing preceding periods as appropriate +type Service internal (resolve : PeriodId -> Equinox.Decider) = + + let calcBalance state = + let createEventsBalance items : Events.Balance = { items = items } + async { return createEventsBalance state } + let genBalance state = async { let! bal = calcBalance state in return Some bal } + + /// Walks back as far as necessary to ensure any preceding Periods that are not yet Closed are, then closes the target if necessary + /// Yields the accumulated balance to be carried forward into the next period + let rec close periodId : Async = + let rules : Rules = + { getIncomingBalance = fun () -> close periodId + decideIngestion = fun () _state -> (), (), [] + decideCarryForward = fun () -> genBalance } // always close + let decider = resolve periodId + decider.TransactEx((fun c -> decideIngestWithCarryForward rules () c.State), fun r _c -> Option.get r.carryForward) + + /// Runs the decision function on the specified Period, closing and bringing forward balances from preceding Periods if necessary + let tryTransact periodId getIncoming (decide : 'request -> Fold.State -> 'request * 'result * Events.Event list) request shouldClose : Async> = + let rules : Rules<'request, 'result> = + { getIncomingBalance = getIncoming + decideIngestion = fun request state -> let residual, result, events = decide request state in residual, result, events + decideCarryForward = fun res state -> async { if shouldClose res then return! genBalance state else return None } } // also close, if we should + let decider = resolve periodId + decider.TransactEx((fun c -> decideIngestWithCarryForward rules request c.State), fun r _c -> r) + + /// Runs the decision function on the specified Period, closing and bringing forward balances from preceding Periods if necessary + /// Processing completes when `decide` yields None for the residual of the 'request + member _.Transact(periodId, decide : 'request -> Fold.State -> 'request option * 'result * Events.Event list, request : 'request) : Async<'result> = + let rec aux periodId getIncoming req = async { + let decide req state = decide (Option.get req) state + match! tryTransact periodId getIncoming decide req Option.isSome with + | { residual = None; result = Some r } -> return r + | { residual = r; carryForward = cf } -> return! aux (PeriodId.next periodId) (fun () -> async { return Option.get cf }) r } + let getIncoming () = + match PeriodId.tryPrev periodId with + | None -> calcBalance [||] + | Some prevPeriodId -> close prevPeriodId + aux periodId getIncoming (Some request) + + /// Exposes the full state to a reader (which is appropriate for a demo but is an anti-pattern in the general case) + /// NOTE the resolve function below uses Equinox.AllowStale - you may want to remove that option of you write a read + /// function like this for real (i.e. if the active Period is constantly being appended to in another process/ + /// machine) then this read function will continually serve the same value, as it has been instructed not to + /// make a store round-trip + member _.Read periodId = + let decider = resolve periodId + decider.Query id + +let private create resolveStream = + let resolve = streamName >> resolveStream (Some Equinox.AllowStale) >> Equinox.createDecider + Service resolve + +module MemoryStore = + + let create store = + let cat = Equinox.MemoryStore.MemoryStoreCategory(store, Events.codec, Fold.fold, Fold.initial) + let resolveStream opt sn = cat.Resolve(sn, ?option = opt) + create resolveStream + +module Cosmos = + + open Equinox.CosmosStore + + // Not using snapshots, on the basis that the writes are all coming from this process, so the cache will be sufficient + // to make reads cheap enough, with the benefit of writes being cheaper as you're not paying to maintain the snapshot + let accessStrategy = AccessStrategy.Unoptimized + let create (context, cache) = + let cacheStrategy = CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.) + let cat = CosmosStoreCategory(context, Events.codec, Fold.fold, Fold.initial, cacheStrategy, accessStrategy) + let resolveStream opt sn = cat.Resolve(sn, ?option = opt) + create resolveStream diff --git a/equinox-patterns/Domain/Types.fs b/equinox-patterns/Domain/Types.fs new file mode 100644 index 000000000..c9d861c84 --- /dev/null +++ b/equinox-patterns/Domain/Types.fs @@ -0,0 +1,56 @@ +namespace Patterns.Domain + +open FSharp.UMX + +/// Identifies a single period within a temporally linked chain of periods +/// Each Period commences with a Balance `BroughtForward` based on what the predecessor Period +/// has decided should be `CarriedForward` +/// TODO prefix the name with a relevant Domain term +type PeriodId = int +and [] periodId +module PeriodId = + + let parse (value : int) : PeriodId = %value + let tryPrev (value : PeriodId) : PeriodId option = match %value with 0 -> None | x -> Some %(x - 1) + let next (value : PeriodId) : PeriodId = %(%value + 1) + let toString (value : PeriodId) : string = string %value + +/// TODO replace the terms `tranche`, `trancheId` and type `TrancheId` with domain term that represents the nature of the separation +/// i.e. it might be a TenantId or a FulfilmentCenterId +type ItemTrancheId = string +and [] itemTrancheId +module ItemTrancheId = + + let parse (value : string) : ItemTrancheId = %value + let toString (value : ItemTrancheId) : string = %value + +/// Identifies an Epoch that holds a set of Items (within a Tranche) +/// TODO replace `Item` with a Domain term referencing the specific element being managed +type ItemEpochId = int +and [] itemEpochId +module ItemEpochId = + + let unknown = -1 + let initial = 0 + let next (value : ItemEpochId) : ItemEpochId = % (%value + 1) + let value (value : ItemEpochId) : int = %value + let toString (value : ItemEpochId) : string = string %value + +/// Identifies an Item stored within an Epoch +/// TODO replace `Item` with a Domain term referencing the specific element being managed + +type ItemId = string +and [] itemId +module ItemId = + + let parse (value : string) : ItemId = %value + let toString (value : ItemId) : string = %value + +/// Identifies a group of chained Epochs (pe Tranche within the series) +/// TODO replace `Item` with a Domain term referencing the specific element being managed +type [] itemSeriesId +type ItemSeriesId = string +module ItemSeriesId = + + let wellKnownId : ItemSeriesId = % "0" + let toString (value : ItemSeriesId) : string = %value diff --git a/equinox-patterns/Patterns.sln b/equinox-patterns/Patterns.sln new file mode 100644 index 000000000..43e15bedb --- /dev/null +++ b/equinox-patterns/Patterns.sln @@ -0,0 +1,48 @@ + +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", "{CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}" +EndProject +Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Domain.Tests", "Domain.Tests\Domain.Tests.fsproj", "{136AA8F5-A808-46F0-9246-3A3271F3539E}" +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 + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Debug|Any CPU.Build.0 = Debug|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Debug|x64.ActiveCfg = Debug|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Debug|x64.Build.0 = Debug|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Debug|x86.ActiveCfg = Debug|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Debug|x86.Build.0 = Debug|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Release|Any CPU.ActiveCfg = Release|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Release|Any CPU.Build.0 = Release|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Release|x64.ActiveCfg = Release|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Release|x64.Build.0 = Release|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Release|x86.ActiveCfg = Release|Any CPU + {136AA8F5-A808-46F0-9246-3A3271F3539E}.Release|x86.Build.0 = Release|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Debug|Any CPU.Build.0 = Debug|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Debug|x64.ActiveCfg = Debug|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Debug|x64.Build.0 = Debug|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Debug|x86.ActiveCfg = Debug|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Debug|x86.Build.0 = Debug|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Release|Any CPU.ActiveCfg = Release|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Release|Any CPU.Build.0 = Release|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Release|x64.ActiveCfg = Release|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Release|x64.Build.0 = Release|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Release|x86.ActiveCfg = Release|Any CPU + {CD2BEF24-B21D-4454-9769-8BD7C23A7FB6}.Release|x86.Build.0 = Release|Any CPU + EndGlobalSection +EndGlobal diff --git a/equinox-patterns/README.md b/equinox-patterns/README.md new file mode 100644 index 000000000..dc7dfea03 --- /dev/null +++ b/equinox-patterns/README.md @@ -0,0 +1,26 @@ +# Equinox Patterns + +This template provides a grab-bag of example Deciders, including illustrations of the following generic techniques: + +## `Period` with Rolling Balance carried forward + +Consists of: + +- `Period`: a Decider that manages ingestion into a chain of periods with each one + a) Carrying Forward a Balance from its immediate predecessor `Period`; then + b) being open for transactions for a period of time; ultimately + c) carrying forward a closing balance to its successor Period + +Notes: +- A given `Period` can thus be read without any need to load any preceding periods, as by definition, all relevant information has been `CarriedForward` +- Any `Period` is guaranteed to have been initialized; thus the preceding epochs can safely be archived the moment the first event has been written to a given `Period` + +## Epochs/Series with deduplication + +Consists of: + +- `Epoch`: A given atomic set of items that have been ingested. May be closed at an arbitrary point in time by any writer. +- `Series`: Records the identifier of the current active Epoch of the series. + +Notes: +- The deduplication logic is such that the ingestion logic can, given a starting Epoch Id, can 100% guarantee exactly a single copy of the item will be stored in the series as a whole. diff --git a/tests/Equinox.Templates.Tests/DotnetBuild.fs b/tests/Equinox.Templates.Tests/DotnetBuild.fs index cfab3a937..c02b8ed00 100644 --- a/tests/Equinox.Templates.Tests/DotnetBuild.fs +++ b/tests/Equinox.Templates.Tests/DotnetBuild.fs @@ -47,6 +47,7 @@ type DotnetBuild(output : ITestOutputHelper, folder : EquinoxTemplatesFixture) = let [] ``*pending*`` () = run "proProjector" ["--source cosmos"; "--kafka"; "--synthesizeSequence"] #endif + let [] eqxPatterns () = run "eqxPatterns" [] let [] eqxTestbed () = run "eqxTestbed" [] let [] eqxShipping () = run "eqxShipping" [] [)>]