mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 12:42:38 +00:00
fix ghost session entries
This commit is contained in:
@@ -1 +1 @@
|
||||
500
|
||||
502
|
||||
|
||||
@@ -89,6 +89,11 @@ namespace AutopilotMonitor.Functions.Services
|
||||
// Safety net: backfill any sessions missing from SessionsIndex
|
||||
await _storageService.BackfillSessionIndexAsync();
|
||||
|
||||
// One-time cleanup: remove ghost SessionsIndex entries caused by the
|
||||
// StoreSessionAsync Replace-mode IndexRowKey bug (now fixed).
|
||||
// TODO: Remove after 2026-06-01
|
||||
await _storageService.CleanupGhostSessionIndexEntriesAsync();
|
||||
|
||||
// Backfill OnboardedAt for tenants that don't have it yet
|
||||
await BackfillTenantOnboardedAtAsync();
|
||||
|
||||
|
||||
@@ -466,6 +466,83 @@ namespace AutopilotMonitor.Functions.Services
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One-time cleanup: removes ghost entries from SessionsIndex where a SessionId has
|
||||
/// multiple index rows (caused by a bug where StoreSessionAsync's Replace mode deleted
|
||||
/// IndexRowKey, preventing old index entries from being cleaned up when StartedAt shifted).
|
||||
/// Fixed in the same release — this cleanup can be removed once all environments have run it.
|
||||
/// TODO: Remove after 2026-06-01 (3 months grace period).
|
||||
/// </summary>
|
||||
public async Task<int> CleanupGhostSessionIndexEntriesAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
var indexTable = _tableServiceClient.GetTableClient(Constants.TableNames.SessionsIndex);
|
||||
var sessionsTable = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions);
|
||||
|
||||
// Build a map of SessionId → list of index RowKeys (across all tenants)
|
||||
var sessionIndexEntries = new Dictionary<string, List<(string PartitionKey, string RowKey, int EventCount)>>();
|
||||
|
||||
var query = indexTable.QueryAsync<TableEntity>(
|
||||
select: new[] { "PartitionKey", "RowKey", "SessionId", "EventCount" });
|
||||
|
||||
await foreach (var entity in query)
|
||||
{
|
||||
var sessionId = entity.GetString("SessionId") ?? ExtractSessionIdFromIndexRowKey(entity.RowKey);
|
||||
var key = $"{entity.PartitionKey}_{sessionId}";
|
||||
var eventCount = entity.GetInt32("EventCount") ?? 0;
|
||||
|
||||
if (!sessionIndexEntries.ContainsKey(key))
|
||||
sessionIndexEntries[key] = new List<(string, string, int)>();
|
||||
|
||||
sessionIndexEntries[key].Add((entity.PartitionKey, entity.RowKey, eventCount));
|
||||
}
|
||||
|
||||
int deletedCount = 0;
|
||||
|
||||
foreach (var (key, entries) in sessionIndexEntries)
|
||||
{
|
||||
if (entries.Count <= 1)
|
||||
continue; // No duplicates
|
||||
|
||||
// Keep the entry with the highest EventCount (most up-to-date);
|
||||
// delete all others as ghosts.
|
||||
var sorted = entries.OrderByDescending(e => e.EventCount).ToList();
|
||||
var keep = sorted[0];
|
||||
|
||||
for (int i = 1; i < sorted.Count; i++)
|
||||
{
|
||||
var ghost = sorted[i];
|
||||
try
|
||||
{
|
||||
await indexTable.DeleteEntityAsync(ghost.PartitionKey, ghost.RowKey);
|
||||
deletedCount++;
|
||||
_logger.LogInformation(
|
||||
"Deleted ghost SessionsIndex entry: {PartitionKey}/{RowKey} (EventCount={GhostCount}, kept entry has EventCount={KeptCount})",
|
||||
ghost.PartitionKey, ghost.RowKey, ghost.EventCount, keep.EventCount);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to delete ghost index entry {PartitionKey}/{RowKey}",
|
||||
ghost.PartitionKey, ghost.RowKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (deletedCount > 0)
|
||||
{
|
||||
_logger.LogInformation("Ghost SessionsIndex cleanup completed: {Count} ghost entries removed", deletedCount);
|
||||
}
|
||||
|
||||
return deletedCount;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Ghost SessionsIndex cleanup failed");
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Checks if the SessionsIndex table is empty.
|
||||
/// Used by startup backfill to determine if a full migration is needed.
|
||||
|
||||
@@ -294,6 +294,7 @@ namespace AutopilotMonitor.Functions.Services
|
||||
string geoRegion = string.Empty;
|
||||
string geoCity = string.Empty;
|
||||
string geoLoc = string.Empty;
|
||||
string? existingIndexRowKey = null;
|
||||
|
||||
try
|
||||
{
|
||||
@@ -323,6 +324,7 @@ namespace AutopilotMonitor.Functions.Services
|
||||
geoRegion = existingEntity.GetString("GeoRegion") ?? string.Empty;
|
||||
geoCity = existingEntity.GetString("GeoCity") ?? string.Empty;
|
||||
geoLoc = existingEntity.GetString("GeoLoc") ?? string.Empty;
|
||||
existingIndexRowKey = existingEntity.GetString("IndexRowKey");
|
||||
|
||||
// Guard: never regress a terminal status (Succeeded/Failed) back to InProgress.
|
||||
// StoreSessionAsync uses UpsertEntity (Replace mode) which overwrites all fields.
|
||||
@@ -407,6 +409,12 @@ namespace AutopilotMonitor.Functions.Services
|
||||
if (!string.IsNullOrEmpty(geoLoc))
|
||||
entity["GeoLoc"] = geoLoc;
|
||||
|
||||
// Preserve IndexRowKey through the Replace so that:
|
||||
// 1. Concurrent UpdateSessionStatusAsync calls can find it (prevents index stale-status)
|
||||
// 2. UpsertSessionIndexAsync can detect and delete old index entries when startedAt shifts
|
||||
if (!string.IsNullOrEmpty(existingIndexRowKey))
|
||||
entity["IndexRowKey"] = existingIndexRowKey;
|
||||
|
||||
await tableClient.UpsertEntityAsync(entity);
|
||||
|
||||
// Dual-write: upsert into SessionsIndex for time-sorted listing
|
||||
@@ -933,6 +941,19 @@ namespace AutopilotMonitor.Functions.Services
|
||||
|
||||
// Dual-write: merge the same fields into SessionsIndex
|
||||
var indexRowKey = session.GetString("IndexRowKey");
|
||||
|
||||
// Fallback: if IndexRowKey was lost (e.g. StoreSessionAsync Replace race),
|
||||
// compute it from StartedAt + SessionId so the index still gets updated.
|
||||
if (string.IsNullOrEmpty(indexRowKey))
|
||||
{
|
||||
var sessionStartedAt = session.GetDateTimeOffset("StartedAt")?.UtcDateTime;
|
||||
if (sessionStartedAt.HasValue)
|
||||
{
|
||||
indexRowKey = ComputeIndexRowKey(sessionStartedAt.Value, sessionId);
|
||||
_logger.LogWarning("Session {SessionId}: IndexRowKey was null, computed fallback from StartedAt", sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
await MergeSessionIndexAsync(tenantId, indexRowKey, update);
|
||||
|
||||
_logger.LogInformation($"Updated session {sessionId} status to {status}");
|
||||
@@ -1041,6 +1062,15 @@ namespace AutopilotMonitor.Functions.Services
|
||||
|
||||
// Dual-write: merge the same fields into SessionsIndex
|
||||
var forceIndexRowKey = freshSession.GetString("IndexRowKey");
|
||||
if (string.IsNullOrEmpty(forceIndexRowKey))
|
||||
{
|
||||
var forceStartedAt = freshSession.GetDateTimeOffset("StartedAt")?.UtcDateTime;
|
||||
if (forceStartedAt.HasValue)
|
||||
{
|
||||
forceIndexRowKey = ComputeIndexRowKey(forceStartedAt.Value, sessionId);
|
||||
_logger.LogWarning("Session {SessionId}: IndexRowKey was null in force-write path, computed fallback from StartedAt", sessionId);
|
||||
}
|
||||
}
|
||||
await MergeSessionIndexAsync(tenantId, forceIndexRowKey, forceUpdate);
|
||||
|
||||
_logger.LogInformation($"Force-updated session {sessionId} status to {status} (unconditional merge after ETag exhaustion)");
|
||||
@@ -1137,6 +1167,12 @@ namespace AutopilotMonitor.Functions.Services
|
||||
|
||||
// Dual-write: merge the same fields into SessionsIndex
|
||||
var indexRowKey = entity.Value.GetString("IndexRowKey");
|
||||
if (string.IsNullOrEmpty(indexRowKey))
|
||||
{
|
||||
var incStartedAt = entity.Value.GetDateTimeOffset("StartedAt")?.UtcDateTime;
|
||||
if (incStartedAt.HasValue)
|
||||
indexRowKey = ComputeIndexRowKey(incStartedAt.Value, sessionId);
|
||||
}
|
||||
await MergeSessionIndexAsync(tenantId, indexRowKey, update);
|
||||
|
||||
return;
|
||||
|
||||
Reference in New Issue
Block a user