After a job-history identity reset, the collector re-reads once instead of on every run (#4487)

Darling/Darling.Tests/DarlingCollectorRunnerTests.cs

@@ -827,6 +827,13 @@ public void TheBoundedServerWatermarkForDatabaseSql_PredicatesOnThePartitioningC
     /// #4197: job_history's numeric (instance_id) watermark read carried NO bound at all before this
     /// issue — this pins that it now has the same collection_time bound as its timestamp siblings, on the
     /// PARTITIONING column rather than on instance_id itself (which does not partition the table).
+    ///
+    /// <para>#4487: BOTH forms are now scoped to the server's NEWEST collected batch (an inner
+    /// <c>MAX(collection_time)</c> subquery), not an all-time max — an all-time max survives an identity
+    /// reseed at the OLD epoch's higher number forever, which is the field bug. So the UNBOUNDED form now
+    /// legitimately mentions <c>collection_time</c> too (in its own inner subquery, with no <c>&gt;</c>
+    /// lower bound), and this pin asserts the bounded/unbounded distinction by the presence of the $2
+    /// parameter rather than by "unbounded never says collection_time".</para>
     /// </summary>
     [Fact]
     public void TheBoundedServerWatermarkInstanceIdSql_PredicatesOnThePartitioningColumn()
@@ -835,9 +842,12 @@ public void TheBoundedServerWatermarkInstanceIdSql_PredicatesOnThePartitioningCo
         var unbounded = DarlingCollectorRunner.BuildServerWatermarkInstanceIdSql("job_history", "instance_id", bounded: false);
 
         Assert.Contains("collection_time > $2", bounded, StringComparison.Ordinal);
-        Assert.DoesNotContain("collection_time", unbounded, StringComparison.Ordinal);
+        Assert.DoesNotContain("$2", unbounded, StringComparison.Ordinal);
         Assert.Contains("MAX(instance_id)", bounded, StringComparison.Ordinal);
         Assert.Contains("MAX(instance_id)", unbounded, StringComparison.Ordinal);
+        /* #4487: both forms now scope to the newest collected batch via an inner MAX(collection_time). */
+        Assert.Contains("collection_time = (SELECT MAX(collection_time)", bounded, StringComparison.Ordinal);
+        Assert.Contains("collection_time = (SELECT MAX(collection_time)", unbounded, StringComparison.Ordinal);
     }
 
     /// <summary>

Darling/Darling.Tests/JobHistoryIdentityEpochTests.cs

@@ -88,11 +88,93 @@ public void SteadyStateFilter_GuardsOnTheTargetsOwnMaximum_AndStillBindsTheWater
         /* The honest-identity arm is still the pre-#3885 predicate, character for character. */
         Assert.Contains(DevWatermarkPredicate, query.Text, StringComparison.Ordinal);
 
-        /* One parameter, still BigInt: instance_id is a bigint IDENTITY and the millions are real. */
-        var p = Assert.Single(query.Parameters);
-        Assert.Equal("@last_instance_id", p.Name);
+        /* Two parameters: instance_id is still a bigint IDENTITY and the millions are real; #4487 adds
+           the failback bound alongside it (null here — this context sets no timestamp Watermark). */
+        Assert.Equal(2, query.Parameters.Count);
+        var p = Assert.Single(query.Parameters, x => x.Name == "@last_instance_id");
         Assert.Equal(11_500_000L, p.Value);
         Assert.Equal(CollectorParameterType.BigInt, p.Type);
