perf: eliminate redundant table scans, improve concurrency, optimize SignalR

- Remove full event-count recount (partition scan) from UpdateSessionStatusAsync;
  IncrementSessionEventCountAsync already maintains the count atomically
- Remove redundant GetEarliestSessionEventTimestampAsync calls from
  UpdateSessionStatusAsync and IncrementSessionEventCountAsync; the caller
  (IngestEventsFunction) already computes and passes the timestamp
- Switch UpdateSessionStatusAsync from Replace to Merge mode, writing only
  changed fields to drastically reduce ETag conflicts under concurrency
- Increase ETag retry limit from 3 to 5 with jitter-based exponential backoff
- Add server-side time-range filter (90 days) to UsageMetrics queries,
  avoiding loading the entire sessions table into memory
- Remove full session object from SignalR eventStream message; session delta
  updates now flow via the existing newevents tenant-group message
- Add newevents handler to session detail page for live session delta updates
- Auto-rejoin SignalR groups on reconnect to prevent silent update loss

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/Backend/AutopilotMonitor.Functions/Functions/IngestEventsFunction.cs

@@ -448,6 +448,9 @@ namespace AutopilotMonitor.Functions.Functions
                     }).ToList<object>()
                     : null;
 
+                // Signal-only pattern: no full session object — the detail page already receives
+                // the session delta via the "newevents" message (tenant group) and fetches fresh
+                // events from Table Storage on receipt. Sending the full session here was redundant.
                 var eventsMessage = new SignalRMessageAction("eventStream")
                 {
                     GroupName = $"session-{request.TenantId}-{request.SessionId}",
@@ -455,7 +458,6 @@ namespace AutopilotMonitor.Functions.Functions
                         sessionId = request.SessionId,
                         tenantId = request.TenantId,
                         newEventCount = processedCount,
-                        session = updatedSession,
                         newRuleResults = slimRuleResults
                     } }
                 };

src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs

@@ -243,7 +243,7 @@ namespace AutopilotMonitor.Functions.Services
         /// <summary>
         /// Gets all sessions for a tenant
         /// </summary>
-        public async Task<List<SessionSummary>> GetSessionsAsync(string tenantId, int maxResults = 100)
+        public async Task<List<SessionSummary>> GetSessionsAsync(string tenantId, int maxResults = 100, DateTime? since = null)
         {
             SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
 
@@ -252,10 +252,16 @@ namespace AutopilotMonitor.Functions.Services
                 var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions);
                 var sessions = new List<SessionSummary>();
 
-                // Query sessions by tenant (PartitionKey)
+                // Query sessions by tenant (PartitionKey), optionally filtered by StartedAt
                 // maxPerPage capped at 1000 (Azure Table Storage limit), pagination is automatic
+                var filter = $"PartitionKey eq '{tenantId}'";
+                if (since.HasValue)
+                {
+                    filter += $" and StartedAt ge datetime'{since.Value:yyyy-MM-ddTHH:mm:ss.fffffffZ}'";
+                }
+
                 var query = tableClient.QueryAsync<TableEntity>(
-                    filter: $"PartitionKey eq '{tenantId}'",
+                    filter: filter,
                     maxPerPage: Math.Min(maxResults, 1000)
                 );
 
@@ -279,16 +285,23 @@ namespace AutopilotMonitor.Functions.Services
         /// <summary>
         /// Gets all sessions across all tenants (for galactic admin mode)
         /// </summary>
