diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/BlockedDeviceServiceCrossInstanceTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/BlockedDeviceServiceCrossInstanceTests.cs
index 08774c5a..ee089615 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/BlockedDeviceServiceCrossInstanceTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/BlockedDeviceServiceCrossInstanceTests.cs
@@ -17,7 +17,10 @@ namespace AutopilotMonitor.Functions.Tests;
/// without ever re-consulting storage. A block added on instance B (via DeviceBlockFunction)
/// therefore stayed invisible to instance A — kill signals never reached affected devices.
///
-/// Fix: cache miss → storage point-read fallback + stale-positive revalidate window.
+/// Fix: cache miss → storage point-read fallback + stale-entry revalidate window. Since the
+/// negative-cache cost fix (2026-06-09), the "not blocked" answer is cached too: cross-instance
+/// blocks now surface within the revalidation window (same bound that always applied to
+/// unblock / Block→Kill upgrades) instead of on the next call.
///
public class BlockedDeviceServiceCrossInstanceTests
{
@@ -30,26 +33,32 @@ public class BlockedDeviceServiceCrossInstanceTests
[Theory]
[InlineData("Block")] // Pauses uploads; session remains alive
[InlineData("Kill")] // Forces remote self-destruct
- public async Task IsBlockedAsync_CacheMissAfterTenantLoaded_FallsThroughToStorage(string action)
+ public async Task IsBlockedAsync_BlockFromOtherInstance_SurfacesAfterRevalidationWindow(string action)
{
// Arrange — instance A loads the tenant's block list while the device is NOT yet blocked.
// Action is action-agnostic: the staleness bug hit both Block and Kill the same way,
- // so the fallback must surface whichever action storage holds.
+ // so the revalidate path must surface whichever action storage holds.
var repo = new FakeDeviceSecurityRepository();
var svc = CreateService(repo);
var beforeBlock = await svc.IsBlockedAsync(TenantA, SerialPF55);
- Assert.False(beforeBlock.isBlocked); // baseline: load happens, cache empty for this device
+ Assert.False(beforeBlock.isBlocked); // baseline: caches the NEGATIVE answer (cost fix)
// Act — another Function App instance writes the block directly to storage (no local-cache
- // update on this instance, mirroring the actual multi-instance bug).
+ // update on this instance). Within the revalidation window the cached negative answer is
+ // served without a storage read; once the entry ages past the window, the revalidate
+ // point-read must find the block.
repo.SetBlock(TenantA, SerialPF55, DateTime.UtcNow.AddHours(12), action);
- var afterBlock = await svc.IsBlockedAsync(TenantA, SerialPF55);
+ var withinWindow = await svc.IsBlockedAsync(TenantA, SerialPF55);
+ Assert.False(withinWindow.isBlocked); // fresh negative entry — block not yet visible
- // Assert — cache miss path must point-read storage, find the block, and return it.
- Assert.True(afterBlock.isBlocked);
- Assert.Equal(action, afterBlock.action);
+ BackdateCacheEntry(svc, TenantA, SerialPF55, TimeSpan.FromMinutes(5));
+ var afterWindow = await svc.IsBlockedAsync(TenantA, SerialPF55);
+
+ // Assert — the stale-entry revalidate path must point-read storage, find the block, and return it.
+ Assert.True(afterWindow.isBlocked);
+ Assert.Equal(action, afterWindow.action);
}
[Fact]
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/BlockedDeviceServiceTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/BlockedDeviceServiceTests.cs
new file mode 100644
index 00000000..bfe7cc2a
--- /dev/null
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/BlockedDeviceServiceTests.cs
@@ -0,0 +1,124 @@
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Shared.DataAccess;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+///
+/// Locks in the negative-cache contract of : a healthy
+/// (never-blocked) device must NOT pay a BlockedDevices point-read on every ingest request —
+/// the "not blocked" answer is cached and revalidated within the same window as positive
+/// entries, so cross-instance blocks still propagate within seconds.
+///
+public class BlockedDeviceServiceTests
+{
+ private const string TenantId = "11111111-1111-1111-1111-111111111111";
+ private const string Serial = "SN-12345";
+
+ private static Mock NewRepo(bool blocked = false, DateTime? unblockAt = null)
+ {
+ var repo = new Mock();
+ repo.Setup(r => r.GetBlockedDevicesAsync(TenantId))
+ .ReturnsAsync(new List());
+ repo.Setup(r => r.IsDeviceBlockedAsync(TenantId, Serial))
+ .ReturnsAsync((blocked, unblockAt, "Block", (string?)null));
+ return repo;
+ }
+
+ [Fact]
+ public async Task NotBlockedDevice_SecondCallWithinWindow_DoesNotHitStorage()
+ {
+ var repo = NewRepo(blocked: false);
+ var service = new BlockedDeviceService(repo.Object, NullLogger.Instance);
+
+ var first = await service.IsBlockedAsync(TenantId, Serial);
+ var second = await service.IsBlockedAsync(TenantId, Serial);
+ var third = await service.IsBlockedAsync(TenantId, Serial);
+
+ Assert.False(first.isBlocked);
+ Assert.False(second.isBlocked);
+ Assert.False(third.isBlocked);
+ // Exactly ONE storage point-read (the first call's cache-miss refresh); the negative
+ // answer is served from cache afterwards. This is the per-request cost fix.
+ repo.Verify(r => r.IsDeviceBlockedAsync(TenantId, Serial), Times.Once);
+ }
+
+ [Fact]
+ public async Task NegativeEntry_AfterRevalidationWindow_SeesBlockFromOtherInstance()
+ {
+ var repo = NewRepo(blocked: false);
+ // Zero window → every subsequent call revalidates, simulating an expired entry.
+ var service = new BlockedDeviceService(
+ repo.Object, NullLogger.Instance, TimeSpan.Zero);
+
+ var first = await service.IsBlockedAsync(TenantId, Serial);
+ Assert.False(first.isBlocked);
+
+ // Another instance blocks the device → storage now answers "blocked".
+ repo.Setup(r => r.IsDeviceBlockedAsync(TenantId, Serial))
+ .ReturnsAsync((true, DateTime.UtcNow.AddHours(1), "Block", (string?)null));
+
+ var second = await service.IsBlockedAsync(TenantId, Serial);
+
+ Assert.True(second.isBlocked);
+ repo.Verify(r => r.IsDeviceBlockedAsync(TenantId, Serial), Times.Exactly(2));
+ }
+
+ [Fact]
+ public async Task BlockDevice_OverridesCachedNegativeEntry()
+ {
+ var repo = NewRepo(blocked: false);
+ var service = new BlockedDeviceService(repo.Object, NullLogger.Instance);
+
+ // Prime the negative cache.
+ var before = await service.IsBlockedAsync(TenantId, Serial);
+ Assert.False(before.isBlocked);
+
+ // Local block must flip the cached entry to blocked (AddOrUpdate sets IsBlocked = true) —
+ // without that, the fresh negative entry would keep answering "not blocked" for 30s.
+ await service.BlockDeviceAsync(TenantId, Serial, durationHours: 1, blockedByEmail: "alice@contoso.com");
+
+ var after = await service.IsBlockedAsync(TenantId, Serial);
+ Assert.True(after.isBlocked);
+ Assert.Equal("Block", after.action);
+ }
+
+ [Fact]
+ public async Task SessionAwareBlock_OverNegativeEntry_StaysSessionAware()
+ {
+ // Codex finding: a session-aware block overwriting a cached NEGATIVE entry must not
+ // widen to a whole-device block (negative entries carry BlockedSessionIds = null,
+ // which the merge previously kept — and null means "whole device" downstream).
+ var repo = NewRepo(blocked: false);
+ var service = new BlockedDeviceService(repo.Object, NullLogger.Instance);
+
+ // Prime the negative cache, then apply a session-aware auto-block locally.
+ _ = await service.IsBlockedAsync(TenantId, Serial);
+ await service.BlockDeviceAsync(TenantId, Serial, durationHours: 1,
+ blockedByEmail: "alice@contoso.com", blockedSessionId: "33333333-3333-3333-3333-333333333333");
+
+ // The blocked session itself is blocked...
+ var sameSession = await service.IsBlockedAsync(TenantId, Serial, "33333333-3333-3333-3333-333333333333");
+ Assert.True(sameSession.isBlocked);
+ Assert.Equal("33333333-3333-3333-3333-333333333333", sameSession.blockedSessionIds);
+
+ // ...but a NEW session on the same device auto-unblocks — with the widened
+ // whole-device shape this returned blocked=true and never auto-unblocked.
+ var newSession = await service.IsBlockedAsync(TenantId, Serial, "44444444-4444-4444-4444-444444444444");
+ Assert.False(newSession.isBlocked);
+ }
+
+ [Fact]
+ public async Task BlockedDevice_IsReportedBlocked_OnCacheMissRefresh()
+ {
+ var repo = NewRepo(blocked: true, unblockAt: DateTime.UtcNow.AddHours(2));
+ var service = new BlockedDeviceService(repo.Object, NullLogger.Instance);
+
+ var result = await service.IsBlockedAsync(TenantId, Serial);
+
+ Assert.True(result.isBlocked);
+ Assert.NotNull(result.unblockAt);
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IngestStatusPrefetchTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IngestStatusPrefetchTests.cs
new file mode 100644
index 00000000..cb85faa9
--- /dev/null
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/IngestStatusPrefetchTests.cs
@@ -0,0 +1,53 @@
+using AutopilotMonitor.Functions.Functions.Ingest;
+using AutopilotMonitor.Shared.Models;
+using Azure.Data.Tables;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+///
+/// Pins the wire-format contract of the ingest status prefetch (Codex finding): Sessions
+/// stores Status as a STRING (status.ToString()), so the prefetch must parse
+/// strings — an int-only pattern match never fired and silently disabled the point-read
+/// saving for the stall-heal probe.
+///
+public class IngestStatusPrefetchTests
+{
+ private static TableEntity Row(object? status = null)
+ {
+ var row = new TableEntity("11111111-1111-1111-1111-111111111111", "22222222-2222-2222-2222-222222222222");
+ if (status != null) row["Status"] = status;
+ return row;
+ }
+
+ [Theory]
+ [InlineData("InProgress", SessionStatus.InProgress)]
+ [InlineData("Stalled", SessionStatus.Stalled)]
+ [InlineData("succeeded", SessionStatus.Succeeded)] // mapper parses case-insensitively — mirror it
+ public void StringStatus_TheActualWireFormat_IsParsed(string stored, SessionStatus expected)
+ {
+ Assert.Equal(expected, IngestTelemetryFunction.TryReadSessionStatus(Row(stored)));
+ }
+
+ [Fact]
+ public void IntStatus_LegacyDefensiveFallback_IsCast()
+ {
+ var defined = (int)SessionStatus.Stalled;
+ Assert.Equal(SessionStatus.Stalled, IngestTelemetryFunction.TryReadSessionStatus(Row(defined)));
+ }
+
+ [Theory]
+ [InlineData("NotARealStatus")]
+ [InlineData("")]
+ public void UnparseableStatus_ReturnsNull_SoCallerFallsBackToOwnRead(string stored)
+ {
+ Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(Row(stored)));
+ }
+
+ [Fact]
+ public void MissingStatusOrRow_ReturnsNull()
+ {
+ Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(Row(status: null)));
+ Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(null));
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineArrayConditionTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineArrayConditionTests.cs
index cdd1e0f4..47f207cc 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineArrayConditionTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineArrayConditionTests.cs
@@ -230,7 +230,7 @@ public class RuleEngineArrayConditionTests
ruleRepo.Setup(r => r.GetRuleResultsAsync(It.IsAny(), It.IsAny())).ReturnsAsync(new List());
var sessionRepo = new Mock();
- sessionRepo.Setup(s => s.GetSessionEventsAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
+ sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
var ruleService = new AnalyzeRuleService(ruleRepo.Object, NullLogger.Instance);
var engine = new RuleEngine(ruleService, ruleRepo.Object, sessionRepo.Object, NullLogger.Instance);
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineCorrelationEvidenceSlimTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineCorrelationEvidenceSlimTests.cs
index caa895d4..481d68fc 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineCorrelationEvidenceSlimTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineCorrelationEvidenceSlimTests.cs
@@ -147,7 +147,7 @@ public class RuleEngineCorrelationEvidenceSlimTests
ruleRepo.Setup(r => r.GetRuleResultsAsync(It.IsAny(), It.IsAny())).ReturnsAsync(new List());
var sessionRepo = new Mock();
- sessionRepo.Setup(s => s.GetSessionEventsAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
+ sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
var ruleService = new AnalyzeRuleService(ruleRepo.Object, NullLogger.Instance);
var engine = new RuleEngine(ruleService, ruleRepo.Object, sessionRepo.Object, NullLogger.Instance);
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineDetectionFailureTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineDetectionFailureTests.cs
index 302bd2d8..a16ec53f 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineDetectionFailureTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineDetectionFailureTests.cs
@@ -266,7 +266,7 @@ public class RuleEngineDetectionFailureTests
ruleRepo.Setup(r => r.GetRuleResultsAsync(It.IsAny(), It.IsAny())).ReturnsAsync(new List());
var sessionRepo = new Mock();
- sessionRepo.Setup(s => s.GetSessionEventsAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
+ sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
var ruleService = new AnalyzeRuleService(ruleRepo.Object, NullLogger.Instance);
var engine = new RuleEngine(ruleService, ruleRepo.Object, sessionRepo.Object, NullLogger.Instance);
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs
index b8f1c053..aea83be1 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs
@@ -19,15 +19,15 @@ public class RuleEngineFailurePropagationTests
private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
[Fact]
- public async Task GetSessionEventsAsync_throwing_propagates_to_caller()
+ public async Task GetSessionEventsStrictAsync_throwing_propagates_to_caller()
{
var ruleRepo = new Mock();
ruleRepo.Setup(r => r.GetAnalyzeRulesAsync(It.IsAny())).ReturnsAsync(new List());
ruleRepo.Setup(r => r.GetRuleStatesAsync(It.IsAny())).ReturnsAsync(new Dictionary());
var sessionRepo = new Mock();
- sessionRepo.Setup(s => s.GetSessionEventsAsync(TenantId, SessionId, It.IsAny()))
- .ThrowsAsync(new InvalidOperationException("simulated GetSessionEventsAsync failure"));
+ sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny()))
+ .ThrowsAsync(new InvalidOperationException("simulated GetSessionEventsStrictAsync failure"));
var ruleService = new AnalyzeRuleService(ruleRepo.Object, NullLogger.Instance);
var engine = new RuleEngine(ruleService, ruleRepo.Object, sessionRepo.Object, NullLogger.Instance);
@@ -35,7 +35,7 @@ public class RuleEngineFailurePropagationTests
var ex = await Assert.ThrowsAsync(
() => engine.AnalyzeSessionAsync(TenantId, SessionId));
- Assert.Contains("simulated GetSessionEventsAsync failure", ex.Message);
+ Assert.Contains("simulated GetSessionEventsStrictAsync failure", ex.Message);
}
[Fact]
@@ -69,7 +69,7 @@ public class RuleEngineFailurePropagationTests
.ThrowsAsync(new InvalidOperationException("simulated GetRuleResultsAsync failure"));
var sessionRepo = new Mock();
- sessionRepo.Setup(s => s.GetSessionEventsAsync(TenantId, SessionId, It.IsAny()))
+ sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny()))
.ReturnsAsync(new List
{
new() { TenantId = TenantId, SessionId = SessionId, EventType = "phase_changed" }
@@ -88,16 +88,15 @@ public class RuleEngineFailurePropagationTests
public async Task Empty_event_list_returns_empty_outcome_without_throwing()
{
// Confirms the legitimate "no events" case is preserved: AnalyzeSessionAsync returns
- // an empty outcome cleanly. NOTE: TableStorageService.GetSessionEventsAsync currently
- // swallows storage exceptions and also returns an empty list — that fail-soft behavior
- // is a separate architectural defect (queue worker will treat the storage failure as
- // success). Tracked as a follow-up: storage helpers should fail loud.
+ // an empty outcome cleanly. Since the engine reads via GetSessionEventsStrictAsync
+ // (storage failures throw), an empty list here genuinely means "session without
+ // events" — the historical fail-soft ambiguity is resolved.
var ruleRepo = new Mock();
ruleRepo.Setup(r => r.GetAnalyzeRulesAsync(It.IsAny())).ReturnsAsync(new List());
ruleRepo.Setup(r => r.GetRuleStatesAsync(It.IsAny())).ReturnsAsync(new Dictionary());
var sessionRepo = new Mock();
- sessionRepo.Setup(s => s.GetSessionEventsAsync(TenantId, SessionId, It.IsAny()))
+ sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny()))
.ReturnsAsync(new List());
var ruleService = new AnalyzeRuleService(ruleRepo.Object, NullLogger.Instance);
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEnginePreconditionTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEnginePreconditionTests.cs
index 79ebad00..3398eb7e 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEnginePreconditionTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEnginePreconditionTests.cs
@@ -304,7 +304,7 @@ public class RuleEnginePreconditionTests
ruleRepo.Setup(r => r.GetRuleResultsAsync(It.IsAny(), It.IsAny())).ReturnsAsync(new List());
var sessionRepo = new Mock();
- sessionRepo.Setup(s => s.GetSessionEventsAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
+ sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
var ruleService = new AnalyzeRuleService(ruleRepo.Object, NullLogger.Instance);
var engine = new RuleEngine(ruleService, ruleRepo.Object, sessionRepo.Object, NullLogger.Instance);
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionGuardTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionGuardTests.cs
index e6f1e508..54b0906c 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionGuardTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/SessionDeletionGuardTests.cs
@@ -193,6 +193,39 @@ public class SessionDeletionGuardTests
Times.Never);
}
+ [Fact]
+ public async Task EnsureWritableAndGetRowAsync_returns_the_loaded_row_when_unlocked()
+ {
+ // Hot-path reuse contract (V2 ingest): the guard hands back the row it had to load anyway
+ // so callers can read Status & co. without a second point-read.
+ var guard = NewGuard(out var reader);
+ var row = new TableEntity(TenantId, SessionId)
+ {
+ ["DeletionState"] = SessionDeletionState.None,
+ ["Status"] = "InProgress", // Sessions stores Status as string (status.ToString())
+ };
+ reader.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny()))
+ .ReturnsAsync(row);
+
+ var returned = await guard.EnsureWritableAndGetRowAsync(TenantId, SessionId, callerContext: "V2.IngestTelemetry");
+
+ Assert.Same(row, returned);
+ }
+
+ [Fact]
+ public async Task EnsureWritableAndGetRowAsync_returns_null_when_row_missing_and_no_tombstone()
+ {
+ var guard = NewGuard(out var reader);
+ reader.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny()))
+ .ReturnsAsync((TableEntity?)null);
+ reader.Setup(r => r.GetActiveSessionTombstoneAsync(TenantId, SessionId, It.IsAny()))
+ .ReturnsAsync((TableEntity?)null);
+
+ var returned = await guard.EnsureWritableAndGetRowAsync(TenantId, SessionId, callerContext: "V2.IngestTelemetry");
+
+ Assert.Null(returned);
+ }
+
private static SessionDeletionGuard NewGuard(out Mock reader)
{
reader = new Mock();
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SessionTimeoutRuleTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SessionTimeoutRuleTests.cs
index 1749b04c..e3fae9fa 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/SessionTimeoutRuleTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/SessionTimeoutRuleTests.cs
@@ -125,7 +125,7 @@ public class SessionTimeoutRuleTests
ruleRepo.Setup(r => r.GetRuleResultsAsync(It.IsAny(), It.IsAny())).ReturnsAsync(new List());
var sessionRepo = new Mock();
- sessionRepo.Setup(s => s.GetSessionEventsAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
+ sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny())).ReturnsAsync(events);
var ruleService = new AnalyzeRuleService(ruleRepo.Object, NullLogger.Instance);
var engine = new RuleEngine(ruleService, ruleRepo.Object, sessionRepo.Object, NullLogger.Instance);
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityInventoryLoaderTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityInventoryLoaderTests.cs
index f809a80f..f4d105a1 100644
--- a/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityInventoryLoaderTests.cs
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityInventoryLoaderTests.cs
@@ -191,7 +191,7 @@ public class VulnerabilityInventoryLoaderTests
{
var sessionRepo = new Mock();
sessionRepo
- .Setup(r => r.GetSessionEventsAsync(TenantId, SessionId, It.IsAny()))
+ .Setup(r => r.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny()))
.ReturnsAsync(events);
return new VulnerabilityInventoryLoader(
diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs
index fb38f062..8469fe26 100644
--- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs
+++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs
@@ -89,7 +89,7 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
public Task> FetchAndClearPendingActionsAsync(string tenantId, string sessionId)
=> _storage.FetchAndClearPendingActionsAsync(tenantId, sessionId);
- public Task IncrementSessionEventCountAsync(
+ public Task IncrementSessionEventCountAsync(
string tenantId, string sessionId, int increment,
DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null,
EnrollmentPhase? currentPhase = null,
@@ -151,6 +151,9 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
public Task> GetSessionEventsAsync(string tenantId, string sessionId, int maxResults = 1000)
=> _storage.GetSessionEventsAsync(tenantId, sessionId, maxResults);
+ public Task> GetSessionEventsStrictAsync(string tenantId, string sessionId, int maxResults = 1000)
+ => _storage.GetSessionEventsStrictAsync(tenantId, sessionId, maxResults);
+
public Task> GetSessionEventsByTypeAsync(string tenantId, string sessionId, string eventType, int maxResults = 200)
=> _storage.GetSessionEventsByTypeAsync(tenantId, sessionId, eventType, maxResults);
diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs
index 82f9989b..0451a44a 100644
--- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs
@@ -200,9 +200,10 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
// UpdateSessionImeAgentVersionAsync upsert, …) would land rows past the lock and
// leave orphan data the manifest cannot describe. One read per batch; absent
// Sessions row → silent pass (caller handles session-not-found in its own write).
+ Azure.Data.Tables.TableEntity? guardSessionRow;
try
{
- await _deletionGuard.EnsureWritableAsync(bodyTenantId, sessionId, "V2.IngestTelemetry");
+ guardSessionRow = await _deletionGuard.EnsureWritableAndGetRowAsync(bodyTenantId, sessionId, "V2.IngestTelemetry");
}
catch (SessionDeletionLockedException locked)
{
@@ -212,10 +213,14 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
return AsOutput(await WriteSessionLockedAsync(req, locked));
}
+ // Reuse the guard's full-row read: its Status feeds the stall-heal check inside
+ // EventIngestProcessor, saving one Sessions point-read per batch.
+ var preFetchedStatus = TryReadSessionStatus(guardSessionRow);
+
// Partition + persist. Events are routed through EventIngestProcessor which runs the
// full pipeline (rule engine / app-install aggregation / SignalR / webhooks / ...);
// Signal + Transition go straight to their repositories.
- var outcome = await PersistItemsAsync(items, bodyTenantId, sessionId, validation);
+ var outcome = await PersistItemsAsync(items, bodyTenantId, sessionId, validation, preFetchedStatus);
_logger.LogInformation(
"IngestTelemetry: tenant={Tenant} session={Session} events={E} signals={S} transitions={T} unknown={U}",
@@ -336,6 +341,26 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
return response;
}
+ ///
+ /// Extracts Status from a Sessions row the deletion guard already loaded. Sessions
+ /// writes Status as a STRING (status.ToString() in UpdateSessionStatusAsync — never
+ /// an int), so this mirrors the canonical mapper's parse: Enum.TryParse,
+ /// case-insensitive. Returns null for missing/unparseable values (incl. a defensive int
+ /// fallback for any legacy numeric shape) — callers then fall back to their own read.
+ ///
+ internal static SessionStatus? TryReadSessionStatus(Azure.Data.Tables.TableEntity? sessionRow)
+ {
+ if (sessionRow == null || !sessionRow.TryGetValue("Status", out var statusValue))
+ return null;
+
+ return statusValue switch
+ {
+ string s when Enum.TryParse(s, ignoreCase: true, out var parsed) => parsed,
+ int i when Enum.IsDefined(typeof(SessionStatus), i) => (SessionStatus)i,
+ _ => null,
+ };
+ }
+
///
/// Partitions the incoming batch by and persists each
/// kind through its destination path. Events go through
@@ -347,7 +372,8 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
IReadOnlyList items,
string tenantId,
string sessionId,
- SecurityValidationResult validation)
+ SecurityValidationResult validation,
+ SessionStatus? preFetchedStatus)
{
var events = new List();
var signals = new List();
@@ -407,7 +433,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
TenantId = tenantId,
Events = events,
};
- var processed = await _eventProcessor.ProcessEventsAsync(eventRequest, validation);
+ var processed = await _eventProcessor.ProcessEventsAsync(eventRequest, validation, preFetchedStatus);
eventCount = processed.EventsProcessed;
adminAction = processed.AdminAction;
diff --git a/src/Backend/AutopilotMonitor.Functions/Program.cs b/src/Backend/AutopilotMonitor.Functions/Program.cs
index 9876476d..cafbc722 100644
--- a/src/Backend/AutopilotMonitor.Functions/Program.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Program.cs
@@ -8,6 +8,7 @@ using AutopilotMonitor.Functions.Functions.Sessions;
using AutopilotMonitor.Functions.Helpers;
using AutopilotMonitor.Functions.Middleware;
using AutopilotMonitor.Functions.Services;
+using Microsoft.ApplicationInsights.Extensibility;
using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
@@ -52,7 +53,15 @@ builder.Services.Configure(options =>
});
builder.Services
- .AddApplicationInsightsTelemetryWorkerService()
+ .AddApplicationInsightsTelemetryWorkerService(options =>
+ {
+ // The DEFAULT worker-side adaptive sampler has no excluded types: under load it thinned
+ // the enriched request copies from RequestTelemetryMiddleware (TenantId/CorrelationId —
+ // the per-call audit record) to ~15%, while the unenriched HOST request items stayed
+ // unsampled (host.json excludedTypes only governs the host pipeline). Protection was
+ // inverted. Disabled here and re-added below with Request + Event excluded.
+ options.EnableAdaptiveSampling = false;
+ })
.ConfigureFunctionsApplicationInsights();
// Drop successful Azure Storage dependencies (Table/Queue/Blob) from the worker telemetry
@@ -61,6 +70,19 @@ builder.Services
// (HTTP/Graph/SQL/SignalR) are preserved. See StorageDependencyFilterProcessor for the contract.
builder.Services.AddApplicationInsightsTelemetryProcessor();
+// Re-introduce worker-side adaptive sampling with Request + Event excluded: the enriched
+// request items stay complete (with host.json "Host.Results": "Error" they are the ONLY
+// success-request record), and customEvents are low-volume ops signals worth keeping whole.
+// Traces/dependencies/exceptions remain adaptively sampled. The sampler is appended to the
+// processor chain AFTER the DI-registered processors above, so the dropped storage chatter
+// no longer pressures the sampling rate of everything else.
+builder.Services.Configure(config =>
+{
+ var chain = config.DefaultTelemetrySink.TelemetryProcessorChainBuilder;
+ chain.UseAdaptiveSampling(maxTelemetryItemsPerSecond: 5, excludedTypes: "Request;Event");
+ chain.Build();
+});
+
// Configure JWT Authentication for Multi-Tenant Azure AD
builder.Services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
.AddMicrosoftIdentityWebApi(options =>
diff --git a/src/Backend/AutopilotMonitor.Functions/Services/BlockedDeviceService.cs b/src/Backend/AutopilotMonitor.Functions/Services/BlockedDeviceService.cs
index fa198652..d4eb66a8 100644
--- a/src/Backend/AutopilotMonitor.Functions/Services/BlockedDeviceService.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Services/BlockedDeviceService.cs
@@ -18,11 +18,14 @@ namespace AutopilotMonitor.Functions.Services
private readonly IDeviceSecurityRepository _securityRepo;
private readonly ILogger _logger;
- // Positive cache entries are re-validated against storage after this window so that
- // cross-instance mutations (manual unblock, Action upgrade Block->Kill, UnblockAt change)
- // propagate. Negative answers from a cache miss go through the storage point-read
- // on every call (see IsBlockedAsync), so no TTL is needed there.
- private static readonly TimeSpan EntryRevalidateAfter = TimeSpan.FromSeconds(30);
+ // Cache entries (positive AND negative) are re-validated against storage after this
+ // window so that cross-instance mutations (new block from another instance, manual
+ // unblock, Action upgrade Block->Kill, UnblockAt change) propagate within seconds.
+ // Negative entries matter for cost: without them every ingest request from a healthy
+ // (never-blocked) device paid one BlockedDevices point-read on the response path.
+ private static readonly TimeSpan DefaultEntryRevalidateAfter = TimeSpan.FromSeconds(30);
+
+ private readonly TimeSpan _entryRevalidateAfter;
// Cache key: "tenantId|serialNumber" (upper-cased serial number for case-insensitive matching)
// Cache value: BlockCacheEntry with UnblockAt, Action, BlockedSessionIds, LastCheckedUtc.
@@ -34,9 +37,17 @@ namespace AutopilotMonitor.Functions.Services
private readonly ConcurrentDictionary _loadedTenants = new(StringComparer.OrdinalIgnoreCase);
public BlockedDeviceService(IDeviceSecurityRepository securityRepo, ILogger logger)
+ : this(securityRepo, logger, DefaultEntryRevalidateAfter)
+ {
+ }
+
+ /// Test seam: lets tests shrink the revalidation window without a clock abstraction.
+ internal BlockedDeviceService(
+ IDeviceSecurityRepository securityRepo, ILogger logger, TimeSpan entryRevalidateAfter)
{
_securityRepo = securityRepo;
_logger = logger;
+ _entryRevalidateAfter = entryRevalidateAfter;
}
///
@@ -52,14 +63,13 @@ namespace AutopilotMonitor.Functions.Services
/// update the local instance's cache. Two safety nets bridge other instances to storage:
///
/// - Cache miss after the tenant was loaded → storage point-read on the spot,
- /// then promote into the cache. Lets a newly-set Block/Kill on another instance reach
- /// this instance immediately instead of waiting for the lazy load that never re-runs.
- /// - Positive cache hit older than → storage
- /// point-read to re-confirm. Lets manual Unblock, Action upgrade (Block→Kill), or
- /// UnblockAt changes from another instance propagate within seconds.
+ /// then promote the answer (positive OR negative) into the cache. The negative entry
+ /// is the hot-path cost fix: healthy devices answer from memory instead of paying a
+ /// point-read per ingest request.
+ /// - Cache hit older than the revalidation window → storage point-read to
+ /// re-confirm. Bounds cross-instance propagation (new Block/Kill, manual Unblock,
+ /// Block→Kill upgrade, UnblockAt change) to the window for both directions.
///
- /// Each safety net is one indexed point-read on the BlockedDevices table — negligible
- /// next to ingest cost.
///
///
public async Task<(bool isBlocked, DateTime? unblockAt, string action, string? blockedSessionIds)> IsBlockedAsync(
@@ -79,9 +89,9 @@ namespace AutopilotMonitor.Functions.Services
if (_cache.TryGetValue(cacheKey, out entry))
{
- // Stale positive entry → revalidate against storage so cross-instance mutations
- // (manual unblock, Block→Kill upgrade, UnblockAt change) propagate.
- if (DateTime.UtcNow - entry.LastCheckedUtc > EntryRevalidateAfter)
+ // Stale entry (positive or negative) → revalidate against storage so cross-instance
+ // mutations (new block, manual unblock, Block→Kill upgrade, UnblockAt change) propagate.
+ if (DateTime.UtcNow - entry.LastCheckedUtc > _entryRevalidateAfter)
{
entry = await RefreshCacheEntryFromStorageAsync(tenantId, serialNumber, cacheKey);
if (entry == null)
@@ -93,11 +103,18 @@ namespace AutopilotMonitor.Functions.Services
// Cache miss after tenant was loaded — another instance may have added a block
// since our LoadTenantBlockListAsync ran. Fall through to storage for one point-read
// so we don't blindly return "not blocked" for the lifetime of this instance.
+ // The refresh caches the negative answer, so the next requests within the
+ // revalidation window skip storage entirely.
entry = await RefreshCacheEntryFromStorageAsync(tenantId, serialNumber, cacheKey);
if (entry == null)
return (false, null, "Block", null);
}
+ // Fresh negative entry — device known not-blocked, no storage round-trip needed.
+ // Must run before the UnblockAt check below (negative entries carry no UnblockAt).
+ if (!entry.IsBlocked)
+ return (false, null, "Block", null);
+
if (DateTime.UtcNow >= entry.UnblockAt)
{
// Block has expired — remove from cache
@@ -147,7 +164,14 @@ namespace AutopilotMonitor.Functions.Services
if (!isBlocked || unblockAt == null)
{
- _cache.TryRemove(cacheKey, out _);
+ // Cache the negative answer so healthy devices don't pay a point-read per request.
+ // Revalidated after the same window as positive entries, so a block set on another
+ // instance still propagates within seconds.
+ _cache[cacheKey] = new BlockCacheEntry
+ {
+ IsBlocked = false,
+ LastCheckedUtc = DateTime.UtcNow,
+ };
return null;
}
@@ -185,14 +209,22 @@ namespace AutopilotMonitor.Functions.Services
},
(_, existing) =>
{
+ // A cached negative entry carries no real block scope — its null
+ // BlockedSessionIds must NOT be read as "whole-device block" below, or a
+ // session-aware auto-block would silently widen to the whole device (and
+ // skip the new-session auto-unblock) until the next revalidation.
+ var wasNegative = !existing.IsBlocked;
+ existing.IsBlocked = true;
existing.UnblockAt = unblockAt;
existing.Action = action ?? "Block";
- // Merge session IDs; whole-device block (null) takes precedence
- if (blockedSessionId != null && existing.BlockedSessionIds != null)
+ // Merge session IDs; a positive whole-device block (null) takes precedence
+ if (wasNegative)
+ existing.BlockedSessionIds = blockedSessionId; // take the new block's scope verbatim
+ else if (blockedSessionId != null && existing.BlockedSessionIds != null)
existing.BlockedSessionIds = TableDeviceSecurityRepository.MergeSessionId(existing.BlockedSessionIds, blockedSessionId);
else if (blockedSessionId == null)
existing.BlockedSessionIds = null; // Manual/whole-device block overrides session-aware
- // else: existing is null (whole-device) — keep it null
+ // else: existing is a positive whole-device block — keep it null
existing.LastCheckedUtc = DateTime.UtcNow;
return existing;
});
@@ -270,6 +302,11 @@ namespace AutopilotMonitor.Functions.Services
private class BlockCacheEntry
{
+ ///
+ /// False = cached negative answer ("device not blocked", confirmed against storage at
+ /// ). Negative entries carry no meaningful UnblockAt/Action.
+ ///
+ public bool IsBlocked { get; set; } = true;
public DateTime UnblockAt { get; set; }
public string Action { get; set; } = "Block";
public string? BlockedSessionIds { get; set; }
diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionGuard.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionGuard.cs
index b7b9c5f9..91ee0776 100644
--- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionGuard.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionGuard.cs
@@ -68,19 +68,28 @@ namespace AutopilotMonitor.Functions.Services.Deletion
/// with as the current-state
/// label. Genuine 404 (no marker) → silent pass for the fresh-enrollment path.
///
- public async Task EnsureWritableAsync(string tenantId, string sessionId, string callerContext, CancellationToken cancellationToken = default)
+ public Task EnsureWritableAsync(string tenantId, string sessionId, string callerContext, CancellationToken cancellationToken = default)
+ => EnsureWritableAndGetRowAsync(tenantId, sessionId, callerContext, cancellationToken);
+
+ ///
+ /// Same contract as , but returns the full Sessions row
+ /// the guard had to load anyway (null when the row does not exist and no tombstone marker
+ /// is active). Lets hot-path callers reuse the read — e.g. telemetry ingest feeds the
+ /// row's Status into the stall-heal check instead of issuing a second point-read.
+ ///
+ public async Task EnsureWritableAndGetRowAsync(string tenantId, string sessionId, string callerContext, CancellationToken cancellationToken = default)
{
var sessionRow = await _reader.GetSessionRowAsync(tenantId, sessionId, cancellationToken);
if (sessionRow != null)
{
ThrowIfLocked(sessionRow, callerContext);
- return;
+ return sessionRow;
}
// Sessions row absent → check the tombstone marker. Marker present → still locked.
// Marker absent → safe to proceed (fresh registration or post-retention slot reuse).
var tombstone = await _reader.GetActiveSessionTombstoneAsync(tenantId, sessionId, cancellationToken);
- if (tombstone == null) return;
+ if (tombstone == null) return null;
var manifestId = tombstone.GetString(AutopilotMonitor.Shared.Models.Deletion.SessionTombstoneRecord.Columns.ManifestId);
_logger.LogInformation(
diff --git a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Classification.cs b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Classification.cs
index 63b74f52..270969fa 100644
--- a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Classification.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Classification.cs
@@ -105,7 +105,8 @@ namespace AutopilotMonitor.Functions.Services
};
private async Task<(bool statusTransitioned, bool whiteGloveStatusTransitioned, string? failureReason)>
- UpdateSessionStatusAsync(IngestEventsRequest request, string sessionPrefix, EventClassification c)
+ UpdateSessionStatusAsync(IngestEventsRequest request, string sessionPrefix, EventClassification c,
+ SessionStatus? preFetchedStatus = null)
{
bool statusTransitioned = false;
bool whiteGloveStatusTransitioned = false;
@@ -207,8 +208,18 @@ namespace AutopilotMonitor.Functions.Services
}
else if (c.HasNonPeriodicRealEvent && !statusTransitioned && !whiteGloveStatusTransitioned)
{
- var currentSession = await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
- if (currentSession?.Status == SessionStatus.Stalled)
+ // Stall-heal probe. The pre-fetched status (deletion-guard read, a few ms old) may
+ // ONLY substitute the point-read when this batch performed no status write at all —
+ // the GatherCompletion and WhiteGlove(Resumed) branches above write a new status
+ // without setting the transitioned flags, and a pre-write "Stalled" would then
+ // falsely heal the session back to InProgress over the just-written status.
+ var batchWroteStatus = c.CompletionEvent != null || c.FailureEvent != null
+ || c.EspFailureEvent != null || c.GatherCompletionEvent != null
+ || c.WhiteGloveEvent != null || c.WhiteGloveResumedEvent != null;
+ var currentStatus = !batchWroteStatus && preFetchedStatus.HasValue
+ ? preFetchedStatus
+ : (await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId))?.Status;
+ if (currentStatus == SessionStatus.Stalled)
{
var healed = await _sessionRepo.UpdateSessionStatusAsync(
request.TenantId, request.SessionId, SessionStatus.InProgress,
diff --git a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs
index 8bc375e9..fd0e2332 100644
--- a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs
@@ -96,9 +96,15 @@ namespace AutopilotMonitor.Functions.Services
/// parse and tenant-mismatch check are the caller's responsibility — the V2 function
/// does them before it even knows the item is an Event).
///
+ /// Parsed event batch (single session).
+ /// Security validation result of the carrying HTTP request.
+ /// Session status from a read the caller already performed
+ /// just before this call (V2 passes the deletion-guard row's status). Used only as a
+ /// point-read saver on paths that tolerate a few-ms-old snapshot; null → read on demand.
public async Task ProcessEventsAsync(
IngestEventsRequest request,
- SecurityValidationResult validation)
+ SecurityValidationResult validation,
+ SessionStatus? preFetchedStatus = null)
{
var sessionPrefix = $"[Session: {request.SessionId.Substring(0, Math.Min(8, request.SessionId.Length))}]";
_logger.LogInformation(
@@ -178,7 +184,7 @@ namespace AutopilotMonitor.Functions.Services
}
var (statusTransitioned, whiteGloveStatusTransitioned, failureReason) =
- await UpdateSessionStatusAsync(request, sessionPrefix, classification);
+ await UpdateSessionStatusAsync(request, sessionPrefix, classification, preFetchedStatus);
// A terminal batch (one that drives Succeeded/Failed) takes its RebootCount from the
// authoritative reconcile below, NOT the per-batch increment — otherwise the reboot
@@ -189,9 +195,14 @@ namespace AutopilotMonitor.Functions.Services
|| classification.EspFailureEvent != null
|| classification.GatherCompletionEvent != null;
+ // The increment's post-merge snapshot serves as the "updatedSession" for the common
+ // case (non-terminal batch, no diagnostics upload) — it already reflects this batch's
+ // status transition (written above) plus the counter merge, saving the follow-up
+ // GetSessionAsync that used to run on every batch.
+ SessionSummary? updatedSession = null;
if (processedCount > 0)
{
- await _sessionRepo.IncrementSessionEventCountAsync(
+ updatedSession = await _sessionRepo.IncrementSessionEventCountAsync(
request.TenantId,
request.SessionId,
processedCount,
@@ -312,7 +323,12 @@ namespace AutopilotMonitor.Functions.Services
}
}
- var updatedSession = await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
+ // Re-read only when a write AFTER the increment made the snapshot stale: terminal
+ // batches (ReconcileSessionRebootCountAsync) and diagnostics uploads (blob fields) —
+ // or when no increment ran / it returned null (missing row, exhausted ETag retries).
+ if (isTerminalBatch || classification.DiagnosticsUploadedEvent != null)
+ updatedSession = null;
+ updatedSession ??= await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
if (updatedSession != null && updatedSession.Status == SessionStatus.InProgress)
{
diff --git a/src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs b/src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs
index 2aa3cf97..f027d106 100644
--- a/src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs
@@ -51,14 +51,13 @@ namespace AutopilotMonitor.Functions.Services
var outcome = new AnalysisOutcome();
var activeRules = await _ruleService.GetActiveRulesForTenantAsync(tenantId);
- var allEvents = await _sessionRepo.GetSessionEventsAsync(tenantId, sessionId);
+ // Strict read: storage failures propagate (→ queue retry / poison) instead of
+ // degrading to an empty list — an empty result here therefore really means a
+ // session without events, never a swallowed transient fault.
+ var allEvents = await _sessionRepo.GetSessionEventsStrictAsync(tenantId, sessionId);
if (allEvents.Count == 0)
{
- // NOTE: TableStorageService.GetSessionEventsAsync swallows Storage exceptions
- // and returns an empty list. A transient read failure is therefore indistinguishable
- // from a genuinely empty session here. Tracked as a follow-up: storage helpers
- // should fail loud so the queue worker retries instead of deleting the message.
_logger.LogInformation($"No events found for session {sessionId}, skipping analysis");
return outcome;
}
diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs
index 871b3b88..a1619306 100644
--- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs
@@ -1675,8 +1675,10 @@ namespace AutopilotMonitor.Functions.Services
/// Uses Merge mode to safely handle concurrent updates.
/// The caller provides earliestEventTimestamp from the current batch;
/// no redundant Events-table scan is performed here.
+ /// Returns the post-merge snapshot (RMW read + applied increments) so the ingest hot path
+ /// can skip its follow-up GetSessionAsync; null on missing row / exhausted retries / error.
///
- public async Task IncrementSessionEventCountAsync(string tenantId, string sessionId, int increment, DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null, EnrollmentPhase? currentPhase = null, int platformScriptIncrement = 0, int remediationScriptIncrement = 0, int rebootIncrement = 0)
+ public async Task IncrementSessionEventCountAsync(string tenantId, string sessionId, int increment, DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null, EnrollmentPhase? currentPhase = null, int platformScriptIncrement = 0, int remediationScriptIncrement = 0, int rebootIncrement = 0)
{
SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
@@ -1693,7 +1695,7 @@ namespace AutopilotMonitor.Functions.Services
if (!entityResponse.HasValue)
{
_logger.LogWarning("Session {SessionId} not found when incrementing event count — session may have been cleaned up or not yet registered", sessionId);
- return;
+ return null;
}
var entity = entityResponse.Value!;
var currentCount = entity.GetInt32("EventCount") ?? 0;
@@ -1752,7 +1754,15 @@ namespace AutopilotMonitor.Functions.Services
// Dual-write: keep SessionsIndex in sync (StartedAt-shift → full upsert, else merge).
await SyncSessionIndexAsync(tenantId, sessionId, entity, update, currentStartedAt, earliestEventTimestamp);
- return;
+ // Apply the merged fields onto the RMW read and map it through the shared
+ // mapper — the post-merge snapshot the caller would otherwise re-read.
+ foreach (var kvp in update)
+ {
+ if (kvp.Key is "PartitionKey" or "RowKey" or "Timestamp" or "odata.etag")
+ continue;
+ entity[kvp.Key] = kvp.Value;
+ }
+ return MapToSessionSummary(entity);
}
catch (Azure.RequestFailedException ex) when (ex.Status == 412)
{
@@ -1760,7 +1770,7 @@ namespace AutopilotMonitor.Functions.Services
if (retryCount >= maxRetries)
{
_logger.LogWarning($"Failed to increment event count for session {sessionId} after {maxRetries} retries due to ETag conflicts");
- return;
+ return null;
}
// Exponential backoff with jitter to decorrelate concurrent retries
var baseDelay = 50 * (int)Math.Pow(2, retryCount - 1);
@@ -1769,9 +1779,11 @@ namespace AutopilotMonitor.Functions.Services
catch (Exception ex)
{
_logger.LogError(ex, $"Failed to increment event count for session {sessionId}");
- return;
+ return null;
}
}
+
+ return null;
}
///
@@ -1846,34 +1858,22 @@ namespace AutopilotMonitor.Functions.Services
}
///
- /// Gets all events for a specific session
+ /// Gets all events for a specific session. Fail-soft: storage exceptions degrade to an
+ /// empty list (read-only UI/metrics surfaces prefer a blank timeline over a 500) — which
+ /// makes a read FAILURE indistinguishable from a genuinely empty session. Callers whose
+ /// retry semantics depend on observing failures (queue workers) must use
+ /// instead.
///
public async Task> GetSessionEventsAsync(string tenantId, string sessionId, int maxResults = 1000)
{
+ // Validate before the catch — malformed GUIDs are caller bugs and must not be
+ // masked as "no events" (same behaviour as before the strict/soft split).
SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
try
{
- var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Events);
- var events = new List();
-
- // Events are stored with PartitionKey = "{TenantId}_{SessionId}"
- var partitionKey = $"{tenantId}_{sessionId}";
-
- var query = tableClient.QueryAsync(
- filter: $"PartitionKey eq '{partitionKey}'",
- maxPerPage: maxResults
- );
-
- await foreach (var entity in query)
- {
- events.Add(MapToEnrollmentEvent(entity));
- }
-
- // Sort by Sequence ascending — the authoritative event order
- // (assigned atomically via Interlocked.Increment on the agent)
- return events.OrderBy(e => e.Sequence).ToList();
+ return await GetSessionEventsStrictAsync(tenantId, sessionId, maxResults);
}
catch (Exception ex)
{
@@ -1882,6 +1882,39 @@ namespace AutopilotMonitor.Functions.Services
}
}
+ ///
+ /// Strict variant of : storage exceptions PROPAGATE.
+ /// Used on the queue-worker paths (rule analysis, vulnerability correlation) where a
+ /// swallowed read failure would convert a retryable transient fault into silent data
+ /// loss — the worker would delete its message after "analyzing" a truncated stream.
+ /// A genuinely empty session still returns an empty list (successful query, zero rows).
+ ///
+ public async Task> GetSessionEventsStrictAsync(string tenantId, string sessionId, int maxResults = 1000)
+ {
+ SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+ SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
+
+ var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Events);
+ var events = new List();
+
+ // Events are stored with PartitionKey = "{TenantId}_{SessionId}"
+ var partitionKey = $"{tenantId}_{sessionId}";
+
+ var query = tableClient.QueryAsync(
+ filter: $"PartitionKey eq '{partitionKey}'",
+ maxPerPage: maxResults
+ );
+
+ await foreach (var entity in query)
+ {
+ events.Add(MapToEnrollmentEvent(entity));
+ }
+
+ // Sort by Sequence ascending — the authoritative event order
+ // (assigned atomically via Interlocked.Increment on the agent)
+ return events.OrderBy(e => e.Sequence).ToList();
+ }
+
///
/// Reads a single page of session events. Honours the supplied
/// and resumes from when supplied. The returned page's
diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityInventoryLoader.cs b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityInventoryLoader.cs
index 53c53dde..8a6e5214 100644
--- a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityInventoryLoader.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityInventoryLoader.cs
@@ -11,7 +11,9 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability
{
///
/// Default backed by
- /// . Extracted from
+ /// (strict: a storage read
+ /// failure propagates to the caller — the queue worker retries instead of dropping the
+ /// vulnerability report after "loading" an empty inventory). Extracted from
/// GetVulnerabilityReportFunction.LoadInventoryItemsFromEventsAsync so the queue
/// worker and the manual rescan endpoint share one loader.
///
@@ -37,7 +39,7 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability
{
var items = new List>();
- var allEvents = await _sessionRepo.GetSessionEventsAsync(tenantId, sessionId).ConfigureAwait(false);
+ var allEvents = await _sessionRepo.GetSessionEventsStrictAsync(tenantId, sessionId).ConfigureAwait(false);
var inventoryEvents = allEvents
.Where(e => e.EventType == Constants.EventTypes.SoftwareInventoryAnalysis);
diff --git a/src/Backend/AutopilotMonitor.Functions/host.json b/src/Backend/AutopilotMonitor.Functions/host.json
index b73e502e..498b0692 100644
--- a/src/Backend/AutopilotMonitor.Functions/host.json
+++ b/src/Backend/AutopilotMonitor.Functions/host.json
@@ -8,7 +8,7 @@
"AutopilotMonitor.Functions.Security": "Information",
"Host": "Error",
"Host.Aggregator": "Error",
- "Host.Results": "Information",
+ "Host.Results": "Error",
"Host.Startup": "Error",
"Microsoft.Azure.WebJobs": "Error",
"Microsoft.Azure.WebJobs.Host": "Error",
diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs
index 995374ad..8177f255 100644
--- a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs
+++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs
@@ -73,7 +73,13 @@ namespace AutopilotMonitor.Shared.DataAccess
DateTime? stalledAt = null, bool clearStalledAt = false, bool clearFailureReason = false,
string? failureSource = null, string? adminMarkedAction = null,
string? failureSnapshotJson = null);
- Task IncrementSessionEventCountAsync(
+ ///
+ /// Increments per-session counters via read-modify-write. Returns the post-merge session
+ /// snapshot (the RMW read with the applied increments) so hot-path callers can skip a
+ /// follow-up ; null when the row is missing or the update
+ /// could not be applied (caller falls back to its own read).
+ ///
+ Task IncrementSessionEventCountAsync(
string tenantId, string sessionId, int increment,
DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null,
EnrollmentPhase? currentPhase = null,
@@ -132,6 +138,13 @@ namespace AutopilotMonitor.Shared.DataAccess
Task StoreEventAsync(EnrollmentEvent evt);
Task> StoreEventsBatchAsync(List events);
Task> GetSessionEventsAsync(string tenantId, string sessionId, int maxResults = 1000);
+ ///
+ /// Strict variant of : storage failures PROPAGATE
+ /// instead of degrading to an empty list. Queue-worker paths (rule analysis,
+ /// vulnerability correlation) require this so a transient read failure is retried via
+ /// visibility timeout instead of silently producing an empty analysis.
+ ///
+ Task> GetSessionEventsStrictAsync(string tenantId, string sessionId, int maxResults = 1000);
Task> GetSessionEventsByTypeAsync(string tenantId, string sessionId, string eventType, int maxResults = 200);
///