Review 1.2: bound the activity collector and avoid duplicate rows

The open transactions and long requests had no TOP, so a busy server could
return (and store in the snapshot every minute) thousands of rows with
statement texts. Now at most 100 transactions (large log first, then the
oldest) and the 50 longest requests.

The transactions were joined to sys.dm_exec_requests, which has one row
per request: a MARS session with an open transaction gave duplicate rows,
and the MERGE into insight.LongTransactions then failed on the primary
key every minute. The first request is now taken with OUTER APPLY TOP 1,
and the save keeps one row per transaction.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

src/SQLInsight.Collectors/Concurrency/ActivityCollector.cs

@@ -14,11 +14,12 @@ public sealed class ActivityCollector(int transactionMinutes = 10, int transacti
     private const string NotAgent = "s.is_user_process = 1 AND s.session_id <> @@SPID AND (s.program_name IS NULL OR s.program_name NOT LIKE N'SQLInsight Pro%')";
 
     /// <summary>
-    /// One row per open transaction of a user session. The statement is the running one, or else the session's last.
+    /// One row per open transaction of a user session, at most 100: those holding more log than the threshold first, then
+    /// the oldest. The statement is the running one (the first request of a MARS session), or else the session's last.
     /// Times are converted to UTC on the server.
     /// </summary>
     internal const string TransactionsSql = """
-        SELECT t.session_id, t.transaction_id, t.name, DATEADD(minute, DATEDIFF(minute, GETDATE(), GETUTCDATE()), t.begin_time),
+        SELECT TOP 100 t.session_id, t.transaction_id, t.name, DATEADD(minute, DATEDIFF(minute, GETDATE(), GETUTCDATE()), t.begin_time),
                t.used, t.reserved, t.db, t.dbs, s.login_name, s.host_name, s.program_name, s.status, r.wait_type,
                CAST(CASE WHEN r.sql_handle IS NOT NULL
                          THEN SUBSTRING(q.text, r.statement_start_offset / 2 + 1,
@@ -32,14 +33,16 @@ public sealed class ActivityCollector(int transactionMinutes = 10, int transacti
               LEFT JOIN sys.dm_tran_database_transactions dt ON dt.transaction_id = st.transaction_id AND dt.database_id <> 32767
               GROUP BY st.session_id, at.transaction_id) t
         JOIN sys.dm_exec_sessions s ON s.session_id = t.session_id
-        LEFT JOIN sys.dm_exec_requests r ON r.session_id = t.session_id
+        OUTER APPLY (SELECT TOP 1 x.sql_handle, x.statement_start_offset, x.statement_end_offset, x.wait_type FROM sys.dm_exec_requests x
+                     WHERE x.session_id = t.session_id ORDER BY x.request_id) r
         OUTER APPLY (SELECT TOP 1 c.most_recent_sql_handle FROM sys.dm_exec_connections c WHERE c.session_id = t.session_id) c
         OUTER APPLY sys.dm_exec_sql_text(COALESCE(r.sql_handle, c.most_recent_sql_handle)) q
         WHERE {notAgent} AND t.begin_time IS NOT NULL
+        ORDER BY CASE WHEN t.used + t.reserved >= {logBytes} THEN 0 ELSE 1 END, t.begin_time
         """;
 
     internal const string LongRequestsSql = """
-        SELECT r.session_id, DATEADD(minute, DATEDIFF(minute, GETDATE(), GETUTCDATE()), r.start_time), r.total_elapsed_time, r.command, r.status,
+        SELECT TOP 50 r.session_id, DATEADD(minute, DATEDIFF(minute, GETDATE(), GETUTCDATE()), r.start_time), r.total_elapsed_time, r.command, r.status,
                r.wait_type, r.wait_time, DB_NAME(r.database_id), s.login_name, s.host_name, s.program_name, r.blocking_session_id,
                CAST(SUBSTRING(q.text, r.statement_start_offset / 2 + 1,
                     CASE WHEN r.statement_end_offset = -1 THEN 1000 ELSE (r.statement_end_offset - r.statement_start_offset) / 2 + 1 END) AS nvarchar(1000))
@@ -47,6 +50,7 @@ public sealed class ActivityCollector(int transactionMinutes = 10, int transacti
         JOIN sys.dm_exec_sessions s ON s.session_id = r.session_id
         OUTER APPLY sys.dm_exec_sql_text(r.sql_handle) q
         WHERE {notAgent} AND r.total_elapsed_time > {thresholdMs} AND r.percent_complete = 0
+        ORDER BY r.total_elapsed_time DESC
         """;
 
     public override string Name => "activity";
@@ -56,7 +60,10 @@ public sealed class ActivityCollector(int transactionMinutes = 10, int transacti
     protected override async Task<ActivitySnapshot> CollectAsync(CollectorContext context, ActivitySnapshot? previous)
     {
         var ct = context.CancellationToken;
-        var transactions = await Db.QueryAsync(context.Connection, TransactionsSql.Replace("{notAgent}", NotAgent, StringComparison.Ordinal), r => new OpenTransaction(
+        var logBytes = (long)transactionLogMegabytes * 1024 * 1024;
+        var transactionsSql = TransactionsSql.Replace("{notAgent}", NotAgent, StringComparison.Ordinal)
+            .Replace("{logBytes}", logBytes.ToString(CultureInfo.InvariantCulture), StringComparison.Ordinal);
+        var transactions = await Db.QueryAsync(context.Connection, transactionsSql, r => new OpenTransaction(
             r.Int(0), r.Long(1), r.NullableString(2), Utc(r.DateTime(3)), r.Long(4), r.Long(5), r.NullableString(6), r.Int(7),
             r.NullableString(8), r.NullableString(9), r.NullableString(10), r.NullableString(11), r.NullableString(12), r.NullableString(13)), ct).ConfigureAwait(false);
         var thresholdMs = (long)requestMinutes * 60_000;

src/SQLInsight.Repository/InsightRepository.Activity.cs

@@ -17,7 +17,8 @@ public sealed partial class InsightRepository
 {
     private static async Task SaveActivityAsync(SqlConnection connection, int instanceId, ActivitySnapshot snapshot)
     {
-        var kept = snapshot.Transactions.Where(t => ActivityAnalysis.IsWorthKeeping(t, snapshot.CapturedAt)).ToList();
+        // One row per transaction: a duplicate in the source would make the MERGE insert the same key twice.
+        var kept = snapshot.Transactions.Where(t => ActivityAnalysis.IsWorthKeeping(t, snapshot.CapturedAt)).DistinctBy(t => (t.SessionId, t.TransactionId)).ToList();
         if (kept.Count == 0)
         {
             return;