Fix poller race condition that missed long-running builds

src/Tiger.Tests/BuildIngestionServiceTests.cs

@@ -326,4 +326,5 @@ private void InsertSampleBuild(int buildId)
             SourceBranch = "main",
         });
     }
+
 }

src/Tiger.Tests/BuildPollerTests.cs

@@ -74,42 +74,73 @@ public async Task StartStop_IsRunning()
     }
 
     [Fact]
-    public async Task PollsAndCallsOnNewBuilds()
+    public void FilterNewBuilds_NewBuildsAreIncluded()
     {
+        var poller = CreatePoller();
         var builds = new List<AzdoBuild>
         {
-            new() { Id = 5, BuildNumber = "5", Status = "completed", Result = "succeeded", Uri = "", SourceBranch = "main", DefinitionName = "def" },
-            new() { Id = 10, BuildNumber = "10", Status = "completed", Result = "failed", Uri = "", SourceBranch = "main", DefinitionName = "def" },
-            new() { Id = 3, BuildNumber = "3", Status = "inProgress", Uri = "", SourceBranch = "main", DefinitionName = "def" },
+            MakeBuild(1, finishTime: new DateTime(2025, 1, 1, 12, 0, 0)),
+            MakeBuild(2, finishTime: new DateTime(2025, 1, 1, 13, 0, 0)),
         };
 
-        var config = new TigerConfig
-        {
-            PollIntervalSeconds = 3600,
-            Sources = [new AzdoSource { Organization = "org", Project = "proj" }],
-        };
+        var result = poller.FilterNewBuilds("org", builds);
+        Assert.Equal(2, result.Count);
+    }
 
-        var captured = new List<(string org, string proj, List<AzdoBuild> builds)>();
-        var poller = new BuildPoller(config, _db, new AzdoClientFactory((org, proj) => throw new NotImplementedException()))
-        {
-            OnNewBuilds = (org, proj, newBuilds) =>
-            {
-                captured.Add((org, proj, newBuilds));
-                return Task.CompletedTask;
-            }
-        };
+    [Fact]
+    public void FilterNewBuilds_AlreadyIngestedBuildsAreSkipped()
+    {
+        var poller = CreatePoller();
+        var build = MakeBuild(1, finishTime: new DateTime(2025, 1, 1, 12, 0, 0));
 
-        // Directly test PollSourceAsync would be ideal but it's private.
-        // Instead test watermark behavior which is the core logic.
-        poller.SetWatermark("org", "proj", 0);
-        Assert.Equal(0, poller.GetWatermark("org", "proj"));
-        poller.SetWatermark("org", "proj", 10);
-        Assert.Equal(10, poller.GetWatermark("org", "proj"));
+        // Pre-insert the build
+        var service = new BuildIngestionService(_db);
+        service.InsertBuild("org", "proj", build);
+
+        var result = poller.FilterNewBuilds("org", [build]);
+        Assert.Empty(result);
+    }
+
+    [Fact]
+    public void FilterNewBuilds_LongRunningBuildNotMissed()
+    {
+        // Simulates the scenario where a long-running build completes after
+        // other builds with higher IDs have already been ingested.
+        var poller = CreatePoller();
+        var service = new BuildIngestionService(_db);
+
+        // Build 100 was ingested in a previous poll cycle
+        var earlyBuild = MakeBuild(100, finishTime: new DateTime(2025, 1, 1, 10, 0, 0));
+        service.InsertBuild("org", "proj", earlyBuild);
+
+        // Build 200 was also ingested (higher ID)
+        var laterBuild = MakeBuild(200, finishTime: new DateTime(2025, 1, 1, 11, 0, 0));
+        service.InsertBuild("org", "proj", laterBuild);
+
+        // Build 150 was long-running and just completed — it's new
+        var longRunning = MakeBuild(150, finishTime: new DateTime(2025, 1, 1, 12, 0, 0));
+
+        // The API returns all three (completed), our filter should pick up only 150
+        var result = poller.FilterNewBuilds("org", [earlyBuild, longRunning, laterBuild]);
+        Assert.Single(result);
+        Assert.Equal(150, result[0].Id);
     }
 
     private BuildPoller CreatePoller()
     {
         var config = new TigerConfig { Sources = [] };
         return new BuildPoller(config, _db, new AzdoClientFactory((org, proj) => throw new NotImplementedException()));
     }
+
+    private static AzdoBuild MakeBuild(int id, DateTime? finishTime = null) => new()
+    {
+        Id = id,
+        BuildNumber = $"2025.{id}",
+        Status = "completed",
+        Result = "failed",
+        Uri = "",
+        SourceBranch = "refs/heads/main",
+        DefinitionName = "test-def",
+        FinishTime = finishTime,
+    };
 }

src/Tiger/AzdoClient.cs

@@ -113,13 +113,17 @@ private string GetBuildUri(int buildId) =>
         return null;
     }
 
