diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/AppInstallDoFieldsFallbackTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/AppInstallDoFieldsFallbackTests.cs new file mode 100644 index 00000000..699ece15 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/AppInstallDoFieldsFallbackTests.cs @@ -0,0 +1,162 @@ +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared.Models; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// DO ingest fallback (IME ≥ 1.104 removed the [DO TEL] log line; the agent's DO collector +/// only emits do_telemetry for downloads whose completion a poll itself observed). +/// download_progress events carry the full do* set on every poll and must fold +/// into the summary's Do* columns so interrupted / between-poll-completed downloads still +/// reach the Delivery Optimization aggregates: +/// - progress-only sessions populate Do* incl. a valid DoDownloadMode (DoAggregator gate), +/// - a later do_telemetry stays authoritative, +/// - replayed progress events are idempotent (Math.Max), +/// - DownloadBytes prefers transferred bytes over file size. +/// +public class AppInstallDoFieldsFallbackTests +{ + private static readonly DateTime T0 = new(2026, 8, 19, 10, 0, 0, DateTimeKind.Utc); + + private static Dictionary Aggregate(params EnrollmentEvent[] events) + { + var summaries = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var evt in events) + EventIngestProcessor.AggregateAppInstallEvent(evt, "tenant", "session", summaries); + return summaries; + } + + private static EnrollmentEvent DownloadProgress( + long bytesDownloaded, long doTotal, long doPeers, long doHttp, int doMode, + long doFileSize = 0, int doPct = 0, long doCacheServer = 0, DateTime? ts = null) + => new EnrollmentEvent + { + EventType = "download_progress", + Timestamp = ts ?? T0, + Data = new Dictionary + { + ["appName"] = "App", + ["bytesDownloaded"] = bytesDownloaded, + ["doFileSize"] = doFileSize, + ["doTotalBytesDownloaded"] = doTotal, + ["doBytesFromPeers"] = doPeers, + ["doBytesFromHttp"] = doHttp, + ["doPercentPeerCaching"] = doPct, + ["doDownloadMode"] = doMode, + ["doBytesFromCacheServer"] = doCacheServer, + }, + }; + + private static EnrollmentEvent DoTelemetry( + long doFileSize, long doTotal, long doPeers, long doHttp, int doMode, + string? duration = null) + { + var data = new Dictionary + { + ["appName"] = "App", + ["doFileSize"] = doFileSize, + ["doTotalBytesDownloaded"] = doTotal, + ["doBytesFromPeers"] = doPeers, + ["doBytesFromHttp"] = doHttp, + ["doPercentPeerCaching"] = 50, + ["doDownloadMode"] = doMode, + }; + if (duration != null) data["doDownloadDuration"] = duration; + return new EnrollmentEvent { EventType = "do_telemetry", Timestamp = T0, Data = data }; + } + + [Fact] + public void ProgressOnly_PopulatesDoColumns_AndSurvivesDoAggregator() + { + // An interrupted download: progress events observed, do_telemetry never fired. + var s = Aggregate( + DownloadProgress(bytesDownloaded: 1000, doTotal: 1000, doPeers: 600, doHttp: 400, doMode: 1, doFileSize: 5000, doPct: 60), + DownloadProgress(bytesDownloaded: 3000, doTotal: 3000, doPeers: 2000, doHttp: 1000, doMode: 1, doFileSize: 5000, doPct: 66) + )["App"].Summary; + + Assert.Equal(3000, s.DoTotalBytesDownloaded); + Assert.Equal(2000, s.DoBytesFromPeers); + Assert.Equal(1000, s.DoBytesFromHttp); + Assert.Equal(5000, s.DoFileSize); + Assert.Equal(66, s.DoPercentPeerCaching); + Assert.Equal(1, s.DoDownloadMode); + Assert.Equal(3000, s.DownloadBytes); // from bytesDownloaded, not the file size + + // The whole point: DoDownloadMode >= 0 keeps the row inside the DO aggregate. + var aggregate = DoAggregator.Compute(new[] { s }); + Assert.Equal(1, aggregate.DoAppCount); + Assert.Equal(2000, aggregate.BytesFromPeers); + } + + [Fact] + public void ProgressReplay_IsIdempotent_MaxFoldNeverRegresses() + { + // Out-of-order / replayed batch: the older (smaller) progress arrives again last. + var s = Aggregate( + DownloadProgress(bytesDownloaded: 3000, doTotal: 3000, doPeers: 2000, doHttp: 1000, doMode: 1), + DownloadProgress(bytesDownloaded: 1000, doTotal: 1000, doPeers: 600, doHttp: 400, doMode: 1) + )["App"].Summary; + + Assert.Equal(3000, s.DoTotalBytesDownloaded); + Assert.Equal(2000, s.DoBytesFromPeers); + Assert.Equal(1000, s.DoBytesFromHttp); + } + + [Fact] + public void DoTelemetry_StaysAuthoritative_OverEarlierProgress() + { + // Progress saw a partial state; the final telemetry read has the complete numbers + // (which can be LOWER per bucket than a transient progress max — telemetry wins). + var s = Aggregate( + DownloadProgress(bytesDownloaded: 4000, doTotal: 4000, doPeers: 3500, doHttp: 500, doMode: 1, doPct: 87), + DoTelemetry(doFileSize: 5000, doTotal: 5000, doPeers: 3000, doHttp: 2000, doMode: 1, duration: "00:01:40") + )["App"].Summary; + + Assert.Equal(5000, s.DoTotalBytesDownloaded); + Assert.Equal(3000, s.DoBytesFromPeers); // telemetry overwrote the higher progress value + Assert.Equal(2000, s.DoBytesFromHttp); + Assert.Equal(50, s.DoPercentPeerCaching); + Assert.Equal("00:01:40", s.DoDownloadDuration); + Assert.Equal(100, s.DownloadDurationSeconds); + } + + [Fact] + public void DoTelemetry_WithUnsetMode_DoesNotRegressAKnownMode() + { + // A telemetry event whose DownloadMode property was absent parses as -1; it must not + // re-hide the row from DoAggregator after progress already established the mode. + var s = Aggregate( + DownloadProgress(bytesDownloaded: 1000, doTotal: 1000, doPeers: 600, doHttp: 400, doMode: 3), + DoTelemetry(doFileSize: 1000, doTotal: 1000, doPeers: 600, doHttp: 400, doMode: -1) + )["App"].Summary; + + Assert.Equal(3, s.DoDownloadMode); + } + + [Fact] + public void DownloadBytes_PrefersTransferredBytes_OverFileSize() + { + // Historic bug: DownloadBytes was max-ed with doFileSize, inflating the transfer + // measure. With a real transfer total present the file size must not win. + var s = Aggregate( + DoTelemetry(doFileSize: 9000, doTotal: 5000, doPeers: 3000, doHttp: 2000, doMode: 1) + )["App"].Summary; + + Assert.Equal(5000, s.DownloadBytes); + Assert.Equal(9000, s.DoFileSize); // the file size column itself keeps the real value + } + + [Fact] + public void DownloadBytes_FallsBackToFileSize_WhenNoTransferTotal() + { + var evt = new EnrollmentEvent + { + EventType = "do_telemetry", + Timestamp = T0, + Data = new Dictionary { ["appName"] = "App", ["doFileSize"] = 7000L, ["doDownloadMode"] = 1 }, + }; + var s = Aggregate(evt)["App"].Summary; + Assert.Equal(7000, s.DownloadBytes); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.AppInstall.cs b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.AppInstall.cs index 92889f6a..7f78fbd3 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.AppInstall.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.AppInstall.cs @@ -184,52 +184,109 @@ namespace AutopilotMonitor.Functions.Services : evt.Data?.ContainsKey("bytes_downloaded") == true ? "bytes_downloaded" : null; if (bytesKey != null && long.TryParse(evt.Data![bytesKey]?.ToString(), out var bytes)) summary.DownloadBytes = Math.Max(summary.DownloadBytes, bytes); + // DO fallback: IME >= 1.104 removed the [DO TEL] log line, and the agent's + // DO collector only emits do_telemetry for downloads whose completion a poll + // itself observed — fast, interrupted or between-poll-completed downloads + // never get one. The progress events carry the full do* set on every poll, + // so fold them in monotonically; a do_telemetry (below) stays authoritative. + if (evt.Data != null) + ApplyDoFields(summary, evt.Data, isAuthoritative: false); break; case "do_telemetry": if (evt.Data != null) - { - if (evt.Data.ContainsKey("doFileSize") && long.TryParse(evt.Data["doFileSize"]?.ToString(), out var doFs)) - { - summary.DownloadBytes = Math.Max(summary.DownloadBytes, doFs); - summary.DoFileSize = doFs; - } - if (evt.Data.ContainsKey("doTotalBytesDownloaded") && long.TryParse(evt.Data["doTotalBytesDownloaded"]?.ToString(), out var doTotalDl)) - summary.DoTotalBytesDownloaded = doTotalDl; - if (evt.Data.ContainsKey("doBytesFromPeers") && long.TryParse(evt.Data["doBytesFromPeers"]?.ToString(), out var doPeers)) - summary.DoBytesFromPeers = doPeers; - if (evt.Data.ContainsKey("doBytesFromHttp") && long.TryParse(evt.Data["doBytesFromHttp"]?.ToString(), out var doHttp)) - summary.DoBytesFromHttp = doHttp; - if (evt.Data.ContainsKey("doPercentPeerCaching") && int.TryParse(evt.Data["doPercentPeerCaching"]?.ToString(), out var doPct)) - summary.DoPercentPeerCaching = doPct; - if (evt.Data.ContainsKey("doDownloadMode") && int.TryParse(evt.Data["doDownloadMode"]?.ToString(), out var doMode)) - summary.DoDownloadMode = doMode; - if (evt.Data.ContainsKey("doDownloadDuration")) - { - var doDurStr = evt.Data["doDownloadDuration"]?.ToString() ?? string.Empty; - summary.DoDownloadDuration = doDurStr; - if (TimeSpan.TryParse(doDurStr, out var doDurTs) && doDurTs.TotalSeconds >= 1) - summary.DownloadDurationSeconds = Math.Max(summary.DownloadDurationSeconds, (int)doDurTs.TotalSeconds); - } - if (evt.Data.ContainsKey("doBytesFromLanPeers") && long.TryParse(evt.Data["doBytesFromLanPeers"]?.ToString(), out var doLan)) - summary.DoBytesFromLanPeers = doLan; - if (evt.Data.ContainsKey("doBytesFromGroupPeers") && long.TryParse(evt.Data["doBytesFromGroupPeers"]?.ToString(), out var doGroup)) - summary.DoBytesFromGroupPeers = doGroup; - if (evt.Data.ContainsKey("doBytesFromInternetPeers") && long.TryParse(evt.Data["doBytesFromInternetPeers"]?.ToString(), out var doInet)) - summary.DoBytesFromInternetPeers = doInet; - if (evt.Data.ContainsKey("doBytesFromLinkLocalPeers") && long.TryParse(evt.Data["doBytesFromLinkLocalPeers"]?.ToString(), out var doLinkLocal)) - summary.DoBytesFromLinkLocalPeers = doLinkLocal; - if (evt.Data.ContainsKey("doBytesFromCacheServer") && long.TryParse(evt.Data["doBytesFromCacheServer"]?.ToString(), out var doCache)) - summary.DoBytesFromCacheServer = doCache; - if (evt.Data.ContainsKey("doCacheHost")) - summary.DoCacheHost = evt.Data["doCacheHost"]?.ToString() ?? string.Empty; - } + ApplyDoFields(summary, evt.Data, isAuthoritative: true); break; } RecalculateAppDurations(state); } + /// + /// Folds the do* Delivery Optimization fields of one event into the summary. + /// Two writers share this so they cannot drift: + /// + /// do_telemetry ( = true) — the collector's + /// final per-app read: last-write-wins per present key (pre-existing semantics), except + /// DoDownloadMode never regresses to the -1 "unset" sentinel — that would re-hide + /// the row from DoAggregator after a progress event already established a mode. + /// download_progress (false) — per-poll observations: byte counters and file + /// size fold via Math.Max (monotonic per download, so replays and out-of-order + /// batches stay idempotent); mode/percent/cacheHost are latest-observation writes. + /// + /// DownloadBytes prefers actually-transferred bytes (doTotalBytesDownloaded) + /// over doFileSize — the file size is only the fallback when no transfer total exists + /// (it used to unconditionally inflate the transfer measure). + /// + internal static void ApplyDoFields(AppInstallSummary summary, Dictionary data, bool isAuthoritative) + { + static bool TryGetLong(Dictionary d, string key, out long value) + { + value = 0; + return d.TryGetValue(key, out var raw) && long.TryParse(raw?.ToString(), out value); + } + static bool TryGetInt(Dictionary d, string key, out int value) + { + value = 0; + return d.TryGetValue(key, out var raw) && int.TryParse(raw?.ToString(), out value); + } + static long Fold(bool authoritative, long current, long incoming) + => authoritative ? incoming : Math.Max(current, incoming); + + var hasFileSize = TryGetLong(data, "doFileSize", out var doFs); + var hasTotalDl = TryGetLong(data, "doTotalBytesDownloaded", out var doTotalDl); + + if (hasFileSize) + summary.DoFileSize = Fold(isAuthoritative, summary.DoFileSize, doFs); + if (hasTotalDl) + summary.DoTotalBytesDownloaded = Fold(isAuthoritative, summary.DoTotalBytesDownloaded, doTotalDl); + + // Transfer measure: real transferred bytes when known, file size only as fallback. + // Progress events already fed DownloadBytes via their bytesDownloaded field, so only + // the authoritative telemetry contributes here. + if (isAuthoritative && (hasFileSize || hasTotalDl)) + summary.DownloadBytes = Math.Max(summary.DownloadBytes, hasTotalDl && doTotalDl > 0 ? doTotalDl : doFs); + + if (TryGetLong(data, "doBytesFromPeers", out var doPeers)) + summary.DoBytesFromPeers = Fold(isAuthoritative, summary.DoBytesFromPeers, doPeers); + if (TryGetLong(data, "doBytesFromHttp", out var doHttp)) + summary.DoBytesFromHttp = Fold(isAuthoritative, summary.DoBytesFromHttp, doHttp); + if (TryGetLong(data, "doBytesFromLanPeers", out var doLan)) + summary.DoBytesFromLanPeers = Fold(isAuthoritative, summary.DoBytesFromLanPeers, doLan); + if (TryGetLong(data, "doBytesFromGroupPeers", out var doGroup)) + summary.DoBytesFromGroupPeers = Fold(isAuthoritative, summary.DoBytesFromGroupPeers, doGroup); + if (TryGetLong(data, "doBytesFromInternetPeers", out var doInet)) + summary.DoBytesFromInternetPeers = Fold(isAuthoritative, summary.DoBytesFromInternetPeers, doInet); + if (TryGetLong(data, "doBytesFromLinkLocalPeers", out var doLinkLocal)) + summary.DoBytesFromLinkLocalPeers = Fold(isAuthoritative, summary.DoBytesFromLinkLocalPeers, doLinkLocal); + if (TryGetLong(data, "doBytesFromCacheServer", out var doCache)) + summary.DoBytesFromCacheServer = Fold(isAuthoritative, summary.DoBytesFromCacheServer, doCache); + + if (TryGetInt(data, "doPercentPeerCaching", out var doPct)) + summary.DoPercentPeerCaching = doPct; + + // -1 is the "unset" sentinel DoAggregator filters on — never let it clobber a + // known mode (a telemetry event without the DownloadMode property parses as -1). + if (TryGetInt(data, "doDownloadMode", out var doMode) && doMode >= 0) + summary.DoDownloadMode = doMode; + + if (data.TryGetValue("doCacheHost", out var cacheHostRaw)) + { + var cacheHost = cacheHostRaw?.ToString(); + if (isAuthoritative || !string.IsNullOrEmpty(cacheHost)) + summary.DoCacheHost = cacheHost ?? string.Empty; + } + + // Only do_telemetry carries a duration; progress events never have the key. + if (data.ContainsKey("doDownloadDuration")) + { + var doDurStr = data["doDownloadDuration"]?.ToString() ?? string.Empty; + summary.DoDownloadDuration = doDurStr; + if (TimeSpan.TryParse(doDurStr, out var doDurTs) && doDurTs.TotalSeconds >= 1) + summary.DownloadDurationSeconds = Math.Max(summary.DownloadDurationSeconds, (int)doDurTs.TotalSeconds); + } + } + internal static void RecalculateAppDurations(AppInstallAggregationState state) { var summary = state.Summary;