fix(monitoring): make Counter and Meter increments O(1) amortized

extensions/Sisk.Monitoring/Counter.cs

@@ -5,7 +5,7 @@ namespace Sisk.Monitoring;
 /// </summary>
 public sealed class Counter {
 
-    private readonly List<CounterEntry> _counts;
+    private readonly PriorityQueue<double, DateTime> _counts;
     private readonly object _sync = new ();
 
     /// <summary>
@@ -21,11 +21,9 @@ public Counter () {
     /// <param name="initialValue">The initial value of the counter.</param>
     /// <param name="initialDuration">The duration for which the initial value remains valid.</param>
     public Counter ( double initialValue, TimeSpan initialDuration ) {
-        if (initialValue == 0) {
-            _counts = new ();
-        }
-        else {
-            _counts = new () { new CounterEntry ( initialValue, DateTime.Now + initialDuration ) };
+        _counts = new ();
+        if (initialValue != 0) {
+            _counts.Enqueue ( initialValue, DateTime.Now + initialDuration );
         }
         DefaultDuration = initialDuration;
     }
@@ -44,7 +42,7 @@ public double Current {
         get {
             lock (_sync) {
                 PruneExpiredEntries ();
-                return _counts.Sum ( c => c.Value );
+                return _counts.UnorderedItems.Sum ( c => c.Element );
             }
         }
     }
@@ -57,7 +55,7 @@ public double Current {
     public void Increment ( double by, TimeSpan duration ) {
         lock (_sync) {
             PruneExpiredEntries ();
-            _counts.Add ( new CounterEntry ( by, DateTime.Now + duration ) );
+            _counts.Enqueue ( by, DateTime.Now + duration );
         }
     }
 
@@ -88,8 +86,8 @@ public void Reset () {
     internal CounterIncrementSnapshot [] ExportState () {
         lock (_sync) {
             PruneExpiredEntries ();
-            return _counts
-                .Select ( entry => new CounterIncrementSnapshot ( entry.Value, entry.ExpiresAt ) )
+            return _counts.UnorderedItems
+                .Select ( entry => new CounterIncrementSnapshot ( entry.Element, entry.Priority ) )
                 .ToArray ();
         }
     }
@@ -99,16 +97,17 @@ internal void ImportState ( CounterIncrementSnapshot [] increments ) {
 
         lock (_sync) {
             _counts.Clear ();
-            _counts.AddRange ( increments
+            _counts.EnqueueRange ( increments
                 .Where ( increment => increment.ExpiresAt > now )
-                .Select ( increment => new CounterEntry ( increment.Value, increment.ExpiresAt ) ) );
+                .Select ( increment => (increment.Value, increment.ExpiresAt) ) );
         }
     }
 
     void PruneExpiredEntries () {
         DateTime now = DateTime.Now;
-        _counts.RemoveAll ( entry => entry.ExpiresAt <= now );
-    }
 
-    record struct CounterEntry ( double Value, DateTime ExpiresAt );
+        while (_counts.TryPeek ( out _, out DateTime expiresAt ) && expiresAt <= now) {
+            _counts.Dequeue ();
+        }
+    }
 }

extensions/Sisk.Monitoring/Meter.cs

@@ -9,6 +9,7 @@ public sealed class Meter {
     private readonly TimeSpan _slidingWindowDuration;
     private readonly Dictionary<long, double> _buckets = new ();
     private readonly object _sync = new ();
+    private long _prunedThreshold = long.MinValue;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="Meter"/> class with a default 24-hour sliding window.
@@ -126,9 +127,15 @@ public IEnumerable<MeterReading> Read ( TimeSpan tick ) {
 
     void PruneExpiredBuckets ( DateTime now ) {
         long threshold = ToBucketKey ( now - _slidingWindowDuration );
+        if (threshold == _prunedThreshold) {
+            return;
+        }
 
-        foreach (long key in _buckets.Keys.Where ( k => k < threshold ).ToArray ()) {
-            _buckets.Remove ( key );
+        _prunedThreshold = threshold;
+        foreach (long key in _buckets.Keys) {
+            if (key < threshold) {
+                _buckets.Remove ( key );
+            }
         }
     }
 

tests/Sisk.Monitoring.Tests/CounterMeterTests.cs

@@ -0,0 +1,67 @@
+using System.Diagnostics;
+
+namespace Sisk.Monitoring.Tests;
+
+[TestClass]
+public sealed class CounterMeterTests {
+
+    [TestMethod]
+    public void Counter_SumsLiveIncrementsAndDropsExpiredOnes () {
+        var counter = new Counter ();
+
+        counter.Increment ( 5, TimeSpan.FromMilliseconds ( 50 ) );
+        counter.Increment ( 3, TimeSpan.FromHours ( 1 ) );
+        counter.Increment ( 2, TimeSpan.FromHours ( 2 ) );
+        Assert.AreEqual ( 10, counter.Current );
+
+        Thread.Sleep ( 150 );
+        Assert.AreEqual ( 5, counter.Current );
+
+        counter.Reset ();
+        Assert.AreEqual ( 0, counter.Current );
+    }
+
+    [TestMethod]
+    public void Counter_ExpiresIncrementsOutOfInsertionOrder () {
+        var counter = new Counter ();
+
+        counter.Increment ( 1, TimeSpan.FromHours ( 1 ) );
+        counter.Increment ( 10, TimeSpan.FromMilliseconds ( 50 ) );
+        counter.Increment ( 100, TimeSpan.FromHours ( 1 ) );
+
+        Thread.Sleep ( 150 );
+        Assert.AreEqual ( 101, counter.Current );
+    }
+
+    [TestMethod]
+    public void Counter_IncrementCostDoesNotGrowWithLiveEntries () {
+        const int increments = 100_000;
+        var counter = new Counter ();
+
+        var stopwatch = Stopwatch.StartNew ();
+        for (int i = 0; i < increments; i++) {
+            counter.Increment ();
+        }
+        stopwatch.Stop ();
+
+        Assert.AreEqual ( increments, counter.Current );
+        Assert.IsTrue (
+            stopwatch.Elapsed < TimeSpan.FromSeconds ( 1 ),
+            $"{increments} increments took {stopwatch.Elapsed.TotalMilliseconds:N0} ms; per-increment cost must not grow with the number of live entries." );
+    }
+
+    [TestMethod]
+    public void Meter_SumsRecentIncrementsAndDropsExpiredBuckets () {
+        var meter = new Meter ( TimeSpan.FromSeconds ( 2 ) );
+
+        meter.Increment ( 4 );
+        meter.Increment ();
+        Assert.AreEqual ( 5, meter.Read ( TimeSpan.FromSeconds ( 2 ) ).Sum ( reading => reading.Value ) );
+
+        Thread.Sleep ( 3200 );
+        Assert.AreEqual ( 0, meter.Read ( TimeSpan.FromSeconds ( 2 ) ).Sum ( reading => reading.Value ) );
+
+        meter.Increment ( 7 );
+        Assert.AreEqual ( 7, meter.Read ( TimeSpan.FromSeconds ( 2 ) ).Sum ( reading => reading.Value ) );
+    }
+}