-    public async Task<List<AzdoBuild>> GetRecentBuildsAsync(int? definitionId = null, int top = 10, CancellationToken ct = default)
+    public async Task<List<AzdoBuild>> GetRecentBuildsAsync(int? definitionId = null, int top = 10, string? statusFilter = null, CancellationToken ct = default)
     {
         var url = $"_apis/build/builds?api-version=7.1&$top={top}";
         if (definitionId is not null)
         {
             url += $"&definitions={definitionId}";
         }
+        if (statusFilter is not null)
+        {
+            url += $"&statusFilter={Uri.EscapeDataString(statusFilter)}";
+        }
 
         var response = await HttpClient.GetAsync(url, ct);
         response.EnsureSuccessStatusCode();
@@ -175,13 +179,17 @@ public async Task<List<AzdoBuild>> GetCompletedBuildsSinceAsync(
         return all;
     }
 
-    public async Task<List<AzdoBuild>> GetBuildsForRepositoryAsync(string repository, int top = 10, string? reasonFilter = null, CancellationToken ct = default)
+    public async Task<List<AzdoBuild>> GetBuildsForRepositoryAsync(string repository, int top = 10, string? reasonFilter = null, string? statusFilter = null, CancellationToken ct = default)
     {
         var url = $"_apis/build/builds?api-version=7.1&$top={top}&repositoryId={Uri.EscapeDataString(repository)}&repositoryType=GitHub";
         if (reasonFilter is not null)
         {
             url += $"&reasonFilter={Uri.EscapeDataString(reasonFilter)}";
         }
+        if (statusFilter is not null)
+        {
+            url += $"&statusFilter={Uri.EscapeDataString(statusFilter)}";
+        }
 
         var response = await HttpClient.GetAsync(url, ct);
         response.EnsureSuccessStatusCode();

src/Tiger/BuildPoller.cs

@@ -89,29 +89,30 @@ private async Task PollLoopAsync(CancellationToken ct)
 
     private async Task PollSourceAsync(AzdoSource source, CancellationToken ct)
     {
-        var watermark = GetWatermark(source.Organization, source.Project);
         var client = _clientFactory.Create(source.Organization, source.Project);
 
-        // Fetch recent completed builds, filtered by repository if configured
+        // Fetch recent completed builds, filtered by repository if configured.
+        // We use statusFilter=completed so only terminal builds are returned,
+        // avoiding the old watermark race where in-progress builds caused the
+        // watermark to advance past long-running builds.
         List<AzdoBuild> builds;
         if (source.Repositories.Count > 0)
         {
             builds = [];
             foreach (var repo in source.Repositories)
             {
-                var repoBuilds = await client.GetBuildsForRepositoryAsync(repo, top: 50);
+                var repoBuilds = await client.GetBuildsForRepositoryAsync(repo, top: 50, statusFilter: "completed");
                 builds.AddRange(repoBuilds);
             }
         }
         else
         {
-            builds = await client.GetRecentBuildsAsync(top: 50);
+            builds = await client.GetRecentBuildsAsync(top: 50, statusFilter: "completed");
         }
 
-        var newBuilds = builds
-            .Where(b => b.Id > watermark && b.Status == "completed")
-            .OrderBy(b => b.Id)
-            .ToList();
+        // Filter to builds not yet in the DB. Already-ingested builds are
+        // skipped because INSERT OR IGNORE on tasks makes re-insertion a no-op.
+        var newBuilds = FilterNewBuilds(source.Organization, builds);
 
         if (newBuilds.Count == 0) return;
 
@@ -123,14 +124,37 @@ private async Task PollSourceAsync(AzdoSource source, CancellationToken ct)
             await OnNewBuilds(source.Organization, source.Project, newBuilds);
         }
 
-        // Update watermark to the highest build ID we processed
-        var newWatermark = newBuilds.Max(b => b.Id);
-        SetWatermark(source.Organization, source.Project, newWatermark);
-
         _log?.Success("Poller",
             $"Ingested {newBuilds.Count} builds for {source.Organization}/{source.Project}");
     }
 
+    /// <summary>
+    /// Returns builds that are not yet in the database.
+    /// </summary>
+    internal List<AzdoBuild> FilterNewBuilds(string organization, List<AzdoBuild> builds)
+    {
+        var result = new List<AzdoBuild>();
+        foreach (var build in builds)
+        {
+            if (!BuildExists(organization, build.Id))
+            {
+                result.Add(build);
+            }
+        }
+        return result;
+    }
+
+    private bool BuildExists(string organization, int buildId)
+    {
+        return _db.WithCommand(cmd =>
+        {
+            cmd.CommandText = "SELECT 1 FROM builds WHERE organization = @org AND build_id = @buildId";
+            cmd.Parameters.AddWithValue("@org", organization);
+            cmd.Parameters.AddWithValue("@buildId", buildId);
+            return cmd.ExecuteScalar() is not null;
+        });
+    }
+
     internal int GetWatermark(string organization, string project)
     {
         return _db.WithCommand(cmd =>