+        var bound = Assert.Single(query.Parameters, x => x.Name == "@min_run_datetime");
+        Assert.Null(bound.Value);
+        Assert.Equal(CollectorParameterType.DateTime2, bound.Type);
+        Assert.Contains("@min_run_datetime IS NULL", query.Text, StringComparison.Ordinal);
+    }
+
+    /// <summary>
+    /// #4487: the failback (A→B→A) bound. A store holding an older, higher-id epoch (A) and a newer,
+    /// lower-id epoch (B) supplies BOTH watermarks in the same round trip the honest arm already had —
+    /// context.Watermark is job_history's own timestamp watermark, read for the archival-emptied distinction,
+    /// and repurposed here as the failback anchor with no second host read. BuildQuery must derive
+    /// @min_run_datetime as that stored maximum minus FailbackLookbackDays, and the honest arm's text must
+    /// carry the run_date/run_datetime bound tied to that parameter, not a copy that could drift from
+    /// ArchivalEmptyWindowPredicate's own decode.
+    /// </summary>
+    [Fact]
+    public void SteadyStateFilter_BindsTheFailbackBound_WhenTheHostSuppliesAStoredWatermark()
+    {
+        var storedNewestRunDateTime = new DateTime(2026, 9, 27, 3, 0, 0, DateTimeKind.Unspecified);
+        var context = MakeContext(numericWatermark: 200L);
+        context.Watermark = storedNewestRunDateTime;
+
+        var query = JobHistoryCollector.Instance.BuildQuery(context);
+
+        Assert.Equal(7, JobHistoryCollector.FailbackLookbackDays);
+        var bound = Assert.Single(query.Parameters, x => x.Name == "@min_run_datetime");
+        Assert.Equal(storedNewestRunDateTime.AddDays(-7), bound.Value);
+        Assert.Equal(CollectorParameterType.DateTime2, bound.Type);
+
+        /* The exact decoded-run_datetime shape, tied to the CAPPED @min_run_datetime rather than a bare
+           parameter or GETDATE() — the same idiom ArchivalEmptyFilter uses, anchored differently. */
+        const string capExpression = "CASE WHEN @min_run_datetime > DATEADD(DAY, -7, GETDATE()) THEN DATEADD(DAY, -7, GETDATE()) ELSE @min_run_datetime END";
+        Assert.Contains($"CONVERT(integer, CONVERT(varchar(8), {capExpression}, 112))", query.Text, StringComparison.Ordinal);
+        Assert.Contains($") >= {capExpression}", query.Text, StringComparison.Ordinal);
+        /* Only inside the HONEST arm (paired with the un-guarded instance_id predicate), not the
+           regressed arm, which keeps its own independent 24h GETDATE() window untouched. */
+        Assert.Contains(DevWatermarkPredicate + "\r\n              AND ", query.Text, StringComparison.Ordinal);
+    }
+
+    /// <summary>
+    /// #4487 review finding: a future-dated stored run must not starve the steady arm. The cap forces
+    /// @min_run_datetime's EFFECT in the SQL down to FailbackLookbackDays before the TARGET's own clock
+    /// whenever the parameter value would otherwise be later than that — asserted here as the literal
+    /// CASE expression the collector emits, since the clamp runs on the target's server-side GETDATE(),
+    /// not on a value this test can observe from the parameter alone.
+    /// </summary>
+    [Fact]
+    public void SteadyStateFilter_CapsTheFailbackBound_AtTheTargetsOwnRecentPast()
+    {
+        var context = MakeContext(numericWatermark: 200L);
+        context.Watermark = new DateTime(2099, 1, 1, 0, 0, 0, DateTimeKind.Unspecified);
+
+        var query = JobHistoryCollector.Instance.BuildQuery(context);
+
+        /* The parameter itself still carries the raw (uncapped) stored-watermark-minus-lookback value —
+           the SQL, not the parameter, is where the cap is applied, so a future-dated store never reaches
+           the target as a future bound. */
+        var bound = Assert.Single(query.Parameters, x => x.Name == "@min_run_datetime");
+        Assert.Equal(new DateTime(2098, 12, 25, 0, 0, 0, DateTimeKind.Unspecified), bound.Value);
+
+        Assert.Contains(
+            "CASE WHEN @min_run_datetime > DATEADD(DAY, -7, GETDATE()) THEN DATEADD(DAY, -7, GETDATE()) ELSE @min_run_datetime END",
+            query.Text, StringComparison.Ordinal);
+    }
+
+    /// <summary>
+    /// #4487: no stored timestamp watermark (a true first run, or a Lite store emptied by archival) means
+    /// nothing to fail back FROM, so the bound must impose no restriction — @min_run_datetime is null and
+    /// the predicate's IS NULL branch is what the guard relies on for it, pinned here at the parameter
+    /// level so the SQL-shape pin above and this null-input case cannot silently diverge.
+    /// </summary>
+    [Fact]
+    public void SteadyStateFilter_MinRunDateTimeIsNull_WhenTheHostHasNoStoredWatermarkYet()
+    {
+        var context = MakeContext(numericWatermark: 200L);
+        Assert.Null(context.Watermark);
+
+        var query = JobHistoryCollector.Instance.BuildQuery(context);
+
+        var bound = Assert.Single(query.Parameters, x => x.Name == "@min_run_datetime");
+        Assert.Null(bound.Value);
     }
 
     [Fact]