-        public async Task<List<SessionSummary>> GetAllSessionsAsync(int maxResults = 100)
+        public async Task<List<SessionSummary>> GetAllSessionsAsync(int maxResults = 100, DateTime? since = null)
         {
             try
             {
                 var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions);
                 var sessions = new List<SessionSummary>();
 
-                // Query all sessions without tenant filter
+                // Query all sessions, optionally filtered by StartedAt to limit dataset size
                 // maxPerPage is capped at 1000 (Azure Table Storage limit); pagination is automatic
-                var query = tableClient.QueryAsync<TableEntity>(maxPerPage: Math.Min(maxResults, 1000));
+                string? filter = since.HasValue
+                    ? $"StartedAt ge datetime'{since.Value:yyyy-MM-ddTHH:mm:ss.fffffffZ}'"
+                    : null;
+
+                var query = tableClient.QueryAsync<TableEntity>(
+                    filter: filter,
+                    maxPerPage: Math.Min(maxResults, 1000)
+                );
 
                 await foreach (var entity in query)
                 {
@@ -329,14 +342,19 @@ namespace AutopilotMonitor.Functions.Services
         }
 
         /// <summary>
-        /// Updates the session status and current phase
+        /// Updates the session status and current phase.
+        /// Uses Merge mode to write only changed fields, reducing ETag conflicts under concurrency.
+        /// The caller (IngestEventsFunction) provides earliestEventTimestamp from the current batch;
+        /// no redundant Events-table scan is performed here.
+        /// Event count is maintained atomically by IncrementSessionEventCountAsync and is not
+        /// recounted here — avoiding an expensive full-partition scan on every status change.
         /// </summary>
         public async Task<bool> UpdateSessionStatusAsync(string tenantId, string sessionId, SessionStatus status, EnrollmentPhase? currentPhase = null, string? failureReason = null, DateTime? completedAt = null, DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null)
         {
             SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
             SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
 
-            const int maxRetries = 3;
+            const int maxRetries = 5;
             int retryCount = 0;
 
             while (retryCount < maxRetries)
@@ -345,7 +363,7 @@ namespace AutopilotMonitor.Functions.Services
                 {
                     var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions);
 
-                    // Get the existing entity
+                    // Read the existing entity to check idempotency guards and compute derived fields
                     var entity = await tableClient.GetEntityAsync<TableEntity>(tenantId, sessionId);
                     var session = entity.Value;
 
@@ -361,6 +379,9 @@ namespace AutopilotMonitor.Functions.Services
                         }
                     }
 
+                    // Build a Merge update with only the fields that actually change
+                    var update = new TableEntity(tenantId, sessionId);
+
                     // Status promotion rule: a Pending session (WhiteGlove pre-provisioning complete)
                     // must not regress to InProgress via phase-change events from the resumed boot.
                     // The Pending → InProgress transition happens exclusively in StoreSessionAsync
@@ -368,35 +389,24 @@ namespace AutopilotMonitor.Functions.Services
                     if (status == SessionStatus.InProgress && existingStatusStr == SessionStatus.Pending.ToString())
                     {
                         _logger.LogInformation($"Session {sessionId} is Pending (WhiteGlove), preserving status (InProgress blocked by promotion rule)");
-                        // Phase, timestamps, and event count are still updated below.
+                        // Phase, timestamps are still updated below.
                     }
                     else
                     {
-                        // Update status
-                        session["Status"] = status.ToString();
+                        update["Status"] = status.ToString();
                     }
 
                     // Update current phase if provided
                     if (currentPhase.HasValue)
                     {
-                        session["CurrentPhase"] = (int)currentPhase.Value;
+                        update["CurrentPhase"] = (int)currentPhase.Value;
                     }
 
+                    // Align StartedAt with the earliest event timestamp provided by the caller
                     var currentStartedAt = session.GetDateTimeOffset("StartedAt")?.UtcDateTime ?? DateTime.MaxValue;
-                    var effectiveEarliest = earliestEventTimestamp;
-
-                    // Self-heal StartedAt by also looking at persisted events in case an older event
-                    // arrived in a previous request/batch.
-                    var earliestStoredEventTimestamp = await GetEarliestSessionEventTimestampAsync(tenantId, sessionId);
-                    if (earliestStoredEventTimestamp.HasValue &&
-                        (!effectiveEarliest.HasValue || earliestStoredEventTimestamp.Value < effectiveEarliest.Value))
+                    if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < currentStartedAt)
                     {
-                        effectiveEarliest = earliestStoredEventTimestamp.Value;
-                    }
-
-                    if (effectiveEarliest.HasValue && effectiveEarliest.Value < currentStartedAt)
-                    {
-                        session["StartedAt"] = effectiveEarliest.Value;
+                        update["StartedAt"] = earliestEventTimestamp.Value;
                     }
 
                     // Set completion time if succeeded or failed
