@@ -55,6 +55,21 @@ public static class HttpHelper
5555 // 30–60s; bigger waits are clamped to keep tool latency bounded.
5656 private const int MaxRetryWaitSeconds = 60 ;
5757
58+ // Cost Management / Consumption / Billing share an aggressive per-tenant throttle pool.
59+ // App Insights showed 87/89 (97.8%) of /query calls returning 429 inside a single 34s burst
60+ // — the LLM (esp. via BulkAzureRequest with parallelism=20) was fan-firing parallel queries
61+ // that all collided. This semaphore globally serializes those calls to a small concurrency,
62+ // turning a retry storm into ordered execution. Other ARM/Graph/LogAnalytics calls are
63+ // unaffected and still parallelize freely.
64+ private static readonly SemaphoreSlim CostMgmtGate = new ( 2 , 2 ) ;
65+ private const int CostMgmtQueueNotifyMs = 250 ;
66+
67+ private static bool IsCostManagementUrl ( string url ) =>
68+ url . Contains ( "/Microsoft.CostManagement/" , StringComparison . OrdinalIgnoreCase )
69+ || url . Contains ( "/Microsoft.Consumption/" , StringComparison . OrdinalIgnoreCase )
70+ || url . Contains ( "/Microsoft.Billing/" , StringComparison . OrdinalIgnoreCase )
71+ || url . Contains ( "/Microsoft.CostManagementExports/" , StringComparison . OrdinalIgnoreCase ) ;
72+
5873 /// <summary>
5974 /// Sends an HTTP request with silent retry on 429 (up to 5 attempts). On each 429 we honor,
6075 /// in priority order: Cost Management's <c>x-ms-ratelimit-microsoft.costmanagement-qpu-retry-after</c>,
@@ -79,6 +94,49 @@ public static async Task<string> SendWithRetryAsync(
7994 var totalWaitSec = 0.0 ;
8095 var retryCount = 0 ;
8196 HttpResponseMessage res = null ! ;
97+
98+ // Resolve the per-turn SSE reporter ONCE up front — used for queue waits,
99+ // in-flight heartbeats, and retry backoffs alike. Returns no-op when absent.
100+ var turnKey = Activity . Current ? . GetBaggageItem ( "finops.turn.id" ) ;
101+ RetryReporters . TryGetValue ( turnKey ?? "" , out var report ) ;
102+
103+ // Gate Cost Management / Consumption / Billing calls behind a small global
104+ // semaphore (2 concurrent). Without this, the LLM (esp. via BulkAzureRequest
105+ // parallelism=20) fan-fires parallel /query calls that all collide on the
106+ // per-tenant throttle. While queued we emit cooling_down so the UI shows
107+ // "waiting in queue" instead of a frozen tool row.
108+ var isCostMgmt = IsCostManagementUrl ( url ) ;
109+ var heldGate = false ;
110+ if ( isCostMgmt )
111+ {
112+ if ( ! await CostMgmtGate . WaitAsync ( CostMgmtQueueNotifyMs ) )
113+ {
114+ // Couldn't grab the gate immediately — surface a queue-wait event
115+ // and keep retrying every ~3s so the ghost row stays alive in the UI.
116+ var queuedSw = Stopwatch . StartNew ( ) ;
117+ while ( ! await CostMgmtGate . WaitAsync ( 3000 ) )
118+ {
119+ Logger ? . LogInformation ( "HTTP queued {Tool} waitedSec={Wait:F1} url={Url}" ,
120+ telemetryPrefix , queuedSw . Elapsed . TotalSeconds , url ) ;
121+ if ( report is not null )
122+ {
123+ try { await report ( 0 , 5 , url , telemetryPrefix + " (queued)" , 0 ) ; }
124+ catch ( Exception emitEx ) { Logger ? . LogWarning ( emitEx , "SSE queued emit failed for {Tool}" , telemetryPrefix ) ; }
125+ }
126+ }
127+ // One final emit on acquire so the UI clears stale wait time.
128+ if ( report is not null )
129+ {
130+ try { await report ( 0 , 1 , url , telemetryPrefix + " (queued)" , 0 ) ; }
131+ catch { /* swallow */ }
132+ }
133+ activity ? . SetTag ( $ "{ telemetryPrefix } .queued_sec", queuedSw . Elapsed . TotalSeconds ) ;
134+ }
135+ heldGate = true ;
136+ }
137+
138+ try
139+ {
82140 for ( var attempt = 0 ; attempt < MaxThrottleRetries ; attempt ++ )
83141 {
84142 using var req = new HttpRequestMessage ( method , url ) ;
@@ -92,7 +150,35 @@ public static async Task<string> SendWithRetryAsync(
92150 if ( jsonBody is not null )
93151 req . Content = new StringContent ( jsonBody , Encoding . UTF8 , "application/json" ) ;
94152
95- res = await Http . SendAsync ( req ) ;
153+ // Heartbeat: if the request itself takes >5s (slow CM backend, async report
154+ // generation, etc.) emit periodic cooling_down events keyed by url so the
155+ // UI ghost row stays alive instead of looking frozen. Runs concurrently
156+ // with the actual request; cancelled the instant the response arrives.
157+ using var hbCts = new CancellationTokenSource ( ) ;
158+ var heartbeat = report is null ? Task . CompletedTask : Task . Run ( async ( ) =>
159+ {
160+ try
161+ {
162+ await Task . Delay ( 5000 , hbCts . Token ) ;
163+ while ( ! hbCts . IsCancellationRequested )
164+ {
165+ try { await report ( 0 , 6 , url , telemetryPrefix + " (slow)" , 0 ) ; }
166+ catch ( Exception emitEx ) { Logger ? . LogWarning ( emitEx , "SSE slow emit failed for {Tool}" , telemetryPrefix ) ; }
167+ await Task . Delay ( 5000 , hbCts . Token ) ;
168+ }
169+ }
170+ catch ( OperationCanceledException ) { /* expected on response */ }
171+ } , hbCts . Token ) ;
172+
173+ try
174+ {
175+ res = await Http . SendAsync ( req ) ;
176+ }
177+ finally
178+ {
179+ hbCts . Cancel ( ) ;
180+ try { await heartbeat ; } catch { /* heartbeat already swallows */ }
181+ }
96182
97183 // Retry on 429 (throttle) and transient 5xx (502/503/504 — typical ARM regional
98184 // failover or backend hiccups). Other non-success codes return to the caller.
@@ -119,8 +205,7 @@ public static async Task<string> SendWithRetryAsync(
119205 // Copilot CLI subprocess JSON-RPC tool callback) where RootId
120206 // does not. ChatEndpoints stamps "finops.turn.id" (userId:sessionId)
121207 // on the chat activity before SendAsync.
122- var turnKey = Activity . Current ? . GetBaggageItem ( "finops.turn.id" ) ;
123- if ( turnKey is not null && RetryReporters . TryGetValue ( turnKey , out var report ) )
208+ if ( report is not null )
124209 {
125210 try { await report ( attempt + 1 , waitSeconds , url , telemetryPrefix , status ) ; }
126211 catch ( Exception emitEx ) { Logger ? . LogWarning ( emitEx , "SSE cooling_down emit failed for {Tool}" , telemetryPrefix ) ; }
@@ -132,6 +217,11 @@ public static async Task<string> SendWithRetryAsync(
132217 }
133218 await Task . Delay ( TimeSpan . FromSeconds ( waitSeconds ) ) ;
134219 }
220+ }
221+ finally
222+ {
223+ if ( heldGate ) CostMgmtGate . Release ( ) ;
224+ }
135225
136226 var responseBody = await res . Content . ReadAsStringAsync ( ) ;
137227 totalSw . Stop ( ) ;
0 commit comments