Decide a re-created statement entry's counters row-coherently in pg_statement_stats (#4428)

Darling/Darling.Tests/PgStatementStatsRowCoherentResetTests.cs

@@ -0,0 +1,250 @@
+/*
+ * Copyright (c) 2026 Erik Darling, Darling Data LLC
+ *
+ * This file is part of the SQL Server Performance Monitor.
+ *
+ * Licensed under the MIT License. See LICENSE file in the project root for full license information.
+ */
+
+using System;
+using System.Collections;
+using System.Data.Common;
+using System.Threading;
+using System.Threading.Tasks;
+using PerformanceMonitor.Collectors;
+using Xunit;
+
+namespace Darling.Tests;
+
+/// <summary>
+/// #4428: the same row-coherent restart decision <c>QueryStatsRowCoherentResetTests</c> pins for
+/// <c>query_stats</c>, applied to <see cref="PgStatementStatsCollector"/> — with one difference the
+/// PostgreSQL side forces: the series-age signal (<c>stats_since</c>, pg_stat_statements 1.11+) is a
+/// TIMESTAMP on the monitored server's own clock, not an age relative to the collector host's clock, so
+/// placement compares it ONLY against the previous pass's <c>target_now</c> — captured in the SAME read
+/// — never against the collector's own <see cref="CollectorContext.CollectionTime"/>.
+/// </summary>
+public sealed class PgStatementStatsRowCoherentResetTests
+{
+    private const int ServerId = 1;
+    private static DateTime T0 => new(2026, 9, 1, 12, 0, 0, DateTimeKind.Unspecified);
+
+    /// <summary>One row, ordinals matching PgStatementStatsCollector's BuildQuery, 0-29.</summary>
+    private static object[] Row(
+        long queryId, long dbid, long userid, bool toplevel, long calls, double totalExecTimeMs,
+        long rowsReturned, DateTime? statsReset, DateTime? statsSince, DateTime targetNow)
+        => new object[]
+        {
+            queryId, dbid, userid, toplevel, calls, totalExecTimeMs,
+            0.0, 0.0, 0.0, rowsReturned,
+            0L, 0L, 0L, 0L, 0L, 0L, 0.0, 0.0,
+            (object?)null!, (object?)null!, (object?)null!, (object?)null!,
+            0L, 0L, 0L,
+            (object?)null!, (object?)null!,
+            (object?)statsReset ?? DBNull.Value,
+            (object?)statsSince ?? DBNull.Value,
+            targetNow,
+        };
+
+    private static async Task<System.Collections.Generic.List<PgStatementStatsCollector.Row>> RunReadAsync(
+        ICollectorDeltaCalculator deltas, DateTime collectionTime, object[] row)
+    {
+        var reader = new FakeReader(row);
+        var context = new CollectorContext
+        {
+            ServerId = ServerId,
+            ServerName = "target-a",
+            CollectionTime = collectionTime,
+            Deltas = deltas,
+        };
+
+        return await PgStatementStatsCollector.Instance.ReadAsync(reader, context, CancellationToken.None);
+    }
+
+    /// <summary>
+    /// #4431's shape, adapted: calls re-grown past the old value, total time DROPPED (the counter that a
+    /// per-family delta alone would call a reset), stats_since inside the gap on the TARGET clock. The
+    /// whole row credits current values over the real interval.
+    /// </summary>
+    [Fact]
+    public async Task Reset_StatsSinceInsideGap_CreditsCurrentValues()
+    {
+        var deltas = new CollectorDeltaCalculator();
+        var t0Target = T0;
+        var t1Target = T0.AddSeconds(60);
+
+        await RunReadAsync(deltas, T0,
+            Row(1, 1, 1, true, calls: 1, totalExecTimeMs: 57_695_259, rowsReturned: 1,
+                statsReset: null, statsSince: null, targetNow: t0Target));
+
+        /* Restart between t0Target and t1Target: total time FELL (57,695,259 -> 703,943), while calls and
+           rows RE-GREW past their own old values (1 -> 16) — the #4431 field shape. A per-family delta
+           alone would read this as "total time reset, calls +15"; row-coherent placement instead credits
+           every counter as its current value over the real interval. stats_since sits inside the gap on
+           the TARGET clock. */
+        var restartedAt = t0Target.AddSeconds(30);
+        var rows = await RunReadAsync(deltas, T0.AddSeconds(60),
+            Row(1, 1, 1, true, calls: 16, totalExecTimeMs: 703_943, rowsReturned: 16,
+                statsReset: null, statsSince: restartedAt, targetNow: t1Target));
+
+        var restarted = Assert.Single(rows);
+        Assert.Equal(16, restarted.DeltaCalls);
+        Assert.Equal(703_943, restarted.DeltaTotalExecTimeMs);
+        Assert.Equal(16, restarted.DeltaRows);
+        Assert.Equal(60, restarted.SampleIntervalSeconds);
+    }
+
+    /// <summary>
+    /// stats_since BEFORE the previous pass's target_now: the restart happened before we last looked, so
+    /// it is not placeable inside the gap. A decreasing row reports (0, 0).
+    /// </summary>
+    [Fact]
+    public async Task Reset_StatsSinceBeforePreviousTargetNow_ReportsZeroZero()
+    {
+        var deltas = new CollectorDeltaCalculator();
+        var t0Target = T0;
+        var t1Target = T0.AddSeconds(60);
+
+        await RunReadAsync(deltas, T0,
+            Row(1, 1, 1, true, calls: 100, totalExecTimeMs: 5000, rowsReturned: 100,
+                statsReset: null, statsSince: null, targetNow: t0Target));
+
+        /* stats_since is BEFORE t0Target — the entry existed before our previous look, so a decrease here
+           cannot be a "credit the current values" restart. */
+        var beforePreviousLook = t0Target.AddSeconds(-10);
+        var rows = await RunReadAsync(deltas, T0.AddSeconds(60),
+            Row(1, 1, 1, true, calls: 5, totalExecTimeMs: 900, rowsReturned: 5,
+                statsReset: null, statsSince: beforePreviousLook, targetNow: t1Target));
+
+        var reset = Assert.Single(rows);
+        Assert.Equal(0, reset.DeltaCalls);
+        Assert.Equal(0, reset.DeltaTotalExecTimeMs);
+        Assert.Equal(0, reset.DeltaRows);
+        Assert.Equal(0, reset.SampleIntervalSeconds);
+    }
+
+    /// <summary>stats_since absent (PostgreSQL 16, or Aurora without it): strict (0, 0) for a decreasing row.</summary>
+    [Fact]
+    public async Task Reset_StatsSinceAbsent_ReportsZeroZero()
+    {
+        var deltas = new CollectorDeltaCalculator();
+
+        await RunReadAsync(deltas, T0,
+            Row(1, 1, 1, true, calls: 100, totalExecTimeMs: 5000, rowsReturned: 100,
+                statsReset: null, statsSince: null, targetNow: T0));
+
+        var rows = await RunReadAsync(deltas, T0.AddSeconds(60),
+            Row(1, 1, 1, true, calls: 5, totalExecTimeMs: 900, rowsReturned: 5,
+                statsReset: null, statsSince: null, targetNow: T0.AddSeconds(60)));
+
+        var reset = Assert.Single(rows);
+        Assert.Equal(0, reset.DeltaCalls);
+        Assert.Equal(0, reset.DeltaTotalExecTimeMs);
+        Assert.Equal(0, reset.DeltaRows);
+        Assert.Equal(0, reset.SampleIntervalSeconds);
+    }
+
+    /// <summary>
+    /// Clock skew: the target's clock runs several minutes AHEAD of the collector host's clock. An entry
+    /// created BEFORE the previous pass on the TARGET's clock must not be credited, even though its
+    /// stats_since reads "after" the previous pass's context.CollectionTime on the collector's clock —
+    /// proving placement never compares against the collector's own clock.
+    /// </summary>
+    [Fact]
+    public async Task Reset_TargetClockSkew_NeverComparedAgainstCollectorClock()
+    {
+        var deltas = new CollectorDeltaCalculator();
+
+        /* Target clock runs 5 minutes ahead of the collector host: collector sees T0, target's now() is
+           T0 + 5 minutes. */
+        var skew = TimeSpan.FromMinutes(5);
+        var t0Target = T0 + skew;
+        var t1Target = T0.AddSeconds(60) + skew;
+
+        await RunReadAsync(deltas, T0,
+            Row(1, 1, 1, true, calls: 100, totalExecTimeMs: 5000, rowsReturned: 100,
+                statsReset: null, statsSince: null, targetNow: t0Target));
+
+        /* The restart's stats_since is T0 + 90s on the COLLECTOR's clock — later than the collector's own
+           T0 (context.CollectionTime for pass one) — but on the TARGET's clock it is BEFORE t0Target
+           (T0 + 300s), i.e. before the previous pass. A calculator that compared stats_since against the
+           collector's clock would wrongly credit this; comparing against target_now correctly refuses it. */
+        var statsSinceOnCollectorClock = T0.AddSeconds(90);
+        var rows = await RunReadAsync(deltas, T0.AddSeconds(60),
+            Row(1, 1, 1, true, calls: 5, totalExecTimeMs: 900, rowsReturned: 5,
+                statsReset: null, statsSince: statsSinceOnCollectorClock, targetNow: t1Target));
+
+        var reset = Assert.Single(rows);
+        Assert.Equal(0, reset.DeltaCalls);
+        Assert.Equal(0, reset.DeltaTotalExecTimeMs);
+        Assert.Equal(0, reset.DeltaRows);
+        Assert.Equal(0, reset.SampleIntervalSeconds);
+    }
+
+    /// <summary>No reset — every counter only ever increases — takes the ordinary per-family path unchanged.</summary>
+    [Fact]
+    public async Task NoReset_OrdinaryIncrease_Unchanged()
+    {
+        var deltas = new CollectorDeltaCalculator();
+
+        await RunReadAsync(deltas, T0,
+            Row(1, 1, 1, true, calls: 100, totalExecTimeMs: 5000, rowsReturned: 100,
+                statsReset: null, statsSince: null, targetNow: T0));
+
+        var rows = await RunReadAsync(deltas, T0.AddSeconds(60),
+            Row(1, 1, 1, true, calls: 120, totalExecTimeMs: 5200, rowsReturned: 110,
+                statsReset: null, statsSince: null, targetNow: T0.AddSeconds(60)));
+
+        var ordinary = Assert.Single(rows);
+        Assert.Equal(20, ordinary.DeltaCalls);
+        Assert.Equal(200, ordinary.DeltaTotalExecTimeMs);
+        Assert.Equal(10, ordinary.DeltaRows);
+        Assert.Equal(60, ordinary.SampleIntervalSeconds);
+    }
+
+    /// <summary>A minimal multi-call <see cref="DbDataReader"/> over one boxed row, ordinal access only.</summary>
+    private sealed class FakeReader : DbDataReader
+    {
+        private readonly object[] _values;
+        private int _row = -1;
+
+        public FakeReader(object[] values) => _values = values;
+
+        private object Raw(int ordinal) => _values[ordinal];
+
+        public override bool IsDBNull(int ordinal) => Raw(ordinal) is DBNull;
+        public override DateTime GetDateTime(int ordinal) => (DateTime)Raw(ordinal);
+        public override long GetInt64(int ordinal) => Convert.ToInt64(Raw(ordinal));
+        public override int GetInt32(int ordinal) => Convert.ToInt32(Raw(ordinal));
+        public override bool GetBoolean(int ordinal) => (bool)Raw(ordinal);
+        public override double GetDouble(int ordinal) => Convert.ToDouble(Raw(ordinal));
+        public override string GetString(int ordinal) => (string)Raw(ordinal);
+        public override T GetFieldValue<T>(int ordinal) => (T)Raw(ordinal);
+        public override object GetValue(int ordinal) => Raw(ordinal);
+        public override int FieldCount => _values.Length;
+        public override bool Read() => ++_row == 0;
+        public override Task<bool> ReadAsync(CancellationToken cancellationToken) => Task.FromResult(Read());
+        public override bool HasRows => true;
+
+        public override int Depth => 0;
+        public override bool IsClosed => false;
+        public override int RecordsAffected => 0;
+        public override object this[int ordinal] => Raw(ordinal);
+        public override object this[string name] => throw new NotSupportedException("ordinal access only");
+        public override int GetOrdinal(string name) => throw new NotSupportedException("ordinal access only");
+        public override string GetName(int ordinal) => throw new NotSupportedException("ordinal access only");
+        public override bool NextResult() => false;
+        public override IEnumerator GetEnumerator() => throw new NotSupportedException();
+        public override int GetValues(object[] values) => throw new NotSupportedException();
+        public override string GetDataTypeName(int ordinal) => throw new NotSupportedException();
+        public override Type GetFieldType(int ordinal) => Raw(ordinal).GetType();
+        public override byte GetByte(int ordinal) => throw new NotSupportedException();
+        public override long GetBytes(int ordinal, long dataOffset, byte[]? buffer, int bufferOffset, int length) => throw new NotSupportedException();
+        public override char GetChar(int ordinal) => throw new NotSupportedException();
+        public override long GetChars(int ordinal, long dataOffset, char[]? buffer, int bufferOffset, int length) => throw new NotSupportedException();
+        public override decimal GetDecimal(int ordinal) => throw new NotSupportedException();
+        public override float GetFloat(int ordinal) => throw new NotSupportedException();
+        public override Guid GetGuid(int ordinal) => throw new NotSupportedException();
+        public override short GetInt16(int ordinal) => throw new NotSupportedException();
+    }
+}