@@ -404,14 +414,14 @@ namespace AutopilotMonitor.Functions.Services
                     {
                         // Use the provided completedAt timestamp (from event) if available, otherwise use current time
                         var effectiveCompletedAt = completedAt ?? DateTime.UtcNow;
-                        session["CompletedAt"] = effectiveCompletedAt;
+                        update["CompletedAt"] = effectiveCompletedAt;
 
                         // Store accurate DurationSeconds based on first event timestamp (not registration StartedAt)
                         // The agent registers before events arrive, so registration StartedAt is always earlier
                         // than the actual enrollment start — using the first event gives the true enrollment duration.
-                        var durationStart = effectiveEarliest ?? currentStartedAt;
+                        var durationStart = earliestEventTimestamp ?? currentStartedAt;
                         if (durationStart < effectiveCompletedAt)
-                            session["DurationSeconds"] = (int)(effectiveCompletedAt - durationStart).TotalSeconds;
+                            update["DurationSeconds"] = (int)(effectiveCompletedAt - durationStart).TotalSeconds;
                     }
 
                     // Track the most recent event timestamp for excessive data sender detection
@@ -419,28 +429,18 @@ namespace AutopilotMonitor.Functions.Services
                     {
                         var currentLastEventAt = session.GetDateTimeOffset("LastEventAt")?.UtcDateTime;
                         if (!currentLastEventAt.HasValue || latestEventTimestamp.Value > currentLastEventAt.Value)
-                            session["LastEventAt"] = latestEventTimestamp.Value;
+                            update["LastEventAt"] = latestEventTimestamp.Value;
                     }
 
                     // Set failure reason if failed
                     if (status == SessionStatus.Failed && !string.IsNullOrEmpty(failureReason))
                     {
-                        session["FailureReason"] = failureReason;
+                        update["FailureReason"] = failureReason;
                     }
 
-                    // Update event count (full recount for accuracy on status changes)
-                    var eventsTableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Events);
-                    var partitionKey = $"{tenantId}_{sessionId}";
-                    var eventCountFilter = $"PartitionKey eq '{partitionKey}'";
-                    var eventCount = 0;
-                    await foreach (var _ in eventsTableClient.QueryAsync<TableEntity>(eventCountFilter, select: new[] { "RowKey" }))
-                    {
-                        eventCount++;
-                    }
-                    session["EventCount"] = eventCount;
-
-                    // Update entity with optimistic concurrency
-                    await tableClient.UpdateEntityAsync(session, session.ETag, TableUpdateMode.Replace);
+                    // Merge mode: only the fields set above are written; all other fields remain untouched.
+                    // This drastically reduces ETag conflicts when concurrent requests update different fields.
+                    await tableClient.UpdateEntityAsync(update, session.ETag, TableUpdateMode.Merge);
 
                     _logger.LogInformation($"Updated session {sessionId} status to {status}");
                     return true;
@@ -454,8 +454,9 @@ namespace AutopilotMonitor.Functions.Services
                         return false;
                     }
 
-                    // Exponential backoff: 50ms, 100ms, 200ms
-                    await Task.Delay(50 * (int)Math.Pow(2, retryCount - 1));
+                    // Exponential backoff with jitter to decorrelate concurrent retries
+                    var baseDelay = 50 * (int)Math.Pow(2, retryCount - 1);
+                    await Task.Delay(baseDelay + Random.Shared.Next(0, baseDelay));
                     _logger.LogDebug($"Retrying session {sessionId} update (attempt {retryCount}/{maxRetries})");
                 }
                 catch (Exception ex)
@@ -471,13 +472,15 @@ namespace AutopilotMonitor.Functions.Services
         /// <summary>
         /// Increments the session event count without touching status or phase fields.
         /// Uses Merge mode to safely handle concurrent updates.
