dohertj2/lmxopcua
1
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(scripted-alarms): retire orphaned ScriptedAlarmActor + F9b evaluator (T11)

Browse Source
This commit is contained in:
Joseph Doherty
2026-06-10 15:22:26 -04:00
parent 5256761368
commit fc0d43a3dc
10 changed files with 7 additions and 1109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynScriptedAlarmEvaluator.cs
-124
View File
@@ -1,124 +0,0 @@
using System.Collections.Concurrent;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using SerilogLogger = Serilog.ILogger;
namespace ZB.MOM.WW.OtOpcUa.Host.Engines;
/// <summary>
/// F9b — production <see cref="IScriptedAlarmEvaluator"/> binding. Compiles each unique
/// predicate once via <see cref="ScriptEvaluator{TContext, TResult}"/> against
/// <see cref="AlarmPredicateContext"/> and caches the resulting evaluator. Predicates are
/// pure functions returning <c>bool</c>: <see cref="AlarmPredicateContext.SetVirtualTag"/>
/// throws so a misbehaving script can't smuggle a side effect into alarm evaluation.
///
/// Failure modes (compile error, sandbox violation, runtime exception, timeout) all surface
/// as <see cref="ScriptedAlarmEvalResult.Failure"/>; <see cref="ScriptedAlarmActor"/>
/// preserves the prior state on failure (does not flip Active/Inactive).
/// </summary>
public sealed class RoslynScriptedAlarmEvaluator : IScriptedAlarmEvaluator, IDisposable
{
private readonly ConcurrentDictionary<string, ScriptEvaluator<AlarmPredicateContext, bool>> _cache
= new(StringComparer.Ordinal);
private readonly ILogger<RoslynScriptedAlarmEvaluator> _logger;
private readonly SerilogLogger _scriptRoot;
private readonly TimeSpan _runTimeout;
private bool _disposed;
/// <summary>Initializes a new instance of the Roslyn scripted alarm evaluator.</summary>
/// <param name="logger">Logger for diagnostic messages (host diagnostics).</param>
/// <param name="scriptRoot">Root script logger; user <c>ctx.Logger.*</c> output flows through this to the Script-log page.</param>
/// <param name="runTimeout">Optional timeout for script evaluation; defaults to 2 seconds.</param>
public RoslynScriptedAlarmEvaluator(
ILogger<RoslynScriptedAlarmEvaluator> logger,
ScriptRootLogger scriptRoot,
TimeSpan? runTimeout = null)
{
_logger = logger;
_scriptRoot = (scriptRoot ?? throw new ArgumentNullException(nameof(scriptRoot))).Logger;
_runTimeout = runTimeout ?? TimeSpan.FromSeconds(2);
}
/// <summary>Evaluates a scripted alarm predicate against provided dependencies.</summary>
/// <param name="alarmId">The alarm identifier for logging purposes.</param>
/// <param name="predicate">The predicate expression to evaluate.</param>
/// <param name="dependencies">Variables available to the predicate expression.</param>
/// <returns>Evaluation result with success flag and active state or failure reason.</returns>
public ScriptedAlarmEvalResult Evaluate(string alarmId, string predicate, IReadOnlyDictionary<string, object?> dependencies)
{
if (_disposed) return ScriptedAlarmEvalResult.Failure("evaluator disposed");
if (string.IsNullOrWhiteSpace(predicate)) return ScriptedAlarmEvalResult.Failure("empty predicate");
ScriptEvaluator<AlarmPredicateContext, bool> evaluator;
try
{
evaluator = _cache.GetOrAdd(predicate, ScriptEvaluator<AlarmPredicateContext, bool>.Compile);
}
catch (CompilationErrorException ex)
{
_logger.LogWarning(ex, "Alarm {Id}: predicate compile failed", alarmId);
return ScriptedAlarmEvalResult.Failure($"compile error: {ex.Message}");
}
catch (ScriptSandboxViolationException ex)
{
_logger.LogWarning(ex, "Alarm {Id}: predicate sandbox violation", alarmId);
return ScriptedAlarmEvalResult.Failure($"sandbox violation: {ex.Message}");
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Alarm {Id}: predicate compile threw", alarmId);
return ScriptedAlarmEvalResult.Failure($"compile failure: {ex.Message}");
}
var readCache = BuildReadCache(dependencies);
// Per-evaluation script logger: bind both ScriptId and AlarmId from the alarm id so the
// Script-log page can attribute each line to the owning scripted alarm.
var scriptLog = _scriptRoot
.ForContext(ScriptLoggerFactory.ScriptIdProperty, alarmId)
.ForContext(ScriptLoggerFactory.AlarmIdProperty, alarmId);
var context = new AlarmPredicateContext(readCache, scriptLog);
try
{
using var cts = new CancellationTokenSource(_runTimeout);
var active = evaluator.RunAsync(context, cts.Token).GetAwaiter().GetResult();
return ScriptedAlarmEvalResult.Ok(active);
}
catch (OperationCanceledException)
{
return ScriptedAlarmEvalResult.Failure($"predicate timed out after {_runTimeout.TotalSeconds:F1}s");
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Alarm {Id}: predicate execution threw", alarmId);
return ScriptedAlarmEvalResult.Failure($"predicate threw: {ex.Message}");
}
}
private static IReadOnlyDictionary<string, DataValueSnapshot> BuildReadCache(
IReadOnlyDictionary<string, object?> deps)
{
var nowUtc = DateTime.UtcNow;
var cache = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
foreach (var kv in deps)
{
cache[kv.Key] = new DataValueSnapshot(kv.Value, StatusCode: 0u, SourceTimestampUtc: nowUtc, ServerTimestampUtc: nowUtc);
}
return cache;
}
/// <summary>Disposes the evaluator and all cached script evaluators.</summary>
public void Dispose()
{
if (_disposed) return;
_disposed = true;
foreach (var ev in _cache.Values)
{
try { ev.Dispose(); } catch { /* best-effort */ }
}
_cache.Clear();
}
}
src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs
-8
View File
@@ -114,14 +114,6 @@ if (hasDriver)
sp.GetRequiredService<ScriptRootLogger>()));
builder.Services.AddSingleton<IVirtualTagEvaluator>(sp => sp.GetRequiredService<RoslynVirtualTagEvaluator>());
// F9b — same pattern for scripted-alarm predicates. The actor preserves prior state on
// any Failure result, so a misbehaving script can't flip Active/Inactive spuriously.
builder.Services.AddSingleton<RoslynScriptedAlarmEvaluator>(sp =>
new RoslynScriptedAlarmEvaluator(
sp.GetRequiredService<ILoggerFactory>().CreateLogger<RoslynScriptedAlarmEvaluator>(),
sp.GetRequiredService<ScriptRootLogger>()));
builder.Services.AddSingleton<IScriptedAlarmEvaluator>(sp => sp.GetRequiredService<RoslynScriptedAlarmEvaluator>());
// Script-log fan-out (Layer 0). The DPS publisher resolves the ActorSystem lazily so it never
// races Akka startup. ScriptRootLogger wraps the composed pipeline (rolling scripts-*.log +
// error mirror to the main log + script-logs DPS topic) for unambiguous DI resolution; Task 3
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/EfAlarmActorStateStore.cs
-121
View File
@@ -1,121 +0,0 @@
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
/// <summary>
/// Production-side <see cref="IAlarmActorStateStore"/> backed by the
/// <see cref="ScriptedAlarmState"/> table in the central config DB. The actor's
/// 3-state enum projects into the table's two persisted dimensions: Acked + an
/// internal "_lastActiveState" recorded via a synthetic mapping (Inactive ⇒ Acked,
/// Active ⇒ Unacked, Acknowledged ⇒ Acked). ActiveState itself is deliberately NOT
/// persisted — re-derives from the evaluator on startup (Phase 7 decision #14).
/// </summary>
public sealed class EfAlarmActorStateStore : IAlarmActorStateStore
{
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
private readonly ILogger<EfAlarmActorStateStore> _logger;
/// <summary>Initializes a new instance of the EfAlarmActorStateStore.</summary>
/// <param name="dbFactory">The factory for creating database contexts.</param>
/// <param name="logger">The logger instance.</param>
public EfAlarmActorStateStore(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
ILogger<EfAlarmActorStateStore> logger)
{
_dbFactory = dbFactory;
_logger = logger;
}
/// <summary>Loads the alarm state snapshot from the database.</summary>
/// <param name="alarmId">The identifier of the alarm.</param>
/// <param name="ct">The cancellation token.</param>
/// <returns>The alarm state snapshot, or null if not found.</returns>
public async Task<AlarmActorStateSnapshot?> LoadAsync(string alarmId, CancellationToken ct)
{
using var db = await _dbFactory.CreateDbContextAsync(ct).ConfigureAwait(false);
var row = await db.ScriptedAlarmStates.AsNoTracking()
.FirstOrDefaultAsync(r => r.ScriptedAlarmId == alarmId, ct)
.ConfigureAwait(false);
if (row is null) return null;
var state = MapAckedToActorState(row.AckedState);
return new AlarmActorStateSnapshot(
AlarmId: alarmId,
State: state,
LastTransitionUtc: row.UpdatedAtUtc,
LastAckUser: row.LastAckUser);
}
/// <summary>Saves the alarm state snapshot to the database.</summary>
/// <param name="snapshot">The alarm state snapshot to save.</param>
/// <param name="ct">The cancellation token.</param>
public async Task SaveAsync(AlarmActorStateSnapshot snapshot, CancellationToken ct)
{
using var db = await _dbFactory.CreateDbContextAsync(ct).ConfigureAwait(false);
var row = await db.ScriptedAlarmStates
.FirstOrDefaultAsync(r => r.ScriptedAlarmId == snapshot.AlarmId, ct)
.ConfigureAwait(false);
var ackedState = MapActorStateToAcked(snapshot.State);
if (row is null)
{
db.ScriptedAlarmStates.Add(new ScriptedAlarmState
{
ScriptedAlarmId = snapshot.AlarmId,
EnabledState = "Enabled",
AckedState = ackedState,
ConfirmedState = "Confirmed",
ShelvingState = "Unshelved",
LastAckUser = snapshot.LastAckUser,
LastAckUtc = string.Equals(snapshot.State, "Acknowledged", StringComparison.Ordinal)
? snapshot.LastTransitionUtc
: null,
UpdatedAtUtc = snapshot.LastTransitionUtc,
CommentsJson = "[]",
});
}
else
{
row.AckedState = ackedState;
row.LastAckUser = snapshot.LastAckUser ?? row.LastAckUser;
if (string.Equals(snapshot.State, "Acknowledged", StringComparison.Ordinal))
row.LastAckUtc = snapshot.LastTransitionUtc;
row.UpdatedAtUtc = snapshot.LastTransitionUtc;
}
try
{
await db.SaveChangesAsync(ct).ConfigureAwait(false);
}
catch (DbUpdateConcurrencyException ex)
{
// Two actors racing to save the same alarm is benign — the last writer wins on
// UpdatedAtUtc, and the next transition on either side will write again. Log
// + drop so a race doesn't crash the dispatcher.
_logger.LogDebug(ex, "EfAlarmActorStateStore: concurrency conflict for {AlarmId}; dropping save",
snapshot.AlarmId);
}
}
private static string MapActorStateToAcked(string actorState) => actorState switch
{
"Active" => "Unacknowledged",
"Acknowledged" => "Acknowledged",
// Inactive maps to Acknowledged — when an alarm clears, nothing is left to ack.
_ => "Acknowledged",
};
private static string MapAckedToActorState(string ackedState)
{
// Only Active distinguishes from Acked — Inactive comes from a re-eval, not from
// the table. Persisted "Unacknowledged" implies the actor was last Active +
// un-acked; we restore it to Active so a restart doesn't drop pending operator work.
return string.Equals(ackedState, "Unacknowledged", StringComparison.Ordinal)
? "Active"
: "Acknowledged";
}
}
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/EfAlarmConditionStateStore.cs
+7 -9
View File
@@ -10,11 +10,9 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
/// <summary>
/// Production-side <see cref="IAlarmStateStore"/> backed by the
/// <see cref="ScriptedAlarmState"/> table in the central config DB. Unlike the narrower
/// <see cref="EfAlarmActorStateStore"/> (which only persists AckedState for the actor's
/// 3-state enum), this store maps the full Part 9 <see cref="AlarmConditionState"/> —
/// Enabled / Acked / Confirmed / Shelving + the ack/confirm audit trail + operator
/// comments.
/// <see cref="ScriptedAlarmState"/> table in the central config DB. This store maps the
/// full Part 9 <see cref="AlarmConditionState"/> — Enabled / Acked / Confirmed / Shelving
/// + the ack/confirm audit trail + operator comments.
/// </summary>
/// <remarks>
/// <para>
@@ -34,7 +32,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
/// <para>
/// <see cref="AlarmConditionState.Comments"/> serializes to/from
/// <see cref="ScriptedAlarmState.CommentsJson"/> via System.Text.Json. An empty list
/// round-trips as <c>"[]"</c> (matching the entity default + <see cref="EfAlarmActorStateStore"/>).
/// round-trips as <c>"[]"</c> (matching the entity default).
/// </para>
/// </remarks>
public sealed class EfAlarmConditionStateStore : IAlarmStateStore
@@ -78,9 +76,9 @@ public sealed class EfAlarmConditionStateStore : IAlarmStateStore
/// <inheritdoc />
/// <remarks>
/// <b>Concurrency assumption</b>: saves for a given <c>alarmId</c> are serialized by the
/// owning host actor (one actor owns the engine per equipment), mirroring
/// <c>EfAlarmActorStateStore</c>. The check-then-insert pattern is therefore safe under
/// that guarantee — two concurrent inserts for the same alarm cannot occur in the live
/// owning host actor (one actor owns the engine per equipment). The check-then-insert
/// pattern is therefore safe under that guarantee — two concurrent inserts for the same
/// alarm cannot occur in the live
/// runtime. The <see cref="DbUpdateConcurrencyException"/> catch handles the edge case of a
/// racing concurrent restart during crash recovery.
/// </remarks>
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmActor.cs
-254
View File
@@ -1,254 +0,0 @@
using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
public enum ScriptedAlarmActorState { Inactive, Active, Acknowledged }
/// <summary>
/// One scripted alarm. Receives dependency value updates, runs the predicate via an
/// injected <see cref="IScriptedAlarmEvaluator"/>, and on transitions publishes both
/// an <see cref="AlarmTransitionEvent"/> on the cluster <c>alerts</c> DPS topic and a
/// <see cref="ScriptLogEntry"/> on <c>script-logs</c>. Manual <see cref="AcknowledgeAlarm"/>
/// + <see cref="ConditionCleared"/> still flow through the same state machine so the
/// legacy callers keep working.
/// </summary>
public sealed class ScriptedAlarmActor : ReceiveActor
{
public const string AlertsTopic = "alerts";
public const string ScriptLogsTopic = "script-logs";
public sealed record DependencyValueChanged(string TagId, object? Value, DateTime TimestampUtc);
public sealed record ConditionMet(string Reason);
public sealed record AcknowledgeAlarm(string Actor);
public sealed record ConditionCleared;
public sealed record StateChanged(string AlarmId, ScriptedAlarmActorState State, DateTime AtUtc);
public sealed record AlarmConfig(
string AlarmId,
string AlarmName,
string EquipmentPath,
int Severity,
string? Predicate);
private readonly AlarmConfig _config;
private readonly IScriptedAlarmEvaluator _evaluator;
private readonly IAlarmActorStateStore _stateStore;
private readonly Func<DPSPublisher>? _publisherFactory;
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly Dictionary<string, object?> _dependencies = new(StringComparer.Ordinal);
private ScriptedAlarmActorState _state = ScriptedAlarmActorState.Inactive;
private string? _lastAckUser;
public sealed record StateRestored(ScriptedAlarmActorState State, string? LastAckUser);
/// <summary>Creates a new Props for a ScriptedAlarmActor with the given configuration and optional dependencies.</summary>
/// <param name="config">The alarm configuration.</param>
/// <param name="evaluator">The alarm evaluator; defaults to null evaluator if not provided.</param>
/// <param name="publisherFactory">Optional factory for creating DPS publishers.</param>
/// <param name="stateStore">Optional state store for persistence; defaults to null store if not provided.</param>
/// <returns>Akka Props for creating the actor.</returns>
public static Props Props(
AlarmConfig config,
IScriptedAlarmEvaluator? evaluator = null,
Func<DPSPublisher>? publisherFactory = null,
IAlarmActorStateStore? stateStore = null) =>
Akka.Actor.Props.Create(() => new ScriptedAlarmActor(
config,
evaluator ?? NullScriptedAlarmEvaluator.Instance,
publisherFactory,
stateStore ?? NullAlarmActorStateStore.Instance));
/// <summary>Legacy single-arg ctor kept for callers that only care about the state machine
/// (no engine evaluation, no DPS fan-out, no persistence). Equivalent to <c>Props(new AlarmConfig(...))</c>.</summary>
/// <param name="alarmId">The alarm identifier, used as both alarm ID and name.</param>
/// <returns>Akka Props for creating the actor with minimal configuration.</returns>
public static Props Props(string alarmId) =>
Props(new AlarmConfig(alarmId, alarmId, EquipmentPath: "", Severity: 500, Predicate: null));
/// <summary>Initializes a new ScriptedAlarmActor with the given configuration and dependencies.</summary>
/// <param name="config">The alarm configuration.</param>
/// <param name="evaluator">The alarm predicate evaluator.</param>
/// <param name="publisherFactory">Optional factory for creating DPS publishers.</param>
/// <param name="stateStore">The state store for loading and saving alarm state.</param>
public ScriptedAlarmActor(
AlarmConfig config,
IScriptedAlarmEvaluator evaluator,
Func<DPSPublisher>? publisherFactory,
IAlarmActorStateStore stateStore)
{
_config = config;
_evaluator = evaluator;
_publisherFactory = publisherFactory;
_stateStore = stateStore;
Receive<DependencyValueChanged>(OnDependencyChanged);
Receive<ConditionMet>(_ => { if (_state == ScriptedAlarmActorState.Inactive) Transition(ScriptedAlarmActorState.Active, user: "system"); });
Receive<AcknowledgeAlarm>(msg => { if (_state == ScriptedAlarmActorState.Active) Transition(ScriptedAlarmActorState.Acknowledged, user: msg.Actor); });
Receive<ConditionCleared>(_ => { if (_state != ScriptedAlarmActorState.Inactive) Transition(ScriptedAlarmActorState.Inactive, user: "system"); });
Receive<StateRestored>(OnStateRestored);
}
/// <inheritdoc />
protected override void PreStart()
{
// Load persisted state — when the store has a row, restore in-memory state before the
// first dependency-change arrives. Async I/O is piped back as StateRestored so we don't
// block the message-loop thread; until it arrives the actor stays at the default Inactive.
var self = Self;
_ = Task.Run(async () =>
{
try
{
var snapshot = await _stateStore.LoadAsync(_config.AlarmId, CancellationToken.None)
.ConfigureAwait(false);
if (snapshot is null) return;
if (!Enum.TryParse<ScriptedAlarmActorState>(snapshot.State, ignoreCase: true, out var parsed))
return;
self.Tell(new StateRestored(parsed, snapshot.LastAckUser));
}
catch (Exception ex)
{
_log.Warning(ex, "ScriptedAlarm {Id}: state-store load failed; booting Inactive",
_config.AlarmId);
}
});
}
private void OnStateRestored(StateRestored msg)
{
// Active is re-derived from the evaluator at the next DependencyValueChanged — we still
// restore Active here so operators don't lose the in-flight transition if a restart races
// ahead of the next eval. The first evaluator tick will correct it if the condition cleared.
_state = msg.State;
_lastAckUser = msg.LastAckUser;
_log.Info("ScriptedAlarm {Id}: restored persisted state {State} (lastAck={User})",
_config.AlarmId, _state, _lastAckUser ?? "(none)");
}
private void OnDependencyChanged(DependencyValueChanged msg)
{
_dependencies[msg.TagId] = msg.Value;
if (string.IsNullOrEmpty(_config.Predicate)) return;
ScriptedAlarmEvalResult result;
try
{
result = _evaluator.Evaluate(_config.AlarmId, _config.Predicate, _dependencies);
}
catch (Exception ex)
{
_log.Warning(ex, "ScriptedAlarm {Id}: evaluator threw", _config.AlarmId);
PublishLog("Error", $"evaluator threw: {ex.Message}");
return;
}
if (!result.Success)
{
PublishLog("Warning", result.Reason ?? "evaluator failure");
return;
}
// Active condition wins regardless of ack state — re-firing is suppressed because
// _state already == Active. Cleared moves Active OR Acknowledged → Inactive.
if (result.Active && _state == ScriptedAlarmActorState.Inactive)
{
Transition(ScriptedAlarmActorState.Active, user: "system");
}
else if (!result.Active && _state != ScriptedAlarmActorState.Inactive)
{
Transition(ScriptedAlarmActorState.Inactive, user: "system");
}
}
private void Transition(ScriptedAlarmActorState next, string user)
{
var prev = _state;
_state = next;
if (next == ScriptedAlarmActorState.Acknowledged) _lastAckUser = user;
_log.Info("ScriptedAlarm {Id}: {From} → {To}", _config.AlarmId, prev, next);
var nowUtc = DateTime.UtcNow;
Context.Parent.Tell(new StateChanged(_config.AlarmId, next, nowUtc));
PersistStateAsync(nowUtc);
var kind = next switch
{
ScriptedAlarmActorState.Active => "Activated",
ScriptedAlarmActorState.Acknowledged => "Acknowledged",
ScriptedAlarmActorState.Inactive => "Cleared",
_ => next.ToString(),
};
OtOpcUaTelemetry.ScriptedAlarmTransition.Add(1,
new KeyValuePair<string, object?>("state", kind.ToLowerInvariant()));
var evt = new AlarmTransitionEvent(
AlarmId: _config.AlarmId,
EquipmentPath: _config.EquipmentPath,
AlarmName: _config.AlarmName,
TransitionKind: kind,
Severity: _config.Severity,
Message: $"{_config.AlarmName} {kind}",
User: user,
TimestampUtc: nowUtc);
PublishOrFallback(AlertsTopic, evt);
PublishLog("Information", $"{_config.AlarmName} {kind} (by {user})");
}
private void PublishLog(string level, string message)
{
var entry = new ScriptLogEntry(
ScriptId: _config.AlarmId,
Level: level,
Message: message,
TimestampUtc: DateTime.UtcNow,
VirtualTagId: null,
AlarmId: _config.AlarmId,
EquipmentId: null);
PublishOrFallback(ScriptLogsTopic, entry);
}
private void PublishOrFallback(string topic, object payload)
{
if (_publisherFactory is not null)
{
_publisherFactory().Publish(topic, payload);
return;
}
DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(topic, payload));
}
private void PersistStateAsync(DateTime nowUtc)
{
var snapshot = new AlarmActorStateSnapshot(
AlarmId: _config.AlarmId,
State: _state.ToString(),
LastTransitionUtc: nowUtc,
LastAckUser: _lastAckUser);
// Fire-and-forget. Save failures get logged but don't block the message loop —
// the worst case is a restart loses one transition, which then re-derives from
// the evaluator's next tick anyway.
_ = Task.Run(async () =>
{
try
{
await _stateStore.SaveAsync(snapshot, CancellationToken.None).ConfigureAwait(false);
}
catch (Exception ex)
{
_log.Warning(ex, "ScriptedAlarm {Id}: state-store save failed", _config.AlarmId);
}
});
}
}