using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests; /// /// Register row 33: the sustained-backlog anchor behind the Health dashboard's /// "Replication backlog" badge. /// /// /// The design constraint these tests exist to pin: a bare "backlog > threshold" badge is /// unusable, because the daily `site_events` retention purge is CDC-captured like any other /// write and jumps the backlog by the deleted batch size before it drains. That spike is /// expected and documented. So the aggregator holds the instant the backlog FIRST rose above /// the threshold, and the badge only appears once the configured dwell has elapsed — healthy /// means the backlog comes back down, which no point-in-time check can express. /// public sealed class LocalDbBacklogSustainedTests { private const long Threshold = 1000; [Fact] public void BacklogAboveThreshold_StartsTheClock() { var (aggregator, time) = NewAggregator(); var start = time.GetUtcNow(); aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1)); Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); } [Fact] public void BacklogStayingAbove_KeepsTheOriginalStartInstant() { // The window must be sustained, not sliding: if each above-threshold report reset the // anchor, the dwell would never elapse and the badge could never appear at all. var (aggregator, time) = NewAggregator(); var start = time.GetUtcNow(); aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1)); time.Advance(TimeSpan.FromMinutes(10)); aggregator.ProcessReport(Report("site-a", 2, backlog: Threshold + 5000)); Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); } [Fact] public void BacklogAtOrBelowThreshold_ClearsTheClock() { // The purge-burst exit path. A spike that drains must discard the anchor well before // the dwell elapses, so the expected daily burst never badges the site. var (aggregator, time) = NewAggregator(); aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 50_000)); time.Advance(TimeSpan.FromMinutes(2)); aggregator.ProcessReport(Report("site-a", 2, backlog: 0)); Assert.Null(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); } [Fact] public void BacklogExactlyAtThreshold_DoesNotStartTheClock() { // The comparison is strictly greater-than, so the threshold is the highest ACCEPTABLE // depth rather than the lowest alarming one. var (aggregator, _) = NewAggregator(); aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold)); Assert.Null(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); } [Fact] public void UnreplicatedSite_NeverAccumulatesAnAnchor() { // Rig site-b/site-c: a null backlog forever. Null means "no reading", never zero, so // it must never start a clock — otherwise every unreplicated site would eventually be // badged for a backlog it does not have. var (aggregator, time) = NewAggregator(); aggregator.ProcessReport(Report("site-b", 1, backlog: null)); time.Advance(TimeSpan.FromHours(6)); aggregator.ProcessReport(Report("site-b", 2, backlog: null)); Assert.Null(aggregator.GetSiteState("site-b")!.LocalDbBacklogAboveSince); } [Fact] public void NullBacklogMidClimb_LeavesTheAnchorUntouched() { // A failed poll on a replicated node reports null. That is not evidence the backlog // drained, so it must not clear the clock — a node that goes blind mid-climb would // otherwise silently reset its own dwell and never be badged. var (aggregator, time) = NewAggregator(); var start = time.GetUtcNow(); aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1)); time.Advance(TimeSpan.FromMinutes(5)); aggregator.ProcessReport(Report("site-a", 2, backlog: null)); Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); } [Fact] public void FirstReportForANewSite_CanStartTheClock() { // The registration path is a separate branch from the update path; a site whose very // first report is already backlogged must not be missed. var (aggregator, time) = NewAggregator(); var start = time.GetUtcNow(); aggregator.ProcessReport(Report("brand-new", 1, backlog: Threshold + 1)); Assert.Equal(start, aggregator.GetSiteState("brand-new")!.LocalDbBacklogAboveSince); } [Fact] public void ThresholdIsConfigurable() { var (aggregator, _) = NewAggregator(threshold: 10); aggregator.ProcessReport(Report("site-a", 1, backlog: 11)); Assert.NotNull(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); } [Fact] public void StaleReport_DoesNotDisturbTheAnchor() { // Out-of-order reports are rejected wholesale by the sequence-number guard; the anchor // must ride along with that rejection rather than being updated from stale data. var (aggregator, time) = NewAggregator(); var start = time.GetUtcNow(); aggregator.ProcessReport(Report("site-a", 5, backlog: Threshold + 1)); time.Advance(TimeSpan.FromMinutes(5)); aggregator.ProcessReport(Report("site-a", 2, backlog: 0)); Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); } private static (CentralHealthAggregator Aggregator, TestTimeProvider Time) NewAggregator( long threshold = Threshold) { var time = new TestTimeProvider(DateTimeOffset.UtcNow); var opts = new HealthMonitoringOptions { OfflineTimeout = TimeSpan.FromSeconds(60), LocalDbBacklogThreshold = threshold, }; var aggregator = new CentralHealthAggregator( Options.Create(opts), NullLogger.Instance, time); return (aggregator, time); } private static SiteHealthReport Report(string siteId, long seq, long? backlog) => new( SiteId: siteId, SequenceNumber: seq, ReportTimestamp: DateTimeOffset.UtcNow, DataConnectionStatuses: new Dictionary(), TagResolutionCounts: new Dictionary(), ScriptErrorCount: 0, AlarmEvaluationErrorCount: 0, StoreAndForwardBufferDepths: new Dictionary(), DeadLetterCount: 0, DeployedInstanceCount: 0, EnabledInstanceCount: 0, DisabledInstanceCount: 0) { LocalDbOplogBacklog = backlog, }; }