+        /// The caller provides earliestEventTimestamp from the current batch;
+        /// no redundant Events-table scan is performed here.
         /// </summary>
         public async Task IncrementSessionEventCountAsync(string tenantId, string sessionId, int increment, DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null)
         {
             SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
             SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
 
-            const int maxRetries = 3;
+            const int maxRetries = 5;
             int retryCount = 0;
 
             while (retryCount < maxRetries)
@@ -493,20 +496,11 @@ namespace AutopilotMonitor.Functions.Services
                         ["EventCount"] = currentCount + increment
                     };
 
+                    // Align StartedAt with the earliest event timestamp provided by the caller
                     var currentStartedAt = entity.Value.GetDateTimeOffset("StartedAt")?.UtcDateTime ?? DateTime.MaxValue;
-                    var effectiveEarliest = earliestEventTimestamp;
-
-                    // Self-heal StartedAt by checking already persisted events for this session.
-                    var earliestStoredEventTimestamp = await GetEarliestSessionEventTimestampAsync(tenantId, sessionId);
-                    if (earliestStoredEventTimestamp.HasValue &&
-                        (!effectiveEarliest.HasValue || earliestStoredEventTimestamp.Value < effectiveEarliest.Value))
+                    if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < currentStartedAt)
                     {
-                        effectiveEarliest = earliestStoredEventTimestamp.Value;
-                    }
-
-                    if (effectiveEarliest.HasValue && effectiveEarliest.Value < currentStartedAt)
-                    {
-                        update["StartedAt"] = effectiveEarliest.Value;
+                        update["StartedAt"] = earliestEventTimestamp.Value;
                     }
 
                     // Track the most recent event timestamp for excessive data sender detection
@@ -528,7 +522,9 @@ namespace AutopilotMonitor.Functions.Services
                         _logger.LogWarning($"Failed to increment event count for session {sessionId} after {maxRetries} retries due to ETag conflicts");
                         return;
                     }
