backend: storage SDK budgets under the agent timeout, ingest 503 on an exhausted budget, request-token fan-outs, single-flight factory budget, ProcessStall monitor (audit F09, D-203)

- StorageClientOptions: Table/Queue 8 s per attempt × 3 attempts + capped backoffs = 25.8 s < the
  agent's 30 s upload timeout (pinned by test); Blob 60 s × 4. Every production client takes its
  options from there. No Polly over the SDK.
- StorageErrors.IsTransient(Exception): OperationCanceled / IO / HttpRequest / Aggregate-of-those
  / RequestFailed status 0; the ingest answers 503 + Retry-After instead of 500.
- The ingest still takes no caller cancellation (a batch is a unit). GetAllSessionsPageAsync
  carries the request token down to BoundedFanOut (Global Admin list); SingleFlightCache hands its
  factory a 30 s budget token and evicts on timeout (tenant ids, app-summary snapshot, session stats).
- ProcessStallMonitor: 1 s tick drift ≥ 1 s ⇒ ProcessStall event (sampling bypass) + metric with
  cause gc / threadpool / unknown from GC.GetTotalPauseDuration and thread-pool counters;
  GcPauseMs on request rows ≥ 100 ms. Measures the instance-local 5–10 s pauses seen in production.
- Gitlinks: internal (plan, D-203, concept doc, B-9hf, lesson), .claude/commands (marker rows).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-09-06 01:51:02 +02:00
parent c925001a8a
commit 5cb9b0b30b
25 changed files with 748 additions and 54 deletions
@@ -4,6 +4,7 @@ using AutopilotMonitor.Shared.Models;
using AutopilotMonitor.Shared.Pagination;
using Microsoft.Extensions.Logging.Abstractions;
using Moq;
using System.Threading;
namespace AutopilotMonitor.Functions.Tests;
@@ -21,7 +22,7 @@ public class AgentEfficiencyMetricsServiceTests
{
var repo = new Mock<ISessionRepository>();
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(new RawPage<SessionSummary>(new List<SessionSummary>(), null));
repo
.Setup(r => r.GetSessionEventsByTypesAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<IReadOnlyCollection<string>>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<int>()))
@@ -67,7 +68,7 @@ public class AgentEfficiencyMetricsServiceTests
private static void SetupSessions(Mock<ISessionRepository> repo, List<SessionSummary> sessions)
{
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(new RawPage<SessionSummary>(sessions, null));
}
@@ -201,8 +202,8 @@ public class AgentEfficiencyMetricsServiceTests
string? capturedTenant = null;
IEnumerable<string>? capturedSelect = null;
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Callback<string?, int?, int, string?, IReadOnlyCollection<string>?, IEnumerable<string>?>((t, _, _, _, _, sel) => { capturedTenant = t; capturedSelect = sel; })
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.Callback<string?, int?, int, string?, IReadOnlyCollection<string>?, IEnumerable<string>?, CancellationToken>((t, _, _, _, _, sel, _) => { capturedTenant = t; capturedSelect = sel; })
.ReturnsAsync(new RawPage<SessionSummary>(new List<SessionSummary>(), null));
var result = await service.ComputeAsync(days: 45, limit: 100, tenantId: Tenant);
@@ -220,7 +221,7 @@ public class AgentEfficiencyMetricsServiceTests
var (service, repo) = CreateService();
int callCount = 0;
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.Callback(() => callCount++)
.ReturnsAsync(new RawPage<SessionSummary>(new List<SessionSummary>(), null));
@@ -51,4 +51,14 @@ public class IngestStorageFailureClassificationTests
Assert.Equal(HttpStatusCode.InternalServerError, status);
Assert.Null(retryAfter);
}
[Fact]
public void A_connection_failure_without_status_is_503_with_retry_after()
{
var (status, _, retryAfter) = IngestTelemetryFunction.ClassifyStorageFailure(
new RequestFailedException(0, "connection reset before a status arrived"));
Assert.Equal(HttpStatusCode.ServiceUnavailable, status);
Assert.Equal(IngestTelemetryFunction.StorageRetryAfterSeconds, retryAfter);
}
}
@@ -4,6 +4,7 @@ using AutopilotMonitor.Shared.Models;
using AutopilotMonitor.Shared.Pagination;
using Microsoft.Extensions.Logging.Abstractions;
using Moq;
using System.Threading;
namespace AutopilotMonitor.Functions.Tests;
@@ -23,7 +24,7 @@ public class PlatformMetricsServiceWindowTests
{
var sessionRepo = new Mock<ISessionRepository>();
sessionRepo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(EmptyPage());
sessionRepo
.Setup(r => r.GetSessionEventsAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<int>()))
@@ -64,8 +65,8 @@ public class PlatformMetricsServiceWindowTests
int? capturedDays = null;
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Callback<string?, int?, int, string?, IReadOnlyCollection<string>?, IEnumerable<string>?>((_, d, _, _, _, _) => capturedDays = d)
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.Callback<string?, int?, int, string?, IReadOnlyCollection<string>?, IEnumerable<string>?, CancellationToken>((_, d, _, _, _, _, _) => capturedDays = d)
.ReturnsAsync(EmptyPage());
var result = await service.ComputePlatformMetricsAsync(days, limit: 50);
@@ -89,8 +90,8 @@ public class PlatformMetricsServiceWindowTests
int? capturedPageSize = null;
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Callback<string?, int?, int, string?, IReadOnlyCollection<string>?, IEnumerable<string>?>((_, _, ps, _, _, _) => capturedPageSize = ps)
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.Callback<string?, int?, int, string?, IReadOnlyCollection<string>?, IEnumerable<string>?, CancellationToken>((_, _, ps, _, _, _, _) => capturedPageSize = ps)
.ReturnsAsync(EmptyPage());
var result = await service.ComputePlatformMetricsAsync(days: days, limit: limit);
@@ -106,8 +107,8 @@ public class PlatformMetricsServiceWindowTests
int? capturedDays = null;
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Callback<string?, int?, int, string?, IReadOnlyCollection<string>?, IEnumerable<string>?>((_, d, _, _, _, _) => capturedDays = d)
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.Callback<string?, int?, int, string?, IReadOnlyCollection<string>?, IEnumerable<string>?, CancellationToken>((_, d, _, _, _, _, _) => capturedDays = d)
.ReturnsAsync(EmptyPage());
var result = await service.ComputePlatformMetricsAsync(0);
@@ -140,7 +141,7 @@ public class PlatformMetricsServiceWindowTests
var (service, repo) = CreateService();
int callCount = 0;
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.Callback(() => callCount++)
.ReturnsAsync(EmptyPage());
@@ -160,7 +161,7 @@ public class PlatformMetricsServiceWindowTests
var (service, repo) = CreateService();
var sessions = Enumerable.Range(1, 7).Select(MakeSession).ToList();
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(new RawPage<SessionSummary>(sessions, null));
var result = await service.ComputePlatformMetricsAsync(days: 77, limit: 50);
@@ -177,7 +178,7 @@ public class PlatformMetricsServiceWindowTests
var (service, repo) = CreateService();
var sessions = Enumerable.Range(1, 25).Select(MakeSession).ToList();
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(new RawPage<SessionSummary>(sessions, null));
var fullFetchSessions = new List<string>();
@@ -216,7 +217,7 @@ public class PlatformMetricsServiceWindowTests
var (service, repo) = CreateService();
var sessions = Enumerable.Range(1, 21).Select(MakeSession).ToList();
repo
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>()))
.Setup(r => r.GetAllSessionsPageAsync(It.IsAny<string?>(), It.IsAny<int?>(), It.IsAny<int>(), It.IsAny<string?>(), It.IsAny<IReadOnlyCollection<string>?>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(new RawPage<SessionSummary>(sessions, null));
var session21 = sessions[20].SessionId;
@@ -0,0 +1,84 @@
using System;
using AutopilotMonitor.Functions.Telemetry;
using Xunit;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// The stall classifier turns two runtime snapshots around a late timer tick into a cause.
/// GC wins when its pause clock covers at least half the drift; queued work or thread injection
/// means starvation; otherwise the drift stays unexplained rather than guessed.
/// </summary>
public class ProcessStallClassifierTests
{
private static ProcessSnapshot Snapshot(
double gcPauseMs = 0, int gen2 = 0, int threads = 8, long pending = 0, long completed = 0)
=> new(
GcPause: TimeSpan.FromMilliseconds(gcPauseMs),
Gen0: 0, Gen1: 0, Gen2: gen2,
ThreadPoolThreads: threads,
PendingWorkItems: pending,
CompletedWorkItems: completed,
WorkingSetBytes: 0, HeapSizeBytes: 0, CommittedBytes: 0,
LastGcGeneration: 0, LastGcPause: TimeSpan.Zero, LastGcCompacted: false);
[Fact]
public void Gc_pause_covering_half_the_drift_is_a_gc_stall()
{
var before = Snapshot(gcPauseMs: 1000, gen2: 3);
var after = Snapshot(gcPauseMs: 6200, gen2: 4, pending: 12);
var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(10), before, after);
Assert.Equal(StallCause.Gc, report.Cause);
Assert.Equal(TimeSpan.FromMilliseconds(5200), report.GcPauseDelta);
Assert.Equal(1, report.Gen2Delta);
}
[Fact]
public void Queued_work_without_gc_is_thread_pool_starvation()
{
var before = Snapshot(gcPauseMs: 1000);
var after = Snapshot(gcPauseMs: 1100, pending: 40);
var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(8), before, after);
Assert.Equal(StallCause.ThreadPool, report.Cause);
}
[Fact]
public void Thread_injection_without_gc_is_thread_pool_starvation()
{
var before = Snapshot(threads: 8);
var after = Snapshot(threads: 11);
var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(3), before, after);
Assert.Equal(StallCause.ThreadPool, report.Cause);
Assert.Equal(3, report.ThreadPoolThreadsDelta);
}
[Fact]
public void Drift_neither_clock_explains_stays_unknown()
{
var before = Snapshot(gcPauseMs: 500, threads: 8, completed: 100);
var after = Snapshot(gcPauseMs: 900, threads: 8, completed: 130);
var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(5), before, after);
Assert.Equal(StallCause.Unknown, report.Cause);
Assert.Equal(30, report.CompletedWorkItemsDelta);
}
[Fact]
public void A_wrapped_pause_clock_never_yields_a_negative_delta()
{
var before = Snapshot(gcPauseMs: 900);
var after = Snapshot(gcPauseMs: 100);
var report = ProcessStallClassifier.Classify(TimeSpan.FromSeconds(2), before, after);
Assert.Equal(TimeSpan.Zero, report.GcPauseDelta);
Assert.Equal(StallCause.Unknown, report.Cause);
}
}
@@ -99,4 +99,47 @@ public class SingleFlightCacheTests
// w1 is gone: its factory runs again.
Assert.Equal("one-again", await cache.GetOrAddAsync("w1", Ttl, () => Task.FromResult("one-again")));
}
[Fact]
public async Task Factory_exceeding_the_budget_faults_every_waiter_and_is_evicted()
{
var cache = new SingleFlightCache<int>(() => DateTime.UtcNow, factoryBudget: TimeSpan.FromMilliseconds(50));
var runs = 0;
async Task<int> Hanging(CancellationToken ct)
{
Interlocked.Increment(ref runs);
await Task.Delay(Timeout.Infinite, ct);
return 1;
}
var first = cache.GetOrAddAsync("k", Ttl, Hanging);
var second = cache.GetOrAddAsync("k", Ttl, Hanging);
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => first);
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => second);
Assert.Equal(1, runs);
Assert.Equal(0, cache.Count);
// The next caller starts a fresh run instead of inheriting the faulted one.
Assert.Equal(7, await cache.GetOrAddAsync("k", Ttl, _ => Task.FromResult(7)));
}
[Fact]
public async Task Factory_finishing_inside_the_budget_is_cached()
{
var cache = new SingleFlightCache<int>(() => DateTime.UtcNow, factoryBudget: TimeSpan.FromSeconds(5));
var runs = 0;
Task<int> Quick(CancellationToken ct)
{
Assert.True(ct.CanBeCanceled);
Interlocked.Increment(ref runs);
return Task.FromResult(3);
}
Assert.Equal(3, await cache.GetOrAddAsync("k", Ttl, Quick));
Assert.Equal(3, await cache.GetOrAddAsync("k", Ttl, Quick));
Assert.Equal(1, runs);
}
}
@@ -0,0 +1,70 @@
using System;
using Azure.Core;
using AutopilotMonitor.Functions.Helpers;
using Xunit;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// The storage SDK budgets are sized so one hot-path operation — every attempt into its
/// timeout, every backoff at its cap — finishes before the agent's 30 s upload timeout. A
/// change to either side that breaks that relation turns a storage brownout into zombie
/// requests plus agent replays (audit 2026-09-05 F09).
/// </summary>
public class StorageClientOptionsTests
{
[Fact]
public void Table_worst_case_stays_under_the_agent_upload_timeout()
{
var worst = StorageClientOptions.WorstCase(StorageClientOptions.Table().Retry);
Assert.True(worst < StorageClientOptions.AgentUploadTimeout, $"worst case {worst} must stay under {StorageClientOptions.AgentUploadTimeout}");
}
[Fact]
public void Queue_worst_case_stays_under_the_agent_upload_timeout()
{
var worst = StorageClientOptions.WorstCase(StorageClientOptions.Queue().Retry);
Assert.True(worst < StorageClientOptions.AgentUploadTimeout, $"worst case {worst} must stay under {StorageClientOptions.AgentUploadTimeout}");
}
[Fact]
public void Blob_worst_case_is_bounded()
{
var worst = StorageClientOptions.WorstCase(StorageClientOptions.Blob().Retry);
Assert.True(worst < TimeSpan.FromMinutes(5), $"worst case {worst} must stay under five minutes");
}
[Fact]
public void Worst_case_counts_every_attempt_and_every_capped_backoff()
{
// 3 attempts × 8 s + backoffs 0.5 s·1.2 and 1 s·1.2 (both under the 4 s cap) = 25.8 s.
var retry = StorageClientOptions.Table().Retry;
Assert.Equal(RetryMode.Exponential, retry.Mode);
Assert.Equal(StorageClientOptions.HotPathMaxRetries, retry.MaxRetries);
Assert.Equal(StorageClientOptions.HotPathNetworkTimeout, retry.NetworkTimeout);
var expected = TimeSpan.FromTicks(StorageClientOptions.HotPathNetworkTimeout.Ticks * 3)
+ TimeSpan.FromTicks((long)(StorageClientOptions.HotPathRetryDelay.Ticks * 1.2))
+ TimeSpan.FromTicks((long)(StorageClientOptions.HotPathRetryDelay.Ticks * 2 * 1.2));
Assert.Equal(expected, StorageClientOptions.WorstCase(retry));
}
[Fact]
public void Blob_worst_case_counts_four_attempts_and_three_backoffs()
{
var blob = StorageClientOptions.Blob().Retry;
// 4 attempts × 60 s + backoffs 0.6 s, 1.2 s, 2.4 s (all under the 4 s cap).
var expected = TimeSpan.FromTicks(StorageClientOptions.BlobNetworkTimeout.Ticks * 4)
+ TimeSpan.FromMilliseconds(600) + TimeSpan.FromMilliseconds(1200) + TimeSpan.FromMilliseconds(2400);
Assert.Equal(expected, StorageClientOptions.WorstCase(blob));
}
[Fact]
public void Queue_options_carry_the_requested_encoding()
{
Assert.Equal(Azure.Storage.Queues.QueueMessageEncoding.Base64,
StorageClientOptions.Queue(Azure.Storage.Queues.QueueMessageEncoding.Base64).MessageEncoding);
Assert.Equal(Azure.Storage.Queues.QueueMessageEncoding.None, StorageClientOptions.Queue().MessageEncoding);
}
}
@@ -1,6 +1,9 @@
using Azure;
using AutopilotMonitor.Functions.Helpers;
using Xunit;
using System;
using System.IO;
using System.Net.Http;
namespace AutopilotMonitor.Functions.Tests;
@@ -59,4 +62,40 @@ public class StorageErrorsTests
Assert.False(StorageErrors.IsPayloadTooLarge(null!));
Assert.False(StorageErrors.IsAlreadyExists(null!));
}
[Fact]
public void No_response_at_all_is_transient()
=> Assert.True(StorageErrors.IsTransient(Failure(0)));
// ---- Exception overload: what an exhausted SDK budget looks like on a path without a caller token ----
[Fact]
public void Sdk_network_timeout_is_transient()
=> Assert.True(StorageErrors.IsTransient((Exception)new TaskCanceledException("exceeded the configured timeout")));
[Fact]
public void Socket_failures_are_transient()
{
Assert.True(StorageErrors.IsTransient(new IOException("connection reset")));
Assert.True(StorageErrors.IsTransient(new HttpRequestException("name resolution failed")));
}
[Fact]
public void Aggregate_of_retried_attempts_is_transient_only_when_every_attempt_was()
{
var allTransient = new AggregateException(new TaskCanceledException(), Failure(503), new IOException());
var mixed = new AggregateException(new TaskCanceledException(), Failure(404));
Assert.True(StorageErrors.IsTransient(allTransient));
Assert.False(StorageErrors.IsTransient(mixed));
Assert.False(StorageErrors.IsTransient(new AggregateException()));
}
[Fact]
public void Ordinary_exceptions_are_not_transient()
{
Assert.False(StorageErrors.IsTransient(new InvalidOperationException()));
Assert.False(StorageErrors.IsTransient((Exception)Failure(412)));
Assert.False(StorageErrors.IsTransient((Exception?)null!));
}
}
@@ -57,8 +57,8 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
public Task<RawPage<SessionSummary>> GetSessionsPageAsync(string tenantId, int? days, int pageSize, string? continuation)
=> _storage.GetSessionsPageAsync(tenantId, days, pageSize, continuation);
public Task<RawPage<SessionSummary>> GetAllSessionsPageAsync(string? tenantIdFilter, int? days, int pageSize, string? continuation, IReadOnlyCollection<string>? allowedTenantIds = null, IEnumerable<string>? select = null)
=> _storage.GetAllSessionsPageAsync(tenantIdFilter, days, pageSize, continuation, allowedTenantIds, select);
public Task<RawPage<SessionSummary>> GetAllSessionsPageAsync(string? tenantIdFilter, int? days, int pageSize, string? continuation, IReadOnlyCollection<string>? allowedTenantIds = null, IEnumerable<string>? select = null, CancellationToken cancellationToken = default)
=> _storage.GetAllSessionsPageAsync(tenantIdFilter, days, pageSize, continuation, allowedTenantIds, select, cancellationToken);
public Task<SessionStats> GetSessionStatsAsync(string tenantId, int days)
=> _storage.GetSessionStatsAsync(tenantId, days);
@@ -335,6 +335,17 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
_logger.LogWarning(ex, "IngestTelemetry: storage request failed with status {Status} ({Code}) — answering {Http}", ex.Status, ex.ErrorCode, (int)status);
return AsOutput(await WriteErrorAsync(req, status, message, retryAfter));
}
catch (Exception ex) when (StorageErrors.IsTransient(ex))
{
// The SDK ran out of its budget (StorageClientOptions: per-attempt timeout ×
// retries, sized under the agent's 30 s) or the connection failed. A later replay
// can succeed, so 503 + Retry-After keeps the agent on its retry ladder while it
// is still listening; the generic 500 below would be replayed too, but without
// the wait hint and logged as a bug.
_logger.LogWarning(ex, "IngestTelemetry: storage call exhausted its budget ({Type}) — answering 503", ex.GetType().Name);
return AsOutput(await WriteErrorAsync(req, HttpStatusCode.ServiceUnavailable,
"Storage temporarily unavailable; retry later", StorageRetryAfterSeconds));
}
catch (Exception ex)
{
_logger.LogError(ex, "IngestTelemetry: unhandled exception");
@@ -72,8 +72,11 @@ namespace AutopilotMonitor.Functions.Functions.Sessions
}
}
// The request token ends the per-tenant fan-out when the browser tab closes or
// the caller gives up — a cross-tenant page nobody reads must not keep 32 queries alive.
var page = await _sessionRepo.GetAllSessionsPageAsync(
parsed.FilterTenantId, parsed.Days, parsed.PageSize, azureToken, allowedTenantIds);
parsed.FilterTenantId, parsed.Days, parsed.PageSize, azureToken, allowedTenantIds,
cancellationToken: req.FunctionContext.CancellationToken);
string? nextLink = null;
if (!string.IsNullOrEmpty(page.NextRawToken))
@@ -0,0 +1,116 @@
using System;
using Azure.Core;
using Azure.Data.Tables;
using Azure.Storage.Blobs;
using Azure.Storage.Queues;
namespace AutopilotMonitor.Functions.Helpers;
/// <summary>
/// One place that sizes the Azure Storage SDK retry and timeout budgets. The SDK defaults
/// (100 s network timeout, 3 retries, exponential backoff from 0.8 s) let one hanging Table call
/// hold a request for minutes, while the agent gives up on its upload after
/// <see cref="AgentUploadTimeout"/> and replays the batch — under a storage brownout that fills
/// every worker slot with requests nobody is waiting for and doubles the load (audit 2026-09-05
/// F09). The hot-path budget therefore keeps the WORST CASE of one SDK operation (every attempt
/// running into its timeout, every backoff at its cap) under the agent's timeout, so a request
/// that cannot finish fails as 503 + Retry-After while the agent is still listening and its own
/// retry ladder takes over. Pinned by StorageClientOptionsTests.
/// <para>
/// Table and Queue share the hot-path budget: the ingest writes tables and enqueues follow-up
/// work on every batch. Blob is not on the ingest path (diagnostics uploads go straight to the
/// SAS) and carries single-shot uploads of tens of MB (IME MSI archive, report ZIPs), so it gets
/// a bounded but generous budget. No extra retry layer (Polly) on top of the SDK: that would
/// multiply retries instead of bounding them.
/// </para>
/// <para>
/// Known gap: a service <c>Retry-After</c> header on a 429/503 can stretch one backoff beyond
/// the computed cap — Azure.Core honours the larger of the two. Storage answers with small
/// values; the bound below is the code-side guarantee, not a hard ceiling on service hints.
/// </para>
/// </summary>
public static class StorageClientOptions
{
/// <summary>
/// The agent's upload HttpClient timeout (<c>MtlsHttpClientFactory.DefaultTimeout</c> on the
/// agent side). Every hot-path storage call chain must stay under it.
/// </summary>
public static readonly TimeSpan AgentUploadTimeout = TimeSpan.FromSeconds(30);
/// <summary>Per-attempt network timeout for Table and Queue calls.</summary>
public static readonly TimeSpan HotPathNetworkTimeout = TimeSpan.FromSeconds(8);
/// <summary>Retries after the first attempt for Table and Queue calls.</summary>
public const int HotPathMaxRetries = 2;
public static readonly TimeSpan HotPathRetryDelay = TimeSpan.FromMilliseconds(500);
public static readonly TimeSpan HotPathMaxRetryDelay = TimeSpan.FromSeconds(4);
/// <summary>Per-attempt network timeout for Blob calls (single-shot uploads, streamed reads).</summary>
public static readonly TimeSpan BlobNetworkTimeout = TimeSpan.FromSeconds(60);
public const int BlobMaxRetries = 3;
/// <summary>
/// Azure.Core adds up to 20 % jitter on top of the exponential delay; the worst-case bound
/// accounts for it.
/// </summary>
internal const double BackoffJitterFactor = 1.2;
public static TableClientOptions Table()
{
var options = new TableClientOptions();
ApplyHotPath(options.Retry);
return options;
}
public static QueueClientOptions Queue(QueueMessageEncoding encoding = QueueMessageEncoding.None)
{
var options = new QueueClientOptions { MessageEncoding = encoding };
ApplyHotPath(options.Retry);
return options;
}
public static BlobClientOptions Blob()
{
var options = new BlobClientOptions();
options.Retry.Mode = RetryMode.Exponential;
options.Retry.MaxRetries = BlobMaxRetries;
options.Retry.Delay = HotPathRetryDelay;
options.Retry.MaxDelay = HotPathMaxRetryDelay;
options.Retry.NetworkTimeout = BlobNetworkTimeout;
return options;
}
private static void ApplyHotPath(RetryOptions retry)
{
retry.Mode = RetryMode.Exponential;
retry.MaxRetries = HotPathMaxRetries;
retry.Delay = HotPathRetryDelay;
retry.MaxDelay = HotPathMaxRetryDelay;
retry.NetworkTimeout = HotPathNetworkTimeout;
}
/// <summary>
/// Upper bound on the wall-clock one SDK operation can take under these options: every
/// attempt runs into <see cref="RetryOptions.NetworkTimeout"/>, every backoff is the
/// exponential delay (with jitter) capped at <see cref="RetryOptions.MaxDelay"/>.
/// </summary>
public static TimeSpan WorstCase(RetryOptions retry)
{
if (retry == null) throw new ArgumentNullException(nameof(retry));
var attempts = retry.MaxRetries + 1;
var total = TimeSpan.FromTicks(retry.NetworkTimeout.Ticks * attempts);
for (var attempt = 0; attempt < retry.MaxRetries; attempt++)
{
var backoff = retry.Mode == RetryMode.Exponential
? TimeSpan.FromTicks((long)(retry.Delay.Ticks * Math.Pow(2, attempt) * BackoffJitterFactor))
: retry.Delay;
if (backoff > retry.MaxDelay) backoff = retry.MaxDelay;
total += backoff;
}
return total;
}
}
@@ -1,4 +1,7 @@
using System;
using System.IO;
using System.Linq;
using System.Net.Http;
using Azure;
namespace AutopilotMonitor.Functions.Helpers;
@@ -21,9 +24,31 @@ public static class StorageErrors
"PropertyValueTooLarge",
];
/// <summary>Throttling, timeouts and service-side failures — a later retry can succeed.</summary>
/// <summary>
/// Throttling, timeouts and service-side failures — a later retry can succeed. Status 0 is
/// the SDK's "no response" (connection refused/reset before any status arrived).
/// </summary>
public static bool IsTransient(RequestFailedException ex)
=> ex != null && Array.IndexOf(TransientStatuses, ex.Status) >= 0;
=> ex != null && (ex.Status == 0 || Array.IndexOf(TransientStatuses, ex.Status) >= 0);
/// <summary>
/// The exception shapes an exhausted SDK budget surfaces as (see <c>StorageClientOptions</c>):
/// the per-attempt network timeout is an <see cref="OperationCanceledException"/>, a socket
/// failure an <see cref="IOException"/> / <see cref="HttpRequestException"/>, and after
/// several attempts Azure.Core wraps them in an <see cref="AggregateException"/>. Only for
/// call paths WITHOUT a caller cancellation token (the ingest has none): there every
/// cancellation is the SDK's own timeout, never the caller giving up.
/// </summary>
public static bool IsTransient(Exception ex) => ex switch
{
null => false,
RequestFailedException failed => IsTransient(failed),
OperationCanceledException => true,
IOException => true,
HttpRequestException => true,
AggregateException aggregate => aggregate.InnerExceptions.Count > 0 && aggregate.InnerExceptions.All(IsTransient),
_ => false,
};
/// <summary>
/// The transaction or entity exceeds a size limit. HTTP 413 in production; some backends
@@ -18,6 +18,9 @@ namespace AutopilotMonitor.Functions.Middleware;
/// </summary>
public class RequestTelemetryMiddleware : IFunctionsWorkerMiddleware
{
/// <summary>GC pause during the request below this is not worth a column on the row.</summary>
internal static readonly TimeSpan GcPauseStampThreshold = TimeSpan.FromMilliseconds(100);
private readonly TelemetryClient _telemetryClient;
// Exact allowlist of device endpoints that READ + validate the X-Tenant-Id header (cert auth
@@ -52,6 +55,7 @@ public class RequestTelemetryMiddleware : IFunctionsWorkerMiddleware
var sw = Stopwatch.StartNew();
var startTime = DateTimeOffset.UtcNow;
var gcPauseAtStart = GC.GetTotalPauseDuration();
Exception? caughtException = null;
try
@@ -170,6 +174,14 @@ public class RequestTelemetryMiddleware : IFunctionsWorkerMiddleware
if (caughtException != null)
requestTelemetry.Properties["ExceptionType"] = caughtException.GetType().Name;
// Process-wide GC pause time that elapsed while this request was in flight — not the
// request's own allocation cost. The point is the opposite: a slow request whose
// duration is mostly GC pause was stalled by the process, not by its own work (the
// ProcessStall event says which kind of stall). Stamped only when worth a look.
var gcPause = GC.GetTotalPauseDuration() - gcPauseAtStart;
if (gcPause >= GcPauseStampThreshold)
requestTelemetry.Properties["GcPauseMs"] = Math.Round(gcPause.TotalMilliseconds).ToString(System.Globalization.CultureInfo.InvariantCulture);
// Bypass the worker's adaptive sampling for this item. The SDK sampling processor
// passes through any item whose SamplingPercentage is already set, so this is the
// per-item opt-out that host.json "excludedTypes" (host process only) cannot provide.
@@ -235,6 +235,7 @@ builder.Services.AddHostedService<AutopilotMonitor.Functions.Services.Offboardin
builder.Services.AddHostedService<TableInitializerService>(); // Initialize all tables at startup
builder.Services.AddSingleton(startupClock);
builder.Services.AddHostedService<StartupTelemetryService>(); // BackendStartupMs / BackendTableInitMs metrics on ApplicationStarted
builder.Services.AddHostedService<AutopilotMonitor.Functions.Telemetry.ProcessStallMonitor>(); // ProcessStall event + metric: process-wide pauses attributed to GC / thread-pool starvation
// Data Access Layer — repository interfaces backed by Table Storage.
// To switch to Cosmos DB: replace AddTableStorageDataAccess() with AddCosmosDataAccess().
@@ -10,6 +10,7 @@ using Azure.Storage.Sas;
using AutopilotMonitor.Shared.Models.Deletion;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using AutopilotMonitor.Functions.Helpers;
namespace AutopilotMonitor.Functions.Services
{
@@ -40,13 +41,13 @@ namespace AutopilotMonitor.Functions.Services
if (!string.IsNullOrEmpty(storageAccountName))
{
var blobUri = new Uri($"https://{storageAccountName}.blob.core.windows.net");
_blobServiceClient = new BlobServiceClient(blobUri, new DefaultAzureCredential());
_blobServiceClient = new BlobServiceClient(blobUri, new DefaultAzureCredential(), StorageClientOptions.Blob());
_usesManagedIdentity = true;
_logger.LogInformation("Blob Storage initialized with Managed Identity (account: {Account})", storageAccountName);
}
else if (!string.IsNullOrEmpty(connectionString))
{
_blobServiceClient = new BlobServiceClient(connectionString);
_blobServiceClient = new BlobServiceClient(connectionString, StorageClientOptions.Blob());
_usesManagedIdentity = false;
_logger.LogInformation("Blob Storage initialized with connection string");
}
@@ -20,10 +20,22 @@ namespace AutopilotMonitor.Functions.Services.Caching
///
/// The factory runs WITHOUT any caller's cancellation token on purpose: the task is shared,
/// so one caller's abort must not fault every other waiter — and a finished result is exactly
/// what the aborted caller's retry wants to find.
/// what the aborted caller's retry wants to find. What it runs on instead is the cache's own
/// <see cref="DefaultFactoryBudget"/>: a factory that hangs (a cross-tenant fan-out on a slow
/// storage account) would otherwise hold every waiter until the SDK's worst case; past the
/// budget the shared task faults with <see cref="OperationCanceledException"/>, the entry is
/// evicted, and the next caller starts fresh. Factories that ignore the token they are handed
/// are not bounded — pass it into the SDK calls.
/// </summary>
internal sealed class SingleFlightCache<T>
{
/// <summary>
/// Wall-clock one factory run may take. Sized for the cross-tenant fan-outs (32 parallel
/// partition queries, each under the hot-path SDK budget) — a single-flight aggregate that
/// needs longer is a scan that belongs in a job, not behind a portal request.
/// </summary>
internal static readonly TimeSpan DefaultFactoryBudget = TimeSpan.FromSeconds(30);
private sealed class Entry
{
public Entry(Lazy<Task<T>> task, DateTime expiresUtc)
@@ -39,6 +51,7 @@ namespace AutopilotMonitor.Functions.Services.Caching
private readonly ConcurrentDictionary<string, Entry> _entries = new(StringComparer.Ordinal);
private readonly Func<DateTime> _utcNow;
private readonly int _maxEntries;
private readonly TimeSpan _factoryBudget;
/// <param name="maxEntries">
/// Upper bound on live entries. When a NEW key is added beyond it, every other key is
@@ -47,20 +60,33 @@ namespace AutopilotMonitor.Functions.Services.Caching
/// </param>
public SingleFlightCache(int maxEntries = int.MaxValue) : this(() => DateTime.UtcNow, maxEntries) { }
/// <summary>Clock seam for tests.</summary>
internal SingleFlightCache(Func<DateTime> utcNow, int maxEntries = int.MaxValue)
/// <summary>Clock and budget seams for tests.</summary>
internal SingleFlightCache(Func<DateTime> utcNow, int maxEntries = int.MaxValue, TimeSpan? factoryBudget = null)
{
if (maxEntries < 1) throw new ArgumentOutOfRangeException(nameof(maxEntries));
_utcNow = utcNow;
_maxEntries = maxEntries;
_factoryBudget = factoryBudget ?? DefaultFactoryBudget;
}
/// <summary>
/// <see cref="GetOrAddAsync(string, TimeSpan, Func{CancellationToken, Task{T}})"/> for a
/// factory with nothing to cancel (in-memory work, or SDK calls that take no token).
/// </summary>
public Task<T> GetOrAddAsync(string key, TimeSpan ttl, Func<Task<T>> factory)
{
if (factory == null) throw new ArgumentNullException(nameof(factory));
return GetOrAddAsync(key, ttl, _ => factory());
}
/// <summary>
/// Returns the cached value for <paramref name="key"/>, or runs <paramref name="factory"/>
/// exactly once for all concurrent callers and caches its result for <paramref name="ttl"/>.
/// A factory failure propagates to every waiter and leaves nothing behind.
/// The token handed to the factory is the cache's budget, never a caller's. A factory
/// failure (including the budget running out) propagates to every waiter and leaves
/// nothing behind.
/// </summary>
public async Task<T> GetOrAddAsync(string key, TimeSpan ttl, Func<Task<T>> factory)
public async Task<T> GetOrAddAsync(string key, TimeSpan ttl, Func<CancellationToken, Task<T>> factory)
{
if (key == null) throw new ArgumentNullException(nameof(key));
if (factory == null) throw new ArgumentNullException(nameof(factory));
@@ -131,12 +157,18 @@ namespace AutopilotMonitor.Functions.Services.Caching
}
}
private static Entry CreateEntry(Func<Task<T>> factory, DateTime now, TimeSpan ttl)
private Entry CreateEntry(Func<CancellationToken, Task<T>> factory, DateTime now, TimeSpan ttl)
{
// Lazy defers the factory to the first awaiter, so it runs on the caller's context
// and any synchronous prefix stays attributable to that request.
var lazy = new Lazy<Task<T>>(factory, LazyThreadSafetyMode.ExecutionAndPublication);
var lazy = new Lazy<Task<T>>(() => RunWithinBudgetAsync(factory), LazyThreadSafetyMode.ExecutionAndPublication);
return new Entry(lazy, now.Add(ttl));
}
private async Task<T> RunWithinBudgetAsync(Func<CancellationToken, Task<T>> factory)
{
using var budget = new CancellationTokenSource(_factoryBudget);
return await factory(budget.Token).ConfigureAwait(false);
}
}
}
@@ -13,6 +13,7 @@ using Azure.Storage.Blobs.Models;
using Azure.Storage.Sas;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using AutopilotMonitor.Functions.Helpers;
namespace AutopilotMonitor.Functions.Services.Diagnostics
{
@@ -66,7 +67,7 @@ namespace AutopilotMonitor.Functions.Services.Diagnostics
if (!string.IsNullOrEmpty(storageAccountName))
{
var blobUri = new Uri($"https://{storageAccountName}.blob.core.windows.net");
_blobServiceClient = new BlobServiceClient(blobUri, new DefaultAzureCredential());
_blobServiceClient = new BlobServiceClient(blobUri, new DefaultAzureCredential(), StorageClientOptions.Blob());
_usesManagedIdentity = true;
_logger.LogInformation(
"HostedDiagnosticsBlobService initialized with Managed Identity (account: {Account}, container: {Container})",
@@ -74,7 +75,7 @@ namespace AutopilotMonitor.Functions.Services.Diagnostics
}
else if (!string.IsNullOrEmpty(connectionString))
{
_blobServiceClient = new BlobServiceClient(connectionString);
_blobServiceClient = new BlobServiceClient(connectionString, StorageClientOptions.Blob());
_usesManagedIdentity = false;
_logger.LogInformation(
"HostedDiagnosticsBlobService initialized with connection string (container: {Container})",
@@ -2,6 +2,7 @@ using System;
using Azure.Identity;
using Azure.Storage.Queues;
using Microsoft.Extensions.Configuration;
using AutopilotMonitor.Functions.Helpers;
namespace AutopilotMonitor.Functions.Services.Queueing
{
@@ -48,12 +49,12 @@ namespace AutopilotMonitor.Functions.Services.Queueing
$"https://{storageAccountName}.queue.core.windows.net/{queueName}");
return new QueueClient(uri, credential, options);
};
_serviceBuilder = () => new QueueServiceClient(serviceUri, credential);
_serviceBuilder = () => new QueueServiceClient(serviceUri, credential, StorageClientOptions.Queue());
}
else if (!string.IsNullOrEmpty(connectionString))
{
_builder = (queueName, options) => new QueueClient(connectionString, queueName, options);
_serviceBuilder = () => new QueueServiceClient(connectionString);
_serviceBuilder = () => new QueueServiceClient(connectionString, StorageClientOptions.Queue());
}
else
{
@@ -82,9 +83,8 @@ namespace AutopilotMonitor.Functions.Services.Queueing
if (string.IsNullOrWhiteSpace(queueName))
throw new ArgumentException("Queue name must not be empty.", nameof(queueName));
var options = base64
? new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }
: new QueueClientOptions();
// Hot-path budget (StorageClientOptions): producers enqueue on every ingest batch.
var options = StorageClientOptions.Queue(base64 ? QueueMessageEncoding.Base64 : QueueMessageEncoding.None);
return _builder(queueName, options);
}
@@ -610,11 +610,11 @@ namespace AutopilotMonitor.Functions.Services
{
var bucketStart = new DateTime(sinceUtc.Ticks - sinceUtc.Ticks % TimeSpan.TicksPerMinute, DateTimeKind.Utc);
var snapshot = await _appSummarySnapshotCache.GetOrAddAsync(
bucketStart.ToString("yyyyMMddHHmm"), AppSummarySnapshotTtl, () => ScanCrossTenantAppSummariesAsync(bucketStart));
bucketStart.ToString("yyyyMMddHHmm"), AppSummarySnapshotTtl, ct => ScanCrossTenantAppSummariesAsync(bucketStart, ct));
return new List<AppInstallSummary>(snapshot);
}
private async Task<List<AppInstallSummary>> ScanCrossTenantAppSummariesAsync(DateTime sinceUtc)
private async Task<List<AppInstallSummary>> ScanCrossTenantAppSummariesAsync(DateTime sinceUtc, CancellationToken cancellationToken)
{
var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.AppInstallSummaries);
var windowFilter = AppSummaryWindowFilter(sinceUtc);
@@ -624,12 +624,12 @@ namespace AutopilotMonitor.Functions.Services
if (tenantIds.Count == 0)
{
// Empty config table (fresh install): the legacy cross-partition scan is the safety net.
return await ScanAppInstallSummariesAsync(tableClient, windowFilter, select, CancellationToken.None);
return await ScanAppInstallSummariesAsync(tableClient, windowFilter, select, cancellationToken);
}
return await BoundedFanOut.RunAsync(tenantIds, BoundedFanOut.CrossTenantConcurrency,
(tenantId, ct) => ScanAppInstallSummariesAsync(tableClient, $"PartitionKey eq '{tenantId}' and {windowFilter}", select, ct),
CancellationToken.None);
cancellationToken);
}
/// <summary>
@@ -1162,7 +1162,8 @@ namespace AutopilotMonitor.Functions.Services
public async Task<RawPage<SessionSummary>> GetAllSessionsPageAsync(
string? tenantIdFilter, int? days, int pageSize, string? continuation,
IReadOnlyCollection<string>? allowedTenantIds = null,
IEnumerable<string>? select = null)
IEnumerable<string>? select = null,
CancellationToken cancellationToken = default)
{
if (pageSize < 1) throw new ArgumentOutOfRangeException(nameof(pageSize));
@@ -1178,7 +1179,7 @@ namespace AutopilotMonitor.Functions.Services
return new RawPage<SessionSummary>(tenantPage.Sessions, tenantPage.HasMore ? tenantPage.NextCursor : null);
}
var page = await FetchAllSessionsPageInternalAsync(maxResults: pageSize, cursor: continuation, days: days, allowedTenantIds: allowedTenantIds, select: select);
var page = await FetchAllSessionsPageInternalAsync(maxResults: pageSize, cursor: continuation, days: days, allowedTenantIds: allowedTenantIds, select: select, cancellationToken: cancellationToken);
return new RawPage<SessionSummary>(page.Sessions, page.HasMore ? page.NextCursor : null);
}
@@ -1278,10 +1279,10 @@ namespace AutopilotMonitor.Functions.Services
var cacheKey = $"{days}|{DescribeTenantBound(allowedTenantIds)}";
return await _sessionStatsCache.GetOrAddAsync(cacheKey, SessionStatsCacheTtl,
() => ComputeAllSessionStatsAsync(days, allowedTenantIds));
ct => ComputeAllSessionStatsAsync(days, allowedTenantIds, ct));
}
private async Task<SessionStats> ComputeAllSessionStatsAsync(int days, IReadOnlyCollection<string>? allowedTenantIds)
private async Task<SessionStats> ComputeAllSessionStatsAsync(int days, IReadOnlyCollection<string>? allowedTenantIds, CancellationToken cancellationToken)
{
var tenantIds = ApplyTenantBound(await GetTenantIdsCachedAsync(), allowedTenantIds);
if (tenantIds.Count == 0)
@@ -1296,7 +1297,7 @@ namespace AutopilotMonitor.Functions.Services
}
var rows = await DrainSessionsIndexWindowAsync(
tenantIds, startUtc: DateTime.UtcNow.AddDays(-days), endUtc: null, SessionStatsProjection, CancellationToken.None);
tenantIds, startUtc: DateTime.UtcNow.AddDays(-days), endUtc: null, SessionStatsProjection, cancellationToken);
return AggregateSessionStats(rows, days);
}
@@ -1461,7 +1462,7 @@ namespace AutopilotMonitor.Functions.Services
/// </summary>
private async Task<(List<SessionSummary> Sessions, bool HasMore, string? NextCursor)> FetchAllSessionsPageInternalAsync(
int maxResults, string? cursor, int? days, IReadOnlyCollection<string>? allowedTenantIds = null,
IEnumerable<string>? select = null)
IEnumerable<string>? select = null, CancellationToken cancellationToken = default)
{
try
{
@@ -1531,7 +1532,7 @@ namespace AutopilotMonitor.Functions.Services
if (tenantSessions.Count >= fetchPerTenant) break;
}
return tenantSessions;
}, CancellationToken.None);
}, cancellationToken);
// Step 3: Merge-sort across tenants by StartedAt descending
allSessions = allSessions.OrderByDescending(s => s.StartedAt).ToList();
@@ -12,6 +12,7 @@ using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using AutopilotMonitor.Functions.Helpers;
namespace AutopilotMonitor.Functions.Services
{
@@ -50,7 +51,7 @@ namespace AutopilotMonitor.Functions.Services
/// storage failure so callers keep their own fallback semantics (empty page vs. legacy scan).
/// </summary>
internal Task<IReadOnlyList<string>> GetTenantIdsCachedAsync()
=> _tenantIdCache.GetOrAddAsync(TenantIdCacheKey, TenantIdCacheTtl, () => QueryTenantIdsAsync(CancellationToken.None));
=> _tenantIdCache.GetOrAddAsync(TenantIdCacheKey, TenantIdCacheTtl, ct => QueryTenantIdsAsync(ct));
private async Task<IReadOnlyList<string>> QueryTenantIdsAsync(CancellationToken cancellationToken)
{
@@ -98,13 +99,13 @@ namespace AutopilotMonitor.Functions.Services
{
// Managed Identity: use DefaultAzureCredential with storage account name
var tableUri = new Uri($"https://{storageAccountName}.table.core.windows.net");
_tableServiceClient = new TableServiceClient(tableUri, new DefaultAzureCredential());
_tableServiceClient = new TableServiceClient(tableUri, new DefaultAzureCredential(), StorageClientOptions.Table());
_logger.LogInformation("Table Storage initialized with Managed Identity (account: {Account})", storageAccountName);
}
else if (!string.IsNullOrEmpty(connectionString))
{
// Fallback: connection string (local dev, legacy)
_tableServiceClient = new TableServiceClient(connectionString);
_tableServiceClient = new TableServiceClient(connectionString, StorageClientOptions.Table());
_logger.LogInformation("Table Storage initialized with connection string");
}
else
@@ -0,0 +1,241 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Globalization;
using System.Threading;
using System.Threading.Tasks;
using AutopilotMonitor.Functions.Services;
using Microsoft.ApplicationInsights;
using Microsoft.ApplicationInsights.Channel;
using Microsoft.ApplicationInsights.DataContracts;
using Microsoft.ApplicationInsights.Metrics;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
namespace AutopilotMonitor.Functions.Telemetry;
/// <summary>
/// Point-in-time reading of the runtime counters a process-wide pause leaves traces in. Two
/// consecutive snapshots around a late timer tick say WHAT paused the process: the GC pause
/// clock (<see cref="GC.GetTotalPauseDuration"/>, exact) or the thread pool (queued work,
/// thread injection).
/// </summary>
public readonly record struct ProcessSnapshot(
TimeSpan GcPause,
int Gen0,
int Gen1,
int Gen2,
int ThreadPoolThreads,
long PendingWorkItems,
long CompletedWorkItems,
long WorkingSetBytes,
long HeapSizeBytes,
long CommittedBytes,
int LastGcGeneration,
TimeSpan LastGcPause,
bool LastGcCompacted)
{
public static ProcessSnapshot Capture()
{
var info = GC.GetGCMemoryInfo();
var lastPause = TimeSpan.Zero;
foreach (var pause in info.PauseDurations)
lastPause += pause;
return new ProcessSnapshot(
GcPause: GC.GetTotalPauseDuration(),
Gen0: GC.CollectionCount(0),
Gen1: GC.CollectionCount(1),
Gen2: GC.CollectionCount(2),
ThreadPoolThreads: ThreadPool.ThreadCount,
PendingWorkItems: ThreadPool.PendingWorkItemCount,
CompletedWorkItems: ThreadPool.CompletedWorkItemCount,
WorkingSetBytes: Environment.WorkingSet,
HeapSizeBytes: info.HeapSizeBytes,
CommittedBytes: info.TotalCommittedBytes,
LastGcGeneration: info.Generation,
LastGcPause: lastPause,
LastGcCompacted: info.Compacted);
}
}
/// <summary>What a stall report says caused the pause.</summary>
public static class StallCause
{
/// <summary>At least half of the drift was GC pause time.</summary>
public const string Gc = "gc";
/// <summary>Work was queued or the pool injected threads: starvation, not GC.</summary>
public const string ThreadPool = "threadpool";
/// <summary>Neither clock explains the drift (CPU steal, host pause, page-in).</summary>
public const string Unknown = "unknown";
}
/// <summary>One classified stall: the drift plus the counter deltas that explain it.</summary>
public sealed record StallReport(
TimeSpan Drift,
string Cause,
TimeSpan GcPauseDelta,
int Gen0Delta,
int Gen1Delta,
int Gen2Delta,
int ThreadPoolThreadsDelta,
long CompletedWorkItemsDelta,
ProcessSnapshot After);
/// <summary>
/// Pure classification so the rule is unit-testable without a live process pause.
/// </summary>
public static class ProcessStallClassifier
{
/// <summary>Fraction of the drift the GC pause clock must cover to blame the GC.</summary>
internal const double GcShareThreshold = 0.5;
public static StallReport Classify(TimeSpan drift, ProcessSnapshot before, ProcessSnapshot after)
{
var gcPauseDelta = after.GcPause - before.GcPause;
if (gcPauseDelta < TimeSpan.Zero) gcPauseDelta = TimeSpan.Zero;
var threadsDelta = after.ThreadPoolThreads - before.ThreadPoolThreads;
string cause;
if (drift > TimeSpan.Zero && gcPauseDelta.Ticks >= drift.Ticks * GcShareThreshold)
cause = StallCause.Gc;
else if (after.PendingWorkItems > 0 || threadsDelta > 0)
cause = StallCause.ThreadPool;
else
cause = StallCause.Unknown;
return new StallReport(
Drift: drift,
Cause: cause,
GcPauseDelta: gcPauseDelta,
Gen0Delta: after.Gen0 - before.Gen0,
Gen1Delta: after.Gen1 - before.Gen1,
Gen2Delta: after.Gen2 - before.Gen2,
ThreadPoolThreadsDelta: threadsDelta,
CompletedWorkItemsDelta: after.CompletedWorkItems - before.CompletedWorkItems,
After: after);
}
}
/// <summary>
/// Detects process-wide pauses in the worker and reports what caused them. Production showed
/// episodes where every in-flight request on ONE instance finished at the same instant after
/// 5–10 s, with no slow storage call, no warning, no cold start and idle CPU (2026-09-03 12:15
/// and 19:00) — a pause of the whole process, invisible to per-minute performance counters.
/// <para>
/// Mechanism: a 1 s timer tick measures its own lateness. A tick that arrives
/// <see cref="StallThreshold"/> or more late means the process did not run continuations for
/// that long. Two <see cref="ProcessSnapshot"/>s around the tick attribute the drift: the GC
/// pause clock is exact, so a drift the GC covers is a GC pause; queued work or thread
/// injection means thread-pool starvation (the tick's own continuation waits in that queue);
/// anything else is <see cref="StallCause.Unknown"/>. Cost: one delay per second per instance.
/// </para>
/// <para>
/// Output: <c>customEvents</c> <see cref="EventName"/> with the deltas (sampling bypassed —
/// the events are rare and every one is wanted) and the pre-aggregated metric
/// <see cref="MetricName"/> (value = drift ms, dimension <c>cause</c>) for the trend. Pairs with
/// the <c>GcPauseMs</c> request dimension stamped by RequestTelemetryMiddleware, which says how
/// much of one slow request was GC pause. Query:
/// <c>customEvents | where name == 'ProcessStall' | project timestamp, cloud_RoleInstance, customDimensions</c>.
/// </para>
/// </summary>
public sealed class ProcessStallMonitor : BackgroundService
{
public const string EventName = "ProcessStall";
public const string MetricName = "ProcessStall";
internal static readonly TimeSpan Tick = TimeSpan.FromSeconds(1);
/// <summary>Lateness of one tick that counts as a stall.</summary>
internal static readonly TimeSpan StallThreshold = TimeSpan.FromSeconds(1);
private readonly TelemetryClient? _telemetry;
private readonly Metric? _metric;
private readonly BackendBuildInfo _buildInfo;
private readonly ILogger<ProcessStallMonitor> _logger;
public ProcessStallMonitor(BackendBuildInfo buildInfo, ILogger<ProcessStallMonitor> logger, TelemetryClient? telemetry = null)
{
_buildInfo = buildInfo;
_logger = logger;
_telemetry = telemetry;
_metric = telemetry?.GetMetric(MetricName, "cause");
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
var previous = ProcessSnapshot.Capture();
var stopwatch = new Stopwatch();
while (!stoppingToken.IsCancellationRequested)
{
stopwatch.Restart();
try
{
await Task.Delay(Tick, stoppingToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
break;
}
var drift = stopwatch.Elapsed - Tick;
var current = ProcessSnapshot.Capture();
if (drift >= StallThreshold)
{
try
{
Report(ProcessStallClassifier.Classify(drift, previous, current));
}
catch (Exception ex)
{
_logger.LogWarning(ex, "ProcessStallMonitor: failed to report a stall");
}
}
previous = current;
}
}
private void Report(StallReport report)
{
var driftMs = Math.Round(report.Drift.TotalMilliseconds);
_metric?.TrackValue(driftMs, report.Cause);
if (_telemetry == null) return;
var after = report.After;
var evt = new EventTelemetry(EventName);
var props = evt.Properties;
props["driftMs"] = Format(driftMs);
props["cause"] = report.Cause;
props["gcPauseMs"] = Format(Math.Round(report.GcPauseDelta.TotalMilliseconds));
props["gen0"] = Format(report.Gen0Delta);
props["gen1"] = Format(report.Gen1Delta);
props["gen2"] = Format(report.Gen2Delta);
props["lastGcGeneration"] = Format(after.LastGcGeneration);
props["lastGcPauseMs"] = Format(Math.Round(after.LastGcPause.TotalMilliseconds));
props["lastGcCompacted"] = after.LastGcCompacted ? "true" : "false";
props["heapMb"] = Format(after.HeapSizeBytes >> 20);
props["committedMb"] = Format(after.CommittedBytes >> 20);
props["workingSetMb"] = Format(after.WorkingSetBytes >> 20);
props["threadPoolThreads"] = Format(after.ThreadPoolThreads);
props["threadPoolThreadsDelta"] = Format(report.ThreadPoolThreadsDelta);
props["pendingWorkItems"] = Format(after.PendingWorkItems);
props["completedWorkItemsDelta"] = Format(report.CompletedWorkItemsDelta);
props["instance"] = Environment.MachineName;
props["version"] = _buildInfo.Version;
// Rare by construction, and every one is evidence: opt out of the worker's adaptive
// sampling the same way the canonical request record does.
((ISupportSampling)evt).SamplingPercentage = 100;
_telemetry.TrackEvent(evt);
}
private static string Format(double value) => value.ToString(CultureInfo.InvariantCulture);
private static string Format(long value) => value.ToString(CultureInfo.InvariantCulture);
}
@@ -66,7 +66,8 @@ namespace AutopilotMonitor.Shared.DataAccess
Task<RawPage<SessionSummary>> GetAllSessionsPageAsync(
string? tenantIdFilter, int? days, int pageSize, string? continuation,
IReadOnlyCollection<string>? allowedTenantIds = null,
IEnumerable<string>? select = null);
IEnumerable<string>? select = null,
CancellationToken cancellationToken = default);
/// <summary>
/// Server-side aggregation for the dashboard stats cards (per-tenant scope).