@@ -194,13 +276,13 @@ public void SelfHeals_OnTheVeryNextRun_NoOperatorAction()
         /* Run one: the store's watermark is the pre-reseed maximum, so the guarded statement's regressed
            arm takes the bounded window and stores rows carrying the NEW epoch's ids. */
         var regressed = JobHistoryCollector.Instance.BuildQuery(MakeContext(numericWatermark: 11_500_000L));
-        Assert.Equal(11_500_000L, Assert.Single(regressed.Parameters).Value);
+        Assert.Equal(11_500_000L, Assert.Single(regressed.Parameters, x => x.Name == "@last_instance_id").Value);
 
         /* Run two: the host re-reads SELECT MAX(instance_id) and now gets the new epoch's maximum, so the
            same statement's honest arm applies - one run of fallback per reseed, not a mode it stays in.
            The text is identical between the runs; only the bound parameter moved. */
         var healed = JobHistoryCollector.Instance.BuildQuery(MakeContext(numericWatermark: 4_000L));
-        Assert.Equal(4_000L, Assert.Single(healed.Parameters).Value);
+        Assert.Equal(4_000L, Assert.Single(healed.Parameters, x => x.Name == "@last_instance_id").Value);
         Assert.Equal(regressed.Text, healed.Text);
         Assert.Contains(DevWatermarkPredicate, healed.Text, StringComparison.Ordinal);
     }

Darling/Darling.Tests/JobHistoryNaturalKeyDedupeTests.cs

