diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/IngestEventsFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/IngestEventsFunction.cs index c0c50041..d7092ef4 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/IngestEventsFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/IngestEventsFunction.cs @@ -448,6 +448,9 @@ namespace AutopilotMonitor.Functions.Functions }).ToList() : 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 } } }; diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs index cff6f115..8880183b 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs @@ -243,7 +243,7 @@ namespace AutopilotMonitor.Functions.Services /// /// Gets all sessions for a tenant /// - public async Task> GetSessionsAsync(string tenantId, int maxResults = 100) + public async Task> 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(); - // 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( - filter: $"PartitionKey eq '{tenantId}'", + filter: filter, maxPerPage: Math.Min(maxResults, 1000) ); @@ -279,16 +285,23 @@ namespace AutopilotMonitor.Functions.Services /// /// Gets all sessions across all tenants (for galactic admin mode) /// - public async Task> GetAllSessionsAsync(int maxResults = 100) + public async Task> GetAllSessionsAsync(int maxResults = 100, DateTime? since = null) { try { var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions); var sessions = new List(); - // 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(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( + filter: filter, + maxPerPage: Math.Min(maxResults, 1000) + ); await foreach (var entity in query) { @@ -329,14 +342,19 @@ namespace AutopilotMonitor.Functions.Services } /// - /// 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. /// public async Task 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(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(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 /// /// 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. /// 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) { diff --git a/src/Backend/AutopilotMonitor.Functions/Services/UsageMetricsService.cs b/src/Backend/AutopilotMonitor.Functions/Services/UsageMetricsService.cs index 3877b2f1..5a1a67d4 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/UsageMetricsService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/UsageMetricsService.cs @@ -92,10 +92,10 @@ namespace AutopilotMonitor.Functions.Services private async Task 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 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; diff --git a/src/Web/autopilot-monitor-web/app/sessions/[sessionId]/page.tsx b/src/Web/autopilot-monitor-web/app/sessions/[sessionId]/page.tsx index b4a2455b..8d833357 100644 --- a/src/Web/autopilot-monitor-web/app/sessions/[sessionId]/page.tsx +++ b/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 }) => { + 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); } diff --git a/src/Web/autopilot-monitor-web/contexts/SignalRContext.tsx b/src/Web/autopilot-monitor-web/contexts/SignalRContext.tsx index 87836b5d..e480c40d 100644 --- a/src/Web/autopilot-monitor-web/contexts/SignalRContext.tsx +++ b/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