fix(alarms): truncation-safe transitions; perf: single-pass alarm parse; configurable poll cadence
This commit is contained in:
Joseph Doherty
@@ -135,6 +135,56 @@ alarm state is gateway-wide, not session-scoped — every client wants the same
|
||||
current set plus updates, and forcing each to own a worker would multiply AVEVA
|
||||
polling load for no benefit.
|
||||
|
||||
### Alarms — a capped snapshot fetch never implies a clear
|
||||
|
||||
Decision (2026-08-15): when the worker's `GetXmlCurrentAlarms2` fetch comes back
|
||||
holding exactly `MxGateway:Alarms:MaxAlarmsPerFetch` records, the worker treats
|
||||
the snapshot as **truncated** and merges it into the retained snapshot instead
|
||||
of replacing it. Alarms the capped reply did carry update normally; alarms it
|
||||
had no room to mention are retained untouched.
|
||||
|
||||
The COM API caps its reply at `maxAlmCnt` and exposes no "more available" flag,
|
||||
so a reply sitting exactly on the cap is indistinguishable from a galaxy that
|
||||
happens to hold exactly that many active alarms. Both are treated as truncated,
|
||||
because the two error directions are not symmetric.
|
||||
|
||||
Nothing in the worker emits a Clear transition. The clear is an **inference**:
|
||||
`WnWrapAlarmConsumer.ComputeTransitions` produces no transition for an alarm
|
||||
that disappears from the snapshot, and `GatewayAlarmMonitor.ApplyReconcile`
|
||||
later diffs its cache against `SnapshotActiveAlarms()` and broadcasts a Clear
|
||||
for every cached alarm the worker no longer reports. Before this decision, a
|
||||
capped fetch shrank that snapshot, so every alarm past the cap was broadcast as
|
||||
cleared while still standing — a silent, galaxy-wide false clear on exactly the
|
||||
alarm floods where the cap is reached.
|
||||
|
||||
Consequences, and how this sits with the existing failover/reconcile design:
|
||||
|
||||
- **The suppression is an eviction guard, not a transition filter.** It lives in
|
||||
the snapshot update inside `PollOnce`, not in `ComputeTransitions`, which was
|
||||
never going to emit anything for a disappearance. The reconcile/dedup
|
||||
machinery (`_clearedByReconcile` tombstones, the NEXT-03 duplicate-Clear
|
||||
suppression) is untouched: it still sees the same shape of snapshot, only
|
||||
with the truncated poll's unmentionable alarms still present.
|
||||
- **It preserves at-least-once, idempotent application.** The failure mode
|
||||
becomes bounded staleness — a genuinely cleared alarm can linger until the
|
||||
first sub-cap fetch evicts it, and the reconcile then broadcasts its Clear
|
||||
late. A late Clear is repaired by the next complete poll; a Clear that never
|
||||
happened is broadcast to every `StreamAlarms` subscriber and cannot be taken
|
||||
back. Consumers already apply transitions as "set this alarm to this state",
|
||||
so a repeated or delayed Clear is absorbed.
|
||||
- **It does not synthesize anything.** Suppressing an inference is the opposite
|
||||
of inventing an event; no transition is fabricated on a truncated poll.
|
||||
- **Failover is unaffected.** `FailoverAlarmConsumer` selects which
|
||||
`IMxAccessAlarmConsumer` is live; the guard is internal to the wnwrap
|
||||
consumer's own snapshot bookkeeping and changes neither the failure counting
|
||||
that triggers failover nor the subtag standby's snapshot, which is built from
|
||||
a bounded watch-list and has no per-fetch cap to hit.
|
||||
- **Operators get told.** A truncated poll logs a rate-limited (once per
|
||||
minute) `AlarmSnapshotTruncated` warning carrying the cap, the record counts,
|
||||
and the running truncated-fetch total — identifiers and counts only, never
|
||||
tag names, values, limits, or comments. A galaxy that truncates persistently
|
||||
is a configuration problem: raise `MxGateway:Alarms:MaxAlarmsPerFetch`.
|
||||
|
||||
## Session-Resilience Epic Scope
|
||||
|
||||
Decision (2026-07-09, archreview TST-04; migrated here 2026-08-07 from the retired
|
||||
|
||||
@@ -417,6 +417,8 @@ behavior.
|
||||
| `MxGateway:Alarms:SubscriptionExpression` | _(empty)_ | AVEVA alarm-subscription expression the monitor subscribes on startup, in canonical `\\<machine>\Galaxy!<area>` form. The literal `Galaxy` provider is correct regardless of the Galaxy database name. When empty and `Enabled` is `true`, the gateway falls back to `\\<MachineName>\Galaxy!<DefaultArea>` if `DefaultArea` is set. |
|
||||
| `MxGateway:Alarms:DefaultArea` | _(empty)_ | Area name used to compose a default subscription when `SubscriptionExpression` is empty. If both are empty while `Enabled` is `true`, the monitor faults with a configuration diagnostic. |
|
||||
| `MxGateway:Alarms:ReconcileIntervalSeconds` | `30` | How often the monitor reconciles its in-process alarm cache against the worker's authoritative active-alarm snapshot, catching transitions the live poll-and-diff feed missed. Floored at 5 seconds. |
|
||||
| `MxGateway:Alarms:PollIntervalMilliseconds` | `500` | Cadence at which the worker's STA polls the AVEVA alarm consumer (`GetXmlCurrentAlarms2`) for the active-alarm snapshot the live feed diffs. Must be `>= 100`: every poll is a COM call plus an XML parse on the same STA that serves reads and writes, so a tighter cadence starves the command path. The gateway conveys the value to the worker via the `MXGATEWAY_ALARM_POLL_INTERVAL_MS` environment variable; a missing or unusable value leaves the worker on the 500 ms default rather than failing the session. |
|
||||
| `MxGateway:Alarms:MaxAlarmsPerFetch` | `1024` | Cap the worker passes to `GetXmlCurrentAlarms2`'s `maxAlmCnt`. Must be `>= 64`. It doubles as the **truncation threshold**: a fetch returning exactly this many records is treated as truncated, because the COM API caps its reply with no "more available" flag. On a truncated poll the worker retains the alarms the capped reply could not mention instead of letting their absence read as a clear, and logs a rate-limited `AlarmSnapshotTruncated` warning (identifiers and counts only). Raise this on galaxies whose steady-state active-alarm count approaches the cap — a galaxy permanently above it holds stale entries in the snapshot until a sub-cap poll. Conveyed to the worker via the `MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH` environment variable; a missing or unusable value leaves the worker on the 1024 default. |
|
||||
|
||||
The alarm monitor is independent of client sessions: `AcknowledgeAlarm` and
|
||||
`StreamAlarms` are session-less RPCs served by the monitor.
|
||||
|
||||
@@ -46,6 +46,29 @@ public sealed class AlarmsOptions
|
||||
/// </summary>
|
||||
public int ReconcileIntervalSeconds { get; init; } = 30;
|
||||
|
||||
/// <summary>
|
||||
/// Cadence at which the worker's STA polls the AVEVA alarm consumer
|
||||
/// (<c>GetXmlCurrentAlarms2</c>) for the current active-alarm snapshot.
|
||||
/// Default 500 ms; must be at least 100 ms. Every poll is a COM call
|
||||
/// plus an XML parse on the STA that also serves reads and writes, so
|
||||
/// driving it below 100 ms starves the command path. Conveyed to the
|
||||
/// worker through the <c>MXGATEWAY_ALARM_POLL_INTERVAL_MS</c>
|
||||
/// environment variable.
|
||||
/// </summary>
|
||||
public int PollIntervalMilliseconds { get; init; } = 500;
|
||||
|
||||
/// <summary>
|
||||
/// Cap the worker passes to <c>GetXmlCurrentAlarms2</c>'s
|
||||
/// <c>maxAlmCnt</c> argument. Default 1024; must be at least 64. A
|
||||
/// fetch that comes back holding exactly this many records is treated
|
||||
/// as truncated: the worker keeps the alarms the capped fetch could
|
||||
/// not mention in its snapshot rather than letting their absence read
|
||||
/// as a clear. Raise it on galaxies whose steady-state active-alarm
|
||||
/// count approaches the cap. Conveyed to the worker through the
|
||||
/// <c>MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH</c> environment variable.
|
||||
/// </summary>
|
||||
public int MaxAlarmsPerFetch { get; init; } = 1024;
|
||||
|
||||
/// <summary>
|
||||
/// Configuration for the alarm-manager ↔ subtag fallback mechanism:
|
||||
/// operating mode, failure-detection thresholds, discovery, and subtag
|
||||
|
||||
@@ -414,8 +414,26 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
|
||||
|
||||
private static readonly string[] ValidAlarmFallbackModes = ["Auto", "ForceAlarmManager", "ForceSubtag"];
|
||||
|
||||
private const int MinimumAlarmPollIntervalMilliseconds = 100;
|
||||
private const int MinimumMaxAlarmsPerFetch = 64;
|
||||
|
||||
private static void ValidateAlarms(AlarmsOptions options, ValidationBuilder builder)
|
||||
{
|
||||
// Validated regardless of Enabled: both values are stamped onto every
|
||||
// worker launch environment, so a bad value is a misconfiguration even
|
||||
// before the central monitor is switched on.
|
||||
if (options.PollIntervalMilliseconds < MinimumAlarmPollIntervalMilliseconds)
|
||||
{
|
||||
builder.Add(
|
||||
$"MxGateway:Alarms:PollIntervalMilliseconds must be greater than or equal to {MinimumAlarmPollIntervalMilliseconds}.");
|
||||
}
|
||||
|
||||
if (options.MaxAlarmsPerFetch < MinimumMaxAlarmsPerFetch)
|
||||
{
|
||||
builder.Add(
|
||||
$"MxGateway:Alarms:MaxAlarmsPerFetch must be greater than or equal to {MinimumMaxAlarmsPerFetch}.");
|
||||
}
|
||||
|
||||
if (!options.Enabled)
|
||||
{
|
||||
return;
|
||||
|
||||
@@ -29,11 +29,27 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
|
||||
public const string WorkerWriteCompletionWaitEnvironmentVariableName =
|
||||
"MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS";
|
||||
|
||||
/// <summary>
|
||||
/// Conveys MxGateway:Alarms:PollIntervalMilliseconds to the worker: the
|
||||
/// cadence at which the worker's STA polls the AVEVA alarm consumer.
|
||||
/// </summary>
|
||||
public const string WorkerAlarmPollIntervalEnvironmentVariableName =
|
||||
"MXGATEWAY_ALARM_POLL_INTERVAL_MS";
|
||||
|
||||
/// <summary>
|
||||
/// Conveys MxGateway:Alarms:MaxAlarmsPerFetch to the worker: the cap
|
||||
/// passed to GetXmlCurrentAlarms2, which is also the record count at
|
||||
/// which the worker treats a fetch as truncated.
|
||||
/// </summary>
|
||||
public const string WorkerMaxAlarmsPerFetchEnvironmentVariableName =
|
||||
"MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH";
|
||||
|
||||
private readonly IWorkerProcessFactory _processFactory;
|
||||
private readonly IWorkerStartupProbe _startupProbe;
|
||||
private readonly GatewayMetrics _metrics;
|
||||
private readonly TimeProvider _timeProvider;
|
||||
private readonly WorkerOptions _workerOptions;
|
||||
private readonly AlarmsOptions _alarmsOptions;
|
||||
private readonly ILogger<WorkerProcessLauncher> _logger;
|
||||
|
||||
/// <summary>
|
||||
@@ -59,6 +75,7 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
|
||||
ArgumentNullException.ThrowIfNull(metrics);
|
||||
|
||||
_workerOptions = gatewayOptions.Value.Worker;
|
||||
_alarmsOptions = gatewayOptions.Value.Alarms;
|
||||
_processFactory = processFactory;
|
||||
_startupProbe = startupProbe;
|
||||
_metrics = metrics;
|
||||
@@ -185,6 +202,10 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
|
||||
_workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
startInfo.Environment[WorkerWriteCompletionWaitEnvironmentVariableName] =
|
||||
_workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
startInfo.Environment[WorkerAlarmPollIntervalEnvironmentVariableName] =
|
||||
_alarmsOptions.PollIntervalMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
startInfo.Environment[WorkerMaxAlarmsPerFetchEnvironmentVariableName] =
|
||||
_alarmsOptions.MaxAlarmsPerFetch.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
|
||||
commandLine = new WorkerProcessCommandLine(executablePath, arguments);
|
||||
|
||||
|
||||
@@ -163,6 +163,103 @@ public sealed class GatewayOptionsValidatorTests
|
||||
Tls = source.Tls,
|
||||
};
|
||||
|
||||
/// <summary>Verifies the alarm poll cadence and per-fetch cap defaults pass validation.</summary>
|
||||
[Fact]
|
||||
public void Validate_Succeeds_WithDefaultAlarmPollCadenceAndFetchCap()
|
||||
{
|
||||
AlarmsOptions alarms = new();
|
||||
Assert.Equal(500, alarms.PollIntervalMilliseconds);
|
||||
Assert.Equal(1024, alarms.MaxAlarmsPerFetch);
|
||||
|
||||
ValidateOptionsResult result = new GatewayOptionsValidator()
|
||||
.Validate(null, CloneWithAlarms(ValidOptions(), alarms));
|
||||
Assert.True(result.Succeeded);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A poll cadence below the 100 ms floor must fail validation. Both
|
||||
/// values are stamped onto every worker launch environment, so they
|
||||
/// are validated whether or not the central alarm monitor is enabled.
|
||||
/// </summary>
|
||||
/// <param name="pollIntervalMilliseconds">Cadence under test.</param>
|
||||
/// <param name="alarmsEnabled">Whether the central alarm monitor is on.</param>
|
||||
[Theory]
|
||||
[InlineData(99, false)]
|
||||
[InlineData(0, false)]
|
||||
[InlineData(-1, false)]
|
||||
[InlineData(99, true)]
|
||||
public void Validate_Fails_WhenAlarmPollIntervalBelowFloor(
|
||||
int pollIntervalMilliseconds,
|
||||
bool alarmsEnabled)
|
||||
{
|
||||
GatewayOptions options = CloneWithAlarms(
|
||||
ValidOptions(),
|
||||
new AlarmsOptions
|
||||
{
|
||||
Enabled = alarmsEnabled,
|
||||
DefaultArea = "Galaxy",
|
||||
PollIntervalMilliseconds = pollIntervalMilliseconds,
|
||||
});
|
||||
|
||||
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
|
||||
|
||||
Assert.False(result.Succeeded);
|
||||
Assert.Contains(
|
||||
result.Failures!,
|
||||
f => f.Contains("MxGateway:Alarms:PollIntervalMilliseconds", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A per-fetch cap below the 64-record floor must fail validation. The
|
||||
/// cap doubles as the truncation-detection threshold in the worker, so
|
||||
/// a tiny cap would make almost every fetch read as truncated.
|
||||
/// </summary>
|
||||
/// <param name="maxAlarmsPerFetch">Cap under test.</param>
|
||||
/// <param name="alarmsEnabled">Whether the central alarm monitor is on.</param>
|
||||
[Theory]
|
||||
[InlineData(63, false)]
|
||||
[InlineData(0, false)]
|
||||
[InlineData(-1, false)]
|
||||
[InlineData(63, true)]
|
||||
public void Validate_Fails_WhenMaxAlarmsPerFetchBelowFloor(
|
||||
int maxAlarmsPerFetch,
|
||||
bool alarmsEnabled)
|
||||
{
|
||||
GatewayOptions options = CloneWithAlarms(
|
||||
ValidOptions(),
|
||||
new AlarmsOptions
|
||||
{
|
||||
Enabled = alarmsEnabled,
|
||||
DefaultArea = "Galaxy",
|
||||
MaxAlarmsPerFetch = maxAlarmsPerFetch,
|
||||
});
|
||||
|
||||
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
|
||||
|
||||
Assert.False(result.Succeeded);
|
||||
Assert.Contains(
|
||||
result.Failures!,
|
||||
f => f.Contains("MxGateway:Alarms:MaxAlarmsPerFetch", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
/// <summary>Verifies the floor values themselves are accepted.</summary>
|
||||
[Fact]
|
||||
public void Validate_Succeeds_AtAlarmPollCadenceAndFetchCapFloors()
|
||||
{
|
||||
GatewayOptions options = CloneWithAlarms(
|
||||
ValidOptions(),
|
||||
new AlarmsOptions
|
||||
{
|
||||
Enabled = true,
|
||||
DefaultArea = "Galaxy",
|
||||
PollIntervalMilliseconds = 100,
|
||||
MaxAlarmsPerFetch = 64,
|
||||
});
|
||||
|
||||
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
|
||||
Assert.True(result.Succeeded);
|
||||
}
|
||||
|
||||
/// <summary>Verifies an invalid fallback mode is not validated when alarms are disabled.</summary>
|
||||
[Fact]
|
||||
public void Validate_Succeeds_WhenAlarmsDisabled_FallbackNotValidated()
|
||||
|
||||
@@ -47,6 +47,17 @@ public sealed class WorkerProcessLauncherTests
|
||||
"1500",
|
||||
processFactory.LastStartInfo.Environment[
|
||||
WorkerProcessLauncher.WorkerWriteCompletionWaitEnvironmentVariableName]);
|
||||
// MxGateway:Alarms defaults reach the worker's alarm poll loop and its
|
||||
// GetXmlCurrentAlarms2 cap (which is also its truncation threshold)
|
||||
// through the launch environment, not the command line.
|
||||
Assert.Equal(
|
||||
"500",
|
||||
processFactory.LastStartInfo.Environment[
|
||||
WorkerProcessLauncher.WorkerAlarmPollIntervalEnvironmentVariableName]);
|
||||
Assert.Equal(
|
||||
"1024",
|
||||
processFactory.LastStartInfo.Environment[
|
||||
WorkerProcessLauncher.WorkerMaxAlarmsPerFetchEnvironmentVariableName]);
|
||||
Assert.DoesNotContain(Nonce, handle.CommandLine.ToString(), StringComparison.Ordinal);
|
||||
Assert.DoesNotContain(Nonce, string.Join(" ", handle.CommandLine.Arguments), StringComparison.Ordinal);
|
||||
Assert.False(pipeReservation.DisposeCalled);
|
||||
|
||||
@@ -317,6 +317,346 @@ public sealed class WnWrapAlarmConsumerXmlTests
|
||||
Assert.Equal(MxAlarmStateKind.AckAlm, changedTransition.Record.State);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Truncation cliff (Task 23). GetXmlCurrentAlarms2 caps its reply at
|
||||
// maxAlmCnt with no "more available" flag. Before the guard, a capped fetch
|
||||
// shrank the retained snapshot, and every alarm past the cap vanished from
|
||||
// SnapshotActiveAlarms — which the gateway's reconcile pass reads as a
|
||||
// Clear and broadcasts to every StreamAlarms subscriber.
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/// <summary>
|
||||
/// A fetch that comes back holding exactly the cap must be treated as
|
||||
/// truncated; anything below the cap must not.
|
||||
/// </summary>
|
||||
/// <param name="fetchedRecordCount">Records the reply carried.</param>
|
||||
/// <param name="maxAlarmsPerFetch">The cap passed to the fetch.</param>
|
||||
/// <param name="expected">Whether the reply must read as truncated.</param>
|
||||
[Theory]
|
||||
[InlineData(1024, 1024, true)]
|
||||
[InlineData(1023, 1024, false)]
|
||||
[InlineData(0, 1024, false)]
|
||||
[InlineData(64, 64, true)]
|
||||
public void IsTruncatedFetch_AtOrAboveCap_IsTruncated(
|
||||
int fetchedRecordCount,
|
||||
int maxAlarmsPerFetch,
|
||||
bool expected)
|
||||
{
|
||||
Assert.Equal(
|
||||
expected,
|
||||
WnWrapAlarmConsumer.IsTruncatedFetch(fetchedRecordCount, maxAlarmsPerFetch));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// THE truncation-correctness test. A 1024-record fetch against a
|
||||
/// 1024 cap that does not mention a known-active alarm must NOT evict
|
||||
/// that alarm from the retained snapshot: the snapshot is what
|
||||
/// <see cref="WnWrapAlarmConsumer.SnapshotActiveAlarms"/> returns and
|
||||
/// what the gateway's reconcile diffs its cache against, so an
|
||||
/// eviction here is a Clear broadcast for an alarm that never cleared.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ApplySnapshotUpdate_WhenFetchTruncated_RetainsAlarmMissingFromCappedFetch()
|
||||
{
|
||||
const int Cap = 1024;
|
||||
Guid missingGuid = new Guid("11111111-1111-1111-1111-111111111111");
|
||||
|
||||
string xml = BuildAlarmXml(Cap);
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> next =
|
||||
WnWrapAlarmConsumer.ParseSnapshotXml(xml, out int fetchedRecordCount);
|
||||
|
||||
Assert.Equal(Cap, fetchedRecordCount);
|
||||
Assert.Equal(Cap, next.Count);
|
||||
Assert.False(next.ContainsKey(missingGuid));
|
||||
Assert.True(WnWrapAlarmConsumer.IsTruncatedFetch(fetchedRecordCount, Cap));
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> snapshot = new()
|
||||
{
|
||||
[missingGuid] = NewRecord(missingGuid, MxAlarmStateKind.UnackAlm),
|
||||
};
|
||||
|
||||
// The diff itself never emits a Clear for a disappearance — the Clear
|
||||
// is the eviction, one level up. Pin both halves.
|
||||
IReadOnlyList<MxAlarmTransitionEvent> transitions =
|
||||
WnWrapAlarmConsumer.ComputeTransitions(snapshot, next);
|
||||
Assert.DoesNotContain(transitions, t => t.Record.AlarmGuid == missingGuid);
|
||||
|
||||
WnWrapAlarmConsumer.ApplySnapshotUpdate(snapshot, next, truncated: true);
|
||||
|
||||
// NO Clear: the alarm the capped fetch had no room to mention survives,
|
||||
// so the gateway's reconcile still sees it active.
|
||||
Assert.True(snapshot.ContainsKey(missingGuid));
|
||||
Assert.Equal(MxAlarmStateKind.UnackAlm, snapshot[missingGuid].State);
|
||||
Assert.Equal(Cap + 1, snapshot.Count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The control case: the same missing alarm against a sub-cap fetch
|
||||
/// must still be evicted, because a complete fetch IS authoritative
|
||||
/// about absence. Without this, the truncation guard would have
|
||||
/// silently disabled clears altogether.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ApplySnapshotUpdate_WhenFetchBelowCap_EvictsAlarmMissingFromFetch()
|
||||
{
|
||||
const int Cap = 1024;
|
||||
Guid missingGuid = new Guid("11111111-1111-1111-1111-111111111111");
|
||||
|
||||
string xml = BuildAlarmXml(Cap - 1);
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> next =
|
||||
WnWrapAlarmConsumer.ParseSnapshotXml(xml, out int fetchedRecordCount);
|
||||
|
||||
Assert.Equal(Cap - 1, fetchedRecordCount);
|
||||
Assert.False(WnWrapAlarmConsumer.IsTruncatedFetch(fetchedRecordCount, Cap));
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> snapshot = new()
|
||||
{
|
||||
[missingGuid] = NewRecord(missingGuid, MxAlarmStateKind.UnackAlm),
|
||||
};
|
||||
|
||||
WnWrapAlarmConsumer.ApplySnapshotUpdate(snapshot, next, truncated: false);
|
||||
|
||||
// Clear as today: the alarm drops out of the snapshot, and the
|
||||
// gateway's reconcile turns that absence into a Clear.
|
||||
Assert.False(snapshot.ContainsKey(missingGuid));
|
||||
Assert.Equal(Cap - 1, snapshot.Count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A truncated fetch must still apply the alarms it DID carry — the
|
||||
/// guard suppresses eviction, not the update. An alarm whose state
|
||||
/// changed inside a capped reply still transitions.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ApplySnapshotUpdate_WhenFetchTruncated_StillAppliesPresentAlarms()
|
||||
{
|
||||
Guid presentGuid = new Guid("22222222-2222-2222-2222-222222222222");
|
||||
Guid missingGuid = new Guid("11111111-1111-1111-1111-111111111111");
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> snapshot = new()
|
||||
{
|
||||
[presentGuid] = NewRecord(presentGuid, MxAlarmStateKind.UnackAlm),
|
||||
[missingGuid] = NewRecord(missingGuid, MxAlarmStateKind.UnackAlm),
|
||||
};
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> next = new()
|
||||
{
|
||||
[presentGuid] = NewRecord(presentGuid, MxAlarmStateKind.AckAlm),
|
||||
};
|
||||
|
||||
IReadOnlyList<MxAlarmTransitionEvent> transitions =
|
||||
WnWrapAlarmConsumer.ComputeTransitions(snapshot, next);
|
||||
|
||||
MxAlarmTransitionEvent single = Assert.Single(transitions);
|
||||
Assert.Equal(presentGuid, single.Record.AlarmGuid);
|
||||
Assert.Equal(MxAlarmStateKind.UnackAlm, single.PreviousState);
|
||||
Assert.Equal(MxAlarmStateKind.AckAlm, single.Record.State);
|
||||
|
||||
WnWrapAlarmConsumer.ApplySnapshotUpdate(snapshot, next, truncated: true);
|
||||
|
||||
Assert.Equal(MxAlarmStateKind.AckAlm, snapshot[presentGuid].State);
|
||||
Assert.True(snapshot.ContainsKey(missingGuid));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Truncation is detected from the raw ALARM-element count, not the
|
||||
/// parsed dictionary size. A record dropped for a malformed GUID still
|
||||
/// consumed a slot in the capped reply, so counting only survivors
|
||||
/// would let a truncated fetch pass as complete and re-open the cliff.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ParseSnapshotXml_CountsRawAlarmElements_IncludingRecordsDroppedForBadGuid()
|
||||
{
|
||||
const int Cap = 64;
|
||||
string xml = BuildAlarmXml(Cap).Replace(
|
||||
"<GUID>00000000000000000000000000000001</GUID>",
|
||||
"<GUID>not-a-guid</GUID>");
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> records =
|
||||
WnWrapAlarmConsumer.ParseSnapshotXml(xml, out int fetchedRecordCount);
|
||||
|
||||
Assert.Equal(Cap, fetchedRecordCount);
|
||||
Assert.Equal(Cap - 1, records.Count);
|
||||
Assert.True(WnWrapAlarmConsumer.IsTruncatedFetch(fetchedRecordCount, Cap));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The per-fetch cap comes from the launcher-set environment variable;
|
||||
/// a missing, unparseable, or below-floor value must fall back to the
|
||||
/// 1024 default rather than throw. A bad environment value must never
|
||||
/// stop the alarm consumer from starting.
|
||||
/// </summary>
|
||||
/// <param name="environmentValue">Raw environment value under test.</param>
|
||||
/// <param name="expected">Expected resolved cap.</param>
|
||||
[Theory]
|
||||
[InlineData(null, 1024)]
|
||||
[InlineData("", 1024)]
|
||||
[InlineData("not-a-number", 1024)]
|
||||
[InlineData("0", 1024)]
|
||||
[InlineData("-5", 1024)]
|
||||
[InlineData("63", 1024)]
|
||||
[InlineData("64", 64)]
|
||||
[InlineData("4096", 4096)]
|
||||
public void ResolveMaxAlarmsPerFetch_WithEnvironmentValue_FallsBackToDefaultWhenUnusable(
|
||||
string? environmentValue,
|
||||
int expected)
|
||||
{
|
||||
string? original = Environment.GetEnvironmentVariable(
|
||||
WnWrapAlarmConsumer.MaxAlarmsPerFetchEnvironmentVariableName);
|
||||
try
|
||||
{
|
||||
Environment.SetEnvironmentVariable(
|
||||
WnWrapAlarmConsumer.MaxAlarmsPerFetchEnvironmentVariableName,
|
||||
environmentValue);
|
||||
|
||||
Assert.Equal(expected, WnWrapAlarmConsumer.ResolveMaxAlarmsPerFetch());
|
||||
}
|
||||
finally
|
||||
{
|
||||
Environment.SetEnvironmentVariable(
|
||||
WnWrapAlarmConsumer.MaxAlarmsPerFetchEnvironmentVariableName,
|
||||
original);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds a well-formed ALARM_RECORDS payload with
|
||||
/// <paramref name="count"/> distinct alarms. GUIDs are the dashless
|
||||
/// 32-char hex form wnwrap actually emits, numbered from 1 so a test
|
||||
/// can name one (e.g. ...0001) to corrupt.
|
||||
/// </summary>
|
||||
/// <param name="count">Number of ALARM elements to emit.</param>
|
||||
/// <returns>The XML payload.</returns>
|
||||
private static string BuildAlarmXml(int count)
|
||||
{
|
||||
System.Text.StringBuilder sb = new System.Text.StringBuilder();
|
||||
sb.Append("<?xml version=\"1.0\"?><ALARM_RECORDS COUNT=\"")
|
||||
.Append(count)
|
||||
.Append("\">");
|
||||
for (int index = 1; index <= count; index++)
|
||||
{
|
||||
sb.Append("<ALARM><GUID>")
|
||||
.Append(index.ToString("X32", System.Globalization.CultureInfo.InvariantCulture))
|
||||
.Append("</GUID>")
|
||||
.Append("<DATE>2026/5/1</DATE><TIME>13:26:14.709</TIME>")
|
||||
.Append("<GMTOFFSET>240</GMTOFFSET><DSTADJUST>0</DSTADJUST>")
|
||||
.Append("<PROVIDER_NODE>TEST-NODE</PROVIDER_NODE>")
|
||||
.Append("<PROVIDER_NAME>Galaxy</PROVIDER_NAME>")
|
||||
.Append("<GROUP>TestArea</GROUP>")
|
||||
.Append("<TAGNAME>TestMachine_")
|
||||
.Append(index.ToString(System.Globalization.CultureInfo.InvariantCulture))
|
||||
.Append(".TestAlarm</TAGNAME>")
|
||||
.Append("<TYPE>DSC</TYPE><VALUE>true</VALUE><LIMIT>true</LIMIT>")
|
||||
.Append("<PRIORITY>500</PRIORITY><STATE>UNACK_ALM</STATE>")
|
||||
.Append("<OPERATOR_NODE></OPERATOR_NODE><OPERATOR_NAME></OPERATOR_NAME>")
|
||||
.Append("<ALARM_COMMENT>Test alarm</ALARM_COMMENT></ALARM>");
|
||||
}
|
||||
sb.Append("</ALARM_RECORDS>");
|
||||
return sb.ToString();
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Single-pass parse parity (Task 23). The per-field SelectSingleNode calls
|
||||
// were replaced by one walk over ChildNodes; these pin the semantics that
|
||||
// walk has to reproduce.
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/// <summary>
|
||||
/// An absent child element and a present-but-empty one must both yield
|
||||
/// <see cref="string.Empty"/>, exactly as
|
||||
/// <c>SelectSingleNode(name)?.InnerText ?? string.Empty</c> did.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ParseSnapshotXml_WithAbsentAndEmptyChildren_YieldsEmptyStrings()
|
||||
{
|
||||
string xml =
|
||||
"<?xml version=\"1.0\"?><ALARM_RECORDS COUNT=\"1\">" +
|
||||
"<ALARM><GUID>BCC4705395424D65BDAABCDEA6A32A73</GUID>" +
|
||||
"<TAGNAME></TAGNAME>" +
|
||||
"<STATE>UNACK_ALM</STATE></ALARM>" +
|
||||
"</ALARM_RECORDS>";
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> records = WnWrapAlarmConsumer.ParseSnapshotXml(xml);
|
||||
MxAlarmSnapshotRecord record = records[new Guid("BCC47053-9542-4D65-BDAA-BCDEA6A32A73")];
|
||||
|
||||
Assert.Equal(string.Empty, record.TagName); // present but empty
|
||||
Assert.Equal(string.Empty, record.ProviderNode); // absent
|
||||
Assert.Equal(string.Empty, record.Value); // absent
|
||||
Assert.Equal(string.Empty, record.AlarmComment); // absent
|
||||
Assert.Equal(0, record.Priority); // absent → ParseInt("") → 0
|
||||
Assert.Equal(MxAlarmStateKind.UnackAlm, record.State);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Duplicate child elements must resolve to the FIRST occurrence —
|
||||
/// <c>SelectSingleNode</c> returned the first match, so a last-wins
|
||||
/// walk would silently change which value a malformed payload yields.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ParseSnapshotXml_WithDuplicateChildElements_TakesFirstOccurrence()
|
||||
{
|
||||
string xml =
|
||||
"<?xml version=\"1.0\"?><ALARM_RECORDS COUNT=\"1\">" +
|
||||
"<ALARM><GUID>BCC4705395424D65BDAABCDEA6A32A73</GUID>" +
|
||||
"<TAGNAME>First</TAGNAME><TAGNAME>Second</TAGNAME>" +
|
||||
"<PRIORITY>100</PRIORITY><PRIORITY>900</PRIORITY>" +
|
||||
"<STATE>UNACK_ALM</STATE></ALARM>" +
|
||||
"</ALARM_RECORDS>";
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> records = WnWrapAlarmConsumer.ParseSnapshotXml(xml);
|
||||
MxAlarmSnapshotRecord record = records[new Guid("BCC47053-9542-4D65-BDAA-BCDEA6A32A73")];
|
||||
|
||||
Assert.Equal("First", record.TagName);
|
||||
Assert.Equal(100, record.Priority);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Field names are matched ordinally and case-sensitively, as the
|
||||
/// XPath node test was. A lowercase element must not populate the
|
||||
/// field its uppercase counterpart owns.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ParseSnapshotXml_WithWrongCaseChildElement_DoesNotPopulateField()
|
||||
{
|
||||
string xml =
|
||||
"<?xml version=\"1.0\"?><ALARM_RECORDS COUNT=\"1\">" +
|
||||
"<ALARM><GUID>BCC4705395424D65BDAABCDEA6A32A73</GUID>" +
|
||||
"<tagname>ShouldBeIgnored</tagname>" +
|
||||
"<STATE>UNACK_ALM</STATE></ALARM>" +
|
||||
"</ALARM_RECORDS>";
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> records = WnWrapAlarmConsumer.ParseSnapshotXml(xml);
|
||||
MxAlarmSnapshotRecord record = records[new Guid("BCC47053-9542-4D65-BDAA-BCDEA6A32A73")];
|
||||
|
||||
Assert.Equal(string.Empty, record.TagName);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Non-element children (whitespace text, comments, CDATA) must be
|
||||
/// skipped rather than matched by name, and a nested element must
|
||||
/// contribute its InnerText exactly as <c>InnerText</c> always did.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ParseSnapshotXml_WithCommentsAndWhitespace_IgnoresNonElementChildren()
|
||||
{
|
||||
string xml =
|
||||
"<?xml version=\"1.0\"?><ALARM_RECORDS COUNT=\"1\">\n" +
|
||||
" <ALARM>\n" +
|
||||
" <!-- a comment -->\n" +
|
||||
" <GUID>BCC4705395424D65BDAABCDEA6A32A73</GUID>\n" +
|
||||
" <TAGNAME>TestMachine.TestAlarm</TAGNAME>\n" +
|
||||
" <STATE>UNACK_ALM</STATE>\n" +
|
||||
" </ALARM>\n" +
|
||||
"</ALARM_RECORDS>";
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> records =
|
||||
WnWrapAlarmConsumer.ParseSnapshotXml(xml, out int fetchedRecordCount);
|
||||
|
||||
Assert.Equal(1, fetchedRecordCount);
|
||||
MxAlarmSnapshotRecord record = Assert.Single(records).Value;
|
||||
Assert.Equal("TestMachine.TestAlarm", record.TagName);
|
||||
Assert.Equal(MxAlarmStateKind.UnackAlm, record.State);
|
||||
}
|
||||
|
||||
private static MxAlarmSnapshotRecord NewRecord(Guid guid, MxAlarmStateKind state)
|
||||
{
|
||||
return new MxAlarmSnapshotRecord
|
||||
|
||||
@@ -19,7 +19,26 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
||||
internal const string WriteCompletionWaitEnvironmentVariableName =
|
||||
"MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS";
|
||||
|
||||
private static readonly TimeSpan AlarmPollInterval = TimeSpan.FromMilliseconds(500);
|
||||
/// <summary>
|
||||
/// Environment variable the gateway's WorkerProcessLauncher sets from
|
||||
/// MxGateway:Alarms:PollIntervalMilliseconds. A missing or invalid
|
||||
/// value falls back to <see cref="DefaultAlarmPollInterval"/>.
|
||||
/// </summary>
|
||||
internal const string AlarmPollIntervalEnvironmentVariableName =
|
||||
"MXGATEWAY_ALARM_POLL_INTERVAL_MS";
|
||||
|
||||
/// <summary>
|
||||
/// Floor on the resolved alarm poll cadence. Mirrors the gateway-side
|
||||
/// MxGateway:Alarms:PollIntervalMilliseconds minimum so a value that
|
||||
/// slipped past validation (or an environment set by hand) still can
|
||||
/// not starve the STA's command path.
|
||||
/// </summary>
|
||||
internal static readonly TimeSpan MinimumAlarmPollInterval = TimeSpan.FromMilliseconds(100);
|
||||
|
||||
/// <summary>Default alarm poll cadence when the environment says nothing usable.</summary>
|
||||
internal static readonly TimeSpan DefaultAlarmPollInterval = TimeSpan.FromMilliseconds(500);
|
||||
|
||||
private readonly TimeSpan alarmPollInterval = ResolveAlarmPollInterval();
|
||||
|
||||
private readonly IMxAccessComObjectFactory factory;
|
||||
private readonly IMxAccessEventSink eventSink;
|
||||
@@ -191,6 +210,31 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
||||
: MxAccessCommandExecutor.DefaultWriteCompletionTimeout;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves the alarm poll cadence from the launcher-provided
|
||||
/// environment variable. A missing, unparseable, or out-of-range value
|
||||
/// falls back to <see cref="DefaultAlarmPollInterval"/> rather than
|
||||
/// faulting the worker: an alarm poll that runs at the default cadence
|
||||
/// is always safe, and a session that refuses to start over a
|
||||
/// mistyped environment variable is not.
|
||||
/// </summary>
|
||||
/// <returns>The cadence the alarm poll loop waits between polls.</returns>
|
||||
internal static TimeSpan ResolveAlarmPollInterval()
|
||||
{
|
||||
string? value = Environment.GetEnvironmentVariable(AlarmPollIntervalEnvironmentVariableName);
|
||||
if (!int.TryParse(
|
||||
value,
|
||||
System.Globalization.NumberStyles.Integer,
|
||||
System.Globalization.CultureInfo.InvariantCulture,
|
||||
out int milliseconds))
|
||||
{
|
||||
return DefaultAlarmPollInterval;
|
||||
}
|
||||
|
||||
TimeSpan resolved = TimeSpan.FromMilliseconds(milliseconds);
|
||||
return resolved < MinimumAlarmPollInterval ? DefaultAlarmPollInterval : resolved;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the MXAccess COM session asynchronously.
|
||||
/// </summary>
|
||||
@@ -274,7 +318,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(AlarmPollInterval, cancellationToken).ConfigureAwait(false);
|
||||
await Task.Delay(alarmPollInterval, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Diagnostics;
|
||||
using System.Globalization;
|
||||
using System.Runtime.InteropServices;
|
||||
using System.Xml;
|
||||
@@ -49,11 +50,39 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
private const string DefaultVersion = "1.0";
|
||||
private const int DefaultMaxAlarmsPerFetch = 1024;
|
||||
|
||||
/// <summary>
|
||||
/// Floor on the resolved per-fetch cap. Mirrors the gateway-side
|
||||
/// <c>MxGateway:Alarms:MaxAlarmsPerFetch</c> minimum so a value that
|
||||
/// slipped past startup validation still leaves a usable snapshot
|
||||
/// window.
|
||||
/// </summary>
|
||||
internal const int MinimumMaxAlarmsPerFetch = 64;
|
||||
|
||||
/// <summary>
|
||||
/// Environment variable the gateway's <c>WorkerProcessLauncher</c>
|
||||
/// sets from <c>MxGateway:Alarms:MaxAlarmsPerFetch</c>. A missing,
|
||||
/// unparseable, or below-floor value falls back to
|
||||
/// <see cref="DefaultMaxAlarmsPerFetch"/> — a bad environment value
|
||||
/// must never keep the alarm consumer from starting.
|
||||
/// </summary>
|
||||
internal const string MaxAlarmsPerFetchEnvironmentVariableName =
|
||||
"MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH";
|
||||
|
||||
/// <summary>
|
||||
/// Minimum gap between truncated-fetch warnings. A galaxy parked above
|
||||
/// the cap truncates on *every* poll, so an unthrottled warning would
|
||||
/// be two log lines a second forever.
|
||||
/// </summary>
|
||||
private const long TruncationWarningIntervalMilliseconds = 60_000;
|
||||
|
||||
private readonly object syncRoot = new object();
|
||||
private readonly Dictionary<Guid, MxAlarmSnapshotRecord> latestSnapshot =
|
||||
new Dictionary<Guid, MxAlarmSnapshotRecord>();
|
||||
private readonly int maxAlarmsPerFetch;
|
||||
private readonly Stopwatch truncationWarningClock = Stopwatch.StartNew();
|
||||
|
||||
private long lastTruncationWarningMilliseconds = -TruncationWarningIntervalMilliseconds;
|
||||
private long truncatedFetchCount;
|
||||
private wwAlarmConsumerClass? client;
|
||||
private wwAlarmConsumerClass? ackClient;
|
||||
private bool subscribed;
|
||||
@@ -67,7 +96,7 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
/// every COM call stays on the STA that owns the apartment.
|
||||
/// </summary>
|
||||
public WnWrapAlarmConsumer()
|
||||
: this(new wwAlarmConsumerClass(), DefaultMaxAlarmsPerFetch)
|
||||
: this(new wwAlarmConsumerClass(), ResolveMaxAlarmsPerFetch())
|
||||
{
|
||||
}
|
||||
|
||||
@@ -86,6 +115,27 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
: DefaultMaxAlarmsPerFetch;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves the per-fetch cap from the launcher-provided environment
|
||||
/// variable. A missing, unparseable, or below-floor value falls back
|
||||
/// to <see cref="DefaultMaxAlarmsPerFetch"/> rather than throwing:
|
||||
/// polling at the default cap is always safe, and a session that
|
||||
/// refuses to subscribe over a mistyped environment variable is not.
|
||||
/// </summary>
|
||||
/// <returns>The cap passed to <c>GetXmlCurrentAlarms2</c>.</returns>
|
||||
internal static int ResolveMaxAlarmsPerFetch()
|
||||
{
|
||||
string? value = Environment.GetEnvironmentVariable(MaxAlarmsPerFetchEnvironmentVariableName);
|
||||
return int.TryParse(
|
||||
value,
|
||||
NumberStyles.Integer,
|
||||
CultureInfo.InvariantCulture,
|
||||
out int cap)
|
||||
&& cap >= MinimumMaxAlarmsPerFetch
|
||||
? cap
|
||||
: DefaultMaxAlarmsPerFetch;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fires once per detected alarm-state transition (raise, acknowledge,
|
||||
/// clear, or new-alarm-already-acked-on-arrival), dispatched from
|
||||
@@ -282,6 +332,15 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>
|
||||
/// This is the set the gateway's reconcile pass diffs its cache
|
||||
/// against, so an alarm missing here is what the gateway reads as a
|
||||
/// Clear. After a truncated fetch the snapshot deliberately still
|
||||
/// carries alarms the capped reply could not mention — see
|
||||
/// <see cref="PollOnce"/>. Erring toward "still active" keeps the feed
|
||||
/// at-least-once: a stale entry is repaired by the next sub-cap poll,
|
||||
/// whereas a dropped one is broadcast as a Clear that never happened.
|
||||
/// </remarks>
|
||||
public IReadOnlyList<MxAlarmSnapshotRecord> SnapshotActiveAlarms()
|
||||
{
|
||||
if (disposed) throw new ObjectDisposedException(nameof(WnWrapAlarmConsumer));
|
||||
@@ -300,6 +359,16 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sink for the rate-limited truncated-fetch warning. Defaults to
|
||||
/// <see cref="Console.Error"/>, the stream
|
||||
/// <c>WorkerConsoleLogger</c> already writes to, so the line lands in
|
||||
/// the worker's captured stderr alongside the rest of its log. Tests
|
||||
/// substitute a collector. The message carries identifiers and counts
|
||||
/// only — never alarm values, tag names, or comments.
|
||||
/// </summary>
|
||||
internal Action<string>? TruncationWarningSink { get; set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public void PollOnce()
|
||||
{
|
||||
@@ -316,17 +385,28 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
string xml = xmlObj?.ToString() ?? string.Empty;
|
||||
if (xml.Length == 0) return;
|
||||
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> next = ParseSnapshotXml(xml);
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> next = ParseSnapshotXml(xml, out int fetchedRecordCount);
|
||||
|
||||
// TRUNCATION CLIFF. GetXmlCurrentAlarms2 caps its reply at maxAlmCnt
|
||||
// and gives no "there is more" flag, so a reply holding exactly the
|
||||
// cap is indistinguishable from a galaxy that happens to have exactly
|
||||
// that many active alarms. Treat the ambiguous case as truncated: the
|
||||
// false-positive cost is a snapshot that stays stale for one poll, the
|
||||
// false-negative cost is every alarm past the cap reading as cleared.
|
||||
bool truncated = IsTruncatedFetch(fetchedRecordCount, maxAlarmsPerFetch);
|
||||
|
||||
IReadOnlyList<MxAlarmTransitionEvent> transitions;
|
||||
int retainedCount;
|
||||
lock (syncRoot)
|
||||
{
|
||||
transitions = ComputeTransitions(latestSnapshot, next);
|
||||
latestSnapshot.Clear();
|
||||
foreach (KeyValuePair<Guid, MxAlarmSnapshotRecord> kv in next)
|
||||
{
|
||||
latestSnapshot[kv.Key] = kv.Value;
|
||||
ApplySnapshotUpdate(latestSnapshot, next, truncated);
|
||||
retainedCount = latestSnapshot.Count;
|
||||
}
|
||||
|
||||
if (truncated)
|
||||
{
|
||||
WarnTruncatedFetch(fetchedRecordCount, next.Count, retainedCount);
|
||||
}
|
||||
|
||||
if (transitions.Count == 0) return;
|
||||
@@ -338,6 +418,111 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Decides whether a fetch that came back holding
|
||||
/// <paramref name="fetchedRecordCount"/> records hit the cap.
|
||||
/// <c>GetXmlCurrentAlarms2</c> caps its reply at <c>maxAlmCnt</c> and
|
||||
/// offers no "more available" flag, so a reply at exactly the cap is
|
||||
/// indistinguishable from a galaxy that happens to hold exactly that
|
||||
/// many active alarms; both are treated as truncated. Exposed as
|
||||
/// <c>internal static</c> so the rule is unit-testable without the
|
||||
/// wnwrapConsumer COM object.
|
||||
/// </summary>
|
||||
/// <param name="fetchedRecordCount">ALARM records the reply carried.</param>
|
||||
/// <param name="maxAlarmsPerFetch">The cap that was passed to the fetch.</param>
|
||||
/// <returns><see langword="true"/> when the reply must be treated as truncated.</returns>
|
||||
internal static bool IsTruncatedFetch(int fetchedRecordCount, int maxAlarmsPerFetch)
|
||||
{
|
||||
return fetchedRecordCount >= maxAlarmsPerFetch;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Folds a freshly fetched snapshot into the retained one.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// A complete fetch is authoritative about <em>absence</em>:
|
||||
/// replacing the snapshot wholesale is exactly what lets a cleared
|
||||
/// alarm drop out of <see cref="SnapshotActiveAlarms"/>, which is
|
||||
/// the evidence the gateway's reconcile pass turns into a Clear.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// A truncated fetch is authoritative about <em>presence</em>
|
||||
/// only. Merging rather than replacing means alarms the capped
|
||||
/// reply did carry update normally, while alarms it had no room to
|
||||
/// mention are retained untouched — so the gateway infers no Clear
|
||||
/// from their absence. The cost is bounded staleness: a genuinely
|
||||
/// cleared alarm can linger in the snapshot until the first
|
||||
/// sub-cap fetch evicts it. That direction is the safe one for an
|
||||
/// at-least-once alarm feed — a late Clear is repaired by the next
|
||||
/// complete poll, whereas a Clear that never happened is broadcast
|
||||
/// to every subscriber and cannot be taken back.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="snapshot">The retained snapshot, updated in place.</param>
|
||||
/// <param name="next">The snapshot just parsed from the fetch.</param>
|
||||
/// <param name="truncated">Whether the fetch hit the per-fetch cap.</param>
|
||||
internal static void ApplySnapshotUpdate(
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> snapshot,
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> next,
|
||||
bool truncated)
|
||||
{
|
||||
if (snapshot is null) throw new ArgumentNullException(nameof(snapshot));
|
||||
if (next is null) throw new ArgumentNullException(nameof(next));
|
||||
|
||||
if (!truncated)
|
||||
{
|
||||
snapshot.Clear();
|
||||
}
|
||||
|
||||
foreach (KeyValuePair<Guid, MxAlarmSnapshotRecord> kv in next)
|
||||
{
|
||||
snapshot[kv.Key] = kv.Value;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Emits the truncated-fetch warning at most once per
|
||||
/// <see cref="TruncationWarningIntervalMilliseconds"/>, counting every
|
||||
/// truncated poll in between so the operator can tell a one-off burst
|
||||
/// from a galaxy permanently parked above the cap.
|
||||
/// </summary>
|
||||
/// <param name="fetchedRecordCount">ALARM records the capped reply carried.</param>
|
||||
/// <param name="parsedRecordCount">Records that survived GUID parsing.</param>
|
||||
/// <param name="retainedCount">Snapshot size after the merge.</param>
|
||||
private void WarnTruncatedFetch(int fetchedRecordCount, int parsedRecordCount, int retainedCount)
|
||||
{
|
||||
long total;
|
||||
long elapsed;
|
||||
lock (syncRoot)
|
||||
{
|
||||
total = ++truncatedFetchCount;
|
||||
elapsed = truncationWarningClock.ElapsedMilliseconds;
|
||||
if (elapsed - lastTruncationWarningMilliseconds < TruncationWarningIntervalMilliseconds)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
lastTruncationWarningMilliseconds = elapsed;
|
||||
}
|
||||
|
||||
// Identifiers and counts only — no tag names, values, limits, or
|
||||
// comments, per the gateway's "don't log tag values by default" rule.
|
||||
string message = string.Format(
|
||||
CultureInfo.InvariantCulture,
|
||||
"level=Warning event=AlarmSnapshotTruncated maxAlarmsPerFetch={0} fetchedRecords={1} "
|
||||
+ "parsedRecords={2} retainedSnapshotSize={3} truncatedFetchesSinceStart={4} "
|
||||
+ "note=absence-implies-clear suppressed for this poll; raise MxGateway:Alarms:MaxAlarmsPerFetch",
|
||||
maxAlarmsPerFetch,
|
||||
fetchedRecordCount,
|
||||
parsedRecordCount,
|
||||
retainedCount,
|
||||
total);
|
||||
|
||||
Action<string> sink = TruncationWarningSink ?? Console.Error.WriteLine;
|
||||
sink(message);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Pure snapshot-to-transitions diff. Compares the previous polled
|
||||
/// snapshot to the next snapshot and produces one
|
||||
@@ -355,6 +540,21 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
/// <item><description>A GUID present in both with a different <see cref="MxAlarmSnapshotRecord.State"/> produces a transition carrying the prior state.</description></item>
|
||||
/// <item><description>A GUID present in <paramref name="previous"/> but absent from <paramref name="next"/> produces no transition. AVEVA drops cleared alarms from the active set; the snapshot simply stops mentioning them.</description></item>
|
||||
/// </list>
|
||||
/// <para>
|
||||
/// The absence rule is deliberately inert here: this diff never
|
||||
/// turns a disappearance into a Clear. The clear is inferred one
|
||||
/// level up, by the gateway's reconcile pass diffing its cache
|
||||
/// against <see cref="SnapshotActiveAlarms"/>, which reads the
|
||||
/// snapshot <see cref="PollOnce"/> maintains from these same
|
||||
/// dictionaries. That is why the truncation guard lives in
|
||||
/// <see cref="PollOnce"/>'s snapshot update rather than in this
|
||||
/// method: suppressing "absence implies Clear" means declining to
|
||||
/// *evict* alarms a capped fetch could not mention, not declining
|
||||
/// to emit a transition this method was never going to emit.
|
||||
/// Passing a truncated <paramref name="next"/> straight through
|
||||
/// stays correct for the alarms it does carry — first sightings
|
||||
/// and state changes are computed from presence alone.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="previous">The snapshot from the previous poll (or empty on first call).</param>
|
||||
/// <param name="next">The snapshot just parsed from <c>GetXmlCurrentAlarms2</c>.</param>
|
||||
@@ -393,9 +593,29 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
/// <param name="xml">The XML snapshot payload.</param>
|
||||
/// <returns>The parsed alarm snapshot records, keyed by alarm GUID.</returns>
|
||||
public static Dictionary<Guid, MxAlarmSnapshotRecord> ParseSnapshotXml(string xml)
|
||||
{
|
||||
return ParseSnapshotXml(xml, out _);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="ParseSnapshotXml(string)"/> plus the raw
|
||||
/// <c>ALARM</c>-element count, which <see cref="PollOnce"/> compares
|
||||
/// against the per-fetch cap to detect a truncated reply. The raw
|
||||
/// count — not the dictionary size — is the right signal: records
|
||||
/// dropped for a malformed GUID still consumed a slot in the capped
|
||||
/// reply, so counting only the survivors would under-report and let a
|
||||
/// truncated fetch pass as complete.
|
||||
/// </summary>
|
||||
/// <param name="xml">The XML snapshot payload.</param>
|
||||
/// <param name="alarmRecordCount">Number of <c>ALARM</c> elements in the payload.</param>
|
||||
/// <returns>The parsed alarm snapshot records, keyed by alarm GUID.</returns>
|
||||
internal static Dictionary<Guid, MxAlarmSnapshotRecord> ParseSnapshotXml(
|
||||
string xml,
|
||||
out int alarmRecordCount)
|
||||
{
|
||||
Dictionary<Guid, MxAlarmSnapshotRecord> records =
|
||||
new Dictionary<Guid, MxAlarmSnapshotRecord>();
|
||||
alarmRecordCount = 0;
|
||||
|
||||
if (string.IsNullOrWhiteSpace(xml)) return records;
|
||||
|
||||
@@ -404,45 +624,96 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
|
||||
XmlNodeList? alarmNodes = doc.SelectNodes("/ALARM_RECORDS/ALARM");
|
||||
if (alarmNodes is null) return records;
|
||||
|
||||
alarmRecordCount = alarmNodes.Count;
|
||||
|
||||
foreach (XmlNode alarmNode in alarmNodes)
|
||||
{
|
||||
string guidHex = TextOf(alarmNode, "GUID");
|
||||
// One pass over the child elements instead of 17 SelectSingleNode
|
||||
// XPath evaluations per record. At the 1024-record cap that is the
|
||||
// difference between ~17k XPath expression compilations plus child
|
||||
// walks and one walk per record, on the STA that also serves reads
|
||||
// and writes.
|
||||
//
|
||||
// Parity with the XPath it replaces: SelectSingleNode(name) picks
|
||||
// the FIRST child ELEMENT of that name (ordinal, case-sensitive)
|
||||
// and yields its InnerText; an absent child yields string.Empty,
|
||||
// and so does a present-but-empty one. Hence: element nodes only,
|
||||
// first-match-wins (null means "not seen yet"), and a single
|
||||
// coalesce to string.Empty at the end — which keeps the absent and
|
||||
// empty cases indistinguishable exactly as before.
|
||||
string? guidHex = null;
|
||||
string? xmlDate = null;
|
||||
string? xmlTime = null;
|
||||
string? gmtOffset = null;
|
||||
string? dstAdjust = null;
|
||||
string? providerNode = null;
|
||||
string? providerName = null;
|
||||
string? group = null;
|
||||
string? tagName = null;
|
||||
string? type = null;
|
||||
string? value = null;
|
||||
string? limit = null;
|
||||
string? priority = null;
|
||||
string? state = null;
|
||||
string? operatorNode = null;
|
||||
string? operatorName = null;
|
||||
string? alarmComment = null;
|
||||
|
||||
foreach (XmlNode child in alarmNode.ChildNodes)
|
||||
{
|
||||
if (child.NodeType != XmlNodeType.Element) continue;
|
||||
|
||||
switch (child.Name)
|
||||
{
|
||||
case "GUID": guidHex ??= child.InnerText; break;
|
||||
case "DATE": xmlDate ??= child.InnerText; break;
|
||||
case "TIME": xmlTime ??= child.InnerText; break;
|
||||
case "GMTOFFSET": gmtOffset ??= child.InnerText; break;
|
||||
case "DSTADJUST": dstAdjust ??= child.InnerText; break;
|
||||
case "PROVIDER_NODE": providerNode ??= child.InnerText; break;
|
||||
case "PROVIDER_NAME": providerName ??= child.InnerText; break;
|
||||
case "GROUP": group ??= child.InnerText; break;
|
||||
case "TAGNAME": tagName ??= child.InnerText; break;
|
||||
case "TYPE": type ??= child.InnerText; break;
|
||||
case "VALUE": value ??= child.InnerText; break;
|
||||
case "LIMIT": limit ??= child.InnerText; break;
|
||||
case "PRIORITY": priority ??= child.InnerText; break;
|
||||
case "STATE": state ??= child.InnerText; break;
|
||||
case "OPERATOR_NODE": operatorNode ??= child.InnerText; break;
|
||||
case "OPERATOR_NAME": operatorName ??= child.InnerText; break;
|
||||
case "ALARM_COMMENT": alarmComment ??= child.InnerText; break;
|
||||
}
|
||||
}
|
||||
|
||||
if (!TryParseHexGuid(guidHex, out Guid guid)) continue;
|
||||
|
||||
string xmlDate = TextOf(alarmNode, "DATE");
|
||||
string xmlTime = TextOf(alarmNode, "TIME");
|
||||
int gmtOffset = ParseInt(TextOf(alarmNode, "GMTOFFSET"));
|
||||
int dstAdjust = ParseInt(TextOf(alarmNode, "DSTADJUST"));
|
||||
DateTime tsUtc = AlarmRecordTransitionMapper.ParseTransitionTimestampUtc(
|
||||
xmlDate, xmlTime, gmtOffset, dstAdjust);
|
||||
xmlDate ?? string.Empty,
|
||||
xmlTime ?? string.Empty,
|
||||
ParseInt(gmtOffset ?? string.Empty),
|
||||
ParseInt(dstAdjust ?? string.Empty));
|
||||
|
||||
records[guid] = new MxAlarmSnapshotRecord
|
||||
{
|
||||
AlarmGuid = guid,
|
||||
TransitionTimestampUtc = tsUtc,
|
||||
ProviderNode = TextOf(alarmNode, "PROVIDER_NODE"),
|
||||
ProviderName = TextOf(alarmNode, "PROVIDER_NAME"),
|
||||
Group = TextOf(alarmNode, "GROUP"),
|
||||
TagName = TextOf(alarmNode, "TAGNAME"),
|
||||
Type = TextOf(alarmNode, "TYPE"),
|
||||
Value = TextOf(alarmNode, "VALUE"),
|
||||
Limit = TextOf(alarmNode, "LIMIT"),
|
||||
Priority = ParseInt(TextOf(alarmNode, "PRIORITY")),
|
||||
State = AlarmRecordTransitionMapper.ParseStateKind(TextOf(alarmNode, "STATE")),
|
||||
OperatorNode = TextOf(alarmNode, "OPERATOR_NODE"),
|
||||
OperatorName = TextOf(alarmNode, "OPERATOR_NAME"),
|
||||
AlarmComment = TextOf(alarmNode, "ALARM_COMMENT"),
|
||||
ProviderNode = providerNode ?? string.Empty,
|
||||
ProviderName = providerName ?? string.Empty,
|
||||
Group = group ?? string.Empty,
|
||||
TagName = tagName ?? string.Empty,
|
||||
Type = type ?? string.Empty,
|
||||
Value = value ?? string.Empty,
|
||||
Limit = limit ?? string.Empty,
|
||||
Priority = ParseInt(priority ?? string.Empty),
|
||||
State = AlarmRecordTransitionMapper.ParseStateKind(state ?? string.Empty),
|
||||
OperatorNode = operatorNode ?? string.Empty,
|
||||
OperatorName = operatorName ?? string.Empty,
|
||||
AlarmComment = alarmComment ?? string.Empty,
|
||||
};
|
||||
}
|
||||
return records;
|
||||
}
|
||||
|
||||
private static string TextOf(XmlNode parent, string childName)
|
||||
{
|
||||
XmlNode? node = parent.SelectSingleNode(childName);
|
||||
return node?.InnerText ?? string.Empty;
|
||||
}
|
||||
|
||||
private static int ParseInt(string text)
|
||||
{
|
||||
return int.TryParse(text, NumberStyles.Integer, CultureInfo.InvariantCulture, out int n)
|
||||
|
||||
Reference in New Issue
Block a user