-                    await Task.Delay(50 * (int)Math.Pow(2, retryCount - 1));
+                    // Exponential backoff with jitter to decorrelate concurrent retries
+                    var baseDelay = 50 * (int)Math.Pow(2, retryCount - 1);
+                    await Task.Delay(baseDelay + Random.Shared.Next(0, baseDelay));
                 }
                 catch (Exception ex)
                 {

src/Backend/AutopilotMonitor.Functions/Services/UsageMetricsService.cs

@@ -92,10 +92,10 @@ namespace AutopilotMonitor.Functions.Services
 
         private async Task<PlatformUsageMetrics> ComputeUsageMetricsInternalAsync()
         {
-            // Query all sessions (galactic admin only!)
-            // Note: This queries ALL sessions across ALL tenants - can handle millions of records
-            // Since this is on-demand only, we can afford to process large datasets
-            var allSessions = await _storageService.GetAllSessionsAsync(maxResults: 1000000);
+            // Query sessions from the last 90 days only — covers all displayed metrics (today/7d/30d)
+            // with a generous margin. Avoids loading the entire table into memory.
+            // "Total" counters use PlatformStats (cumulative, already tracked separately).
+            var allSessions = await _storageService.GetAllSessionsAsync(maxResults: 1000000, since: DateTime.UtcNow.AddDays(-90));
 
             var now = DateTime.UtcNow;
             var today = now.Date;
@@ -210,8 +210,8 @@ namespace AutopilotMonitor.Functions.Services
 
         private async Task<PlatformUsageMetrics> ComputeTenantUsageMetricsInternalAsync(string tenantId)
         {
-            // Query sessions for specific tenant only
-            var tenantSessions = await _storageService.GetSessionsAsync(tenantId, maxResults: 1000000);
+            // Query sessions for specific tenant only — last 90 days
+            var tenantSessions = await _storageService.GetSessionsAsync(tenantId, maxResults: 1000000, since: DateTime.UtcNow.AddDays(-90));
 
             var now = DateTime.UtcNow;
             var today = now.Date;

src/Web/autopilot-monitor-web/app/sessions/[sessionId]/page.tsx

@@ -225,19 +225,17 @@ export default function SessionDetailPage() {
   useEffect(() => {
     // Listen for event stream signal — backend sends a lightweight signal (no event payload).
     // Frontend fetches fresh events from Table Storage on receipt: canonical truth, no gaps.
-    const handleEventStream = (data: { sessionId: string; tenantId: string; newEventCount: number; session?: Session; newRuleResults?: RuleResult[] }) => {
+    const handleEventStream = (data: { sessionId: string; tenantId: string; newEventCount: number; newRuleResults?: RuleResult[] }) => {
       console.log('Event stream signal received via SignalR:', data);
       if (data.sessionId !== sessionIdRef.current) return;
 
       // Fetch full events from storage (single source of truth), but debounce bursts.
+      // Session updates arrive via the "newevents" message (tenant group) — no session
+      // object in this signal to keep payloads minimal.
       scheduleFetchEvents();
 
-      // Session update from SignalR (no extra fetch needed)
-      if (data.session) {
-        setSession(data.session);
-        if (data.session.tenantId) {
-          setSessionTenantId(prev => prev || data.session!.tenantId);
-        }
+      if (data.tenantId) {
+        setSessionTenantId(prev => prev || data.tenantId);
       }
 
       // Rule results from SignalR (only on enrollment completion)
@@ -246,10 +244,25 @@ export default function SessionDetailPage() {
       }
     };
 
+    // Listen for session delta updates via the tenant group ("newevents").
+    // This replaces the full session object that was previously sent inside "eventStream".
+    const handleNewEvents = (data: { sessionId: string; tenantId: string; sessionUpdate?: Partial<Session> }) => {
+      if (data.sessionId !== sessionIdRef.current) return;
+
+      if (data.sessionUpdate) {
+        setSession(prev => prev ? { ...prev, ...data.sessionUpdate } : prev);
+      }
+      if (data.tenantId) {
+        setSessionTenantId(prev => prev || data.tenantId);
+      }
+    };
+
     on('eventStream', handleEventStream);
+    on('newevents', handleNewEvents);
 
     return () => {
       off('eventStream', handleEventStream);
+      off('newevents', handleNewEvents);
       if (eventRefreshTimeoutRef.current) {
         clearTimeout(eventRefreshTimeoutRef.current);
       }

src/Web/autopilot-monitor-web/contexts/SignalRContext.tsx

@@ -70,10 +70,40 @@ export function SignalRProvider({ children }: { children: React.ReactNode }) {
       setConnectionState(signalR.HubConnectionState.Reconnecting);
     });
 
-    newConnection.onreconnected((connectionId) => {
+    newConnection.onreconnected(async (connectionId) => {
       setConnectionState(signalR.HubConnectionState.Connected);
       retryCountRef.current = 0; // Reset retry count on successful reconnect
-      joinedGroupsRef.current.clear(); // Clear joined groups on reconnect - need to re-join
+
+      // Auto-rejoin all previously joined groups after reconnect.
+      // The server-side connection ID changed, so all group memberships are lost.
+      const previousGroups = Array.from(joinedGroupsRef.current);
+      joinedGroupsRef.current.clear();
+
+      for (const groupName of previousGroups) {
+        try {
+          const token = await getAccessToken();
+          const response = await fetch(`${API_BASE_URL}/api/realtime/groups/join`, {
+            method: 'POST',
+            headers: {
+              'Content-Type': 'application/json',
+              'Authorization': `Bearer ${token}`
+            },
+            body: JSON.stringify({ connectionId, groupName })
+          });
+
+          if (response.ok) {
+            joinedGroupsRef.current.add(groupName);
+          } else {
+            console.warn(`[SignalR] Failed to rejoin group ${groupName} after reconnect (status ${response.status})`);
+          }
+        } catch (error) {
+          console.warn(`[SignalR] Error rejoining group ${groupName} after reconnect:`, error);
+        }
+      }
+
+      if (previousGroups.length > 0) {
+        console.log(`[SignalR] Rejoined ${joinedGroupsRef.current.size}/${previousGroups.length} groups after reconnect`);
+      }
     });
 
     // Start connection