From 365f8ef0637e8f40e7c38b45531df8ed7c1601bf Mon Sep 17 00:00:00 2001 From: Ruben Bartelink Date: Tue, 5 Mar 2019 12:58:23 +0000 Subject: [PATCH] Remove optionality from Cosmos caching (#107) resolves #104 --- CHANGELOG.md | 7 +++++-- samples/Infrastructure/Services.fs | 8 ++++---- samples/Infrastructure/Storage.fs | 6 +++--- samples/Store/Integration/CartIntegration.fs | 6 +++--- .../ContactPreferencesIntegration.fs | 6 +++--- .../Store/Integration/FavoritesIntegration.fs | 4 ++-- samples/Tutorial/Cosmos.fsx | 2 +- samples/Tutorial/Todo.fsx | 2 +- src/Equinox.Cosmos/Cosmos.fs | 17 ++++++++++++----- src/Equinox.EventStore/EventStore.fs | 7 ++++++- .../CosmosIntegration.fs | 12 ++++++------ .../EventStoreIntegration.fs | 8 ++++---- 12 files changed, 50 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d9e71ad04..35e2a7750 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,9 +2,8 @@ The repo is versioned based on [SemVer 2.0](https://semver.org/spec/v2.0.0.html) using the tiny-but-mighty [MinVer](https://github.com/adamralph/minver) from [@adamralph](https://github.com/adamralph). [See here](https://github.com/adamralph/minver#how-it-works) for more information on how it works. -All notable changes to this project will be documented in this file. +All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). -The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). The `Unreleased` section name is replaced by the expected version of next release. A stable version's log contains all changes between that version and the previous stable version (can duplicate the prereleases logs). _NB at the present time, this project does not adhere strictly to Semantic Versioning - small binary-breaking changes may occur without a change in Major at the until this notice is removed (it will be!)._ @@ -13,6 +12,10 @@ _NB at the present time, this project does not adhere strictly to Semantic Versi ### Added ### Changed + +- Make `caching` non-optional in CosmosStreamResolver; add `NoCaching` cache mode for `Equinox.Cosmos` [#104](https://github.com/jet/equinox/issues/104) @jakzale +- Reorder `caching` and `access` in `GesStreamResolver` to match `CosmosStreamResolver` [#107](https://github.com/jet/equinox/issues/107) + ### Removed ### Fixed diff --git a/samples/Infrastructure/Services.fs b/samples/Infrastructure/Services.fs index bfb2879a0..a593416f6 100644 --- a/samples/Infrastructure/Services.fs +++ b/samples/Infrastructure/Services.fs @@ -15,13 +15,13 @@ type StreamResolver(storage) = match storage with | Storage.StorageConfig.Memory store -> Equinox.MemoryStore.MemoryResolver(store, fold, initial).Resolve - | Storage.StorageConfig.Es (gateway, cache, unfolds) -> + | Storage.StorageConfig.Es (gateway, caching, unfolds) -> let accessStrategy = if unfolds then Equinox.EventStore.AccessStrategy.RollingSnapshots snapshot |> Some else None - Equinox.EventStore.GesResolver<'event,'state>(gateway, codec, fold, initial, ?access = accessStrategy, ?caching = cache).Resolve - | Storage.StorageConfig.Cosmos (gateway, cache, unfolds, databaseId, collectionId) -> + Equinox.EventStore.GesResolver<'event,'state>(gateway, codec, fold, initial, ?caching = caching, ?access = accessStrategy).Resolve + | Storage.StorageConfig.Cosmos (gateway, caching, unfolds, databaseId, collectionId) -> let store = Equinox.Cosmos.CosmosStore(gateway, databaseId, collectionId) let accessStrategy = if unfolds then Equinox.Cosmos.AccessStrategy.Snapshot snapshot |> Some else None - Equinox.Cosmos.CosmosResolver<'event,'state>(store, codec, fold, initial, ?access = accessStrategy, ?caching = cache).Resolve + Equinox.Cosmos.CosmosResolver<'event,'state>(store, codec, fold, initial, caching, ?access = accessStrategy).Resolve type ServiceBuilder(storageConfig, handlerLog) = let resolver = StreamResolver(storageConfig) diff --git a/samples/Infrastructure/Storage.fs b/samples/Infrastructure/Storage.fs index 72a90b978..a409a0ea2 100644 --- a/samples/Infrastructure/Storage.fs +++ b/samples/Infrastructure/Storage.fs @@ -56,7 +56,7 @@ let defaultBatchSize = 500 type StorageConfig = | Memory of Equinox.MemoryStore.VolatileStore | Es of Equinox.EventStore.GesGateway * Equinox.EventStore.CachingStrategy option * unfolds: bool - | Cosmos of Equinox.Cosmos.CosmosGateway * Equinox.Cosmos.CachingStrategy option * unfolds: bool * databaseId: string * collectionId: string + | Cosmos of Equinox.Cosmos.CosmosGateway * Equinox.Cosmos.CachingStrategy * unfolds: bool * databaseId: string * collectionId: string module MemoryStore = let config () = @@ -129,6 +129,6 @@ module Cosmos = let cacheStrategy = if cache then let c = Caching.Cache("equinox-tool", sizeMb = 50) - CachingStrategy.SlidingWindow (c, TimeSpan.FromMinutes 20.) |> Some - else None + CachingStrategy.SlidingWindow (c, TimeSpan.FromMinutes 20.) + else CachingStrategy.NoCaching StorageConfig.Cosmos (createGateway conn (defaultBatchSize,pageSize), cacheStrategy, unfolds, dbName, collName) \ No newline at end of file diff --git a/samples/Store/Integration/CartIntegration.fs b/samples/Store/Integration/CartIntegration.fs index 6094a4020..207beee53 100644 --- a/samples/Store/Integration/CartIntegration.fs +++ b/samples/Store/Integration/CartIntegration.fs @@ -19,14 +19,14 @@ let createServiceMemory log store = let codec = Equinox.EventStore.Integration.EventStoreIntegration.genCodec() let resolveGesStreamWithRollingSnapshots gateway = - GesResolver(gateway, codec, fold, initial, AccessStrategy.RollingSnapshots snapshot).Resolve + GesResolver(gateway, codec, fold, initial, access = AccessStrategy.RollingSnapshots snapshot).Resolve let resolveGesStreamWithoutCustomAccessStrategy gateway = GesResolver(gateway, codec, fold, initial).Resolve let resolveCosmosStreamWithProjection gateway = - CosmosResolver(gateway, codec, fold, initial, AccessStrategy.Snapshot snapshot).Resolve + CosmosResolver(gateway, codec, fold, initial, CachingStrategy.NoCaching, AccessStrategy.Snapshot snapshot).Resolve let resolveCosmosStreamWithoutCustomAccessStrategy gateway = - CosmosResolver(gateway, codec, fold, initial).Resolve + CosmosResolver(gateway, codec, fold, initial, CachingStrategy.NoCaching).Resolve let addAndThenRemoveItemsManyTimesExceptTheLastOne context cartId skuId (service: Backend.Cart.Service) count = service.FlowAsync(cartId, fun _ctx execute -> diff --git a/samples/Store/Integration/ContactPreferencesIntegration.fs b/samples/Store/Integration/ContactPreferencesIntegration.fs index 8708cd772..68348be37 100644 --- a/samples/Store/Integration/ContactPreferencesIntegration.fs +++ b/samples/Store/Integration/ContactPreferencesIntegration.fs @@ -18,14 +18,14 @@ let createServiceMemory log store = let codec = genCodec() let resolveStreamGesWithOptimizedStorageSemantics gateway = - GesResolver(gateway 1, codec, fold, initial, AccessStrategy.EventsAreState).Resolve + GesResolver(gateway 1, codec, fold, initial, access = AccessStrategy.EventsAreState).Resolve let resolveStreamGesWithoutAccessStrategy gateway = GesResolver(gateway defaultBatchSize, codec, fold, initial).Resolve let resolveStreamCosmosWithKnownEventTypeSemantics gateway = - CosmosResolver(gateway 1, codec, fold, initial, AccessStrategy.AnyKnownEventType).Resolve + CosmosResolver(gateway 1, codec, fold, initial, CachingStrategy.NoCaching, AccessStrategy.AnyKnownEventType).Resolve let resolveStreamCosmosWithoutCustomAccessStrategy gateway = - CosmosResolver(gateway defaultBatchSize, codec, fold, initial).Resolve + CosmosResolver(gateway defaultBatchSize, codec, fold, initial, CachingStrategy.NoCaching).Resolve type Tests(testOutputHelper) = let testOutput = TestOutputAdapter testOutputHelper diff --git a/samples/Store/Integration/FavoritesIntegration.fs b/samples/Store/Integration/FavoritesIntegration.fs index 396e6503a..f2d115d17 100644 --- a/samples/Store/Integration/FavoritesIntegration.fs +++ b/samples/Store/Integration/FavoritesIntegration.fs @@ -19,11 +19,11 @@ let createServiceMemory log store = let codec = genCodec() let createServiceGes gateway log = - let resolveStream = GesResolver(gateway, codec, fold, initial, AccessStrategy.RollingSnapshots snapshot).Resolve + let resolveStream = GesResolver(gateway, codec, fold, initial, access = AccessStrategy.RollingSnapshots snapshot).Resolve Backend.Favorites.Service(log, resolveStream) let createServiceCosmos gateway log = - let resolveStream = CosmosResolver(gateway, codec, fold, initial, AccessStrategy.Snapshot snapshot).Resolve + let resolveStream = CosmosResolver(gateway, codec, fold, initial, CachingStrategy.NoCaching, AccessStrategy.Snapshot snapshot).Resolve Backend.Favorites.Service(log, resolveStream) type Tests(testOutputHelper) = diff --git a/samples/Tutorial/Cosmos.fsx b/samples/Tutorial/Cosmos.fsx index 0a7f285fd..02d1a46b9 100644 --- a/samples/Tutorial/Cosmos.fsx +++ b/samples/Tutorial/Cosmos.fsx @@ -61,7 +61,7 @@ module Store = module FavoritesCategory = let codec = Equinox.UnionCodec.JsonUtf8.Create(Newtonsoft.Json.JsonSerializerSettings()) - let resolve = CosmosResolver(Store.store, codec, Favorites.fold, Favorites.initial, caching=Store.cacheStrategy).Resolve + let resolve = CosmosResolver(Store.store, codec, Favorites.fold, Favorites.initial, Store.cacheStrategy).Resolve let service = Favorites.Service(Log.log, FavoritesCategory.resolve) diff --git a/samples/Tutorial/Todo.fsx b/samples/Tutorial/Todo.fsx index b2b09b04d..2a4f5c02c 100644 --- a/samples/Tutorial/Todo.fsx +++ b/samples/Tutorial/Todo.fsx @@ -114,7 +114,7 @@ module Store = module TodosCategory = let codec = Equinox.UnionCodec.JsonUtf8.Create(Newtonsoft.Json.JsonSerializerSettings()) let access = Equinox.Cosmos.AccessStrategy.Snapshot (isOrigin,compact) - let resolve = CosmosResolver(Store.store, codec, fold, initial, access=access, caching=Store.cacheStrategy).Resolve + let resolve = CosmosResolver(Store.store, codec, fold, initial, Store.cacheStrategy, access=access).Resolve let service = Service(log, TodosCategory.resolve) diff --git a/src/Equinox.Cosmos/Cosmos.fs b/src/Equinox.Cosmos/Cosmos.fs index a9f4819b9..175009182 100644 --- a/src/Equinox.Cosmos/Cosmos.fs +++ b/src/Equinox.Cosmos/Cosmos.fs @@ -874,6 +874,13 @@ type CosmosStore(gateway: CosmosGateway, collections: CosmosCollections, [] type CachingStrategy = + /// Do not apply any caching strategy for this Stream. + /// NB opting not to leverage caching when using CosmosDb can have significant implications for the scalability + /// of your application, both in terms of latency and running costs. + /// While the cost of a cache miss can be ameliorated to varying degrees by employing an appropriate `AccessStrategy` + /// [that works well and has been validated for your scenario with real data], even a cache with a low Hit Rate provides + /// a direct benefit in terms of the number of Request Unit (RU)s that need to be provisioned to your CosmosDb instances. + | NoCaching /// Retain a single 'state per streamName, together with the associated etag /// NB while a strategy like EventStore.Caching.SlidingWindowPrefixed is obviously easy to implement, the recommended approach is to /// track all relevant data in the state, and/or have the `unfold` function ensure _all_ relevant events get held in the `u`nfolds in tip @@ -890,11 +897,11 @@ type AccessStrategy<'event,'state> = /// Trust every event type as being an origin | AnyKnownEventType -type CosmosResolver<'event, 'state>(store : CosmosStore, codec, fold, initial, []?access, []?caching) = +type CosmosResolver<'event, 'state>(store : CosmosStore, codec, fold, initial, caching, []?access) = let readCacheOption = match caching with - | None -> None - | Some (CachingStrategy.SlidingWindow(cache, _)) -> Some(cache, null) + | CachingStrategy.NoCaching -> None + | CachingStrategy.SlidingWindow(cache, _) -> Some(cache, null) let isOrigin, projectOption = match access with | None -> (fun _ -> false), None @@ -905,8 +912,8 @@ type CosmosResolver<'event, 'state>(store : CosmosStore, codec, fold, initial, [ let folder = Folder<'event, 'state>(cosmosCat, fold, initial, isOrigin, ?unfold=projectOption, ?readCache = readCacheOption) let category : Store.ICategory<_,_,CollectionStream> = match caching with - | None -> folder :> _ - | Some (CachingStrategy.SlidingWindow(cache, window)) -> + | CachingStrategy.NoCaching -> folder :> _ + | CachingStrategy.SlidingWindow(cache, window) -> Caching.applyCacheUpdatesWithSlidingExpiration cache null window folder let resolveStream (streamId, maybeCollectionInitializationGate) = diff --git a/src/Equinox.EventStore/EventStore.fs b/src/Equinox.EventStore/EventStore.fs index 209a641dd..abceb942c 100644 --- a/src/Equinox.EventStore/EventStore.fs +++ b/src/Equinox.EventStore/EventStore.fs @@ -445,7 +445,12 @@ type CachingStrategy = /// Prefix is used to segregate multiple folds per stream when they are stored in the cache | SlidingWindowPrefixed of Caching.Cache * window: TimeSpan * prefix: string -type GesResolver<'event,'state>(gateway : GesGateway, codec, fold, initial, []?access, []?caching) = +type GesResolver<'event,'state> + ( gateway : GesGateway, codec, fold, initial, + /// Caching can be overkill for EventStore esp considering the degree to which its intrinsic caching is a first class feature + /// e.g., A key benefit is that reads of streams more than a few pages long get completed in constant time after the initial load + []?caching, + []?access) = do match access with | Some (AccessStrategy.EventsAreState) when Option.isSome caching -> "Equinox.EventStore does not support (and it would make things _less_ efficient even if it did)" diff --git a/tests/Equinox.Cosmos.Integration/CosmosIntegration.fs b/tests/Equinox.Cosmos.Integration/CosmosIntegration.fs index fb57243f6..1b63b727f 100644 --- a/tests/Equinox.Cosmos.Integration/CosmosIntegration.fs +++ b/tests/Equinox.Cosmos.Integration/CosmosIntegration.fs @@ -19,21 +19,21 @@ module Cart = let codec = genCodec() let createServiceWithoutOptimization connection batchSize log = let store = createCosmosStore connection batchSize - let resolveStream = CosmosResolver(store, codec, fold, initial).Resolve + let resolveStream = CosmosResolver(store, codec, fold, initial, CachingStrategy.NoCaching).Resolve Backend.Cart.Service(log, resolveStream) let createServiceWithoutOptimizationAndMaxItems connection batchSize maxEventsPerSlice log = let store = createCosmosStoreWithMaxEventsPerSlice connection batchSize maxEventsPerSlice - let resolveStream = CosmosResolver(store, codec, fold, initial).Resolve + let resolveStream = CosmosResolver(store, codec, fold, initial, CachingStrategy.NoCaching).Resolve Backend.Cart.Service(log, resolveStream) let projection = "Compacted",snd snapshot let createServiceWithProjection connection batchSize log = let store = createCosmosStore connection batchSize - let resolveStream = CosmosResolver(store, codec, fold, initial, AccessStrategy.Snapshot snapshot).Resolve + let resolveStream = CosmosResolver(store, codec, fold, initial, CachingStrategy.NoCaching, AccessStrategy.Snapshot snapshot).Resolve Backend.Cart.Service(log, resolveStream) let createServiceWithProjectionAndCaching connection batchSize log cache = let store = createCosmosStore connection batchSize let sliding20m = CachingStrategy.SlidingWindow (cache, TimeSpan.FromMinutes 20.) - let resolveStream = CosmosResolver(store, codec, fold, initial, AccessStrategy.Snapshot snapshot, sliding20m).Resolve + let resolveStream = CosmosResolver(store, codec, fold, initial, sliding20m, AccessStrategy.Snapshot snapshot).Resolve Backend.Cart.Service(log, resolveStream) module ContactPreferences = @@ -41,10 +41,10 @@ module ContactPreferences = let codec = genCodec() let createServiceWithoutOptimization createGateway defaultBatchSize log _ignoreWindowSize _ignoreCompactionPredicate = let gateway = createGateway defaultBatchSize - let resolveStream = CosmosResolver(gateway, codec, fold, initial).Resolve + let resolveStream = CosmosResolver(gateway, codec, fold, initial, CachingStrategy.NoCaching).Resolve Backend.ContactPreferences.Service(log, resolveStream) let createService createGateway log = - let resolveStream = CosmosResolver(createGateway 1, codec, fold, initial, AccessStrategy.AnyKnownEventType).Resolve + let resolveStream = CosmosResolver(createGateway 1, codec, fold, initial, CachingStrategy.NoCaching, AccessStrategy.AnyKnownEventType).Resolve Backend.ContactPreferences.Service(log, resolveStream) #nowarn "1182" // From hereon in, we may have some 'unused' privates (the tests) diff --git a/tests/Equinox.EventStore.Integration/EventStoreIntegration.fs b/tests/Equinox.EventStore.Integration/EventStoreIntegration.fs index 65fd8d619..92624aa0f 100644 --- a/tests/Equinox.EventStore.Integration/EventStoreIntegration.fs +++ b/tests/Equinox.EventStore.Integration/EventStoreIntegration.fs @@ -30,14 +30,14 @@ module Cart = let createServiceWithoutOptimization log gateway = Backend.Cart.Service(log, GesResolver(gateway, codec, fold, initial).Resolve) let createServiceWithCompaction log gateway = - let resolveStream = GesResolver(gateway, codec, fold, initial, AccessStrategy.RollingSnapshots snapshot).Resolve + let resolveStream = GesResolver(gateway, codec, fold, initial, access = AccessStrategy.RollingSnapshots snapshot).Resolve Backend.Cart.Service(log, resolveStream) let createServiceWithCaching log gateway cache = let sliding20m = CachingStrategy.SlidingWindow (cache, TimeSpan.FromMinutes 20.) - Backend.Cart.Service(log, GesResolver(gateway, codec, fold, initial, caching = sliding20m).Resolve) + Backend.Cart.Service(log, GesResolver(gateway, codec, fold, initial, sliding20m).Resolve) let createServiceWithCompactionAndCaching log gateway cache = let sliding20m = CachingStrategy.SlidingWindow (cache, TimeSpan.FromMinutes 20.) - Backend.Cart.Service(log, GesResolver(gateway, codec, fold, initial, AccessStrategy.RollingSnapshots snapshot, sliding20m).Resolve) + Backend.Cart.Service(log, GesResolver(gateway, codec, fold, initial, sliding20m, AccessStrategy.RollingSnapshots snapshot).Resolve) module ContactPreferences = let fold, initial = Domain.ContactPreferences.Folds.fold, Domain.ContactPreferences.Folds.initial @@ -46,7 +46,7 @@ module ContactPreferences = let gateway = createGesGateway connection defaultBatchSize Backend.ContactPreferences.Service(log, GesResolver(gateway, codec, fold, initial).Resolve) let createService log connection = - let resolveStream = GesResolver(createGesGateway connection 1, codec, fold, initial, AccessStrategy.EventsAreState).Resolve + let resolveStream = GesResolver(createGesGateway connection 1, codec, fold, initial, access = AccessStrategy.EventsAreState).Resolve Backend.ContactPreferences.Service(log, resolveStream) #nowarn "1182" // From hereon in, we may have some 'unused' privates (the tests)