Update to FsCodec 3rc10, Equinox 4rc10, Propulsion 3rc5 (#130)

Author: bartelink

CHANGELOG.md

@@ -10,6 +10,10 @@ The `Unreleased` section name is replaced by the expected version of next releas
 
 ### Added
 ### Changed
+
+- Target `Equinox` v `4.0.0-rc.9`, `Propulsion` v `3.0.0-rc.3` [#128](https://github.com/jet/dotnet-templates/pull/128)
+- `module Config` -> `module Store`/`module Factory` [#128](https://github.com/jet/dotnet-templates/pull/128)
+
 ### Removed 
 ### Fixed
 

equinox-patterns/Domain.Tests/ExactlyOnceIngesterTests.fs

@@ -11,7 +11,7 @@ let linger, maxItemsPerEpoch = System.TimeSpan.FromMilliseconds 1., 5
 
 let createSut =
     // While we use ~ 200ms when hitting Cosmos, there's no value in doing so in the context of these property based tests
-    ListIngester.Config.create_ linger maxItemsPerEpoch
+    ListIngester.Factory.create_ linger maxItemsPerEpoch
 
 type GuidStringN<[<Measure>]'m> = GuidStringN of string<'m> with static member op_Explicit(GuidStringN x) = x
 let (|Ids|) = Array.map (function GuidStringN x -> x)
@@ -28,7 +28,7 @@ type Custom =
 
 let [<Property>] properties shouldUseSameSut (Gap gap) (initialEpochId, NonEmptyArray (Ids initialItems)) (NonEmptyArray (Ids items)) = async {
 
-    let store = Equinox.MemoryStore.VolatileStore() |> Config.Store.Memory
+    let store = Equinox.MemoryStore.VolatileStore() |> Store.Context.Memory
 
     let mutable nextEpochId = initialEpochId
     for _ in 1 .. gap do nextEpochId <- ExactlyOnceIngester.Internal.next nextEpochId

equinox-patterns/Domain.Tests/PeriodsCarryingForward.fs

@@ -9,8 +9,8 @@ open Xunit
 
 [<Fact>]
 let ``Happy path`` () =
-    let store = Equinox.MemoryStore.VolatileStore() |> Config.Store.Memory
-    let service = Config.create store
+    let store = Equinox.MemoryStore.VolatileStore() |> Store.Context.Memory
+    let service = Factory.create store
     let decide items _state =
         let apply = Array.truncate 2 items
         let overflow = Array.skip apply.Length items

equinox-patterns/Domain/Domain.fsproj

@@ -7,7 +7,7 @@
   <ItemGroup>
     <Compile Include="ExactlyOnceIngester.fs" />
     <Compile Include="Infrastructure.fs" />
-    <Compile Include="Config.fs" />
+    <Compile Include="Store.fs" />
     <Compile Include="Types.fs" />
     <Compile Include="Period.fs" />
     <Compile Include="ListEpoch.fs" />
@@ -16,9 +16,9 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Equinox.MemoryStore" Version="4.0.0-rc.7" />
-    <PackageReference Include="Equinox.CosmosStore" Version="4.0.0-rc.7" />
-    <PackageReference Include="FsCodec.SystemTextJson" Version="3.0.0-rc.9" />
+    <PackageReference Include="Equinox.MemoryStore" Version="4.0.0-rc.10" />
+    <PackageReference Include="Equinox.CosmosStore" Version="4.0.0-rc.10" />
+    <PackageReference Include="FsCodec.SystemTextJson" Version="3.0.0-rc.10" />
   </ItemGroup>
 
 </Project>

equinox-patterns/Domain/ExactlyOnceIngester.fs

@@ -7,32 +7,32 @@ module Patterns.Domain.ExactlyOnceIngester
 
 open FSharp.UMX // %
 
-type IngestResult<'req, 'res> = { accepted : 'res[]; closed : bool; residual : 'req[] }
+type IngestResult<'req, 'res> = { accepted: 'res[]; closed: bool; residual: 'req[] }
 
 module Internal =
 
     let unknown<[<Measure>]'m> = UMX.tag -1
-    let next<[<Measure>]'m> (value : int<'m>) = UMX.tag<'m>(UMX.untag value + 1)
+    let next<[<Measure>]'m> (value: int<'m>) = UMX.tag<'m>(UMX.untag value + 1)
 
 /// Ensures any given item is only added to the series exactly once by virtue of the following protocol:
 /// 1. Caller obtains an origin epoch via ActiveIngestionEpochId, storing that alongside the source item
 /// 2. Caller deterministically obtains that origin epoch to supply to Ingest/TryIngest such that retries can be idempotent
 type Service<[<Measure>]'id, 'req, 'res, 'outcome> internal
-    (   log : Serilog.ILogger,
-        readActiveEpoch : unit -> Async<int<'id>>,
-        markActiveEpoch : int<'id> -> Async<unit>,
-        ingest : int<'id> * 'req [] -> Async<IngestResult<'req, 'res>>,
-        mapResults : 'res [] -> 'outcome seq,
+    (   log: Serilog.ILogger,
+        readActiveEpoch: unit -> Async<int<'id>>,
+        markActiveEpoch: int<'id> -> Async<unit>,
+        ingest: int<'id> * 'req [] -> Async<IngestResult<'req, 'res>>,
+        mapResults: 'res [] -> 'outcome seq,
         linger) =
 
-    let uninitializedSentinel : int = %Internal.unknown
+    let uninitializedSentinel: int = %Internal.unknown
     let mutable currentEpochId_ = uninitializedSentinel
     let currentEpochId () = if currentEpochId_ <> uninitializedSentinel then Some %currentEpochId_ else None
 
-    let tryIngest (reqs : (int<'id> * 'req)[][]) =
+    let tryIngest (reqs: (int<'id> * 'req)[][]) =
         let rec aux ingestedItems items = async {
             let epochId = items |> Seq.map fst |> Seq.min
-            let epochItems, futureEpochItems = items |> Array.partition (fun (e, _ : 'req) -> e = epochId)
+            let epochItems, futureEpochItems = items |> Array.partition (fun (e, _: 'req) -> e = epochId)
             let! res = ingest (epochId, Array.map snd epochItems)
             let ingestedItemIds = Array.append ingestedItems res.accepted
             let logLevel =
@@ -56,7 +56,7 @@ type Service<[<Measure>]'id, 'req, 'res, 'outcome> internal
 
     /// In the overall processing using an Ingester, we frequently have a Scheduler running N streams concurrently
     /// If each thread works in isolation, they'll conflict with each other as they feed the Items into the batch in epochs.Ingest
-    /// Instead, we enable concurrent requests to coalesce by having requests converge in this AsyncBatchingGate
+    /// Instead, we enable concurrent requests to coalesce by having requests converge in this Batcher
     /// This has the following critical effects:
     /// - Traffic to CosmosDB is naturally constrained to a single flight in progress
     ///   (BatchingGate does not release next batch for execution until current has succeeded or throws)
@@ -65,11 +65,11 @@ type Service<[<Measure>]'id, 'req, 'res, 'outcome> internal
     ///   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 = Equinox.Core.AsyncBatchingGate(tryIngest, linger)
+    let batchedIngest = Equinox.Core.Batching.Batcher(tryIngest, linger)
 
     /// Run the requests over a chain of epochs.
     /// Returns the subset that actually got handled this time around (i.e., exclusive of items that did not trigger writes per the idempotency rules).
-    member _.IngestMany(originEpoch, reqs) : Async<'outcome seq> = async {
+    member _.IngestMany(originEpoch, reqs): Async<'outcome seq> = async {
         if Array.isEmpty reqs then return Seq.empty else
 
         let! results = batchedIngest.Execute [| for x in reqs -> originEpoch, x |]
@@ -80,7 +80,7 @@ type Service<[<Measure>]'id, 'req, 'res, 'outcome> internal
     /// The fact that any Ingest call for a given item (or set of items) always commences from the same origin is key to exactly once insertion guarantee.
     /// Caller should first store this alongside the item in order to deterministically be able to start from the same origin in idempotent retry cases.
     /// Uses cached values as epoch transitions are rare, and caller needs to deal with the inherent race condition in any case
-    member _.ActiveIngestionEpochId() : Async<int<'id>> =
+    member _.ActiveIngestionEpochId(): Async<int<'id>> =
         match currentEpochId () with
         | Some currentEpochId -> async { return currentEpochId }
         | None -> readActiveEpoch()

equinox-patterns/Domain/Infrastructure.fs

@@ -2,31 +2,31 @@
 module Patterns.Domain.Infrastructure
 
 /// Buffers events accumulated from a series of decisions while evolving the presented `state` to reflect said proposed `Events`
-type Accumulator<'e, 's>(originState : 's, fold : 's -> seq<'e> -> 's) =
+type Accumulator<'e, 's>(originState: 's, fold: 's -> seq<'e> -> 's) =
     let mutable state = originState
     let pendingEvents = ResizeArray<'e>()
-    let (|Apply|) (xs : #seq<'e>) = state <- fold state xs; pendingEvents.AddRange xs
+    let (|Apply|) (xs: #seq<'e>) = state <- fold state xs; pendingEvents.AddRange xs
 
     /// Run an Async interpret function that does not yield a result
-    member _.Transact(interpret : 's -> Async<#seq<'e>>) : Async<unit> = async {
+    member _.Transact(interpret: 's -> Async<#seq<'e>>): Async<unit> = async {
         let! Apply = interpret state in return () }
 
     /// Run an Async decision function, buffering and applying any Events yielded
-    member _.Transact(decide : 's -> Async<'r * #seq<'e>>) : Async<'r> = async {
+    member _.Transact(decide: 's -> Async<'r * #seq<'e>>): Async<'r> = async {
         let! r, Apply = decide state in return r }
 
     /// Run a decision function, buffering and applying any Events yielded
-    member _.Transact(decide : 's -> 'r * #seq<'e>) : 'r =
+    member _.Transact(decide: 's -> 'r * #seq<'e>): 'r =
         let r, Apply = decide state in r
 
     /// Accumulated events based on the Decisions applied to date
-    member _.Events : 'e list =
+    member _.Events: 'e list =
         List.ofSeq pendingEvents
 
 //    /// Run a decision function that does not yield a result
-//    member x.Transact(interpret) : unit =
+//    member x.Transact(interpret): unit =
 //        x.Transact(fun state -> (), interpret state)
 
 //    /// Projects from the present state including accumulated events
-//    member _.Query(render : 's -> 'r) : 'r =
+//    member _.Query(render: 's -> 'r): 'r =
 //        render state

equinox-patterns/Domain/ListEpoch.fs

@@ -8,11 +8,11 @@ let streamId = Equinox.StreamId.gen ListEpochId.toString
 module Events =
 
     type Event =
-        | Ingested of           {| ids : ItemId[] |}
+        | Ingested of           {| ids: ItemId[] |}
         | Closed
-        | Snapshotted of        {| ids : ItemId[]; closed : bool |}
+        | Snapshotted of        {| ids: ItemId[]; closed: bool |}
         interface TypeShape.UnionContract.IUnionContract
-    let codec, codecJe = Config.EventCodec.gen<Event>, Config.EventCodec.genJsonElement<Event>
+    let codec, codecJe = Store.Codec.gen<Event>, Store.Codec.genJsonElement<Event>
 
 module Fold =
 
@@ -22,7 +22,7 @@ module Fold =
         | Events.Ingested e ->  Array.append e.ids ids, closed
         | Events.Closed ->      (ids, true)
         | Events.Snapshotted e -> (e.ids, e.closed)
-    let fold : State -> Events.Event seq -> State = Seq.fold evolve
+    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 |}
@@ -41,35 +41,35 @@ let decide shouldClose candidateIds = function
                 let ingestEvent = Events.Ingested {| ids = news |}
                 news, if closing then [ ingestEvent ; Events.Closed ] else [ ingestEvent ]
         let _, closed = Fold.fold state events
-        let res : ExactlyOnceIngester.IngestResult<_, _> = { accepted = added; closed = closed; residual = [||] }
+        let res: ExactlyOnceIngester.IngestResult<_, _> = { accepted = added; closed = closed; residual = [||] }
         res, events
     | currentIds, true ->
         { accepted = [||]; closed = true; residual = candidateIds |> Array.except currentIds (*|> Array.distinct*) }, []
 
 // NOTE see feedSource for example of separating Service logic into Ingestion and Read Services in order to vary the folding and/or state held
 type Service internal
-    (   shouldClose : ItemId[] -> ItemId[] -> bool, // let outer layers decide whether ingestion should trigger closing of the batch
-        resolve : ListEpochId -> Equinox.Decider<Events.Event, Fold.State>) =
+    (   shouldClose: ItemId[] -> ItemId[] -> bool, // let outer layers decide whether ingestion should trigger closing of the batch
+        resolve: ListEpochId -> Equinox.Decider<Events.Event, Fold.State>) =
 
     /// 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(epochId, items) : Async<ExactlyOnceIngester.IngestResult<_, _>> =
+    member _.Ingest(epochId, items): Async<ExactlyOnceIngester.IngestResult<_, _>> =
         let decider = resolve 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 shouldClose items, load = Equinox.AllowStale)
+        decider.Transact(decide shouldClose items, load = Equinox.AnyCachedValue)
 
-    /// Returns all the items currently held in the stream (Not using AllowStale on the assumption this needs to see updates from other apps)
-    member _.Read epochId : Async<Fold.State> =
+    /// Returns all the items currently held in the stream (Not using AnyCachedValue on the assumption this needs to see updates from other apps)
+    member _.Read epochId: Async<Fold.State> =
         let decider = resolve epochId
-        decider.Query id
+        decider.Query(id, Equinox.AllowStale (System.TimeSpan.FromSeconds 1))
 
-module Config =
+module Factory =
 
     let private (|Category|) = function
-        | Config.Store.Memory store -> Config.Memory.create Events.codec Fold.initial Fold.fold store
-        | Config.Store.Cosmos (context, cache) -> Config.Cosmos.createSnapshotted Events.codecJe Fold.initial Fold.fold (Fold.isOrigin, Fold.toSnapshot) (context, cache)
+        | Store.Context.Memory store -> Store.Memory.create Events.codec Fold.initial Fold.fold store
+        | Store.Context.Cosmos (context, cache) -> Store.Cosmos.createSnapshotted Events.codecJe Fold.initial Fold.fold (Fold.isOrigin, Fold.toSnapshot) (context, cache)
     let create maxItemsPerEpoch (Category cat) =
         let shouldClose candidateItems currentItems = Array.length currentItems + Array.length candidateItems >= maxItemsPerEpoch
-        Service(shouldClose, streamId >> Config.resolveDecider cat Category)
+        Service(shouldClose, streamId >> Store.resolveDecider cat Category)

equinox-patterns/Domain/ListIngester.fs

@@ -1,22 +1,22 @@
 module Patterns.Domain.ListIngester
 
-type Service internal (ingester : ExactlyOnceIngester.Service<_, _, _, _>) =
+type Service internal (ingester: ExactlyOnceIngester.Service<_, _, _, _>) =
 
     /// Slot the item into the series of epochs.
     /// Returns items that actually got added (i.e. may be empty if it was an idempotent retry).
-    member _.IngestItems(originEpochId, items : ItemId[]) : Async<seq<ItemId>>=
+    member _.IngestItems(originEpochId, items: ItemId[]): Async<seq<ItemId>>=
         ingester.IngestMany(originEpochId, items)
 
     /// Efficiently determine a valid ingestion origin epoch
     member _.ActiveIngestionEpochId() =
         ingester.ActiveIngestionEpochId()
 
-module Config =
+module Factory =
 
     let create_ linger maxItemsPerEpoch store =
         let log = Serilog.Log.ForContext<Service>()
-        let series = ListSeries.Config.create store
-        let epochs = ListEpoch.Config.create maxItemsPerEpoch store
+        let series = ListSeries.Factory.create store
+        let epochs = ListEpoch.Factory.create maxItemsPerEpoch store
         let ingester = ExactlyOnceIngester.create log linger (series.ReadIngestionEpochId, series.MarkIngestionEpochId) (epochs.Ingest, Array.toSeq)
         Service(ingester)
     let create store =

equinox-patterns/Domain/ListSeries.fs

@@ -13,10 +13,10 @@ let streamId () = Equinox.StreamId.gen ListSeriesId.toString ListSeriesId.wellKn
 module Events =
 
     type Event =
-        | Started of            {| epochId : ListEpochId |}
-        | Snapshotted of        {| active : ListEpochId |}
+        | Started of            {| epochId: ListEpochId |}
+        | Snapshotted of        {| active: ListEpochId |}
         interface TypeShape.UnionContract.IUnionContract
-    let codec, codecJe = Config.EventCodec.gen<Event>, Config.EventCodec.genJsonElement<Event>
+    let codec, codecJe = Store.Codec.gen<Event>, Store.Codec.genJsonElement<Event>
 
 module Fold =
 
@@ -25,33 +25,33 @@ module Fold =
     let private evolve _state = function
         | Events.Started e -> Some e.epochId
         | Events.Snapshotted e -> Some e.active
-    let fold : State -> Events.Event seq -> State = Seq.fold evolve
+    let fold: State -> Events.Event seq -> State = Seq.fold evolve
 
     let isOrigin = function Events.Snapshotted _ -> true | _ -> false
     let toSnapshot s = Events.Snapshotted {| active = Option.get s |}
 
-let interpret epochId (state : Fold.State) =
+let interpret epochId (state: Fold.State) =
     [if state |> Option.forall (fun cur -> cur < epochId) && epochId >= ListEpochId.initial then
         yield Events.Started {| epochId = epochId |}]
 
-type Service internal (resolve : unit -> Equinox.Decider<Events.Event, Fold.State>) =
+type Service internal (resolve: unit -> Equinox.Decider<Events.Event, Fold.State>) =
 
     /// Determines the current active epoch
     /// Uses cached values as epoch transitions are rare, and caller needs to deal with the inherent race condition in any case
-    member _.ReadIngestionEpochId() : Async<ListEpochId> =
+    member _.ReadIngestionEpochId(): Async<ListEpochId> =
         let decider = resolve ()
         decider.Query(Option.defaultValue ListEpochId.initial)
 
     /// 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 epochId : Async<unit> =
+    member _.MarkIngestionEpochId epochId: Async<unit> =
         let decider = resolve ()
-        decider.Transact(interpret epochId, load = Equinox.AllowStale)
+        decider.Transact(interpret epochId, load = Equinox.AnyCachedValue)
 
-module Config =
+module Factory =
 
     let private (|Category|) = function
-        | Config.Store.Memory store -> Config.Memory.create Events.codec Fold.initial Fold.fold store
-        | Config.Store.Cosmos (context, cache) ->
-            Config.Cosmos.createSnapshotted Events.codecJe Fold.initial Fold.fold (Fold.isOrigin, Fold.toSnapshot) (context, cache)
-    let create (Category cat) = Service(streamId >> Config.resolveDecider cat Category)
+        | Store.Context.Memory store -> Store.Memory.create Events.codec Fold.initial Fold.fold store
+        | Store.Context.Cosmos (context, cache) ->
+            Store.Cosmos.createSnapshotted Events.codecJe Fold.initial Fold.fold (Fold.isOrigin, Fold.toSnapshot) (context, cache)
+    let create (Category cat) = Service(streamId >> Store.resolveDecider cat Category)