Reduce SQL Server queue length monitoring query volume (#5555)

* ✨ Reduce SQL Server queue length monitoring query volume

Replace the per-queue length probe (one IF EXISTS + max-min(RowVersion)
statement per tracked queue, every 200ms) with a single catalog-view query
per catalog that reads approximate row counts from sys.partitions. This reads
no queue table data, takes no locks, and needs only SELECT on the queue tables.

Also make the poll interval configurable via the QueueLengthQueryDelayInterval
connection string part (default 200ms), and add optional adaptive back-off up
to QueueLengthQueryMaxDelayInterval while all monitored queues are empty
(disabled by default; base interval is always used while any queue has work,
preserving the fix for #4556).

* ♻️ Address PR review: Unquoted* naming, primary ctor, concurrent pacing

SqlTable: rename Name/Schema/Catalog to Unquoted* to make the
quoted-vs-unquoted contract explicit (review feedback), and convert to a
primary constructor. The now-dead per-table LengthQuery's duplicated
if/else is collapsed into BuildFullTableName/BuildLengthQuery helpers.

QueueLengthProvider: pace the poll interval concurrently with the query
so the effective cadence is max(interval, queryDuration) instead of
interval + queryDuration. The additive query-time term was what let the
cadence drift past the 1s monitoring bucket and starve buckets of
samples (#4556 false-zero sawtooth). With drift removed, restore sane
defaults — base 1s (matches the finest monitoring bucket) ramping to a
10s idle ceiling — superseding the #4557 200ms oversampling workaround.
Adaptive back-off is now ON by default.

* ♻️ Address review: throttle error loop, immediate back-off snap-back, NOLOCK, validation

- Pace after the query (subtracting iteration time) instead of pre-starting Task.Delay:
  a fast persistent failure no longer spins the loop with zero delay, and a back-off
  snap-back to base takes effect on the current wait instead of one backed-off wait later.
- Reject non-positive QueueLengthQuery(Max)DelayInterval up front (a negative value
  previously threw inside Task.Delay mid-loop; zero meant no pacing).
- Record 0 for a not-found table so a dropped queue no longer pins the cadence at base.
- Statement-scoped WITH (NOLOCK) on the catalog views instead of a connection-scoped
  SET TRANSACTION ISOLATION LEVEL that would persist on the pooled connection.
- Guard BuildBulkLengthQuery against an empty table set (would emit invalid `AND ()`).
- Remove now-dead per-table LengthQuery/BuildLengthQuery.


Co-authored-by: Mauro Servienti <mauro.servienti@gmail.com>

Author: ramonsmits

src/ServiceControl.Transports.SqlServer.Tests/QueueLengthQueryTests.cs

@@ -0,0 +1,164 @@
+namespace ServiceControl.Transport.Tests;
+
+using System;
+using System.Linq;
+using NUnit.Framework;
+using Transports.SqlServer;
+
+[TestFixture]
+class QueueLengthQueryTests
+{
+    [Test]
+    public void BuildBulkLengthQuery_emits_a_single_catalog_view_query_for_all_tables()
+    {
+        var tables = new[]
+        {
+            SqlTable.Parse("Sales", "dbo"),
+            SqlTable.Parse("Billing", "dbo"),
+        };
+
+        var query = SqlTable.BuildBulkLengthQuery(catalog: null, tables);
+
+        // One statement, read from the catalog views — no per-table IF EXISTS, no scan of the queue tables.
+        Assert.That(query, Does.Contain("sys.partitions"));
+        Assert.That(query, Does.Contain("p.index_id IN (0, 1)"));
+        Assert.That(query, Does.Not.Contain("INFORMATION_SCHEMA"));
+        Assert.That(query, Does.Not.Contain("RowVersion"));
+
+        // Dirty reads are scoped to this statement via NOLOCK hints, not a connection-scoped SET that
+        // would leak READ UNCOMMITTED onto the pooled connection.
+        Assert.That(query, Does.Contain("WITH (NOLOCK)"));
+        Assert.That(query, Does.Not.Contain("ISOLATION LEVEL"));
+
+        // Both tracked tables are covered by the single query.
+        Assert.That(query, Does.Contain("t.name = 'Sales'"));
+        Assert.That(query, Does.Contain("t.name = 'Billing'"));
+
+        // Exactly one result-producing statement.
+        Assert.That(System.Text.RegularExpressions.Regex.Matches(query, "SELECT").Count, Is.EqualTo(1));
+    }
+
+    [Test]
+    public void BuildBulkLengthQuery_with_no_tables_stays_valid_and_returns_no_rows()
+    {
+        var query = SqlTable.BuildBulkLengthQuery(catalog: null, []);
+
+        // No predicate terms must not produce the invalid `AND ()`; a constant-false predicate keeps the
+        // query syntactically valid while matching nothing.
+        Assert.That(query, Does.Not.Contain("AND ()"));
+        Assert.That(query, Does.Contain("1 = 0"));
+    }
+
+    [Test]
+    public void BuildBulkLengthQuery_prefixes_the_catalog_when_supplied()
+    {
+        var tables = new[] { SqlTable.Parse("Sales@dbo@MyCatalog", "dbo") };
+
+        var query = SqlTable.BuildBulkLengthQuery(catalog: "MyCatalog", tables);
+
+        Assert.That(query, Does.Contain("[MyCatalog].sys.partitions"));
+        Assert.That(query, Does.Contain("[MyCatalog].sys.tables"));
+        Assert.That(query, Does.Contain("[MyCatalog].sys.schemas"));
+    }
+
+    [Test]
+    public void RemoveQueueLengthQueryDelayInterval_parses_and_strips_the_custom_part()
+    {
+        const string connectionString = "Data Source=.;Initial Catalog=nsb;Integrated Security=true;QueueLengthQueryDelayInterval=1000";
+
+        var cleaned = connectionString.RemoveQueueLengthQueryDelayInterval(out var interval);
+
+        Assert.That(interval, Is.EqualTo(TimeSpan.FromSeconds(1)));
+        // The custom key must be stripped so SqlConnection never sees the unknown keyword.
+        Assert.That(cleaned, Does.Not.Contain("QueueLengthQueryDelayInterval").IgnoreCase);
+        Assert.That(cleaned, Does.Contain("Initial Catalog=nsb").IgnoreCase);
+    }
+
+    [Test]
+    public void RemoveQueueLengthQueryDelayInterval_defaults_to_null_when_absent()
+    {
+        const string connectionString = "Data Source=.;Initial Catalog=nsb;Integrated Security=true";
+
+        connectionString.RemoveQueueLengthQueryDelayInterval(out var interval);
+
+        Assert.That(interval, Is.Null);
+    }
+
+    [Test]
+    public void RemoveQueueLengthQueryDelayInterval_throws_on_non_numeric_value()
+    {
+        const string connectionString = "Data Source=.;Initial Catalog=nsb;QueueLengthQueryDelayInterval=soon";
+
+        Assert.That(() => connectionString.RemoveQueueLengthQueryDelayInterval(out _),
+            Throws.Exception.With.Message.Contains("QueueLengthQueryDelayInterval"));
+    }
+
+    [Test]
+    public void RemoveQueueLengthQueryDelayInterval_throws_on_negative_value()
+    {
+        const string connectionString = "Data Source=.;Initial Catalog=nsb;QueueLengthQueryDelayInterval=-5";
+
+        // A negative interval would flow through to Task.Delay and throw mid-loop; reject it up front.
+        Assert.That(() => connectionString.RemoveQueueLengthQueryDelayInterval(out _),
+            Throws.Exception.With.Message.Contains("QueueLengthQueryDelayInterval"));
+    }
+
+    [Test]
+    public void RemoveQueueLengthQueryDelayInterval_throws_on_zero_value()
+    {
+        const string connectionString = "Data Source=.;Initial Catalog=nsb;QueueLengthQueryDelayInterval=0";
+
+        // Zero would poll with no pacing at all; reject it.
+        Assert.That(() => connectionString.RemoveQueueLengthQueryDelayInterval(out _),
+            Throws.Exception.With.Message.Contains("QueueLengthQueryDelayInterval"));
+    }
+
+    [Test]
+    public void RemoveQueueLengthQueryMaxDelayInterval_parses_and_strips_the_custom_part()
+    {
+        const string connectionString = "Data Source=.;Initial Catalog=nsb;QueueLengthQueryMaxDelayInterval=60000";
+
+        var cleaned = connectionString.RemoveQueueLengthQueryMaxDelayInterval(out var interval);
+
+        Assert.That(interval, Is.EqualTo(TimeSpan.FromSeconds(60)));
+        Assert.That(cleaned, Does.Not.Contain("QueueLengthQueryMaxDelayInterval").IgnoreCase);
+    }
+
+    static readonly TimeSpan Base = TimeSpan.FromMilliseconds(200);
+    static readonly TimeSpan Max = TimeSpan.FromSeconds(60);
+
+    [Test]
+    public void NextDelay_snaps_back_to_base_when_any_queue_has_work()
+    {
+        // Even fully backed off, a single non-empty queue returns to full speed immediately.
+        Assert.That(QueueLengthProvider.NextDelay(Max, Base, Max, maxObservedLength: 1), Is.EqualTo(Base));
+    }
+
+    [Test]
+    public void NextDelay_doubles_while_idle()
+    {
+        Assert.That(QueueLengthProvider.NextDelay(Base, Base, Max, maxObservedLength: 0),
+            Is.EqualTo(TimeSpan.FromMilliseconds(400)));
+    }
+
+    [Test]
+    public void NextDelay_caps_at_max_while_idle()
+    {
+        Assert.That(QueueLengthProvider.NextDelay(TimeSpan.FromSeconds(40), Base, Max, maxObservedLength: 0),
+            Is.EqualTo(Max));
+    }
+
+    [Test]
+    public void NextDelay_never_drops_below_base()
+    {
+        Assert.That(QueueLengthProvider.NextDelay(TimeSpan.FromMilliseconds(50), Base, Max, maxObservedLength: 0),
+            Is.EqualTo(Base));
+    }
+
+    [Test]
+    public void NextDelay_with_max_equal_to_base_disables_backoff()
+    {
+        // Operator did not opt in (max == base) -> cadence is constant at base, even while idle.
+        Assert.That(QueueLengthProvider.NextDelay(Base, Base, Base, maxObservedLength: 0), Is.EqualTo(Base));
+    }
+}

src/ServiceControl.Transports.SqlServer/ConnectionStringExtensions.cs

@@ -1,5 +1,6 @@
 namespace ServiceControl.Transports.SqlServer
 {
+    using System;
     using System.Data.Common;
 
     static class ConnectionStringExtensions
@@ -11,6 +12,39 @@ public static string RemoveCustomConnectionStringParts(this string connectionStr
                 .RemoveCustomConnectionStringPart(subscriptionsTableName, out subscriptionTable);
         }
 
+        // Extracts the optional, ServiceControl-specific 'QueueLengthQueryDelayInterval' (milliseconds) from
+        // the connection string and removes it so it is never handed to SqlConnection (which would reject the
+        // unknown keyword). Mirrors the existing convention used by the Azure Service Bus transport.
+        // This is the BASE interval used while any monitored queue has messages.
+        public static string RemoveQueueLengthQueryDelayInterval(this string connectionString, out TimeSpan? interval) =>
+            connectionString.RemoveIntervalMilliseconds(queueLengthQueryDelayInterval, out interval);
+
+        // Extracts the optional 'QueueLengthQueryMaxDelayInterval' (milliseconds) — the upper bound the adaptive
+        // back-off ramps to while every monitored queue is empty. When omitted a default ceiling applies; set it
+        // equal to the base interval to disable back-off.
+        public static string RemoveQueueLengthQueryMaxDelayInterval(this string connectionString, out TimeSpan? interval) =>
+            connectionString.RemoveIntervalMilliseconds(queueLengthQueryMaxDelayInterval, out interval);
+
+        static string RemoveIntervalMilliseconds(this string connectionString, string key, out TimeSpan? interval)
+        {
+            interval = null;
+
+            var builder = new DbConnectionStringBuilder { ConnectionString = connectionString };
+
+            if (builder.TryGetValue(key, out var value))
+            {
+                if (!int.TryParse(value.ToString(), out var milliseconds) || milliseconds <= 0)
+                {
+                    throw new Exception($"Can't parse '{value}' as a valid {key} (expected an integer number of milliseconds greater than zero).");
+                }
+
+                interval = TimeSpan.FromMilliseconds(milliseconds);
+                builder.Remove(key);
+            }
+
+            return builder.ConnectionString;
+        }
+
         public static string RemoveCustomConnectionStringPart(this string connectionString, string partName, out string schema)
         {
             var builder = new DbConnectionStringBuilder
@@ -30,5 +64,7 @@ public static string RemoveCustomConnectionStringPart(this string connectionStri
 
         const string queueSchemaName = "Queue Schema";
         const string subscriptionsTableName = "Subscriptions Table";
+        const string queueLengthQueryDelayInterval = "QueueLengthQueryDelayInterval";
+        const string queueLengthQueryMaxDelayInterval = "QueueLengthQueryMaxDelayInterval";
     }
 }