fix(CosmosStore): Propagate correlationId, causationId (#260)

Author: bartelink

CHANGELOG.md

@@ -68,6 +68,7 @@ The `Unreleased` section name is replaced by the expected version of next releas
 
 ### Fixed
 
+- `Propulsion.CosmosStore.CosmosStoreSource`: Propagate `correlationId`/`causationId` [#260](https://github.com/jet/propulsion/pull/260)
 - `Propulsion.SqlStreamStore`: Replaced incorrect/meaningless stream name for `SqlStreamStoreSource` [#139](https://github.com/jet/propulsion/pull/139)
 - Synced [`AwaitTaskCorrect`](http://www.fssnip.net/7Rc/title/AsyncAwaitTaskCorrect) with official version [3c11142](https://github.com/jet/propulsion/commit/3c11142b75bf3b0ef2181fd106a4b17c0b2313ef)
 - `Projector`,`Ingester`,`Submitter`, `Scheduler`: Deterministic shutdown via Cancellation and/or unhandled exceptions [#154](https://github.com/jet/propulsion/pull/154)

src/Propulsion.CosmosStore/EquinoxSystemTextJsonParser.fs

@@ -41,7 +41,10 @@ module EquinoxSystemTextJsonParser =
         batch.e |> Seq.mapi (fun offset x ->
             let d = batch.MapData x.d
             let m = batch.MapData x.m
-            FsCodec.Core.TimelineEvent.Create(batch.i + int64 offset, x.c, d, m, timestamp = x.t, size = x.c.Length + d.Length + m.Length + 80))
+            let inline len s = if isNull s then 0 else String.length s
+            FsCodec.Core.TimelineEvent.Create(batch.i + int64 offset, x.c, d, m, timestamp = x.t,
+                                              size = x.c.Length + d.Length + m.Length + len x.correlationId + len x.causationId + 80,
+                                              correlationId = x.correlationId, causationId = x.causationId))
 
     /// Attempts to parse a Document/Item from the Store
     /// returns ValueNone if it does not bear the hallmarks of a valid Batch, or the streamFilter predicate rejects

tools/Propulsion.Tool/Args.fs

@@ -140,7 +140,7 @@ module Cosmos =
             | Provisioning.Database (Throughput.Manual rus) ->
                 Log.Information("Provisioning Leases Container at {modeStr:l} level with {rus:n0} RU/s", modeStr, rus)
             | Provisioning.Serverless ->
-                Log.Information("Provisioning Leases Container in {modeStr:l} mode with automatic throughput RU/s as configured in account", modeStr)
+                Log.Information("Provisioning Leases Container in {modeStr:l} mode with throughput as defined at account level", modeStr)
             initAux container.Database.Client (container.Database.Id, container.Id) mode
         | x -> p.Raise $"unexpected subcommand %A{x}"