@@ -0,0 +1,208 @@
+/*
+ * 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.Generic;
+using System.Linq;
+using PerformanceMonitor.Collectors;
+using Xunit;
+
+namespace Darling.Tests;
+
+/// <summary>
+/// Pure pins for job_history's natural-key pre-insert dedupe (#4487): <see cref="JobHistoryCollector"/>
+/// implements <see cref="INaturalKeyDedupedCollector{TRow}"/>, its natural key is the four columns
+/// <c>instance_id</c>, <c>job_id</c>, <c>step_id</c>, <c>run_datetime</c> (excluding <c>server_id</c>,
+/// which the host's stored-key read already scopes to), and <see cref="JobHistoryCollector.DropAlreadyStored"/>
+/// drops exact tuples only — no partial match on fewer than all four columns drops a row.
+///
+/// <para>RED on dev: <c>INaturalKeyDedupedCollector</c> does not exist there and <c>JobHistoryCollector</c>
+/// declares no <c>NaturalKeyColumns</c>/<c>GetNaturalKey</c>/<c>DropAlreadyStored</c> members — a compile
+/// failure against <c>origin/dev</c>, not a runtime mismatch.</para>
+/// </summary>
+public sealed class JobHistoryNaturalKeyDedupeTests
+{
+    private static JobHistoryCollector.Row MakeRow(long instanceId, string jobId, int stepId, DateTime runDateTime) =>
+        new() { InstanceId = instanceId, JobId = jobId, StepId = stepId, RunDateTime = runDateTime, JobName = "j1", RunStatus = 1 };
+
+    [Fact]
+    public void JobHistoryCollector_ImplementsTheNaturalKeyDedupeContract()
+    {
+        Assert.IsAssignableFrom<INaturalKeyDedupedCollector<JobHistoryCollector.Row>>(JobHistoryCollector.Instance);
+    }
+
+    [Fact]
+    public void NaturalKeyColumns_IsTheFourColumns_ExcludingServerId()
+    {
+        Assert.Equal(
+            new[] { "instance_id", "job_id", "step_id", "run_datetime" },
+            JobHistoryCollector.Instance.NaturalKeyColumns);
+    }
+
+    [Fact]
+    public void GetNaturalKey_ReturnsTheFourFields_InTupleOrder()
+    {
+        var runDateTime = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified);
+        var row = MakeRow(101, "job-a", 3, runDateTime);
+
+        var key = JobHistoryCollector.Instance.GetNaturalKey(row);
+
+        Assert.Equal(101L, key.InstanceId);
+        Assert.Equal("job-a", key.JobId);
+        Assert.Equal(3, key.StepId);
+        Assert.Equal(runDateTime, key.RunDateTime);
+    }
+
+    [Theory]
+    [InlineData(7, 0)]
+    [InlineData(10, 10)]
+    [InlineData(19, 10)]
+    [InlineData(0, 0)]
+    public void ToMicroseconds_TruncatesSubMicrosecondTicks_KeepingTheWholeMicrosecond(long extraTicks, long expectedExtraTicks)
+    {
+        var wholeMicrosecond = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Utc);
+        var withSubMicrosecondTicks = wholeMicrosecond.AddTicks(extraTicks);
+
+        var truncated = JobHistoryCollector.ToMicroseconds(withSubMicrosecondTicks);
+
+        Assert.Equal(wholeMicrosecond.AddTicks(expectedExtraTicks), truncated);
+    }
+
+    [Fact]
+    public void ToMicroseconds_PreservesTheDateTimeKind()
+    {
+        var utc = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Utc).AddTicks(7);
+        var unspecified = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified).AddTicks(7);
+
+        Assert.Equal(DateTimeKind.Utc, JobHistoryCollector.ToMicroseconds(utc).Kind);
+        Assert.Equal(DateTimeKind.Unspecified, JobHistoryCollector.ToMicroseconds(unspecified).Kind);
+    }
+
+    [Fact]
+    public void GetNaturalKey_TruncatesSubMicrosecondTicks_SoABatchRowMatchesAWholeMicrosecondStoredKey()
+    {
+        var wholeMicrosecond = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified);
+        var row = MakeRow(500, "job-a", 0, wholeMicrosecond.AddTicks(7));
+
+        var key = JobHistoryCollector.Instance.GetNaturalKey(row);
+
+        Assert.Equal(wholeMicrosecond, key.RunDateTime);
+    }
+
+    [Fact]
+    public void DropAlreadyStored_DropsARowWithSubMicrosecondTicks_AgainstAStoredKeyAtTheWholeMicrosecond()
+    {
+        var wholeMicrosecond = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified);
+        var subMicrosecondRow = MakeRow(500, "job-a", 0, wholeMicrosecond.AddTicks(7));
+        var rows = new List<JobHistoryCollector.Row> { subMicrosecondRow };
+
+        var storedKeys = new HashSet<(long InstanceId, string JobId, int StepId, DateTime RunDateTime)>
+        {
+            (500, "job-a", 0, wholeMicrosecond),
+        };
+
+        var kept = JobHistoryCollector.Instance.DropAlreadyStored(rows, storedKeys);
+
+        Assert.Empty(kept);
+    }
+
+    [Fact]
+    public void DropAlreadyStored_WithAnEmptyStoredSet_ReturnsEveryRow()
+    {
+        var runDateTime = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified);
+        var rows = new List<JobHistoryCollector.Row>
+        {
+            MakeRow(1, "job-a", 0, runDateTime),
+            MakeRow(2, "job-b", 1, runDateTime),
+        };
+        var storedKeys = new HashSet<(long InstanceId, string JobId, int StepId, DateTime RunDateTime)>();
+
+        var kept = JobHistoryCollector.Instance.DropAlreadyStored(rows, storedKeys);
+
+        Assert.Same(rows, kept);
+        Assert.Equal(2, kept.Count);
+    }
+
+    [Fact]
+    public void DropAlreadyStored_DropsOnlyTheExactTuple_KeepingASameKeyRowWithADifferentJobId()
+    {
+        var runDateTime = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified);
+        var storedRow = MakeRow(500, "job-a", 0, runDateTime);
+        var sameEverythingElseDifferentJobId = MakeRow(500, "job-b", 0, runDateTime);
+        var rows = new List<JobHistoryCollector.Row> { storedRow, sameEverythingElseDifferentJobId };
+
+        var storedKeys = new HashSet<(long InstanceId, string JobId, int StepId, DateTime RunDateTime)>
+        {
+            (500, "job-a", 0, runDateTime),
+        };
+
+        var kept = JobHistoryCollector.Instance.DropAlreadyStored(rows, storedKeys);
+
+        Assert.Single(kept);
+        Assert.Same(sameEverythingElseDifferentJobId, kept[0]);
+    }
+
+    [Fact]
+    public void DropAlreadyStored_DropsOnlyTheExactTuple_KeepingASameKeyRowWithADifferentStepId()
+    {
+        var runDateTime = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified);
+        var storedRow = MakeRow(500, "job-a", 0, runDateTime);
+        var sameEverythingElseDifferentStepId = MakeRow(500, "job-a", 1, runDateTime);
+        var rows = new List<JobHistoryCollector.Row> { storedRow, sameEverythingElseDifferentStepId };
+
+        var storedKeys = new HashSet<(long InstanceId, string JobId, int StepId, DateTime RunDateTime)>
+        {
+            (500, "job-a", 0, runDateTime),
+        };
+
+        var kept = JobHistoryCollector.Instance.DropAlreadyStored(rows, storedKeys);
+
+        Assert.Single(kept);
+        Assert.Same(sameEverythingElseDifferentStepId, kept[0]);
+    }
+
+    [Fact]
+    public void DropAlreadyStored_DropsAnExactTuple_AndKeepsARowNotInTheStoredSet()
+    {
+        var runDateTime = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified);
+        var duplicate = MakeRow(500, "job-a", 0, runDateTime);
+        var newRow = MakeRow(600, "job-c", 0, runDateTime);
+        var rows = new List<JobHistoryCollector.Row> { duplicate, newRow };
+
+        var storedKeys = new HashSet<(long InstanceId, string JobId, int StepId, DateTime RunDateTime)>
+        {
+            (500, "job-a", 0, runDateTime),
+        };
+
+        var kept = JobHistoryCollector.Instance.DropAlreadyStored(rows, storedKeys);
+
+        Assert.Single(kept);
+        Assert.Same(newRow, kept[0]);
+    }
+
+    [Fact]
+    public void DropAlreadyStored_PreservesRelativeOrder_AndMutatesNeitherArgument()
+    {
+        var runDateTime = new DateTime(2026, 9, 20, 10, 0, 0, DateTimeKind.Unspecified);
+        var kept1 = MakeRow(1, "job-a", 0, runDateTime);
+        var dropped = MakeRow(2, "job-b", 0, runDateTime);
+        var kept2 = MakeRow(3, "job-c", 0, runDateTime);
+        var rows = new List<JobHistoryCollector.Row> { kept1, dropped, kept2 };
+        var rowsSnapshot = rows.ToList();
+
+        var storedKeys = new HashSet<(long InstanceId, string JobId, int StepId, DateTime RunDateTime)>
+        {
+            (2, "job-b", 0, runDateTime),
+        };
+
+        var kept = JobHistoryCollector.Instance.DropAlreadyStored(rows, storedKeys);
+
+        Assert.Equal(new[] { kept1, kept2 }, kept);
+        Assert.Equal(rowsSnapshot, rows);
+    }
+}