mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 20:53:06 +00:00
feat(agent): bandwidth estimate survives reboots via state persistence
The estimator accumulator previously lived in process memory only - a mid-DeviceSetup reboot (e.g. a Win32 app forcing a restart) silently discarded every rate sample collected so far. One enrollment commonly spans several reboots. - BandwidthStatePersistence (bandwidth-state.json in the state dir): mirrors the OfficeInstallStatePersistence pattern - atomic write via temp+rename, fail-soft load (missing/corrupt/foreign-session -> fresh start, never throws). SessionId-validated on load. - BandwidthEstimator.ExportState/ImportState: sanitizes NaN/Infinity/ negative samples, caps list sizes; baselines deliberately not persisted (a reboot gap always exceeds the per-file delta window). - Save points are event-driven, never per-poll: the dormant transition (downloads just finished - exactly when app-forced reboots happen), after the interim emission, and at collector stop. - The InterimEmitted guard rides in the same file: device_setup_end now fires once per SESSION, not once per process. Persisted AFTER the emission (StartupEventGate MarkEmitted ordering) so a crash between emit and save re-emits instead of suppressing forever. - Wiring: DefaultComponentFactory passes the state directory through DeliveryOptimizationHost; the collector param is optional (null = in-memory only). +13 tests (estimator roundtrip/merge/sanitizing, persistence roundtrip/ session-mismatch/corruption, collector-level resume incl. guard survival); agent suite 1668/1668 green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
+107
@@ -0,0 +1,107 @@
|
||||
#nullable enable
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using AutopilotMonitor.Agent.V2.Core.Logging;
|
||||
using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic;
|
||||
using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
|
||||
using Xunit;
|
||||
|
||||
namespace AutopilotMonitor.Agent.V2.Core.Tests.Persistence
|
||||
{
|
||||
/// <summary>
|
||||
/// BandwidthStatePersistence — the reboot-survivor store for the passive bandwidth
|
||||
/// estimator. Covers the save/load roundtrip (a new instance over the same state
|
||||
/// directory models an agent restart), the session-id ownership check, and fail-soft
|
||||
/// behavior on missing/corrupt files. The estimator-side import semantics (sanitizing,
|
||||
/// merge) live in BandwidthEstimatorTests.
|
||||
/// </summary>
|
||||
public sealed class BandwidthStatePersistenceTests
|
||||
{
|
||||
private static AgentLogger NewLogger(string dir) => new AgentLogger(dir, AgentLogLevel.Info);
|
||||
|
||||
private static BandwidthStateData SampleState(string sessionId) => new BandwidthStateData
|
||||
{
|
||||
SessionId = sessionId,
|
||||
SavedAtUtc = new DateTime(2026, 7, 9, 12, 0, 0, DateTimeKind.Utc),
|
||||
InterimEmitted = true,
|
||||
WanSamplesMbps = new List<double> { 16.0, 15.5, 16.2 },
|
||||
LanSamplesMbps = new List<double> { 400.0 },
|
||||
WanBytesObserved = 18_000_000,
|
||||
LanBytesObserved = 150_000_000,
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public void Roundtrip_survives_an_agent_restart()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
var logger = NewLogger(tmp.Path);
|
||||
|
||||
new BandwidthStatePersistence(tmp.Path, logger).Save(SampleState("session-1"));
|
||||
|
||||
// New instance over the same state directory = agent restarted after a reboot.
|
||||
var loaded = new BandwidthStatePersistence(tmp.Path, logger).Load("session-1");
|
||||
|
||||
Assert.NotNull(loaded);
|
||||
Assert.True(loaded!.InterimEmitted);
|
||||
Assert.Equal(new List<double> { 16.0, 15.5, 16.2 }, loaded.WanSamplesMbps);
|
||||
Assert.Equal(new List<double> { 400.0 }, loaded.LanSamplesMbps);
|
||||
Assert.Equal(18_000_000, loaded.WanBytesObserved);
|
||||
Assert.Equal(150_000_000, loaded.LanBytesObserved);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Load_rejects_state_from_a_different_session()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
var logger = NewLogger(tmp.Path);
|
||||
|
||||
new BandwidthStatePersistence(tmp.Path, logger).Save(SampleState("session-1"));
|
||||
|
||||
Assert.Null(new BandwidthStatePersistence(tmp.Path, logger).Load("session-2"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Load_returns_null_when_no_file_exists()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
Assert.Null(new BandwidthStatePersistence(tmp.Path, NewLogger(tmp.Path)).Load("session-1"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Load_is_fail_soft_on_corrupt_or_empty_file()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
var logger = NewLogger(tmp.Path);
|
||||
var persistence = new BandwidthStatePersistence(tmp.Path, logger);
|
||||
|
||||
File.WriteAllText(persistence.StateFilePath, "{ not valid json !!!");
|
||||
Assert.Null(persistence.Load("session-1"));
|
||||
|
||||
File.WriteAllText(persistence.StateFilePath, "null");
|
||||
Assert.Null(persistence.Load("session-1"));
|
||||
|
||||
File.WriteAllText(persistence.StateFilePath, "{}"); // valid JSON, no session id
|
||||
Assert.Null(persistence.Load("session-1"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Save_overwrites_atomically_and_latest_state_wins()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
var logger = NewLogger(tmp.Path);
|
||||
var persistence = new BandwidthStatePersistence(tmp.Path, logger);
|
||||
|
||||
persistence.Save(SampleState("session-1"));
|
||||
var updated = SampleState("session-1");
|
||||
updated.WanSamplesMbps!.Add(17.1);
|
||||
updated.WanBytesObserved = 25_000_000;
|
||||
persistence.Save(updated);
|
||||
|
||||
var loaded = persistence.Load("session-1");
|
||||
Assert.Equal(4, loaded!.WanSamplesMbps!.Count);
|
||||
Assert.Equal(25_000_000, loaded.WanBytesObserved);
|
||||
Assert.False(File.Exists(persistence.StateFilePath + ".tmp")); // temp file cleaned up
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -210,6 +210,73 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Telemetry
|
||||
Assert.Equal("high", high.TryBuildEstimate().Confidence);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ExportImport_Roundtrip_PreservesEstimate()
|
||||
{
|
||||
var original = new BandwidthEstimator(3);
|
||||
original.AddSnapshot(T0, Jobs(Job("a", 0, 0)));
|
||||
original.AddSnapshot(T0.AddSeconds(3), Jobs(Job("a", 6_000_000, 3_000_000)));
|
||||
original.AddSnapshot(T0.AddSeconds(6), Jobs(Job("a", 12_000_000, 6_000_000)));
|
||||
|
||||
// Restart: fresh estimator seeded from the exported state.
|
||||
var resumed = new BandwidthEstimator(3);
|
||||
resumed.ImportState(original.ExportState());
|
||||
|
||||
var before = original.TryBuildEstimate();
|
||||
var after = resumed.TryBuildEstimate();
|
||||
Assert.Equal(before.WanMbpsP90, after.WanMbpsP90);
|
||||
Assert.Equal(before.WanSampleCount, after.WanSampleCount);
|
||||
Assert.Equal(before.WanBytesObserved, after.WanBytesObserved);
|
||||
Assert.Equal(before.LanMbpsP90, after.LanMbpsP90);
|
||||
Assert.Equal(before.Confidence, after.Confidence);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ImportedState_MergesWithNewSamples()
|
||||
{
|
||||
// Pre-reboot run: two 16 Mbps samples.
|
||||
var preReboot = new BandwidthEstimator(3);
|
||||
preReboot.AddSnapshot(T0, Jobs(Job("a", 0)));
|
||||
preReboot.AddSnapshot(T0.AddSeconds(3), Jobs(Job("a", 6_000_000)));
|
||||
preReboot.AddSnapshot(T0.AddSeconds(6), Jobs(Job("a", 12_000_000)));
|
||||
|
||||
// Post-reboot run resumes the state and adds one more sample from a NEW job.
|
||||
var postReboot = new BandwidthEstimator(3);
|
||||
postReboot.ImportState(preReboot.ExportState());
|
||||
postReboot.AddSnapshot(T0.AddSeconds(300), Jobs(Job("b", 0)));
|
||||
postReboot.AddSnapshot(T0.AddSeconds(303), Jobs(Job("b", 6_000_000)));
|
||||
|
||||
var estimate = postReboot.TryBuildEstimate();
|
||||
Assert.Equal(3, estimate.WanSampleCount);
|
||||
Assert.Equal(18_000_000, estimate.WanBytesObserved);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ImportState_SanitizesGarbage()
|
||||
{
|
||||
var estimator = new BandwidthEstimator(3);
|
||||
estimator.ImportState(new BandwidthEstimatorState
|
||||
{
|
||||
WanSamplesMbps = new List<double> { 16.0, double.NaN, double.PositiveInfinity, -5.0, 0.0 },
|
||||
LanSamplesMbps = null,
|
||||
WanBytesObserved = -1_000, // negative counter ignored
|
||||
LanBytesObserved = 0,
|
||||
});
|
||||
|
||||
var estimate = estimator.TryBuildEstimate();
|
||||
Assert.Equal(1, estimate.WanSampleCount); // only the finite positive sample survived
|
||||
Assert.Equal(16.0, estimate.WanMbpsP90.Value, 3);
|
||||
Assert.Equal(0, estimate.WanBytesObserved);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ImportState_NullIsIgnored()
|
||||
{
|
||||
var estimator = new BandwidthEstimator(3);
|
||||
estimator.ImportState(null);
|
||||
Assert.Null(estimator.TryBuildEstimate());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void EmptyOrNullSnapshots_AreIgnored()
|
||||
{
|
||||
|
||||
+146
@@ -0,0 +1,146 @@
|
||||
#nullable enable
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using AutopilotMonitor.Agent.V2.Core.Logging;
|
||||
using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime;
|
||||
using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic;
|
||||
using AutopilotMonitor.Agent.V2.Core.Orchestration;
|
||||
using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
|
||||
using AutopilotMonitor.DecisionCore.Engine;
|
||||
using AutopilotMonitor.DecisionCore.Signals;
|
||||
using AutopilotMonitor.Shared;
|
||||
using Xunit;
|
||||
|
||||
namespace AutopilotMonitor.Agent.V2.Core.Tests.Telemetry
|
||||
{
|
||||
/// <summary>
|
||||
/// Reboot-survivor behavior of the DeliveryOptimizationCollector's bandwidth estimate:
|
||||
/// a new collector over the same state directory (= the agent restarted after a reboot)
|
||||
/// must resume the persisted samples AND the interim once-guard — device_setup_end fires
|
||||
/// once per SESSION, not once per process, and pre-reboot samples reach the emitted event.
|
||||
/// No PowerShell runspace is needed: these tests never Start() the collector; resume,
|
||||
/// NotifyEspPhaseChanged and Stop() are all runspace-free paths.
|
||||
/// </summary>
|
||||
public sealed class DeliveryOptimizationBandwidthResumeTests
|
||||
{
|
||||
private const string Session = "session-1";
|
||||
private static readonly DateTime T0 = new DateTime(2026, 7, 9, 12, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private sealed class CapturingSink : ISignalIngressSink
|
||||
{
|
||||
public readonly List<IReadOnlyDictionary<string, string>?> Posts = new List<IReadOnlyDictionary<string, string>?>();
|
||||
|
||||
public void Post(
|
||||
DecisionSignalKind kind,
|
||||
DateTime occurredAtUtc,
|
||||
string sourceOrigin,
|
||||
Evidence evidence,
|
||||
IReadOnlyDictionary<string, string>? payload = null,
|
||||
int kindSchemaVersion = 1,
|
||||
object? typedPayload = null)
|
||||
{
|
||||
Posts.Add(payload);
|
||||
}
|
||||
|
||||
public IEnumerable<IReadOnlyDictionary<string, string>?> BandwidthPosts =>
|
||||
Posts.Where(p => p != null && p.TryGetValue(SignalPayloadKeys.EventType, out var t)
|
||||
&& t == Constants.EventTypes.NetworkBandwidthEstimate);
|
||||
}
|
||||
|
||||
private static DeliveryOptimizationCollector NewCollector(string stateDir, CapturingSink sink)
|
||||
{
|
||||
var logger = new AgentLogger(stateDir, AgentLogLevel.Info);
|
||||
return new DeliveryOptimizationCollector(
|
||||
sessionId: Session,
|
||||
tenantId: "tenant-1",
|
||||
post: new InformationalEventPost(sink, new VirtualClock(T0)),
|
||||
logger: logger,
|
||||
intervalSeconds: 3,
|
||||
getPackageStates: () => new AppPackageStateList(logger),
|
||||
onDoTelemetryReceived: null,
|
||||
logDirectory: stateDir,
|
||||
onOfficeDoSample: null,
|
||||
bandwidthStatePersistence: new BandwidthStatePersistence(stateDir, logger));
|
||||
}
|
||||
|
||||
private static void PersistPreRebootState(string stateDir, bool interimEmitted)
|
||||
{
|
||||
var logger = new AgentLogger(stateDir, AgentLogLevel.Info);
|
||||
new BandwidthStatePersistence(stateDir, logger).Save(new BandwidthStateData
|
||||
{
|
||||
SessionId = Session,
|
||||
SavedAtUtc = T0,
|
||||
InterimEmitted = interimEmitted,
|
||||
WanSamplesMbps = new List<double> { 16.0, 15.8, 16.4 },
|
||||
LanSamplesMbps = new List<double>(),
|
||||
WanBytesObserved = 30_000_000,
|
||||
LanBytesObserved = 0,
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Resumed_samples_reach_the_interim_event_after_a_reboot()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
PersistPreRebootState(tmp.Path, interimEmitted: false);
|
||||
|
||||
var sink = new CapturingSink();
|
||||
var collector = NewCollector(tmp.Path, sink);
|
||||
|
||||
// Post-reboot process observes AccountSetup without any NEW download samples —
|
||||
// the emitted interim must carry the pre-reboot data.
|
||||
collector.NotifyEspPhaseChanged("AccountSetup");
|
||||
|
||||
var post = Assert.Single(sink.BandwidthPosts);
|
||||
Assert.Contains("16", post![SignalPayloadKeys.Message]); // pre-reboot ~16 Mbps figure
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Interim_guard_survives_a_reboot()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
PersistPreRebootState(tmp.Path, interimEmitted: true);
|
||||
|
||||
var sink = new CapturingSink();
|
||||
var collector = NewCollector(tmp.Path, sink);
|
||||
|
||||
collector.NotifyEspPhaseChanged("AccountSetup");
|
||||
|
||||
Assert.Empty(sink.BandwidthPosts); // already emitted by the pre-reboot process
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Final_emission_at_stop_carries_resumed_samples_and_resaves_state()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
PersistPreRebootState(tmp.Path, interimEmitted: true);
|
||||
|
||||
var sink = new CapturingSink();
|
||||
var collector = NewCollector(tmp.Path, sink);
|
||||
collector.Stop();
|
||||
|
||||
var post = Assert.Single(sink.BandwidthPosts);
|
||||
Assert.Contains("16", post![SignalPayloadKeys.Message]);
|
||||
|
||||
// State file still loadable for the same session after the stop-path save.
|
||||
var logger = new AgentLogger(tmp.Path, AgentLogLevel.Info);
|
||||
var persisted = new BandwidthStatePersistence(tmp.Path, logger).Load(Session);
|
||||
Assert.NotNull(persisted);
|
||||
Assert.Equal(3, persisted!.WanSamplesMbps!.Count);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Without_persisted_state_nothing_emits_until_samples_exist()
|
||||
{
|
||||
using var tmp = new TempDirectory();
|
||||
var sink = new CapturingSink();
|
||||
var collector = NewCollector(tmp.Path, sink);
|
||||
|
||||
collector.NotifyEspPhaseChanged("AccountSetup");
|
||||
collector.Stop();
|
||||
|
||||
Assert.Empty(sink.BandwidthPosts);
|
||||
}
|
||||
}
|
||||
}
|
||||
+59
@@ -17,6 +17,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
public long LanBytes;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Serializable accumulator state for restart persistence. Baselines are deliberately
|
||||
/// NOT part of it: a reboot gap always exceeds the per-file delta window, so restored
|
||||
/// baselines would be discarded on first sight anyway — only the reduced samples and
|
||||
/// byte counters carry information across restarts.
|
||||
/// </summary>
|
||||
public sealed class BandwidthEstimatorState
|
||||
{
|
||||
public List<double> WanSamplesMbps { get; set; }
|
||||
public List<double> LanSamplesMbps { get; set; }
|
||||
public long WanBytesObserved { get; set; }
|
||||
public long LanBytesObserved { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>Result of <see cref="BandwidthEstimator.TryBuildEstimate"/>.</summary>
|
||||
public sealed class BandwidthEstimate
|
||||
{
|
||||
@@ -160,6 +174,51 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Copies the accumulator state for restart persistence (see <see cref="ImportState"/>).</summary>
|
||||
public BandwidthEstimatorState ExportState()
|
||||
{
|
||||
lock (_sync)
|
||||
{
|
||||
return new BandwidthEstimatorState
|
||||
{
|
||||
WanSamplesMbps = new List<double>(_wanSamplesMbps),
|
||||
LanSamplesMbps = new List<double>(_lanSamplesMbps),
|
||||
WanBytesObserved = _wanBytesObserved,
|
||||
LanBytesObserved = _lanBytesObserved,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Seeds the accumulator from a persisted state (previous agent run of the SAME
|
||||
/// session — the reboot survivor path). Defensive against a tampered/corrupt file:
|
||||
/// non-finite or negative sample values are dropped, list sizes and counters are
|
||||
/// clamped. Intended for a freshly constructed estimator; imported samples simply
|
||||
/// prepend the ones this run will collect.
|
||||
/// </summary>
|
||||
public void ImportState(BandwidthEstimatorState state)
|
||||
{
|
||||
if (state == null) return;
|
||||
lock (_sync)
|
||||
{
|
||||
ImportSamples(state.WanSamplesMbps, _wanSamplesMbps);
|
||||
ImportSamples(state.LanSamplesMbps, _lanSamplesMbps);
|
||||
if (state.WanBytesObserved > 0) _wanBytesObserved += state.WanBytesObserved;
|
||||
if (state.LanBytesObserved > 0) _lanBytesObserved += state.LanBytesObserved;
|
||||
}
|
||||
}
|
||||
|
||||
private static void ImportSamples(List<double> source, List<double> target)
|
||||
{
|
||||
if (source == null) return;
|
||||
for (int i = 0; i < source.Count && target.Count < MaxSamples; i++)
|
||||
{
|
||||
var v = source[i];
|
||||
if (double.IsNaN(v) || double.IsInfinity(v) || v <= 0) continue;
|
||||
target.Add(v);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Returns the estimate, or null when not a single valid rate sample exists.</summary>
|
||||
public BandwidthEstimate TryBuildEstimate()
|
||||
{
|
||||
|
||||
+121
@@ -0,0 +1,121 @@
|
||||
#nullable enable
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using AutopilotMonitor.Agent.V2.Core.Logging;
|
||||
using Newtonsoft.Json;
|
||||
|
||||
namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
{
|
||||
/// <summary>
|
||||
/// Persists the <see cref="BandwidthEstimator"/> accumulator across agent restarts (one
|
||||
/// enrollment commonly spans several reboots). Without it, a reboot before the interim
|
||||
/// snapshot — e.g. a Win32 app forcing a restart mid-DeviceSetup — silently discards every
|
||||
/// rate sample collected so far; if little downloads after the reboot, the session ends
|
||||
/// with a weak or missing estimate although gigabytes were observed.
|
||||
/// <para>
|
||||
/// Save points are cheap and event-driven (never per-poll): when the collector goes
|
||||
/// dormant (all downloads done — which is exactly the moment an app-forced reboot becomes
|
||||
/// possible), after the interim emission, and at collector stop. <c>InterimEmitted</c>
|
||||
/// rides in the same file so the device_setup_end snapshot stays once-per-SESSION rather
|
||||
/// than once-per-process.
|
||||
/// </para>
|
||||
/// The state is only resumed for the SAME session id — the state directory has a
|
||||
/// per-enrollment lifecycle, but the check makes a stale file from an aborted run harmless.
|
||||
/// Fail-soft like <c>OfficeInstallStatePersistence</c>: I/O errors never throw; a missing
|
||||
/// or corrupt file loads as null (fresh start — worst case is the pre-persistence behavior).
|
||||
/// </summary>
|
||||
public class BandwidthStatePersistence
|
||||
{
|
||||
private const string SourceName = "BandwidthStatePersistence";
|
||||
|
||||
private readonly string _stateDirectory;
|
||||
private readonly string _stateFilePath;
|
||||
private readonly AgentLogger _logger;
|
||||
|
||||
public BandwidthStatePersistence(string stateDirectory, AgentLogger logger)
|
||||
{
|
||||
_stateDirectory = Environment.ExpandEnvironmentVariables(stateDirectory);
|
||||
_stateFilePath = Path.Combine(_stateDirectory, "bandwidth-state.json");
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public string StateFilePath => _stateFilePath;
|
||||
|
||||
/// <summary>
|
||||
/// Loads the persisted state, or null when absent, unreadable, or belonging to a
|
||||
/// different session.
|
||||
/// </summary>
|
||||
public BandwidthStateData? Load(string expectedSessionId)
|
||||
{
|
||||
if (!File.Exists(_stateFilePath)) return null;
|
||||
try
|
||||
{
|
||||
var json = File.ReadAllText(_stateFilePath);
|
||||
var state = JsonConvert.DeserializeObject<BandwidthStateData>(json);
|
||||
if (state == null || string.IsNullOrEmpty(state.SessionId))
|
||||
{
|
||||
_logger.Warning($"[{SourceName}] persisted state file was empty or invalid — starting fresh");
|
||||
return null;
|
||||
}
|
||||
if (!string.Equals(state.SessionId, expectedSessionId, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_logger.Warning($"[{SourceName}] persisted state belongs to session {state.SessionId}, not {expectedSessionId} — starting fresh");
|
||||
return null;
|
||||
}
|
||||
return state;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.Warning($"[{SourceName}] failed to load persisted state, starting fresh: {ex.Message}");
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Persists the state (atomic write via temp file). Never throws.</summary>
|
||||
public void Save(BandwidthStateData state)
|
||||
{
|
||||
try
|
||||
{
|
||||
Directory.CreateDirectory(_stateDirectory);
|
||||
var tempPath = _stateFilePath + ".tmp";
|
||||
File.WriteAllText(tempPath, JsonConvert.SerializeObject(state));
|
||||
if (File.Exists(_stateFilePath))
|
||||
{
|
||||
File.Replace(tempPath, _stateFilePath, destinationBackupFileName: null, ignoreMetadataErrors: true);
|
||||
}
|
||||
else
|
||||
{
|
||||
File.Move(tempPath, _stateFilePath);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.Debug($"[{SourceName}] failed to save state: {ex.Message}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Persisted mirror of the estimator accumulator plus the interim once-guard.</summary>
|
||||
public class BandwidthStateData
|
||||
{
|
||||
public string? SessionId { get; set; }
|
||||
public DateTime SavedAtUtc { get; set; }
|
||||
|
||||
/// <summary>True once the device_setup_end interim snapshot went out (this session, any process).</summary>
|
||||
public bool InterimEmitted { get; set; }
|
||||
|
||||
public List<double>? WanSamplesMbps { get; set; }
|
||||
public List<double>? LanSamplesMbps { get; set; }
|
||||
public long WanBytesObserved { get; set; }
|
||||
public long LanBytesObserved { get; set; }
|
||||
|
||||
public BandwidthEstimatorState ToEstimatorState() => new BandwidthEstimatorState
|
||||
{
|
||||
WanSamplesMbps = WanSamplesMbps,
|
||||
LanSamplesMbps = LanSamplesMbps,
|
||||
WanBytesObserved = WanBytesObserved,
|
||||
LanBytesObserved = LanBytesObserved,
|
||||
};
|
||||
}
|
||||
}
|
||||
+52
-1
@@ -86,6 +86,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
private bool _bandwidthEstimateEmitted;
|
||||
private int _interimBandwidthEmitted; // Interlocked guard — phase callback runs on the IME tracker thread
|
||||
|
||||
// Restart persistence for the estimator accumulator (one enrollment spans several
|
||||
// reboots; without it a mid-DeviceSetup reboot discards all samples collected so far).
|
||||
// Null-tolerant: tests and callers without a state directory simply run in-memory-only.
|
||||
private readonly BandwidthStatePersistence _bandwidthStatePersistence;
|
||||
|
||||
public DeliveryOptimizationCollector(
|
||||
string sessionId,
|
||||
string tenantId,
|
||||
@@ -95,7 +100,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
Func<AppPackageStateList> getPackageStates,
|
||||
Action<AppPackageState> onDoTelemetryReceived,
|
||||
string logDirectory,
|
||||
Action<OfficeDoSample> onOfficeDoSample = null)
|
||||
Action<OfficeDoSample> onOfficeDoSample = null,
|
||||
BandwidthStatePersistence bandwidthStatePersistence = null)
|
||||
: base(sessionId, tenantId, post, logger, intervalSeconds)
|
||||
{
|
||||
_getPackageStates = getPackageStates ?? throw new ArgumentNullException(nameof(getPackageStates));
|
||||
@@ -104,6 +110,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
_onOfficeDoSample = onOfficeDoSample;
|
||||
_pollIntervalSeconds = intervalSeconds;
|
||||
_bandwidthEstimator = new BandwidthEstimator(intervalSeconds);
|
||||
_bandwidthStatePersistence = bandwidthStatePersistence;
|
||||
|
||||
// Resume the accumulator from a previous run of the SAME session (reboot survivor):
|
||||
// samples and byte counters carry over, and the interim once-guard stays claimed so
|
||||
// device_setup_end fires once per session, not once per process.
|
||||
var persisted = _bandwidthStatePersistence?.Load(sessionId);
|
||||
if (persisted != null)
|
||||
{
|
||||
_bandwidthEstimator.ImportState(persisted.ToEstimatorState());
|
||||
if (persisted.InterimEmitted) _interimBandwidthEmitted = 1;
|
||||
Logger.Info($"[DeliveryOptimizationCollector] Resumed bandwidth state from previous run — " +
|
||||
$"{persisted.WanSamplesMbps?.Count ?? 0} WAN samples / {persisted.WanBytesObserved} bytes, " +
|
||||
$"interimEmitted={persisted.InterimEmitted}");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Start dormant — timer does not fire until WakeUp() is called.</summary>
|
||||
@@ -120,6 +140,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
// collectors BEFORE the diagnostics ZIP is built, so this one-shot still lands
|
||||
// ahead of diagnostics_collecting in the timeline.
|
||||
EmitBandwidthEstimateOnce();
|
||||
SaveBandwidthState();
|
||||
DisposeRunspace();
|
||||
}
|
||||
|
||||
@@ -225,6 +246,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
Logger.Info("[DeliveryOptimizationCollector] Going dormant — no active downloads, pending enrichment or Office install");
|
||||
_dormant = true;
|
||||
PauseTimer();
|
||||
// Natural persistence point: all downloads are done — which is exactly when an
|
||||
// app-forced reboot becomes possible. Saves the accumulated samples so a reboot
|
||||
// before the interim/final emission does not discard them (reboot survivor).
|
||||
SaveBandwidthState();
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -583,6 +608,32 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
|
||||
if (!"AccountSetup".Equals(phase, StringComparison.OrdinalIgnoreCase)) return;
|
||||
if (Interlocked.CompareExchange(ref _interimBandwidthEmitted, 1, 0) != 0) return;
|
||||
EmitBandwidthEstimate("device_setup_end");
|
||||
// Persist AFTER the emission (StartupEventGate's MarkEmitted ordering): a crash
|
||||
// between emit and save re-emits the interim on the next run instead of silently
|
||||
// suppressing it for the rest of the session. A rare duplicate is harmless —
|
||||
// analyses take the latest event per trigger.
|
||||
SaveBandwidthState();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Persists the estimator accumulator + interim guard (no-op without a persistence
|
||||
/// instance). Called at the dormant transition, after the interim emission and at
|
||||
/// stop — never per-poll. Fail-soft: Save never throws.
|
||||
/// </summary>
|
||||
private void SaveBandwidthState()
|
||||
{
|
||||
if (_bandwidthStatePersistence == null) return;
|
||||
var state = _bandwidthEstimator.ExportState();
|
||||
_bandwidthStatePersistence.Save(new BandwidthStateData
|
||||
{
|
||||
SessionId = SessionId,
|
||||
SavedAtUtc = DateTime.UtcNow,
|
||||
InterimEmitted = _interimBandwidthEmitted != 0,
|
||||
WanSamplesMbps = state.WanSamplesMbps,
|
||||
LanSamplesMbps = state.LanSamplesMbps,
|
||||
WanBytesObserved = state.WanBytesObserved,
|
||||
LanBytesObserved = state.LanBytesObserved,
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -403,7 +403,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
logger: logger,
|
||||
intervalSeconds: collectors.DeliveryOptimizationIntervalSeconds,
|
||||
imeHost: imeLogHost,
|
||||
officeHost: officeHost);
|
||||
officeHost: officeHost,
|
||||
stateDirectory: _stateDirectory);
|
||||
hosts.Add(doHost);
|
||||
}
|
||||
|
||||
|
||||
+6
-2
@@ -32,7 +32,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
AgentLogger logger,
|
||||
int intervalSeconds,
|
||||
ImeLogHost imeHost,
|
||||
OfficeInstallDetectorHost? officeHost = null)
|
||||
OfficeInstallDetectorHost? officeHost = null,
|
||||
string? stateDirectory = null)
|
||||
{
|
||||
if (ingress == null) throw new ArgumentNullException(nameof(ingress));
|
||||
if (clock == null) throw new ArgumentNullException(nameof(clock));
|
||||
@@ -54,7 +55,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
catch (Exception ex) { logger.Warning($"DeliveryOptimizationHost: OnDoTelemetryReceived invocation threw: {ex.Message}"); }
|
||||
},
|
||||
logDirectory: Environment.ExpandEnvironmentVariables(Constants.LogDirectory),
|
||||
onOfficeDoSample: officeHost != null ? officeHost.SubmitDoSample : (Action<OfficeDoSample>?)null);
|
||||
onOfficeDoSample: officeHost != null ? officeHost.SubmitDoSample : (Action<OfficeDoSample>?)null,
|
||||
bandwidthStatePersistence: stateDirectory != null
|
||||
? new BandwidthStatePersistence(stateDirectory, logger)
|
||||
: null);
|
||||
}
|
||||
|
||||
public void Start()
|
||||
|
||||
Reference in New Issue
Block a user