Estimator tweaks

jet · Oct 11, 2023 · c6effae · c6effae
1 parent 03ba20f
commit c6effae
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 7 deletions.
diff --git a/src/Propulsion.CosmosStore/ChangeFeedProcessor.fs b/src/Propulsion.CosmosStore/ChangeFeedProcessor.fs
@@ -158,9 +158,9 @@ type ChangeFeedProcessor =
                     return result.ToArray() }
                 fun (ct: CancellationToken) -> task {
                     while not ct.IsCancellationRequested do
-                        let! leasesState = fetchEstimatorStates (fun s -> leaseTokenToPartitionId s.LeaseToken, s.EstimatedLag) ct
-                        Array.sortInPlaceBy fst leasesState
-                        do! lagMonitorCallback leasesState } )
+                        let! leasesStates = fetchEstimatorStates (fun s -> struct (leaseTokenToPartitionId s.LeaseToken, s.EstimatedLag)) ct
+                        Array.sortInPlaceBy ValueTuple.fst leasesStates
+                        do! lagMonitorCallback leasesStates } )
         let wrap (f: unit -> Task) () = task { return! f () }
         SourcePipeline.Start(log, wrap processor.StartAsync, maybePumpMetrics, wrap processor.StopAsync, observer)
     static member private mkLeaseOwnerIdForProcess() =

diff --git a/src/Propulsion.CosmosStore/CosmosStoreSource.fs b/src/Propulsion.CosmosStore/CosmosStoreSource.fs
@@ -14,7 +14,7 @@ module Log =
             ingestLatency: TimeSpan; ingestQueued: int }
     type LagMetric =
         {   database: string; container: string; group: string
-            rangeLags: (int * int64)[] }
+            rangeLags: struct (int * int64)[] }
     [<RequireQualifiedAccess; NoEquality; NoComparison>]
     type Metric =
         | Read of ReadMetric
@@ -88,7 +88,7 @@ type CosmosStoreSource =
             [<O; D null>] ?notifyError,
             [<O; D null>] ?customize) =
         let databaseId, containerId = monitored.Database.Id, monitored.Id
-        let logLag (interval: TimeSpan) (remainingWork: (int * int64)[]) = task {
+        let logLag (interval: TimeSpan) (remainingWork: struct (int * int64)[]) = task {
             let mutable synced, lagged, count, total = ResizeArray(), ResizeArray(), 0, 0L
             for partitionId, gap as partitionAndGap in remainingWork do
                 total <- total + gap
@@ -102,7 +102,7 @@ type CosmosStoreSource =
         let startFromTail = defaultArg startFromTail false
         let source =
             ChangeFeedProcessor.Start
-              ( log, monitored, leases, processorName, observer, ?notifyError=notifyError, ?customize=customize,
+              ( log, monitored, leases, processorName, observer, ?notifyError = notifyError, ?customize = customize,
                 ?maxItems = maxItems, ?feedPollDelay = tailSleepInterval, ?reportLagAndAwaitNextEstimation = maybeLogLag,
                 startFromTail = startFromTail,
                 leaseAcquireInterval = TimeSpan.FromSeconds 5., leaseRenewInterval = TimeSpan.FromSeconds 5., leaseTtl = TimeSpan.FromSeconds 10.)

diff --git a/src/Propulsion/Pipeline.fs b/src/Propulsion/Pipeline.fs
@@ -34,7 +34,7 @@ type Pipeline(task: Task<unit>, triggerStop) =
     /// Reacts to cancellation by aborting the processing via <c>Stop()</c>; see <c>Await</c> if such semantics are not desired.
     member x.AwaitWithStopOnCancellation() = async {
         let! ct = Async.CancellationToken
-        use _ = ct.Register(fun () -> x.Stop())
+        use _ = ct.Register(Action x.Stop)
         return! x.Await() }
 
     static member Prepare(log: ILogger, pumpScheduler, pumpSubmitter, ?pumpIngester, ?pumpDispatcher) =