Add isolated connectivity starvation and cancellation repros

BUILDGUIDE.md

@@ -648,3 +648,64 @@ $ dotnet run -c Release -f net10.0
 
 $ RUNNER_CONFIG=~/.configs/runnerconfig.jsonc dotnet run -c Release -f net10.0
 ```
+
+### Sustained async connectivity scorecard
+
+Enable `Benchmarks.ConnectivityLoadRunnerConfig` in your external runner config
+and select only this unit:
+
+```bash
+PERF_BENCHMARK=ConnectivityLoad RUNNER_CONFIG=~/.configs/runnerconfig.jsonc \
+  dotnet run -c Release -f net10.0
+```
+
+This benchmark runs 1, 16, and 64 concurrent async worker loops by default against the
+configured SQL Server. Each loop awaits `OpenAsync` and disposes the connection;
+it does not manufacture network delays or synchronously wait on async tasks.
+`Pooling=false` measures repeated physical connections, while `Pooling=true`
+measures mostly warm-pool checkout/return after the initial ramp. Each iteration
+starts with its own pool cleared. No tables or special server are required.
+`Concurrency` and `Pooling` arrays in the load config select the parameter cases.
+For example, `"Concurrency": [16], "Pooling": [false]` runs only physical opens
+at concurrency 16, allowing each case to be measured in its own fresh process.
+
+`DurationSeconds` (default 10) controls how long loops admit new operations.
+Already-started opens finish before the measurement ends; their time and
+successful completions are included. An open failure fails the invocation,
+stops new admissions and drains in-flight operations instead of reporting a
+misleading improvement. Keep a finite connection timeout in the connection string.
+Failed benchmark cases make the `ConnectivityLoad` process exit nonzero.
+`SampleIntervalMilliseconds` (default 10) controls a dedicated observer thread,
+so sampling can continue even when the thread pool is starved. Sampling integrates
+observed occupancy using actual timestamps, including a final sample.
+
+The custom diagnoser adds these values to the BenchmarkDotNet summary and full
+JSON reports, excluding warmup invocations:
+
+| Metric | Meaning |
+| --- | --- |
+| Avg workers | Time-weighted occupied thread-pool workers. |
+| Worker ms/open | Integrated occupied worker time divided by successful opens. |
+| Opens/sec | Successful opens divided by actual measured elapsed time. |
+| Peak workers | Largest sampled occupied-worker count. |
+| Peak TP threads | Largest sampled existing thread-pool count, including idle threads. |
+| Peak process threads | Largest sampled process thread count, including dedicated threads and the observer. |
+| CPU ms/open | Process CPU time divided by successful opens. |
+| Opens / Samples / Measured sec | Totals across measured invocations, for checking normalization and sampling. |
+
+Occupancy is `GetMaxThreads` minus `GetAvailableThreads`, **not a blocked-thread
+counter**: it includes CPU work, other process activity, and connection disposal.
+Interpret lower occupancy alongside throughput and CPU cost. This is a
+closed-loop workload, so improved async can increase throughput without lowering
+average occupancy. Longer runs reduce sampling noise but do not remove systematic
+server/network variation or exercise cold/expired authentication caches by themselves.
+The process-thread peak guards against moving blocking off the thread pool onto
+dedicated threads. All peaks and integrated occupancy are sampling estimates;
+short waits between samples can be missed.
+
+Use fresh benchmark processes for baseline and candidate, keep concurrency,
+duration, sampling interval, SDK/CPU settings, authentication, and server/network
+conditions matched, and repeat runs. The suite's in-process toolchain preserves
+driver switches but does not reset thread-pool history between parameter cases.
+The existing perf comparison script emits custom metric deltas in JSON and a
+separate Markdown table; its execution-time regression gate is unchanged.

TESTGUIDE.md

@@ -5,6 +5,9 @@ tests.
 
 For build prerequisites and general `build.proj` usage, see [BUILDGUIDE.md](BUILDGUIDE.md).
 
+For isolated worker-starvation, cancellation, and transaction-lock reproductions,
+see [Async connectivity reproductions](doc/async-connectivity-reproductions.md).
+
 ## Test Projects
 
 The primary test projects for Microsoft.Data.SqlClient are under

doc/async-connectivity-reproductions.md

@@ -0,0 +1,86 @@
+# Async connectivity reproductions
+
+This focused suite retains five demonstrated async connectivity mechanisms
+(worker starvation, transport cancellation, and transaction lock inversion),
+plus the controls and test infrastructure needed to distinguish defects from
+inadmissible setups. The related issue for each mechanism is linked below.
+It changes no production behavior or public APIs.
+
+## Retained reproductions
+
+Test classes are under
+`src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/`.
+
+| Mechanism | Desired-behavior test | Local baseline and essential controls |
+|---|---|---|
+| Async authentication starvation | `ConnectivityProcessTests.OpenAsync_TokenBurst_WithCappedWorkers_Completes` | Every V2 worker reaches held Login7. After Login7 release, zero workers are available and the token callback cannot enter. Generous-worker V2 and capped V1/non-pooled controls complete. Relevant to [#3459](https://github.com/dotnet/SqlClient/issues/3459). |
+| Additional user-thread OpenAsync behind occupied workers | `ConnectivityProcessTests.OpenAsync_AdditionalOpen_WithCappedWorkers_MakesProgress` | A non-pool caller obtains a Task, but physical login to a separate, unheld endpoint cannot progress until the original Login7 replies release workers. The recovery control completes every open under the unchanged cap. This covers the starvation mechanism in [#3118](https://github.com/dotnet/SqlClient/issues/3118), not its exact TLS/error signature. |
+| Synchronous provider-authentication worker starvation | `ConnectivityAuthenticationPressureTests.Open_ProviderAuthentication_WithCappedWorkerCallers_Completes` | Cold and genuinely expired fake managed-identity tokens stall both pools before provider entry with zero workers available. Increasing workers during cleanup completes the same callers. Warm-cache, generous-worker, dedicated-caller, and direct-async controls distinguish worker pressure from provider/cache failures. Relevant to [#2152](https://github.com/dotnet/SqlClient/issues/2152)/[#2470](https://github.com/dotnet/SqlClient/issues/2470), which primarily report synchronous Open. |
+| Physical TCP attempts continue after public cancellation | `ConnectivityTransportCancellationTests.OpenAsync_RefusedConnect_AfterPublicCancellation_StartsNoNewAttempt` (.NET only; it observes managed-SNI traces) | Gate the first actual refused TCP connect, confirm public cancellation, then release physical work. Attempts grow from 1 to 13; the desired count remains 1. An uncancelled control proves the internal retry path is reachable with configurable retries disabled. Relevant to [#1619](https://github.com/dotnet/SqlClient/issues/1619). |
+| Cancelled-scope rollback/physical-open lock inversion | `ConnectivityTransactionTests.OpenAsync_AmbientTransaction_CancelledAfterBegin_ScopeAndPhysicalWorkComplete` | Real BEGIN completes before a post-BEGIN gate. After public cancellation, rollback owns the connection monitor while physical opening owns the parser monitor. Releasing the gate produces `LOCK_CYCLE_STALLED`. Post-BEGIN, pre-enlistment cancellation, held-BEGIN, and actual sync/async commit/rollback controls establish phase ordering and wire validity. Relevant to [#4696](https://github.com/dotnet/SqlClient/issues/4696). |
+
+**Keep both `ConnectivityProcessTests` anchors.** Their worker caps are accepted
+and verified, and an async-only yield workload first proves that the capped
+runtime can schedule genuinely asynchronous work. The peer runs in the parent,
+outside the capped pool. Caps cover thread-pool workers, not every managed thread.
+Worker availability and pending-task state are diagnostics, not requirements that
+a future fix must preserve exhaustion.
+
+The fake provider uses the real connection-string authentication-provider path,
+but does not contact Azure. Expiry ages physical and shared cached tokens without
+substituting pool clearing. The transaction peer supports only the local
+BEGIN/COMMIT/ROLLBACK subset needed here, not DTC or a general SQL engine.
+Request round-trip/truncation tests protect that new wire support.
+
+## Running
+
+Use the repository's SDK and build instructions in [BUILDGUIDE](../BUILDGUIDE.md).
+The unit project builds and copies `tests/tools/ConnectivityTestHost` automatically.
+Run healthy controls first:
+
+```bash
+dotnet test src/Microsoft.Data.SqlClient/tests/UnitTests/Microsoft.Data.SqlClient.UnitTests.csproj \
+  -f net10.0 --filter 'FullyQualifiedName~Connectivity&category!=failing'
+```
+
+The desired-behavior reproductions are deliberately `category=failing`, excluded
+by the repository's normal filter. Select them explicitly to capture the current
+defects; all eight cases are expected to fail on the current macOS baseline:
+
+```bash
+dotnet test src/Microsoft.Data.SqlClient/tests/UnitTests/Microsoft.Data.SqlClient.UnitTests.csproj \
+  -f net10.0 --filter 'FullyQualifiedName~Connectivity&category=failing' \
+  --logger 'console;verbosity=normal'
+```
+
+To examine either anchor alone, replace the filter with
+`FullyQualifiedName~OpenAsync_TokenBurst_WithCappedWorkers_Completes` or
+`FullyQualifiedName~OpenAsync_AdditionalOpen_WithCappedWorkers_MakesProgress`.
+Use `--list-tests` with the same filter to check selection. Once a driver fix
+passes a desired-behavior test and its controls, remove that test's `failing`
+trait and retain it as ordinary regression coverage.
+
+Every protocol read/phase/completion is bounded by an independent parent
+watchdog. Missed setup phases, EOF, child faults, or rejected runtime limits are
+test failures, not passing reproductions. Teardown releases gates and kills/reaps
+only the owned child when it cannot recover. A lifecycle control explicitly
+verifies missing-phase timeout and reaping. After a timed-out protocol read,
+cleanup must not issue another read on that stream.
+
+## Evidence limits
+
+Local verification uses macOS/.NET 10, ephemeral loopback peers, and fake tokens:
+no SQL Server, Azure credentials, certificates, or machine network changes.
+Trace gates and reflection probe the current source's physical-open tasks,
+authentication caches, transaction promoter, and monitors. Missing members or
+trace boundaries fail explicitly; these probes are not a general released-package
+comparison harness. Monitor ownership and captured thread states establish the
+local lock cycle, but external opposing blocked stacks were not collected.
+These mechanisms do not establish every customer's exact environment or cause.
+
+The historical [#2192](https://github.com/dotnet/SqlClient/issues/2192) defect was
+addressed by [#2915](https://github.com/dotnet/SqlClient/pull/2915), starting with
+6.0.0-preview2; it is not an outstanding reproduction target here. Windows UI,
+native-SNI, real-service/network confirmations, broad correctness coverage,
+benchmarks, and separate timeout-budget/pool-retirement investigations are
+intentionally outside this focused change.

eng/pipelines/perf/scripts/compare_perf.py

@@ -4,6 +4,7 @@
 Reads every ``*-report-full.json`` under a baseline directory and a current
 directory, matches benchmarks by (Type, Method, Parameters), and computes the
 per-benchmark delta in mean execution time and allocated memory.
+Custom diagnoser metrics are also compared by their stable descriptor IDs.
 
 Outputs:
   * a GitHub-flavoured markdown table (``--out-md``), and
@@ -49,6 +50,18 @@ def _load_benchmarks(directory):
                 continue
             memory = bench.get("Memory") or {}
             alloc = memory.get("BytesAllocatedPerOperation")
+            metrics = {}
+            for metric in bench.get("Metrics", []) or []:
+                descriptor = metric.get("Descriptor") or {}
+                metric_id = descriptor.get("Id")
+                value = metric.get("Value")
+                if metric_id is not None and value is not None:
+                    metrics[metric_id] = {
+                        "value": float(value),
+                        "displayName": descriptor.get("DisplayName") or metric_id,
+                        "unit": descriptor.get("Unit") or "",
+                        "greaterIsBetter": descriptor.get("TheGreaterTheBetter"),
+                    }
 
             key = f"{btype}.{method}({params})"
             records[key] = {
@@ -57,6 +70,7 @@ def _load_benchmarks(directory):
                 "parameterSignature": params,
                 "meanNs": float(mean_ns),
                 "allocatedBytes": float(alloc) if alloc is not None else None,
+                "metrics": metrics,
             }
     return records
 
@@ -90,7 +104,25 @@ def build_comparison(baseline_dir, current_dir, threshold_pct):
             "currentMeanMs": (c["meanNs"] / NS_PER_MS) if c else None,
             "baselineAllocBytes": b["allocatedBytes"] if b else None,
             "currentAllocBytes": c["allocatedBytes"] if c else None,
+            "metrics": [],
         }
+        baseline_metrics = b["metrics"] if b else {}
+        current_metrics = c["metrics"] if c else {}
+        for metric_id in sorted(set(baseline_metrics) | set(current_metrics)):
+            bm = baseline_metrics.get(metric_id)
+            cm = current_metrics.get(metric_id)
+            metadata = cm or bm
+            base_value = bm["value"] if bm else None
+            current_value = cm["value"] if cm else None
+            entry["metrics"].append({
+                "id": metric_id,
+                "displayName": metadata["displayName"],
+                "unit": metadata["unit"],
+                "greaterIsBetter": metadata["greaterIsBetter"],
+                "baselineValue": base_value,
+                "currentValue": current_value,
+                "deltaPct": _pct(base_value, current_value),
+            })
 
         if b and c:
             entry["meanDeltaPct"] = _pct(b["meanNs"], c["meanNs"])
@@ -203,6 +235,28 @@ def render_markdown(entries, baseline_version, threshold_pct):
             )
         )
     lines.append("")
+    metric_entries = [(entry, metric) for entry in entries for metric in entry.get("metrics", [])]
+    if metric_entries:
+        lines.extend([
+            "## Diagnoser metrics",
+            "",
+            "Values use the diagnoser's units and normalization; these deltas do not change "
+            "the execution-time regression gate.",
+            "",
+            "| Benchmark | Method | Params | Metric | Unit | Baseline | Current | Delta |",
+            "| --------- | ------ | ------ | ------ | ---- | -------- | ------- | ----- |",
+        ])
+        for entry, metric in metric_entries:
+            baseline = metric["baselineValue"]
+            current = metric["currentValue"]
+            lines.append(
+                f"| {entry['benchmarkName']} | {entry['methodName']} | "
+                f"{entry['parameterSignature'] or '-'} | {metric['displayName']} | "
+                f"{metric['unit'] or '-'} | "
+                f"{f'{baseline:.4f}' if baseline is not None else '-'} | "
+                f"{f'{current:.4f}' if current is not None else '-'} | "
+                f"{_fmt_pct(metric['deltaPct'])} |")
+        lines.append("")
     return "\n".join(lines)
 
 

eng/pipelines/perf/scripts/tests/test_compare_metrics.py

@@ -0,0 +1,90 @@
+# Licensed to the .NET Foundation under one or more agreements.
+# The .NET Foundation licenses this file to you under the MIT license.
+# See the LICENSE file in the project root for more information.
+
+"""Verify custom scorecard metrics survive baseline comparison and Kusto conversion."""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+import compare_perf
+import perf_to_kusto
+
+
+class CompareMetricsTests(unittest.TestCase):
+    """Compare by stable IDs, preserving zero and missing values explicitly."""
+
+    def report(self, directory, metrics):
+        benchmark = {
+            "Type": "ConnectivityLoadRunner",
+            "Method": "OpenAsyncLoop",
+            "Parameters": "Concurrency=16, Pooling=False",
+            "Statistics": {"Mean": 1_000_000},
+            "Metrics": [
+                {
+                    "Descriptor": {
+                        "Id": metric_id,
+                        "DisplayName": display_name,
+                        "Legend": display_name,
+                        "Unit": "workers",
+                        "TheGreaterTheBetter": False,
+                    },
+                    "Value": value,
+                }
+                for metric_id, display_name, value in metrics
+            ],
+        }
+        with open(os.path.join(directory, "load-report-full.json"), "w") as handle:
+            json.dump({"Benchmarks": [benchmark]}, handle)
+        return benchmark
+
+    def test_metrics_match_by_id_not_display_name(self):
+        with tempfile.TemporaryDirectory() as baseline, tempfile.TemporaryDirectory() as current:
+            self.report(baseline, [("AverageOccupiedWorkers", "Old label", 8)])
+            self.report(current, [("AverageOccupiedWorkers", "Avg workers", 2)])
+            entries = compare_perf.build_comparison(baseline, current, 10)
+        self.assertEqual(entries[0]["status"], "unchanged")
+        metric = entries[0]["metrics"][0]
+        self.assertEqual(metric["baselineValue"], 8)
+        self.assertEqual(metric["currentValue"], 2)
+        self.assertEqual(metric["deltaPct"], -75)
+        markdown = compare_perf.render_markdown(entries, "baseline", 10)
+        self.assertIn("## Diagnoser metrics", markdown)
+        self.assertIn("Avg workers", markdown)
+        self.assertIn("-75.00%", markdown)
+
+    def test_zero_and_one_sided_metrics_remain_visible(self):
+        with tempfile.TemporaryDirectory() as baseline, tempfile.TemporaryDirectory() as current:
+            self.report(baseline, [("zero", "Zero", 0), ("removed", "Removed", 4)])
+            self.report(current, [("zero", "Zero", 2), ("new", "New", 3)])
+            entries = compare_perf.build_comparison(baseline, current, 10)
+        metrics = {metric["id"]: metric for metric in entries[0]["metrics"]}
+        self.assertEqual(metrics["zero"]["baselineValue"], 0)
+        self.assertEqual(metrics["zero"]["currentValue"], 2)
+        self.assertIsNone(metrics["zero"]["deltaPct"])
+        self.assertIsNone(metrics["new"]["baselineValue"])
+        self.assertIsNone(metrics["removed"]["currentValue"])
+        markdown = compare_perf.render_markdown(entries, "baseline", 10)
+        self.assertIn("| 0.0000 | 2.0000 | - |", markdown)
+
+    def test_kusto_converter_preserves_scorecard_values(self):
+        with tempfile.TemporaryDirectory() as directory:
+            benchmark = self.report(directory, [("AverageOccupiedWorkers", "Avg workers", 0)])
+        self.assertEqual(perf_to_kusto._driver_specific_metrics(benchmark)["Avg workers"], 0)
+
+    def test_existing_reports_without_metrics_keep_timing_comparison(self):
+        with tempfile.TemporaryDirectory() as baseline, tempfile.TemporaryDirectory() as current:
+            self.report(baseline, [])
+            self.report(current, [])
+            entries = compare_perf.build_comparison(baseline, current, 10)
+        self.assertEqual(entries[0]["metrics"], [])
+        self.assertNotIn("## Diagnoser metrics", compare_perf.render_markdown(entries, "baseline", 10))
+
+
+if __name__ == "__main__":
+    unittest.main()