diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/CascadeVerificationServiceTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/CascadeVerificationServiceTests.cs index 68bfaf65..d6f1c152 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/CascadeVerificationServiceTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/CascadeVerificationServiceTests.cs @@ -372,6 +372,9 @@ public class CascadeVerificationServiceTests public Task GetSessionRowAsync(string tenantId, string sessionId, CancellationToken ct = default) => Task.FromResult(null); + public Task GetSessionRowAsync(string tenantId, string sessionId, IEnumerable select, CancellationToken ct = default) + => Task.FromResult(null); + public Task GetSessionsIndexRowAsync(string tenantId, string indexRowKey, CancellationToken ct = default) => Task.FromResult(null); diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IngestStatusPrefetchTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IngestStatusPrefetchTests.cs index cb85faa9..9e3ff55b 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/IngestStatusPrefetchTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/IngestStatusPrefetchTests.cs @@ -1,4 +1,4 @@ -using AutopilotMonitor.Functions.Functions.Ingest; +using AutopilotMonitor.Functions.Helpers; using AutopilotMonitor.Shared.Models; using Azure.Data.Tables; using Xunit; @@ -26,14 +26,14 @@ public class IngestStatusPrefetchTests [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))); + Assert.Equal(expected, SessionRowProjections.TryReadStatus(Row(stored))); } [Fact] public void IntStatus_LegacyDefensiveFallback_IsCast() { var defined = (int)SessionStatus.Stalled; - Assert.Equal(SessionStatus.Stalled, IngestTelemetryFunction.TryReadSessionStatus(Row(defined))); + Assert.Equal(SessionStatus.Stalled, SessionRowProjections.TryReadStatus(Row(defined))); } [Theory] @@ -41,13 +41,13 @@ public class IngestStatusPrefetchTests [InlineData("")] public void UnparseableStatus_ReturnsNull_SoCallerFallsBackToOwnRead(string stored) { - Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(Row(stored))); + Assert.Null(SessionRowProjections.TryReadStatus(Row(stored))); } [Fact] public void MissingStatusOrRow_ReturnsNull() { - Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(Row(status: null))); - Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(null)); + Assert.Null(SessionRowProjections.TryReadStatus(Row(status: null))); + Assert.Null(SessionRowProjections.TryReadStatus(null)); } } diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryAuthPathTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryAuthPathTests.cs index e3053300..8984714b 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryAuthPathTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryAuthPathTests.cs @@ -2,6 +2,7 @@ using System.Threading; using System.Threading.Tasks; using Azure.Data.Tables; using AutopilotMonitor.Functions.Functions.Ingest; +using AutopilotMonitor.Functions.Helpers; using AutopilotMonitor.Functions.Services.Deletion; using AutopilotMonitor.Shared.Models; using AutopilotMonitor.Shared.Models.Deletion; @@ -37,7 +38,7 @@ namespace AutopilotMonitor.Functions.Tests; /// OrdinalIgnoreCase equality Run performs (Run L169-180). /// session deletion locked → 410: the real throwing /// (Run L205-213), plus the unique -/// that consumes the guard's row. +/// that consumes the guard's row. /// kill-switch / device-blocked → 200-block: NOT duplicated here. The verdict Run() routes /// on (IsBlocked/IsKill/UnblockAt) is produced by /// and is already exhaustively covered at the @@ -138,7 +139,7 @@ public class IngestTelemetryAuthPathTests public async Task DeletionUnlocked_guard_returns_row_that_feeds_the_status_prefetch() { // Run L205-217: on an unlocked session the guard hands back the loaded row, and Run reads - // Status off it via TryReadSessionStatus to seed the stall-heal prefetch (no second read). + // Status off it via SessionRowProjections.TryReadStatus to seed the stall-heal prefetch (no second read). var guard = NewGuard(out var reader); var row = new TableEntity(TenantId, SessionId) { @@ -151,10 +152,10 @@ public class IngestTelemetryAuthPathTests var returned = await guard.EnsureWritableAndGetRowAsync(TenantId, SessionId, callerContext: "V2.IngestTelemetry"); Assert.Same(row, returned); - Assert.Equal(SessionStatus.Succeeded, IngestTelemetryFunction.TryReadSessionStatus(returned)); + Assert.Equal(SessionStatus.Succeeded, SessionRowProjections.TryReadStatus(returned)); } - // ============================================================ TryReadSessionStatus (unique helper) + // ============================================================ TryReadStatus (unique helper) [Theory] [InlineData("InProgress", SessionStatus.InProgress)] @@ -164,27 +165,27 @@ public class IngestTelemetryAuthPathTests public void TryReadSessionStatus_parses_string_status_case_insensitively(string stored, SessionStatus expected) { var row = new TableEntity(TenantId, SessionId) { ["Status"] = stored }; - Assert.Equal(expected, IngestTelemetryFunction.TryReadSessionStatus(row)); + Assert.Equal(expected, SessionRowProjections.TryReadStatus(row)); } [Fact] public void TryReadSessionStatus_returns_null_for_null_row() { - Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(null)); + Assert.Null(SessionRowProjections.TryReadStatus(null)); } [Fact] public void TryReadSessionStatus_returns_null_when_status_column_absent() { var row = new TableEntity(TenantId, SessionId); // no Status column - Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(row)); + Assert.Null(SessionRowProjections.TryReadStatus(row)); } [Fact] public void TryReadSessionStatus_returns_null_for_unparseable_status() { var row = new TableEntity(TenantId, SessionId) { ["Status"] = "NotARealStatus" }; - Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(row)); + Assert.Null(SessionRowProjections.TryReadStatus(row)); } // ============================================================ Harness (mirrors SessionDeletionGuardTests) diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SessionRowProjectionTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SessionRowProjectionTests.cs new file mode 100644 index 00000000..75301462 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/SessionRowProjectionTests.cs @@ -0,0 +1,535 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Azure; +using Azure.Data.Tables; +using AutopilotMonitor.Functions.Helpers; +using AutopilotMonitor.Functions.Security; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Functions.Services.Deletion; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using AutopilotMonitor.Shared.Models.Deletion; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// The agent hot path reads the Sessions row projected (). A +/// select-strict harness — it hands back only the requested columns, unlike the SDK mocks of the +/// other session tests — pins every list against its consumers: the counter merge's write, its +/// index mirror and its snapshot, the ingest-slice mapper, the signal-only read, and the guard +/// row's readers (status prefetch, kill-switch serial, owner policy) come out identical whether +/// the row arrives whole or projected. A column a consumer reads but the list lacks fails here, +/// not in production as a silent null. Fixtures carry every column a real row carries, so the +/// projection has something to drop. +/// +public class SessionRowProjectionTests +{ + private const string TenantId = "11111111-1111-1111-1111-111111111111"; + private const string SessionId = "22222222-2222-2222-2222-222222222222"; + + private static readonly TimeSpan Ahead = TimeSpan.FromHours(9); + private static readonly DateTime ServerStart = new(2026, 1, 1, 8, 0, 0, DateTimeKind.Utc); + private static readonly DateTime DeviceStart = ServerStart + Ahead; + + private const string Thumb1 = "AA11BB22CC33DD44EE55FF6677889900AABBCCDD"; + private const string Thumb2 = "0011223344556677889900AABBCCDDEEFF001122"; + private const string Dev1 = "0f4e1c2d-1111-4aaa-8bbb-000000000001"; + + public enum RowKind { RunningWhiteGlovePart2, Completed } + + public static IEnumerable RowKinds => new[] { new object[] { RowKind.RunningWhiteGlovePart2 }, new object[] { RowKind.Completed } }; + + // ============================================================== counter merge ==== + + [Theory] + [MemberData(nameof(RowKinds))] + public async Task Counter_merge_write_index_mirror_and_snapshot_are_identical_whole_or_projected(RowKind kind) + { + var whole = new Harness(FullRow(kind), honourSelect: false); + var projected = new Harness(FullRow(kind), honourSelect: true); + + var wholeSnapshot = await Increment(whole.Sut); + var projectedSnapshot = await Increment(projected.Sut); + + // The production read passes the list, and the strict harness really dropped columns. + Assert.NotNull(projected.LastSelect); + Assert.Equal(SessionRowProjections.IngestSnapshot, projected.LastSelect!); + Assert.True(projected.LastReadColumnCount < whole.LastReadColumnCount, + $"projected read returned {projected.LastReadColumnCount} columns, whole read {whole.LastReadColumnCount}"); + + Assert.NotNull(whole.Written); + Assert.NotNull(projected.Written); + Assert.Equal(whole.WrittenETag, projected.WrittenETag); // the CAS ETag survives the projection + AssertSameEntity(whole.Written!, projected.Written!); + Assert.NotNull(whole.IndexWritten); + Assert.NotNull(projected.IndexWritten); + AssertSameEntity(whole.IndexWritten!, projected.IndexWritten!); + + Assert.NotNull(wholeSnapshot); + Assert.NotNull(projectedSnapshot); + AssertSameSnapshot(wholeSnapshot!, projectedSnapshot!); + } + + [Fact] + public async Task Counter_merge_on_the_running_row_exercises_the_tracker_and_the_counters() + { + // Guards the fixture itself: every pinned column must have a visible effect on this write, + // otherwise the equivalence above would pass vacuously. + var harness = new Harness(FullRow(RowKind.RunningWhiteGlovePart2), honourSelect: true); + + var snapshot = await Increment(harness.Sut); + + var written = harness.Written!; + Assert.Equal(2 + 3, written.GetInt32("EventCount")); + Assert.Equal(3 + 1, written.GetInt32("PlatformScriptCount")); + Assert.Equal(1 + 1, written.GetInt32("RemediationScriptCount")); + Assert.Equal(1 + 1, written.GetInt32("RebootCount")); + Assert.False(written.ContainsKey("StartedAt"), "a later earliest event must not move the anchor"); + Assert.False(written.ContainsKey("LastEventAt"), "an older latest event must not move LastEventAt"); + Assert.True(written.ContainsKey(TableStorageService.ServerTimeStateColumn), "the tracker continues on a running row"); + Assert.False(written.ContainsKey("StartedAtServer"), "an unchanged twin is not rewritten"); + Assert.Equal("2516000000000000000_" + SessionId, harness.IndexWritten!.RowKey); // the stored IndexRowKey, not a recomputed one + Assert.Equal(SessionStatus.InProgress, snapshot!.Status); + Assert.Equal(5, snapshot.EventCount); + Assert.Equal("Succeeded", snapshot.AdminMarkedAction); + Assert.Equal("2.0.1478", snapshot.AgentVersion); + Assert.Equal("1.99.1", snapshot.ImeAgentVersion); + Assert.True(snapshot.DurationSeconds >= 600, "WhiteGlove Part 2 running: stored Part 1 plus the running Part 2"); + } + + [Fact] + public async Task Counter_merge_on_the_completed_row_leaves_the_tracker_alone() + { + var harness = new Harness(FullRow(RowKind.Completed), honourSelect: true); + + var snapshot = await Increment(harness.Sut); + + Assert.False(harness.Written!.ContainsKey(TableStorageService.ServerTimeStateColumn), "CompletedAt stops the tracker"); + Assert.Equal(SessionStatus.Succeeded, snapshot!.Status); + Assert.Equal(2400, snapshot.DurationSeconds); + Assert.Equal(new DateTime(2026, 1, 1, 17, 40, 0, DateTimeKind.Utc), snapshot.CompletedAt); + } + + // ===================================================================== mapper ==== + + [Theory] + [MemberData(nameof(RowKinds))] + public void Ingest_slice_mapper_matches_the_canonical_mapper_and_needs_only_its_projection(RowKind kind) + { + var sut = new Harness(FullRow(kind), honourSelect: true).Sut; + var row = FullRow(kind); + + var fromSlice = sut.MapToIngestSnapshot(row); + var fromSummary = SessionIngestSnapshot.From(sut.MapToSessionSummary(row)); + var fromProjected = sut.MapToIngestSnapshot(Project(row, SessionRowProjections.IngestSnapshot)); + + AssertSameSnapshot(fromSummary, fromSlice); + AssertSameSnapshot(fromSlice, fromProjected); + } + + [Theory] + [MemberData(nameof(RowKinds))] + public async Task Signal_only_snapshot_read_uses_the_same_projection_and_mapping(RowKind kind) + { + var harness = new Harness(FullRow(kind), honourSelect: true); + + var snapshot = await harness.Sut.GetSessionIngestSnapshotAsync(TenantId, SessionId); + + Assert.Equal(SessionRowProjections.IngestSnapshot, harness.LastSelect!); + Assert.NotNull(snapshot); + AssertSameSnapshot(harness.Sut.MapToIngestSnapshot(FullRow(kind)), snapshot!); + } + + // ================================================================== guard row ==== + + [Fact] + public void Guard_row_readers_see_the_same_off_the_projected_row() + { + var full = FullRow(RowKind.RunningWhiteGlovePart2); + var projected = Project(full, SessionRowProjections.GuardRow.Concat(SessionDeletionGuard.LockColumns)); + + Assert.True(projected.Count < full.Count, "the guard projection must drop columns"); + Assert.Equal(SessionRowProjections.TryReadStatus(full), SessionRowProjections.TryReadStatus(projected)); + Assert.Equal(SessionStatus.InProgress, SessionRowProjections.TryReadStatus(projected)); + Assert.Equal(full.GetString("SerialNumber"), projected.GetString("SerialNumber")); + Assert.Equal(full.GetString("DeletionState"), projected.GetString("DeletionState")); + } + + [Theory] + [MemberData(nameof(OwnerScenarios))] + public void Owner_policy_decides_the_same_off_the_projected_row(string scenario, SessionOwner? rowOwner, SecurityValidationResult caller) + { + var full = FullRow(RowKind.RunningWhiteGlovePart2, rowOwner); + var projected = Project(full, SessionRowProjections.GuardRow.Concat(SessionDeletionGuard.LockColumns)); + var now = ServerStart.AddHours(1); + + var fromFull = SessionOwnershipPolicy.Evaluate(full, caller, now); + var fromProjected = SessionOwnershipPolicy.Evaluate(projected, caller, now); + + Assert.True(fromFull.Outcome == fromProjected.Outcome, $"{scenario}: {fromFull.Outcome} vs {fromProjected.Outcome}"); + Assert.Equal(fromFull.Rejected, fromProjected.Rejected); + Assert.Equal(fromFull.OwnerToStamp?.Kind, fromProjected.OwnerToStamp?.Kind); + Assert.Equal(fromFull.OwnerToStamp?.Thumbprint, fromProjected.OwnerToStamp?.Thumbprint); + Assert.Equal(fromFull.OwnerToStamp?.DeviceId, fromProjected.OwnerToStamp?.DeviceId); + Assert.Equal(fromFull.OwnerToStamp?.BootstrapCode, fromProjected.OwnerToStamp?.BootstrapCode); + Assert.Equal(fromFull.OwnerToStamp?.Serial, fromProjected.OwnerToStamp?.Serial); + } + + public static IEnumerable OwnerScenarios() + { + yield return new object?[] { "legacy row, cert caller with the row's serial", null, Cert(Thumb1, Dev1) }; + yield return new object?[] { "cert-owned row, the same cert", CertOwner(Thumb1, Dev1), Cert(Thumb1, Dev1) }; + yield return new object?[] { "cert-owned row, a foreign cert", CertOwner(Thumb1, Dev1), Cert(Thumb2, "0f4e1c2d-2222-4aaa-8bbb-000000000002", "SN-OTHER") }; + yield return new object?[] { "bootstrap-owned row, the cert that follows", BootstrapOwner("ABCD1234"), Cert(Thumb1, Dev1) }; + yield return new object?[] { "cert-owned row, a bootstrap caller", CertOwner(Thumb1, Dev1), Bootstrap("ABCD1234") }; + } + + [Fact] + public void Lock_check_works_on_the_projected_row() + { + var full = FullRow(RowKind.RunningWhiteGlovePart2); + full["DeletionState"] = SessionDeletionState.Preparing; + full["PendingDeletionManifestId"] = "MANIFEST-1"; + var projected = Project(full, SessionRowProjections.GuardRow.Concat(SessionDeletionGuard.LockColumns)); + var guard = new SessionDeletionGuard(new Mock().Object, NullLogger.Instance); + + var ex = Assert.Throws(() => guard.ThrowIfLocked(projected, "V2.IngestTelemetry")); + + Assert.Equal(SessionDeletionState.Preparing, ex.CurrentState); + Assert.Equal("MANIFEST-1", ex.ManifestId); + } + + [Fact] + public async Task Projected_guard_read_asks_for_the_caller_columns_plus_the_lock_columns() + { + var reader = new Mock(); + IEnumerable? asked = null; + reader.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny>(), It.IsAny())) + .Returns, CancellationToken>((_, _, select, _) => + { + asked = select.ToList(); + return Task.FromResult(Project(FullRow(RowKind.RunningWhiteGlovePart2), select)); + }); + var guard = new SessionDeletionGuard(reader.Object, NullLogger.Instance); + + var row = await guard.EnsureWritableAndGetRowAsync(TenantId, SessionId, "V2.IngestTelemetry", SessionRowProjections.GuardRow); + + Assert.NotNull(asked); + Assert.Equal( + SessionRowProjections.GuardRow.Concat(SessionDeletionGuard.LockColumns).OrderBy(c => c, StringComparer.Ordinal), + asked!.OrderBy(c => c, StringComparer.Ordinal)); + Assert.NotNull(row); + Assert.Equal(SessionStatus.InProgress, SessionRowProjections.TryReadStatus(row)); + Assert.Equal("SN-1", row!.GetString("SerialNumber")); + Assert.False(row.ContainsKey("DeviceName"), "the projected row carries only the requested columns"); + reader.Verify(r => r.GetSessionRowAsync(It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task Projected_guard_read_still_refuses_a_locked_row_and_consults_the_tombstone_when_absent() + { + var locked = new Mock(); + locked.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny>(), It.IsAny())) + .ReturnsAsync(new TableEntity(TenantId, SessionId) + { + ["DeletionState"] = SessionDeletionState.Queued, + ["PendingDeletionManifestId"] = "MANIFEST-2", + }); + var lockedGuard = new SessionDeletionGuard(locked.Object, NullLogger.Instance); + var ex = await Assert.ThrowsAsync( + () => lockedGuard.EnsureWritableAndGetRowAsync(TenantId, SessionId, "RegisterSession", SessionRowProjections.GuardRow)); + Assert.Equal("MANIFEST-2", ex.ManifestId); + + var absent = new Mock(); + absent.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny>(), It.IsAny())) + .ReturnsAsync((TableEntity?)null); + absent.Setup(r => r.GetActiveSessionTombstoneAsync(TenantId, SessionId, It.IsAny())) + .ReturnsAsync((TableEntity?)null); + var absentGuard = new SessionDeletionGuard(absent.Object, NullLogger.Instance); + var row = await absentGuard.EnsureWritableAndGetRowAsync(TenantId, SessionId, "RegisterSession", SessionRowProjections.GuardRow); + Assert.Null(row); + absent.Verify(r => r.GetActiveSessionTombstoneAsync(TenantId, SessionId, It.IsAny()), Times.Once); + } + + [Fact] + public void Projection_lists_carry_what_their_consumers_name() + { + Assert.Contains("Status", SessionRowProjections.GuardRow); + Assert.Contains("SerialNumber", SessionRowProjections.GuardRow); + Assert.All(SessionOwner.Columns.All, c => Assert.Contains(c, SessionRowProjections.GuardRow)); + Assert.Equal(SessionRowProjections.GuardRow.Length, SessionRowProjections.GuardRow.Distinct(StringComparer.Ordinal).Count()); + + Assert.Contains(TableStorageService.ServerTimeStateColumn, SessionRowProjections.IngestSnapshot); + Assert.Contains(TableStorageService.StartedAtServerColumn, SessionRowProjections.IngestSnapshot); + Assert.Contains(TableStorageService.ResumedAtServerColumn, SessionRowProjections.IngestSnapshot); + Assert.Contains("CompletedAt", SessionRowProjections.IngestSnapshot); + Assert.Contains("IndexRowKey", SessionRowProjections.IngestSnapshot); + Assert.Equal(SessionRowProjections.IngestSnapshot.Length, SessionRowProjections.IngestSnapshot.Distinct(StringComparer.Ordinal).Count()); + } + + // ==================================================================== helpers ==== + + /// The counter merge of a second upload: later events, an older latest event, one of each counter. + private static Task Increment(TableStorageService sut) => + sut.IncrementSessionEventCountAsync( + TenantId, SessionId, increment: 3, + earliestEventTimestamp: DeviceStart.AddMinutes(1), // later than StartedAt: no anchor shift + latestEventTimestamp: DeviceStart.AddMinutes(19), // older than LastEventAt: no move + currentPhase: null, + platformScriptIncrement: 1, remediationScriptIncrement: 1, rebootIncrement: 1, + upload: SecondUpload()); + + /// + /// A Sessions row with every column a real row carries. The running kind is WhiteGlove Part 2 + /// in flight (stored Part 1 duration + ResumedAt drive the live duration) with the tracker + /// state of its first upload and the matching twin; the completed kind has CompletedAt, a + /// stored duration and an admin mark. + /// + private static TableEntity FullRow(RowKind kind, SessionOwner? owner = null) + { + var tracker = new ServerTime.Tracker(); + tracker.Absorb(FirstUpload().Select(ServerTime.Event.From).ToList()); + + var row = new TableEntity(TenantId, SessionId) + { + ["StartedAt"] = new DateTimeOffset(DeviceStart), + ["LastEventAt"] = new DateTimeOffset(DeviceStart.AddMinutes(20)), + ["LastIngestAt"] = new DateTimeOffset(ServerStart.AddMinutes(20)), + ["EventCount"] = 2, + ["PlatformScriptCount"] = 3, + ["RemediationScriptCount"] = 1, + ["RebootCount"] = 1, + ["IndexRowKey"] = "2516000000000000000_" + SessionId, // deliberately not the key StartedAt would give + ["CurrentPhase"] = 4, + ["CurrentPhaseDetail"] = "Installing apps", + ["FailureReason"] = string.Empty, + ["DiagnosticsBlobName"] = "diag-" + SessionId + ".zip", + ["DiagnosticsBlobDestination"] = "platform", + ["IsPreProvisioned"] = true, + ["IsHybridJoin"] = false, + ["IsUserDriven"] = false, + ["IsSelfDeployingProfile"] = false, + ["IsCloudPc"] = false, + ["EnrollmentType"] = "v2", + ["PendingActionsJson"] = "[{\"Type\":\"collect_diagnostics\",\"Reason\":\"rule\"}]", + ["AgentVersion"] = "2.0.1478", + ["ImeAgentVersion"] = "1.99.1", + [TableStorageService.ServerTimeStateColumn] = tracker.ToJson(), + ["StartedAtServer"] = new DateTimeOffset(tracker.StartedAtServer!.Value), + // Columns the hot path never needs — the projection must drop them without a trace. + ["DeviceName"] = "DESKTOP-TEST01", + ["SerialNumber"] = "SN-1", + ["Manufacturer"] = "Contoso", + ["Model"] = "Model X", + ["OsName"] = "Windows 11 Enterprise", + ["OsBuild"] = "26100.4000", + ["OsDisplayVersion"] = "24H2", + ["GeoCountry"] = "DE", + ["GeoRegion"] = "HE", + ["GeoCity"] = "Frankfurt", + ["GeoLoc"] = "50.11,8.68", + ["ConnectionType"] = "Ethernet", + ["ValidatedBy"] = "AutopilotV1", + ["VerdictPath"] = "ime_complete", + ["FailureSnapshotJson"] = new string('x', 2048), + ["AvgApiLatencyMs"] = 123.4, + ["ApiRequestCount"] = 17, + ["DeletionState"] = SessionDeletionState.None, + }; + + if (kind == RowKind.RunningWhiteGlovePart2) + { + row["Status"] = "InProgress"; + row["ResumedAt"] = new DateTimeOffset(DeviceStart.AddMinutes(10)); + row["DurationSeconds"] = 600; // Part 1, stored + row["AdminMarkedAction"] = "Succeeded"; + } + else + { + row["Status"] = "Succeeded"; + row["CompletedAt"] = new DateTimeOffset(DeviceStart.AddMinutes(40)); + row["CompletedAtServer"] = new DateTimeOffset(ServerStart.AddMinutes(40)); + row["DurationSeconds"] = 2400; + row["AdminMarkedAction"] = "Succeeded"; + row["ReconcileReason"] = "ime_complete"; + row["CompletionSource"] = "agent"; + } + + if (owner != null) + SessionOwnershipPolicy.ApplyTo(row, owner); + + row.ETag = new ETag("0xEXISTING"); + return row; + } + + private static List FirstUpload() => new() + { + Event(1, "agent_started", "Agent", DeviceStart, sent: DeviceStart.AddSeconds(10), received: ServerStart.AddSeconds(10)), + Event(2, "performance_snapshot", "PerformanceCollector", DeviceStart.AddSeconds(5), sent: DeviceStart.AddSeconds(10), received: ServerStart.AddSeconds(10)), + }; + + private static List SecondUpload() => new() + { + Event(3, "app_install_started", "ImeLogTracker", DeviceStart.AddSeconds(30), sent: DeviceStart.AddSeconds(60), received: ServerStart.AddSeconds(60)), + Event(4, "performance_snapshot", "PerformanceCollector", DeviceStart.AddSeconds(40), sent: DeviceStart.AddSeconds(60), received: ServerStart.AddSeconds(60)), + }; + + private static EnrollmentEvent Event(long sequence, string type, string source, DateTime time, DateTime? sent, DateTime received) => new() + { + EventId = $"e{sequence}", + TenantId = TenantId, + SessionId = SessionId, + Sequence = sequence, + EventType = type, + Source = source, + Timestamp = time, + SentAt = sent, + ReceivedAt = received, + Message = string.Empty, + Data = new Dictionary(), + }; + + private static SecurityValidationResult Cert(string thumb, string? deviceId, string? serial = "SN-1") => new() + { + IsValid = true, + CertificateThumbprint = thumb, + IntuneDeviceId = deviceId, + SerialNumber = serial, + ValidatedBy = ValidatorType.AutopilotV1, + }; + + private static SecurityValidationResult Bootstrap(string code, string? serial = "SN-1") => new() + { + IsValid = true, + IsBootstrapAuth = true, + BootstrapShortCode = code, + SerialNumber = serial, + ValidatedBy = ValidatorType.Bootstrap, + }; + + private static SessionOwner CertOwner(string thumb, string? deviceId, string serial = "SN-1") => + new() { Kind = SessionOwner.Kinds.Cert, Thumbprint = thumb, DeviceId = deviceId, Serial = serial, BoundAt = ServerStart }; + + private static SessionOwner BootstrapOwner(string code, string serial = "SN-1") => + new() { Kind = SessionOwner.Kinds.Bootstrap, BootstrapCode = code, Serial = serial, BoundAt = ServerStart }; + + /// What a select does on the service: keys, ETag and timestamp survive, every other column is gone. + private static TableEntity Project(TableEntity row, IEnumerable? select) + { + if (select == null) + return row; + var keep = new HashSet(select, StringComparer.Ordinal); + var projected = new TableEntity(row.PartitionKey, row.RowKey) { ETag = row.ETag, Timestamp = row.Timestamp }; + foreach (var kv in row) + { + if (keep.Contains(kv.Key)) + projected[kv.Key] = kv.Value; + } + return projected; + } + + private static readonly string[] SystemKeys = { "PartitionKey", "RowKey", "Timestamp", "odata.etag" }; + + private static void AssertSameEntity(TableEntity expected, TableEntity actual) + { + var expectedKeys = expected.Keys.Except(SystemKeys).OrderBy(k => k, StringComparer.Ordinal).ToList(); + var actualKeys = actual.Keys.Except(SystemKeys).OrderBy(k => k, StringComparer.Ordinal).ToList(); + Assert.Equal(expectedKeys, actualKeys); + Assert.Equal(expected.PartitionKey, actual.PartitionKey); + Assert.Equal(expected.RowKey, actual.RowKey); + foreach (var key in expectedKeys) + { + if (key == "LastIngestAt") + { + var delta = expected.GetDateTimeOffset(key)!.Value - actual.GetDateTimeOffset(key)!.Value; + Assert.True(Math.Abs(delta.TotalSeconds) < 10, $"{key}: {delta}"); + continue; + } + Assert.True(Equals(expected[key], actual[key]), $"{key}: {expected[key]} vs {actual[key]}"); + } + } + + private static void AssertSameSnapshot(SessionIngestSnapshot expected, SessionIngestSnapshot actual) + { + Assert.Equal(expected.Status, actual.Status); + Assert.Equal(expected.CurrentPhase, actual.CurrentPhase); + Assert.Equal(expected.CurrentPhaseDetail, actual.CurrentPhaseDetail); + Assert.Equal(expected.FailureReason, actual.FailureReason); + Assert.Equal(expected.EventCount, actual.EventCount); + Assert.Equal(expected.CompletedAt, actual.CompletedAt); + Assert.Equal(expected.DiagnosticsBlobName, actual.DiagnosticsBlobName); + Assert.Equal(expected.IsPreProvisioned, actual.IsPreProvisioned); + Assert.Equal(expected.AdminMarkedAction, actual.AdminMarkedAction); + Assert.Equal(expected.PendingActionsJson, actual.PendingActionsJson); + Assert.Equal(expected.AgentVersion, actual.AgentVersion); + Assert.Equal(expected.ImeAgentVersion, actual.ImeAgentVersion); + // The live duration of a running session is computed from the clock; the two mappings run + // milliseconds apart. + Assert.NotNull(expected.DurationSeconds); + Assert.NotNull(actual.DurationSeconds); + Assert.True(Math.Abs(expected.DurationSeconds!.Value - actual.DurationSeconds!.Value) <= 5, + $"DurationSeconds: {expected.DurationSeconds} vs {actual.DurationSeconds}"); + } + + /// + /// SDK-mock harness over the real . With honourSelect + /// the Sessions point read returns only the columns the caller selected (what the service + /// does); without it the whole row (what the other session harnesses do). Merges on Sessions + /// and SessionsIndex are captured. + /// + private sealed class Harness + { + public TableStorageService Sut { get; } + public TableEntity? Written { get; private set; } + public ETag WrittenETag { get; private set; } + public TableEntity? IndexWritten { get; private set; } + public IEnumerable? LastSelect { get; private set; } + public int LastReadColumnCount { get; private set; } + + public Harness(TableEntity existing, bool honourSelect) + { + var sessions = new Mock(); + sessions.Setup(t => t.GetEntityIfExistsAsync( + TenantId, SessionId, It.IsAny>(), It.IsAny())) + .Returns, CancellationToken>((_, _, select, _) => + { + LastSelect = select?.ToList(); + var served = honourSelect ? Project(existing, select) : existing; + LastReadColumnCount = served.Count; + var response = new Mock>(); + response.SetupGet(r => r.HasValue).Returns(true); + response.SetupGet(r => r.Value).Returns(served); + return Task.FromResult(response.Object); + }); + sessions.Setup(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((e, etag, _, _) => + { + Written = e; + WrittenETag = etag; + return Task.FromResult(new Mock().Object); + }); + + var index = new Mock(); + index.Setup(t => t.UpdateEntityAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((e, _, _, _) => + { + IndexWritten = e; + return Task.FromResult(new Mock().Object); + }); + + var serviceClient = new Mock(); + serviceClient.Setup(s => s.GetTableClient(Constants.TableNames.Sessions)).Returns(sessions.Object); + serviceClient.Setup(s => s.GetTableClient(Constants.TableNames.SessionsIndex)).Returns(index.Object); + Sut = new TableStorageService(serviceClient.Object, NullLogger.Instance); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TableLifecycleBucketTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TableLifecycleBucketTests.cs index 77fe114a..666d7426 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/TableLifecycleBucketTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TableLifecycleBucketTests.cs @@ -252,6 +252,9 @@ public class TableLifecycleBucketTests public Task GetSessionRowAsync(string tenantId, string sessionId, CancellationToken cancellationToken = default) => Task.FromResult(new TableEntity(tenantId, sessionId) { ["IndexRowKey"] = IndexRowKey }); + public Task GetSessionRowAsync(string tenantId, string sessionId, IEnumerable select, CancellationToken cancellationToken = default) + => GetSessionRowAsync(tenantId, sessionId, cancellationToken); + public Task GetSessionsIndexRowAsync(string tenantId, string indexRowKey, CancellationToken cancellationToken = default) => Task.FromResult(new TableEntity(tenantId, indexRowKey) { ["SessionId"] = SessionId }); diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs index 20621e49..1d4693b8 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs @@ -33,6 +33,9 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage public Task GetSessionAsync(string tenantId, string sessionId) => _storage.GetSessionAsync(tenantId, sessionId); + public Task GetSessionIngestSnapshotAsync(string tenantId, string sessionId) + => _storage.GetSessionIngestSnapshotAsync(tenantId, sessionId); + public Task ResolveSessionTenantIdAsync(string sessionId) => _storage.ResolveSessionTenantIdAsync(sessionId); @@ -146,7 +149,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, diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs index aab5ea58..4a46a4ef 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs @@ -237,7 +237,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest Azure.Data.Tables.TableEntity? guardSessionRow; try { - guardSessionRow = await _deletionGuard.EnsureWritableAndGetRowAsync(bodyTenantId, sessionId, "V2.IngestTelemetry"); + guardSessionRow = await _deletionGuard.EnsureWritableAndGetRowAsync(bodyTenantId, sessionId, "V2.IngestTelemetry", SessionRowProjections.GuardRow); } catch (SessionDeletionLockedException locked) { @@ -262,9 +262,9 @@ namespace AutopilotMonitor.Functions.Functions.Ingest return AsOutput(await WriteDeviceBlockedAsync(req, rowSerialVerdict)); } - // 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); + // Reuse the guard's row (projected to SessionRowProjections.GuardRow): its Status feeds + // the stall-heal check inside EventIngestProcessor, saving one Sessions point-read per batch. + var preFetchedStatus = SessionRowProjections.TryReadStatus(guardSessionRow); // SESSION-OWNER-BINDING: is the device behind this certificate/token the device this // session belongs to? Same guard row, zero extra reads. A foreign identity gets 403 @@ -449,26 +449,6 @@ namespace AutopilotMonitor.Functions.Functions.Ingest }); } - /// - /// 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, - }; - } - /// /// Routes every item of the batch by into its storage /// shape. Pure and total: an item whose Kind is not a name @@ -596,7 +576,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest private async Task<(string? adminAction, List? pendingActions)> ReadControlSignalsAsync(string tenantId, string sessionId) { - var session = await _sessionRepo.GetSessionAsync(tenantId, sessionId); + var session = await _sessionRepo.GetSessionIngestSnapshotAsync(tenantId, sessionId); if (session == null) return (null, null); // AdminAction carries the portal-button signal only. The old logic (Status == diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportAgentErrorFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportAgentErrorFunction.cs index fe14bfc8..991c1cc9 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportAgentErrorFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportAgentErrorFunction.cs @@ -4,6 +4,7 @@ using System.Globalization; using System.Linq; using System.Net; using System.Text.Json; +using AutopilotMonitor.Functions.Helpers; using AutopilotMonitor.Functions.Security; using AutopilotMonitor.Functions.Services; using AutopilotMonitor.Functions.Services.Deletion; @@ -467,7 +468,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest if (deletionLocked) return new(OpsEventSeverity.Info, false, false, "deleting", "session is being deleted — nothing written"); - var status = IngestTelemetryFunction.TryReadSessionStatus(sessionRow); + var status = SessionRowProjections.TryReadStatus(sessionRow); var statusText = status?.ToString() ?? "unknown"; return status.HasValue && Helpers.DeviceJourneyCalculator.IsTerminal(status.Value) ? new(OpsEventSeverity.Info, true, true, statusText, $"late cleanup — session already {statusText} when the break arrived") diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/RegisterSessionFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/RegisterSessionFunction.cs index 450da0b4..0d130c3c 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/RegisterSessionFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/RegisterSessionFunction.cs @@ -167,7 +167,7 @@ namespace AutopilotMonitor.Functions.Functions.Sessions try { guardSessionRow = await _deletionGuard.EnsureWritableAndGetRowAsync( - registration.TenantId, registration.SessionId, "RegisterSession"); + registration.TenantId, registration.SessionId, "RegisterSession", SessionRowProjections.GuardRow); } catch (SessionDeletionLockedException locked) { @@ -180,14 +180,14 @@ namespace AutopilotMonitor.Functions.Functions.Sessions }; } - // Pre-read the session row so we can distinguish three lifecycle entries: - // - existing == null → first-time registration (fresh enrollment) - // - existing.Pending → WhiteGlove Part 2 resume (user-driven phase) - // - everything else → agent restart / terminal re-register (no start signal) + // The guard's row (projected to SessionRowProjections.GuardRow) distinguishes three + // lifecycle entries — no second read of the Sessions row: + // - row absent → first-time registration (fresh enrollment) + // - row Pending → WhiteGlove Part 2 resume (user-driven phase) + // - everything else → agent restart / terminal re-register (no start signal) // Used below to scope the opt-in "enrollment started" webhook to real start events. - var preExistingSession = await _sessionRepo.GetSessionAsync(registration.TenantId, registration.SessionId); - bool isFreshRegistration = preExistingSession == null; - bool isWhiteGloveResume = preExistingSession?.Status == SessionStatus.Pending; + bool isFreshRegistration = guardSessionRow == null; + bool isWhiteGloveResume = SessionRowProjections.TryReadStatus(guardSessionRow) == SessionStatus.Pending; // SESSION-OWNER-BINDING: compare the caller's validated identity against the owner // already on the row (reusing the guard's read). A foreign identity is refused with diff --git a/src/Backend/AutopilotMonitor.Functions/Helpers/SessionRowProjections.cs b/src/Backend/AutopilotMonitor.Functions/Helpers/SessionRowProjections.cs new file mode 100644 index 00000000..e37d9410 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Helpers/SessionRowProjections.cs @@ -0,0 +1,67 @@ +using System; +using System.Linq; +using Azure.Data.Tables; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared.Models; + +namespace AutopilotMonitor.Functions.Helpers +{ + /// + /// Column lists for the projected point reads of the Sessions row on the agent hot path, and + /// the reader of the status they carry. Every upload reads the row twice (deletion guard, + /// counter merge); read whole, the row's size set the ingest latency (backend 1.5.1256: + /// +35 ms p50 from ~400 bytes of new columns, the storage server unchanged). Each list is + /// pinned by SessionRowProjectionTests against a select-strict table: a column a + /// consumer reads but the list lacks fails there, not in production as a silent null. + /// + public static class SessionRowProjections + { + /// + /// What the agent routes read off the deletion guard's row: the status prefetch + /// (, the registration's lifecycle entry), the registered serial + /// for the kill switch, and the SESSION-OWNER-BINDING columns + /// (SessionOwnershipPolicy.Evaluate). The guard adds its own lock columns. + /// + public static readonly string[] GuardRow = + new[] { "Status", "SerialNumber" }.Concat(SessionOwner.Columns.All).ToArray(); + + /// + /// What the counter merge (IncrementSessionEventCountAsync) reads for its own + /// read-modify-write and what the SessionIngestSnapshot it returns carries. + /// + public static readonly string[] IngestSnapshot = + { + // Counter read-modify-write, anchor alignment, index dual-write. + "EventCount", "StartedAt", "LastEventAt", "PlatformScriptCount", "RemediationScriptCount", "RebootCount", + "IndexRowKey", + // Server-time tracker (StageServerTimeUpload / StageServerTimeTwin). + "CompletedAt", TableStorageService.ServerTimeStateColumn, + TableStorageService.StartedAtServerColumn, TableStorageService.ResumedAtServerColumn, + // The snapshot: SignalR delta, control signals, pattern health, shutdown close, live duration + // (ComputeEffectiveDuration reads DurationSeconds, IsPreProvisioned, ResumedAt). + "Status", "CurrentPhase", "CurrentPhaseDetail", "FailureReason", "DurationSeconds", + "DiagnosticsBlobName", "IsPreProvisioned", "ResumedAt", "AdminMarkedAction", "PendingActionsJson", + "AgentVersion", "ImeAgentVersion", + }; + + /// + /// Status off a Sessions row (the guard's row). 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. Null for missing or + /// unparseable values (incl. a defensive int fallback for any legacy numeric shape) — callers + /// then fall back to their own read. + /// + public static SessionStatus? TryReadStatus(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, + }; + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/ISessionDeletionInventoryReader.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/ISessionDeletionInventoryReader.cs index 4f41f2d7..f16ce79d 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/ISessionDeletionInventoryReader.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/ISessionDeletionInventoryReader.cs @@ -20,6 +20,13 @@ namespace AutopilotMonitor.Functions.Services.Deletion /// Task GetSessionRowAsync(string tenantId, string sessionId, CancellationToken cancellationToken = default); + /// + /// projected to + /// — the hot-path form (the agent routes read the guard's row this + /// way, see SessionRowProjections.GuardRow). Returns null when the row does not exist. + /// + Task GetSessionRowAsync(string tenantId, string sessionId, IEnumerable select, CancellationToken cancellationToken = default); + /// /// Loads a SessionsIndex row by its (tenantId, indexRowKey) coordinate. Used for the /// FINAL tombstone step's row dump. Returns null when missing (treated as already-gone). diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionGuard.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionGuard.cs index 91ee0776..c8491dbf 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionGuard.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionGuard.cs @@ -1,3 +1,5 @@ +using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Azure.Data.Tables; @@ -80,6 +82,26 @@ namespace AutopilotMonitor.Functions.Services.Deletion public async Task EnsureWritableAndGetRowAsync(string tenantId, string sessionId, string callerContext, CancellationToken cancellationToken = default) { var sessionRow = await _reader.GetSessionRowAsync(tenantId, sessionId, cancellationToken); + return await GuardRowAsync(sessionRow, tenantId, sessionId, callerContext, cancellationToken); + } + + /// Columns the guard reads off the row itself; a projected read adds them to the caller's list. + public static readonly string[] LockColumns = { "DeletionState", "PendingDeletionManifestId" }; + + /// + /// with the + /// row projected to plus — the agent hot + /// path's form (see SessionRowProjections.GuardRow). The returned row carries only + /// those columns. + /// + public async Task EnsureWritableAndGetRowAsync(string tenantId, string sessionId, string callerContext, IReadOnlyCollection select, CancellationToken cancellationToken = default) + { + var sessionRow = await _reader.GetSessionRowAsync(tenantId, sessionId, select.Concat(LockColumns).Distinct(), cancellationToken); + return await GuardRowAsync(sessionRow, tenantId, sessionId, callerContext, cancellationToken); + } + + private async Task GuardRowAsync(TableEntity? sessionRow, string tenantId, string sessionId, string callerContext, CancellationToken cancellationToken) + { if (sessionRow != null) { ThrowIfLocked(sessionRow, callerContext); diff --git a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs index 4674d150..6d2dfa2c 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs @@ -1,4 +1,5 @@ using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.DataAccess; using System.Linq; using AutopilotMonitor.Functions.Services.Notifications; using AutopilotMonitor.Shared.Models; @@ -134,19 +135,21 @@ namespace AutopilotMonitor.Functions.Services }); private SignalRMessageAction[] BuildSignalRMessages( - IngestEventsRequest request, SessionSummary? updatedSession, int processedCount, + IngestEventsRequest request, SessionIngestSnapshot? snapshot, int processedCount, List newRuleResults) { - object? sessionDelta = updatedSession != null ? new { - updatedSession.CurrentPhase, - updatedSession.CurrentPhaseDetail, - updatedSession.Status, - updatedSession.FailureReason, - updatedSession.EventCount, - updatedSession.DurationSeconds, - updatedSession.CompletedAt, - updatedSession.DiagnosticsBlobName, - updatedSession.IsPreProvisioned + // The delta's member names are the portal's live-update contract (sessionUpdate); the + // snapshot carries exactly these fields, projected off the row. + object? sessionDelta = snapshot != null ? new { + snapshot.CurrentPhase, + snapshot.CurrentPhaseDetail, + snapshot.Status, + snapshot.FailureReason, + snapshot.EventCount, + snapshot.DurationSeconds, + snapshot.CompletedAt, + snapshot.DiagnosticsBlobName, + snapshot.IsPreProvisioned } : null; var summaryMessage = new SignalRMessageAction(Constants.SignalRMessages.NewEvents) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs index a43866c4..c08bab47 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs @@ -219,14 +219,15 @@ 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; + // The increment's post-merge snapshot (the ingest slice of the row, read projected — + // SessionRowProjections.IngestSnapshot) serves the common case (non-terminal batch, no + // transition, no diagnostics upload): it already reflects this batch's status transition + // (written above) plus the counter merge. The whole row is read only below, where a + // consumer needs it. + SessionIngestSnapshot? snapshot = null; if (processedCount > 0) { - updatedSession = await _sessionRepo.IncrementSessionEventCountAsync( + snapshot = await _sessionRepo.IncrementSessionEventCountAsync( request.TenantId, request.SessionId, processedCount, @@ -253,7 +254,7 @@ namespace AutopilotMonitor.Functions.Services if (patternHitsEvent.Data!.TryGetValue("imeVersion", out var imeVersionObj)) hitsImeVersion = imeVersionObj?.ToString(); if (string.IsNullOrWhiteSpace(hitsImeVersion)) - hitsImeVersion = updatedSession?.ImeAgentVersion; + hitsImeVersion = snapshot?.ImeAgentVersion; _ = _imePatternHealth.RecordSessionHitsAsync(hitsImeVersion, patternHitsEvent.Data, request.TenantId, request.SessionId); } @@ -274,7 +275,7 @@ namespace AutopilotMonitor.Functions.Services // tripwire must not. Fail-soft: ingest stays a 200 no matter what. if (skewScan != null && (statusTransitioned || whiteGloveStatusTransitioned)) await TryFireCmTraceSkewTripwireAsync(request.TenantId, request.SessionId, - updatedSession?.AgentVersion, skewScan); + snapshot?.AgentVersion, skewScan); } // Observation end: the batch carrying agent_shutting_down is the agent's last word. @@ -284,10 +285,10 @@ namespace AutopilotMonitor.Functions.Services // and the agent keeps watching after enrollment_complete). Non-terminal sessions // (reboot shutdown, WhiteGlove part 1) keep their rows open. Idempotent + fail-soft. if (classification.AgentShutdownEvent != null - && updatedSession != null - && (updatedSession.Status == SessionStatus.Succeeded - || updatedSession.Status == SessionStatus.Failed - || updatedSession.Status == SessionStatus.Incomplete)) + && snapshot != null + && (snapshot.Status == SessionStatus.Succeeded + || snapshot.Status == SessionStatus.Failed + || snapshot.Status == SessionStatus.Incomplete)) { await _metricsRepo.CloseOpenAppInstallsForSessionAsync(request.TenantId, request.SessionId); } @@ -430,12 +431,18 @@ namespace AutopilotMonitor.Functions.Services } } - // Re-read only when a write AFTER the increment made the snapshot stale: terminal - // batches (ReconcileSessionCountersAsync) 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); + // The whole row is read only where a consumer needs it: terminal batches + // (ReconcileSessionCountersAsync wrote after the increment), diagnostics uploads (blob + // fields), status or WhiteGlove transitions (webhooks, SLA) — and when no increment ran + // or it returned null (missing row, exhausted ETag retries). Every other batch works off + // the projected snapshot; here it is derived from the full read so both paths agree. + SessionSummary? updatedSession = null; + if (isTerminalBatch || classification.DiagnosticsUploadedEvent != null + || statusTransitioned || whiteGloveStatusTransitioned || snapshot == null) + { + updatedSession = await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId); + snapshot = updatedSession != null ? SessionIngestSnapshot.From(updatedSession) : null; + } // NOTE: long-running InProgress sessions are handled authoritatively by // MaintenanceService.MarkStalledSessionsAsTimedOutAsync (Stalled at 2h agent-silence, @@ -443,13 +450,13 @@ namespace AutopilotMonitor.Functions.Services // here was pure observability noise (fired on every ingest of a >4h session, strictly // later than maintenance's first action) and was removed. - if (classification.WhiteGloveEvent != null && updatedSession?.IsPreProvisioned != true) + if (classification.WhiteGloveEvent != null && snapshot?.IsPreProvisioned != true) { _logger.LogError( "{SessionPrefix} WhiteGlove status update not persisted after retries and fallback. " + "IsPreProvisioned={IsPreProvisioned}, Status={Status}. " + "Proceeding with 200 to allow agent spool drain.", - sessionPrefix, updatedSession?.IsPreProvisioned, updatedSession?.Status); + sessionPrefix, snapshot?.IsPreProvisioned, snapshot?.Status); } await SendWebhookNotificationsAsync( @@ -468,7 +475,7 @@ namespace AutopilotMonitor.Functions.Services // fired falsely for every post-completion agent event (agent_shutting_down, // diagnostics_uploaded, enrollment_summary_shown) — making the agent believe an // admin had clicked Mark-Succeeded after its own completion. - string? adminAction = updatedSession?.AdminMarkedAction; + string? adminAction = snapshot?.AdminMarkedAction; if (!string.IsNullOrEmpty(adminAction)) { _logger.LogInformation( @@ -477,7 +484,7 @@ namespace AutopilotMonitor.Functions.Services } List? pendingActions = null; - if (updatedSession != null && !string.IsNullOrEmpty(updatedSession.PendingActionsJson)) + if (snapshot != null && !string.IsNullOrEmpty(snapshot.PendingActionsJson)) { var fetched = await _sessionRepo.FetchAndClearPendingActionsAsync(request.TenantId, request.SessionId); if (fetched.Count > 0) @@ -502,7 +509,7 @@ namespace AutopilotMonitor.Functions.Services } } - var signalRMessages = BuildSignalRMessages(request, updatedSession, processedCount, newRuleResults); + var signalRMessages = BuildSignalRMessages(request, snapshot, processedCount, newRuleResults); return new EventIngestResult { diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs index 9357957f..caa0521b 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs @@ -56,6 +56,20 @@ namespace AutopilotMonitor.Functions.Services } } + public virtual async Task GetSessionRowAsync(string tenantId, string sessionId, IEnumerable select, CancellationToken cancellationToken = default) + { + var tableClient = _tableServiceClient.GetTableClient(Shared.Constants.TableNames.Sessions); + try + { + var response = await tableClient.GetEntityAsync(tenantId, sessionId, select: select, cancellationToken: cancellationToken); + return response.Value; + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + return null; + } + } + public async Task GetSessionsIndexRowAsync(string tenantId, string indexRowKey, CancellationToken cancellationToken = default) { var tableClient = _tableServiceClient.GetTableClient(Shared.Constants.TableNames.SessionsIndex); diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs index 9cd755ca..d52d93ef 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs @@ -7,6 +7,7 @@ using AutopilotMonitor.Functions.Pagination; using AutopilotMonitor.Functions.Security; using AutopilotMonitor.Functions.Services.Caching; using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.DataAccess; using AutopilotMonitor.Shared.Models; using AutopilotMonitor.Shared.Models.Deletion; using AutopilotMonitor.Shared.Pagination; @@ -1687,6 +1688,26 @@ namespace AutopilotMonitor.Functions.Services } } + /// + /// The ingest slice of the session row, read projected to + /// (the signal-only upload's control + /// signals). Null when the row is missing; fail-soft like . + /// + public async Task GetSessionIngestSnapshotAsync(string tenantId, string sessionId) + { + try + { + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions); + var entity = await tableClient.GetEntityIfExistsAsync(tenantId, sessionId, select: SessionRowProjections.IngestSnapshot); + return entity.HasValue ? MapToIngestSnapshot(entity.Value!) : null; + } + catch (Exception ex) + { + _logger.LogError(ex, $"Failed to get the ingest snapshot of session {sessionId}"); + return null; + } + } + /// /// Open (non-terminal) sessions of the same physical device within one tenant: Pending, /// InProgress, Stalled or AwaitingUser rows matching the SerialNumber. Server-side @@ -2570,8 +2591,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. + /// The read is projected to (the hot + /// path never pays for the whole row) and the post-merge values of that slice come back as + /// a so the ingest hot path can skip its follow-up read; + /// null on missing row / exhausted retries / error. /// /// Retry semantics: these read-modify-write increments are NOT idempotent under the /// agent's at-least-once retry (rows dedupe, counters double). EventCount and RebootCount @@ -2584,7 +2607,7 @@ namespace AutopilotMonitor.Functions.Services /// tracker in the same write, see . /// /// - 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, IReadOnlyList? upload = null) + 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, IReadOnlyList? upload = null) { SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId)); @@ -2597,7 +2620,7 @@ namespace AutopilotMonitor.Functions.Services try { var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions); - var entityResponse = await tableClient.GetEntityIfExistsAsync(tenantId, sessionId); + var entityResponse = await tableClient.GetEntityIfExistsAsync(tenantId, sessionId, select: SessionRowProjections.IngestSnapshot); if (!entityResponse.HasValue) { _logger.LogWarning("Session {SessionId} not found when incrementing event count — session may have been cleaned up or not yet registered", sessionId); @@ -2672,15 +2695,15 @@ namespace AutopilotMonitor.Functions.Services // Dual-write: keep SessionsIndex in sync (StartedAt-shift → full upsert, else merge). await SyncSessionIndexAsync(tenantId, sessionId, entity, update, currentStartedAt, safeEarliestEventTimestamp); - // 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. + // Apply the merged fields onto the projected RMW read and map the ingest slice — + // 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); + return MapToIngestSnapshot(entity); } catch (Azure.RequestFailedException ex) when (ex.Status == 412) { @@ -3550,6 +3573,43 @@ namespace AutopilotMonitor.Functions.Services // internal (not private) so UsageMetricsProjectionEquivalenceTests can pin that a Sessions // row carrying only UsageMetricsSessionProjection maps to the same usage-relevant fields // as a full row. + /// Stored Status (a string; "InProgress" when absent); Unknown, with a warning, when it does not parse. + private SessionStatus ParseStoredStatus(TableEntity entity, string sessionId) + { + var statusString = entity.GetString("Status") ?? "InProgress"; + if (Enum.TryParse(statusString, ignoreCase: true, out var status)) + return status; + _logger.LogWarning($"Failed to parse status '{statusString}' for session {sessionId}, defaulting to Unknown"); + return SessionStatus.Unknown; + } + + /// + /// The ingest slice of a Sessions row (): read with + /// , mapped exactly as + /// maps the same fields (pinned by + /// SessionRowProjectionTests). + /// + internal SessionIngestSnapshot MapToIngestSnapshot(TableEntity entity) + { + var startedAt = SafeGetDateTime(entity, "StartedAt") ?? DateTime.UtcNow; + var completedAt = SafeGetDateTime(entity, "CompletedAt"); + var status = ParseStoredStatus(entity, entity.RowKey); + return new SessionIngestSnapshot( + status, + SafeGetInt32(entity, "CurrentPhase") ?? 0, + entity.GetString("CurrentPhaseDetail") ?? string.Empty, + entity.GetString("FailureReason") ?? string.Empty, + SafeGetInt32(entity, "EventCount") ?? 0, + ComputeEffectiveDuration(entity, status, startedAt, completedAt), + completedAt, + entity.GetString("DiagnosticsBlobName"), + entity.GetBoolean("IsPreProvisioned") ?? false, + entity.GetString("AdminMarkedAction"), + entity.GetString("PendingActionsJson") ?? string.Empty, + entity.GetString("AgentVersion") ?? string.Empty, + entity.GetString("ImeAgentVersion") ?? string.Empty); + } + internal SessionSummary MapToSessionSummary(TableEntity entity) => MapToSessionSummary(entity, entity.RowKey); @@ -3566,13 +3626,7 @@ namespace AutopilotMonitor.Functions.Services var startedAt = SafeGetDateTime(entity, "StartedAt") ?? DateTime.UtcNow; var completedAt = SafeGetDateTime(entity, "CompletedAt"); - // Parse status with error handling and case-insensitivity - var statusString = entity.GetString("Status") ?? "InProgress"; - if (!Enum.TryParse(statusString, ignoreCase: true, out var status)) - { - _logger.LogWarning($"Failed to parse status '{statusString}' for session {sessionId}, defaulting to Unknown"); - status = SessionStatus.Unknown; - } + var status = ParseStoredStatus(entity, sessionId); return new SessionSummary { diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs index 7caf5a6c..22b7e970 100644 --- a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs @@ -21,6 +21,13 @@ namespace AutopilotMonitor.Shared.DataAccess Task UpdateSessionOwnerAsync(string tenantId, string sessionId, SessionOwner owner); Task GetSessionAsync(string tenantId, string sessionId); /// + /// The ingest slice of the session row (), read projected + /// to its columns; null when the row is missing or unreadable. The signal-only upload reads + /// its control signals (AdminMarkedAction, pending ServerActions) through this instead of a + /// full . + /// + Task GetSessionIngestSnapshotAsync(string tenantId, string sessionId); + /// /// Resolves the tenantId owning — point-read on the /// SessionTenantLookup table with a legacy SessionsIndex-scan fallback that self-heals /// the lookup row. Null when the session is unknown. Used for global-scope cross-tenant @@ -111,14 +118,15 @@ namespace AutopilotMonitor.Shared.DataAccess string? failureSnapshotJson = null, bool allowTerminalReclassification = false, bool espSoftFailure = false, string? completionSource = null); /// - /// 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). + /// Increments per-session counters via read-modify-write. The read is projected to the + /// ingest slice of the row and its post-merge values come back as a + /// (the RMW read with the applied increments) so + /// hot-path callers can skip a follow-up read; null when the row is missing or the update + /// could not be applied (caller falls back to its own ). /// (the batch's stored events) feeds the session's server-time /// tracker in the same write (StartedAtServer / ResumedAtServer while the session runs). /// - Task IncrementSessionEventCountAsync( + Task IncrementSessionEventCountAsync( string tenantId, string sessionId, int increment, DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null, EnrollmentPhase? currentPhase = null, diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/SessionIngestSnapshot.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/SessionIngestSnapshot.cs new file mode 100644 index 00000000..ff91c6c6 --- /dev/null +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/SessionIngestSnapshot.cs @@ -0,0 +1,62 @@ +using System; +using AutopilotMonitor.Shared.Models; + +namespace AutopilotMonitor.Shared.DataAccess +{ + /// + /// The slice of a Sessions row that a non-terminal ingest batch reads: the counter merge + /// returns it as the post-merge state (SignalR delta, control signals to the agent, IME + /// pattern health, the shutdown close) and the signal-only upload reads it directly. + /// Deliberately not a : the hot path reads the row projected to + /// exactly these columns (read whole, the row's size — not the storage server — set the + /// ingest latency), so a summary built from that read would carry silent defaults in every + /// other field. Batches that need the whole row (terminal, status or WhiteGlove transition, + /// diagnostics upload) read a full summary and derive this via . + /// + public sealed class SessionIngestSnapshot + { + public SessionIngestSnapshot( + SessionStatus status, int currentPhase, string currentPhaseDetail, string failureReason, + int eventCount, int? durationSeconds, DateTime? completedAt, string? diagnosticsBlobName, + bool isPreProvisioned, string? adminMarkedAction, string pendingActionsJson, + string agentVersion, string imeAgentVersion) + { + Status = status; + CurrentPhase = currentPhase; + CurrentPhaseDetail = currentPhaseDetail; + FailureReason = failureReason; + EventCount = eventCount; + DurationSeconds = durationSeconds; + CompletedAt = completedAt; + DiagnosticsBlobName = diagnosticsBlobName; + IsPreProvisioned = isPreProvisioned; + AdminMarkedAction = adminMarkedAction; + PendingActionsJson = pendingActionsJson; + AgentVersion = agentVersion; + ImeAgentVersion = imeAgentVersion; + } + + public SessionStatus Status { get; } + public int CurrentPhase { get; } + public string CurrentPhaseDetail { get; } + public string FailureReason { get; } + public int EventCount { get; } + public int? DurationSeconds { get; } + public DateTime? CompletedAt { get; } + public string? DiagnosticsBlobName { get; } + public bool IsPreProvisioned { get; } + /// Portal-button signal to the agent (set only by the Mark-Succeeded/Failed functions). + public string? AdminMarkedAction { get; } + /// Queued ServerActions (JSON); empty when none are pending. + public string PendingActionsJson { get; } + public string AgentVersion { get; } + public string ImeAgentVersion { get; } + + /// The same slice taken off a full summary (the paths that read the whole row). + public static SessionIngestSnapshot From(SessionSummary session) => new SessionIngestSnapshot( + session.Status, session.CurrentPhase, session.CurrentPhaseDetail, session.FailureReason, + session.EventCount, session.DurationSeconds, session.CompletedAt, session.DiagnosticsBlobName, + session.IsPreProvisioned, session.AdminMarkedAction, session.PendingActionsJson, + session.AgentVersion, session.ImeAgentVersion); + } +}