Polish comments

jet · Jul 18, 2024 · e78b739 · e78b739
1 parent 685de1f
commit e78b739
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 4 deletions.
diff --git a/Propulsion.sln.DotSettings b/Propulsion.sln.DotSettings
@@ -1,7 +1,9 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=abend/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=Dispatchable/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=exns/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=ingester/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=Prioritizer/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=slipstreamed/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=slipstreaming/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=stddev/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
diff --git a/src/Propulsion.CosmosStore/CosmosStoreSink.fs b/src/Propulsion.CosmosStore/CosmosStoreSink.fs
@@ -120,8 +120,8 @@ module Internal =
                     | Ok struct (Writer.Result.Ok pos', _stats) ->
                         struct (streams.SetWritePos(stream, pos'), false)
                     | Ok ((Writer.Result.Duplicate pos' | Writer.Result.PartialDuplicate pos'), _stats) ->
-                        streams.SetWritePos(stream, pos') |> ignore // throw away the events (but not the unfolds)
-                        ValueNone, false // Don't declare progress yet, until any unfolds have been handled
+                        streams.SetWritePos(stream, pos') |> ignore // Trim any events that conflict with stuff already in the store
+                        ValueNone, false // Don't declare progress until any unfolds have been handled
                     | Ok (Writer.Result.PrefixMissing _, _stats) ->
                         streams.WritePos(stream), false
                     | Error struct (exn, _stats) ->

diff --git a/src/Propulsion/Streams.fs b/src/Propulsion/Streams.fs
@@ -493,7 +493,7 @@ module Scheduling =
             let sw = Stopwatch.start()
             member _.RecordResults ts = results <- results + Stopwatch.elapsedTicks ts
             member _.RecordDispatch ts = dispatch <- dispatch + Stopwatch.elapsedTicks ts
-            // If we did not dispatch, we attempt ingestion of streams as a standalone task, but need to add to dispatch time to compensate for calcs below
+            // If we did not dispatch, we attempt ingestion of streams as a standalone task, but need to add to dispatch time to compensate for math below
             member x.RecordDispatchNone ts = x.RecordDispatch ts
             member _.RecordMerge ts = merge <- merge + Stopwatch.elapsedTicks ts
             member _.RecordIngest ts = ingest <- ingest + Stopwatch.elapsedTicks ts
@@ -667,7 +667,7 @@ module Scheduling =
 
     module Progress =
 
-        type [<Struct; NoComparison; NoEquality>] BatchState = { markCompleted: unit -> unit; streamToRequiredIndex: Dictionary<FsCodec.StreamName, int64> }
+        type [<Struct; NoComparison; NoEquality>] BatchState = private { markCompleted: unit -> unit; streamToRequiredIndex: Dictionary<FsCodec.StreamName, int64> }
 
         type ProgressState<'Pos>() =
             let pending = Queue<BatchState>()
@@ -690,6 +690,8 @@ module Scheduling =
             member _.MarkStreamProgress(stream, index) =
                 for x in pending do
                     // example: when we reach position 1 on the stream (having handled event 0), and the required position was 1, we remove the requirement
+                    // NOTE Any unfolds that accompany event 0 will also bear Index 0
+                    // NOTE 2: subsequent updates to Unfolds will bear the same Index of 0 until there is an Event with Index 1
                     let mutable requiredIndex = Unchecked.defaultof<_>
                     if x.streamToRequiredIndex.TryGetValue(stream, &requiredIndex) && index >= requiredIndex then
                         x.streamToRequiredIndex.Remove stream |> ignore