Lite.Tests/PgStatementStatsDeltaSkipTests.cs

@@ -56,8 +56,11 @@ public class PgStatementStatsDeltaSkipTests
     /// on the Aurora-only six, and — since #3653 A5 — a NULL <c>statements_stats_reset</c> at ordinal 27,
     /// which the epoch check reads as "unknown" and so never as a change; these tests are about the skip,
     /// not the epoch, and a NULL keeps the epoch inert. <c>ServerEpochTests</c> drives the epoch itself.)
+    /// Since #4428, ordinal 28 is <c>stats_since</c> (NULL by default here — the row-coherent restart
+    /// placement is tested separately in <c>Darling.Tests</c>) and ordinal 29 is <c>target_now</c>, the
+    /// target's own clock, defaulted to the collection time a scenario is about to pass to <c>ReadAsync</c>.
     /// </summary>
-    private static object[] Row(long queryId, long calls, double totalExecTimeMs, long rowsReturned = 0, long databaseId = 1, long userId = 1) => new object[]
+    private static object[] Row(long queryId, long calls, double totalExecTimeMs, long rowsReturned = 0, long databaseId = 1, long userId = 1, DateTime? statsSince = null, DateTime? targetNow = null) => new object[]
     {
         queryId, databaseId, userId, true, calls, totalExecTimeMs,
         0d, 0d, 0d, rowsReturned,
@@ -67,6 +70,8 @@ public class PgStatementStatsDeltaSkipTests
         0L, 0L, 0L,
         DBNull.Value, DBNull.Value,
         DBNull.Value,
+        statsSince.HasValue ? (object)statsSince.Value : DBNull.Value,
+        targetNow ?? T0,
     };
 
     private static async Task<List<PgStatementStatsCollector.Row>> ReadAsync(

Lite.Tests/PgStatementStatsFlavorTests.cs

@@ -30,8 +30,9 @@ namespace Lite.Tests;
 /// </para>
 ///
 /// <para>
-/// The ordinals are the load-bearing detail. Both queries select the same 28 columns in the same order (the
-/// 28th, appended by #3653 A5, is the statements epoch <c>stats_reset</c>, read and not stored) — the
+/// The ordinals are the load-bearing detail. Both queries select the same 30 columns in the same order (the
+/// 28th, appended by #3653 A5, is the statements epoch <c>stats_reset</c>, read and not stored; the 29th and
+/// 30th, appended by #4428, are <c>stats_since</c> and <c>target_now</c>, also read and not stored) — the
 /// vanilla one fills Aurora's six with typed NULL literals — so <c>ReadAsync</c>, <c>PayloadColumns</c> and
 /// <c>WritePayload</c> stay single implementations. A shorter vanilla SELECT would have meant a second reader
 /// whose ordinals could drift from this one, which is exactly the failure the per-major column naming in this
@@ -110,10 +111,13 @@ public void BothFlavorsSelectTheSameColumnsInTheSameOrder()
 
         Assert.Equal(aurora, vanilla);
         /* The SELECT list is the payload minus the four columns computed on the client (the three deltas
-           and, since V128 (#3540), the interval they accrued over) PLUS the one column read and not stored:
-           the statements epoch stats_reset (#3653 A5), last, so every stored ordinal is where it was. */
-        Assert.Equal(PgStatementStatsCollector.Instance.PayloadColumns.Count - 4 + 1, aurora.Count);
-        Assert.Equal("statements_stats_reset", aurora[^1]);
+           and, since V128 (#3540), the interval they accrued over) PLUS the three columns read and not
+           stored: the statements epoch stats_reset (#3653 A5), and — since #4428 — stats_since and
+           target_now, last, so every stored ordinal is where it was. */
+        Assert.Equal(PgStatementStatsCollector.Instance.PayloadColumns.Count - 4 + 3, aurora.Count);
+        Assert.Equal("statements_stats_reset", aurora[^3]);
+        Assert.Equal("stats_since", aurora[^2]);
+        Assert.Equal("target_now", aurora[^1]);
     }
 
     /// <summary>

Lite.Tests/ServerEpochTests.cs

@@ -835,7 +835,12 @@ public void DrainDiscontinuities_ReturnsInOrder_ThenEmpty()
 
     /* ---------------- helpers ---------------- */
 
-    /// <summary>A pg_statement_stats reader row in ordinal order, the statements epoch at 27.</summary>
+    /// <summary>
+    /// A pg_statement_stats reader row in ordinal order, the statements epoch at 27. Since #4428, ordinal
+    /// 28 is <c>stats_since</c> (NULL here — these tests are about the epoch forget, not the row-coherent
+    /// restart placement, which <c>Darling.Tests</c> covers separately) and ordinal 29 is <c>target_now</c>,
+    /// defaulted to the same moment as <paramref name="statsReset"/> when it is a <see cref="DateTime"/>.
+    /// </summary>
     private static object[] StatementRow(long queryId, long calls, object statsReset) => new object[]
     {
         queryId, 1L, 1L, true, calls, 100d,
@@ -846,6 +851,8 @@ public void DrainDiscontinuities_ReturnsInOrder_ThenEmpty()
         0L, 0L, 0L,
         DBNull.Value, DBNull.Value,
         statsReset,
+        DBNull.Value,
+        statsReset is DateTime resetTime ? resetTime : new DateTime(2026, 9, 19, 3, 59, 0, DateTimeKind.Utc),
     };
 
     /// <summary>Reaches the protected restart-seed hook, to stage the host-restart case.</summary>

PerformanceMonitor.Collectors/CollectorDeltaCalculator.cs

@@ -187,6 +187,17 @@ public static DateTime SeedCutoff()
         return updated.Previous;
     }
 
+    /// <inheritdoc />
+    ///
+    /// <para>#4428: the same window-rolling machinery <see cref="PreviousPass(int, string, DateTime?)"/>
+    /// (private, above) already keeps for a collector-clock caller, exposed under the interface's own
+    /// name for a caller that tracks a DIFFERENT clock. <paramref name="group"/> shares the private
+    /// method's dictionary rather than a second one, on the contract stated on the interface member: a
+    /// caller's group name never collides with any <c>collectorName</c> an ordinary delta call passes, so
+    /// one dictionary safely serves both without either clock's window disturbing the other's.</para>
+    public DateTime? PreviousPass(int serverId, string group, DateTime observedTime)
+        => PreviousPass(serverId, group, (DateTime?)observedTime);
+
     /// <summary>
     /// The discontinuity accounts (#3653 A5) a definition handed to <see cref="ClearServer"/> or
     /// <see cref="ClearGroups"/> and no host has logged yet: serverId -> the lines, in the order they were