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

feat(localdb): alarm S&F on LocalDb with primary-gated drain (cutover)

Browse Source 
Moves the alarm store-and-forward buffer out of its own alarm-historian.db
and into the node's consolidated LocalDb, where it replicates to the
redundant pair peer. A node that dies holding undelivered alarm history no
longer takes it to the grave.

Tasks 2 and 3 land together, as the plan anticipated. They are not separable:
the gate is a constructor argument of the rewritten sink, and a commit that
replicated the queue without gating the drain would be a commit in which both
nodes of a pair deliver every alarm event, continuously.

That drain gate is the load-bearing part of the change, and the recon
explains why it is new work rather than a refinement. Exactly-once delivery
across a pair is enforced today on the ENQUEUE side, by
HistorianAdapterActor: only the Primary enqueues, so the Secondary's queue is
empty and its ungated drain has nothing to send. Replicating the table
destroys that invariant.

The gate is a Func<bool> the caller supplies, because the drain runs on a
timer the sink owns rather than on a mailbox, and Core.AlarmHistorian cannot
reference PrimaryGatePolicy in Runtime. Runtime supplies it via a new
IRedundancyRoleView singleton that DriverHostActor publishes its Primary-gate
verdict to -- the same verdict the inbound-write and native-ack gates use, so
there is no second notion of am-I-the-Primary to drift.

Two failure modes are deliberately closed:

- The view is seeded OPEN, matching the policy's own answer for an unknown
  role with no driver peer. A deployment that runs no redundancy never
  publishes to it, and defaulting closed would silently stop its alarm
  history forever.

- A gate that throws is read as not-now, never as permission, and a closed
  gate reports the new HistorianDrainState.NotPrimary rather than Idle. A
  Secondary's rising queue is supposed to look different from a stalled
  drain, and if BOTH nodes report NotPrimary the pair is misconfigured and
  says so instead of quietly filling toward the capacity ceiling.

Row ids are a hash of the payload rather than fresh GUIDs. Both adapters
accept the same fanned transition in the window before the first redundancy
snapshot arrives, and under last-writer-wins an equal key converges those two
accepts into one row instead of duplicating them.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-21 04:13:00 -04:00
parent 8a9cb40a72
commit 71379816e7
14 changed files with 804 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/IAlarmHistorianSink.cs
+14 -2
View File
@@ -4,13 +4,13 @@ namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
/// The historian sink contract — where qualifying alarm events land. Ingestion routes
/// through the HistorianGateway alarm writer (the gateway's <c>SendEvent</c> gRPC path)
/// behind the durable store-and-forward queue. Tests use an in-memory fake; production uses
/// <see cref="SqliteStoreAndForwardSink"/>.
/// <see cref="LocalDbStoreAndForwardSink"/>.
/// </summary>
/// <remarks>
/// <para>
/// <see cref="EnqueueAsync"/> is fire-and-forget from the engine's perspective —
/// the sink MUST NOT block the emitting thread. Production implementations
/// (<see cref="SqliteStoreAndForwardSink"/>) persist to a local SQLite queue
/// (<see cref="LocalDbStoreAndForwardSink"/>) persist to the node's local SQLite queue
/// first, then drain asynchronously to the actual historian. Per the Phase 7 plan,
/// failed downstream writes replay with exponential backoff;
/// operator actions are never blocked waiting on the historian.
@@ -79,6 +79,18 @@ public enum HistorianDrainState
Idle,
Draining,
BackingOff,
/// <summary>
/// Ticking, but not draining: this node does not hold the Primary role, so it leaves the
/// (replicated) queue for the node that does.
/// </summary>
/// <remarks>
/// Distinct from <see cref="Idle"/> so a rising queue depth on a Secondary reads as the
/// designed behaviour rather than a stalled drain. It is also how a misconfigured pair
/// becomes diagnosable: if BOTH nodes report this, no one is draining and alarm history is
/// silently accumulating toward the capacity ceiling.
/// </remarks>
NotPrimary,
}
/// <summary>Returned by the historian alarm writer per event — drain worker uses this to decide retry cadence.</summary>
src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/SqliteStoreAndForwardSink.cs → src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/LocalDbStoreAndForwardSink.cs
+279 -243
View File
@@ -1,53 +1,56 @@
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using Microsoft.Data.Sqlite;
using Serilog;
using ZB.MOM.WW.LocalDb;
namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
/// <summary>
/// Durable SQLite queue on the node absorbs every qualifying alarm event, a drain
/// worker batches rows to the Wonderware historian sidecar via
/// <see cref="IAlarmHistorianWriter"/> on an exponential-backoff cadence, and
/// operator acks never block on the historian being reachable.
/// Durable queue in the node's consolidated LocalDb absorbs every qualifying alarm event, a
/// drain worker batches rows to the historian gateway via <see cref="IAlarmHistorianWriter"/>
/// on an exponential-backoff cadence, and operator acks never block on the historian being
/// reachable.
/// </summary>
/// <remarks>
/// <para>
/// Queue schema:
/// <code>
/// CREATE TABLE Queue (
/// RowId INTEGER PRIMARY KEY AUTOINCREMENT,
/// AlarmId TEXT NOT NULL,
/// EnqueuedUtc TEXT NOT NULL,
/// PayloadJson TEXT NOT NULL,
/// AttemptCount INTEGER NOT NULL DEFAULT 0,
/// LastAttemptUtc TEXT NULL,
/// LastError TEXT NULL,
/// DeadLettered INTEGER NOT NULL DEFAULT 0
/// );
/// </code>
/// Dead-lettered rows stay in place for the configured retention window (default
/// 30 days) so operators can inspect + manually
/// retry before the sweeper purges them. Regular queue capacity is bounded —
/// overflow evicts the oldest non-dead-lettered rows with a WARN log. The
/// durability guarantee is therefore bounded by <see cref="DefaultCapacity"/>:
/// under a sustained historian outage, accepted events may be evicted before
/// delivery. The <see cref="HistorianSinkStatus.EvictedCount"/> counter makes
/// overflow visible to operators without requiring the WARN log to be scraped.
/// Rows live in <c>alarm_sf_events</c> (see <see cref="AlarmSfSchema"/>), which the host
/// registers for replication. That is the point of this type living on
/// <see cref="ILocalDb"/> rather than owning its own file: the buffer mirrors to the
/// redundant pair peer, so a node that dies holding undelivered alarm history no longer
/// takes it to the grave.
/// </para>
/// <para>
/// <b>Only one node of a pair may drain.</b> Replication puts the Primary's queued rows in
/// the Secondary's table too; an ungated drain there would re-deliver every event,
/// continuously. The <c>drainGate</c> constructor argument is how the caller scopes
/// draining to the Primary. Enqueue is gated separately and upstream, by
/// <c>HistorianAdapterActor</c>.
/// </para>
/// <para>
/// Dead-lettered rows stay in place for the configured retention window (default 30 days)
/// so operators can inspect + manually retry before the sweeper purges them. Regular queue
/// capacity is bounded — overflow evicts the oldest non-dead-lettered rows with a WARN log.
/// The durability guarantee is therefore bounded by <see cref="DefaultCapacity"/>: under a
/// sustained historian outage, accepted events may be evicted before delivery. The
/// <see cref="HistorianSinkStatus.EvictedCount"/> counter makes overflow visible to
/// operators without requiring the WARN log to be scraped.
/// </para>
/// <para>
/// Drain runs on a self-rescheduling one-shot <see cref="System.Threading.Timer"/>.
/// Exponential backoff on <see cref="HistorianWriteOutcome.RetryPlease"/>:
/// 1s → 2s → 5s → 15s → 60s cap — the backoff is applied to the timer's next
/// due-time, so a historian outage genuinely slows the drain cadence.
/// <see cref="HistorianWriteOutcome.PermanentFail"/> rows flip
/// the <c>DeadLettered</c> flag on the individual row; neighbors in the batch
/// still retry on their own cadence.
/// 1s → 2s → 5s → 15s → 60s cap — the backoff is applied to the timer's next due-time, so a
/// historian outage genuinely slows the drain cadence.
/// <see cref="HistorianWriteOutcome.PermanentFail"/> rows flip the <c>dead_lettered</c> flag
/// on the individual row; neighbors in the batch still retry on their own cadence.
/// </para>
/// </remarks>
public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposable
{
/// <summary>Default queue capacity — oldest non-dead-lettered rows evicted past this.</summary>
public const long DefaultCapacity = 1_000_000;
/// <summary>Default window dead-lettered rows are retained for before the sweeper purges them.</summary>
public static readonly TimeSpan DefaultDeadLetterRetention = TimeSpan.FromDays(30);
/// <summary>Default max delivery attempts before a perpetually-retrying (poison) row is dead-lettered.</summary>
@@ -62,7 +65,7 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
TimeSpan.FromSeconds(60),
];
private readonly string _connectionString;
private readonly ILocalDb _db;
private readonly IAlarmHistorianWriter _writer;
private readonly ILogger _logger;
private readonly int _batchSize;
@@ -70,8 +73,9 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
private readonly TimeSpan _deadLetterRetention;
private readonly int _maxAttempts;
private readonly Func<DateTime> _clock;
private readonly Func<bool> _drainGate;
private readonly SemaphoreSlim _drainGate = new(1, 1);
private readonly SemaphoreSlim _drainGateLock = new(1, 1);
private Timer? _drainTimer;
private TimeSpan _tickInterval;
private volatile int _backoffIndex;
@@ -85,13 +89,16 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
private string? _lastError;
private HistorianDrainState _drainState = HistorianDrainState.Idle;
private long _evictedCount;
// Tracks the last gate decision so a denial is logged on the transition rather than on every
// tick. Null until the first drain runs.
private bool? _lastGateDecision;
private long _queuedRowCount;
// Probe counter — incremented every time we actually issue a real COUNT(*) for
// capacity enforcement. Public for test instrumentation only.
private long _capacityProbeCount;
// After every Nth enqueue we resync the in-memory counter from storage to defend
// against silent drift (e.g. an external process editing the DB).
// against silent drift (e.g. the peer replicating rows in underneath us).
private const long ResyncEnqueueInterval = 10_000;
private long _enqueuesSinceResync;
@@ -99,9 +106,13 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
public long DebugCapacityProbeCount => Interlocked.Read(ref _capacityProbeCount);
/// <summary>
/// Initializes a new instance of the <see cref="SqliteStoreAndForwardSink"/> class with the specified configuration.
/// Initializes a new instance of the <see cref="LocalDbStoreAndForwardSink"/> class.
/// </summary>
/// <param name="databasePath">The filesystem path to the SQLite database file.</param>
/// <param name="db">
/// The node's local database. Its <c>alarm_sf_events</c> table must already exist and be
/// registered for replication — the host does both in <c>LocalDbSetup.OnReady</c>, before
/// any consumer can resolve this sink.
/// </param>
/// <param name="writer">The alarm historian writer to handle batch forwarding.</param>
/// <param name="logger">The logger for diagnostic output.</param>
/// <param name="batchSize">The maximum number of rows to forward in a single batch. Defaults to 100.</param>
@@ -109,18 +120,24 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
/// <param name="deadLetterRetention">The timespan to retain dead-lettered rows before purging. Defaults to 30 days.</param>
/// <param name="maxAttempts">The maximum number of delivery attempts before a perpetually-retrying (poison) row is dead-lettered. Defaults to 10.</param>
/// <param name="clock">Optional clock function for testing; defaults to <see cref="DateTime.UtcNow"/>.</param>
public SqliteStoreAndForwardSink(
string databasePath,
/// <param name="drainGate">
/// Consulted at the top of every drain tick; <c>false</c> skips the tick entirely, leaving
/// the queue intact. Callers running a redundant pair MUST supply a gate that is true on at
/// most one node, because the queue replicates — see the class remarks. The default
/// (<c>null</c> ⇒ always drain) is the correct single-node and test posture.
/// </param>
public LocalDbStoreAndForwardSink(
ILocalDb db,
IAlarmHistorianWriter writer,
ILogger logger,
int batchSize = 100,
long capacity = DefaultCapacity,
TimeSpan? deadLetterRetention = null,
int maxAttempts = DefaultMaxAttempts,
Func<DateTime>? clock = null)
Func<DateTime>? clock = null,
Func<bool>? drainGate = null)
{
if (string.IsNullOrWhiteSpace(databasePath))
throw new ArgumentException("Database path required.", nameof(databasePath));
_db = db ?? throw new ArgumentNullException(nameof(db));
_writer = writer ?? throw new ArgumentNullException(nameof(writer));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_batchSize = batchSize > 0 ? batchSize : throw new ArgumentOutOfRangeException(nameof(batchSize));
@@ -128,50 +145,11 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
_deadLetterRetention = deadLetterRetention ?? DefaultDeadLetterRetention;
_maxAttempts = maxAttempts > 0 ? maxAttempts : throw new ArgumentOutOfRangeException(nameof(maxAttempts));
_clock = clock ?? (() => DateTime.UtcNow);
// DefaultTimeout gives ADO.NET command-level retry; the PRAGMA busy_timeout
// applied in OpenConnection backs it with SQLite's own busy-handler so an
// enqueue/drain collision waits out the file lock instead of throwing
// SQLITE_BUSY immediately.
_connectionString = new SqliteConnectionStringBuilder
{
DataSource = databasePath,
DefaultTimeout = 5,
}.ToString();
_drainGate = drainGate ?? (static () => true);
InitializeSchema();
_queuedRowCount = ProbeQueuedRowCount();
}
/// <summary>
/// Open a connection with the busy timeout + WAL journal applied. SQLite
/// serializes writers with a file lock; the busy_timeout lets a writer wait
/// out a competing lock (default is 0 — fail fast), and WAL lets readers and
/// the single writer proceed without blocking each other.
/// </summary>
private SqliteConnection OpenConnection()
{
var conn = new SqliteConnection(_connectionString);
conn.Open();
ApplyPragmas(conn);
return conn;
}
/// <summary>Apply busy_timeout + WAL pragmas to an already-open connection (sync).</summary>
private static void ApplyPragmas(SqliteConnection conn)
{
using var pragma = conn.CreateCommand();
pragma.CommandText = "PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL;";
pragma.ExecuteNonQuery();
}
/// <summary>Apply busy_timeout + WAL pragmas to an already-open connection (async).</summary>
private static async Task ApplyPragmasAsync(SqliteConnection conn, CancellationToken ct)
{
using var pragma = conn.CreateCommand();
pragma.CommandText = "PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL;";
await pragma.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
}
/// <summary>
/// Start the background drain worker. Not started automatically so tests can
/// drive <see cref="DrainOnceAsync"/> deterministically.
@@ -188,7 +166,7 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
/// <param name="tickInterval">The base interval between drain attempts.</param>
public void StartDrainLoop(TimeSpan tickInterval)
{
if (_disposed) throw new ObjectDisposedException(nameof(SqliteStoreAndForwardSink));
if (_disposed) throw new ObjectDisposedException(nameof(LocalDbStoreAndForwardSink));
_tickInterval = tickInterval;
_drainTimer?.Dispose();
// One-shot: dueTime = tickInterval, period = Infinite. RescheduleDrain re-arms
@@ -235,29 +213,50 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
catch (ObjectDisposedException) { /* raced with Dispose — nothing to re-arm */ }
}
/// <summary>
/// Derives the row's primary key from its payload.
/// </summary>
/// <remarks>
/// Deterministic rather than a fresh GUID so that the same event enqueued independently on
/// both nodes of a pair converges to one row under last-writer-wins instead of duplicating.
/// That happens for real: <c>HistorianAdapterActor</c> default-writes while its redundancy
/// role is unknown, so in the window before the first snapshot arrives both adapters accept
/// the same fanned transition. Two distinct events cannot collide —
/// <see cref="AlarmHistorianEvent"/> carries a full-precision timestamp alongside the alarm
/// id, kind, message and user, so an equal hash means an equal event.
/// </remarks>
private static string DeriveId(string payloadJson) =>
Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(payloadJson)));
/// <inheritdoc />
public async Task EnqueueAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken)
{
if (evt is null) throw new ArgumentNullException(nameof(evt));
if (_disposed) throw new ObjectDisposedException(nameof(SqliteStoreAndForwardSink));
if (_disposed) throw new ObjectDisposedException(nameof(LocalDbStoreAndForwardSink));
using var conn = new SqliteConnection(_connectionString);
await conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await ApplyPragmasAsync(conn, cancellationToken).ConfigureAwait(false);
await EnforceCapacityFastPathAsync(cancellationToken).ConfigureAwait(false);
await EnforceCapacityFastPathAsync(conn, cancellationToken).ConfigureAwait(false);
var payload = JsonSerializer.Serialize(evt);
using var cmd = conn.CreateCommand();
cmd.CommandText = """
INSERT INTO Queue (AlarmId, EnqueuedUtc, PayloadJson, AttemptCount)
VALUES ($alarmId, $enqueued, $payload, 0);
""";
cmd.Parameters.AddWithValue("$alarmId", evt.AlarmId);
cmd.Parameters.AddWithValue("$enqueued", _clock().ToString("O"));
cmd.Parameters.AddWithValue("$payload", JsonSerializer.Serialize(evt));
await cmd.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
// ON CONFLICT DO NOTHING: re-enqueuing an identical event (the pair's boot-window
// double-accept) must not resurrect a row the drain already bumped or dead-lettered.
var inserted = await _db.ExecuteAsync(
"""
INSERT INTO alarm_sf_events
(id, alarm_id, enqueued_at_utc, payload_json, attempt_count)
VALUES (@Id, @AlarmId, @EnqueuedAtUtc, @PayloadJson, 0)
ON CONFLICT(id) DO NOTHING
""",
new
{
Id = DeriveId(payload),
AlarmId = evt.AlarmId,
EnqueuedAtUtc = _clock().ToString("O"),
PayloadJson = payload,
},
cancellationToken).ConfigureAwait(false);
Interlocked.Increment(ref _queuedRowCount);
if (inserted > 0) Interlocked.Increment(ref _queuedRowCount);
}
/// <summary>
@@ -268,15 +267,17 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
/// through <see cref="EnforceCapacityAsync"/> which still runs a precise
/// COUNT to compute the exact number of rows to evict.
/// </summary>
private async Task EnforceCapacityFastPathAsync(SqliteConnection conn, CancellationToken ct)
private async Task EnforceCapacityFastPathAsync(CancellationToken ct)
{
var enqueuesSinceResync = Interlocked.Increment(ref _enqueuesSinceResync);
var cached = Interlocked.Read(ref _queuedRowCount);
// Periodic resync — bounded amount of drift even under exotic conditions.
// Periodic resync — bounded amount of drift even under exotic conditions. Now doubly
// warranted: replication applies the peer's rows straight into the table, so the counter
// has a second way to fall behind that no local code path can observe.
if (enqueuesSinceResync >= ResyncEnqueueInterval)
{
await ResyncQueuedRowCountAsync(conn, ct).ConfigureAwait(false);
await ResyncQueuedRowCountAsync(ct).ConfigureAwait(false);
cached = Interlocked.Read(ref _queuedRowCount);
Interlocked.Exchange(ref _enqueuesSinceResync, 0);
}
@@ -286,55 +287,70 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
// Cached counter says we're at or above the capacity wall — fall back to the
// precise path which probes COUNT(*) and evicts whatever's needed.
await EnforceCapacityAsync(conn, ct).ConfigureAwait(false);
await EnforceCapacityAsync(ct).ConfigureAwait(false);
}
/// <summary>Synchronously query <c>COUNT(*)</c> of non-dead-lettered rows. Used at startup.</summary>
private long ProbeQueuedRowCount()
{
Interlocked.Increment(ref _capacityProbeCount);
using var conn = OpenConnection();
using var conn = _db.CreateConnection();
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT COUNT(*) FROM Queue WHERE DeadLettered = 0";
cmd.CommandText = "SELECT COUNT(*) FROM alarm_sf_events WHERE dead_lettered = 0";
return (long)(cmd.ExecuteScalar() ?? 0L);
}
/// <summary>Re-sync the in-memory counter from storage (async path).</summary>
private async Task ResyncQueuedRowCountAsync(SqliteConnection conn, CancellationToken ct)
private async Task ResyncQueuedRowCountAsync(CancellationToken ct)
{
Interlocked.Increment(ref _capacityProbeCount);
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT COUNT(*) FROM Queue WHERE DeadLettered = 0";
var live = (long)(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false) ?? 0L);
var live = await CountAsync("dead_lettered = 0", ct).ConfigureAwait(false);
Interlocked.Exchange(ref _queuedRowCount, live);
}
private async Task<long> CountAsync(string predicate, CancellationToken ct)
{
var rows = await _db.QueryAsync(
$"SELECT COUNT(*) FROM alarm_sf_events WHERE {predicate}",
r => r.GetInt64(0),
parameters: null,
ct).ConfigureAwait(false);
return rows.Count > 0 ? rows[0] : 0L;
}
/// <summary>
/// Read up to <see cref="_batchSize"/> queued rows, forward through the writer,
/// remove Ack'd rows, dead-letter PermanentFail rows, and extend the backoff
/// on RetryPlease. Safe to call from multiple threads; the semaphore enforces
/// serial execution.
/// </summary>
/// <remarks>
/// Returns without touching the queue when the drain gate is closed. On a redundant pair
/// that is the Secondary's steady state: its table fills by replication and stays put until
/// it is promoted.
/// </remarks>
/// <param name="ct">Cancellation token for the operation.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task DrainOnceAsync(CancellationToken ct)
{
if (_disposed) return;
if (!await _drainGate.WaitAsync(0, ct).ConfigureAwait(false)) return;
if (!await _drainGateLock.WaitAsync(0, ct).ConfigureAwait(false)) return;
try
{
if (!EvaluateDrainGate())
{
lock (_statusLock) { _drainState = HistorianDrainState.NotPrimary; }
return;
}
lock (_statusLock)
{
_drainState = HistorianDrainState.Draining;
_lastDrainUtc = _clock();
}
// One connection per drain tick — used by purge, read, corrupt-dead-letter,
// and the outcome-applying transaction.
using var conn = OpenConnection();
PurgeAgedDeadLetters(conn);
var batch = ReadBatch(conn);
await PurgeAgedDeadLettersAsync(ct).ConfigureAwait(false);
var batch = await ReadBatchAsync(ct).ConfigureAwait(false);
if (batch.Count == 0)
{
lock (_statusLock) { _drainState = HistorianDrainState.Idle; }
@@ -342,22 +358,24 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
}
// A null/un-deserializable payload can never succeed — dead-letter it
// immediately for its own RowId so it cannot stall the queue head, and
// immediately for its own id so it cannot stall the queue head, and
// exclude it from the batch handed to the writer.
var corruptRowIds = batch.Where(r => r.Event is null).Select(r => r.RowId).ToList();
var corruptIds = batch.Where(r => r.Event is null).Select(r => r.Id).ToList();
var liveRows = batch.Where(r => r.Event is not null).ToList();
var events = liveRows.Select(r => r.Event!).ToList();
if (corruptRowIds.Count > 0)
if (corruptIds.Count > 0)
{
using var corruptTx = conn.BeginTransaction();
foreach (var rowId in corruptRowIds)
DeadLetterRow(conn, corruptTx, rowId, $"corrupt payload at {_clock():O}");
corruptTx.Commit();
Interlocked.Add(ref _queuedRowCount, -corruptRowIds.Count);
await using (var corruptTx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false))
{
foreach (var id in corruptIds)
await DeadLetterRowAsync(corruptTx, id, $"corrupt payload at {_clock():O}", ct).ConfigureAwait(false);
await corruptTx.CommitAsync(ct).ConfigureAwait(false);
}
Interlocked.Add(ref _queuedRowCount, -corruptIds.Count);
_logger.Warning(
"Dead-lettered {Count} historian queue row(s) with un-deserializable payload",
corruptRowIds.Count);
corruptIds.Count);
}
if (events.Count == 0)
@@ -404,41 +422,44 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
}
int rowsLeavingQueue = 0;
using (var tx = conn.BeginTransaction())
await using (var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false))
{
for (var i = 0; i < outcomes.Count; i++)
{
var outcome = outcomes[i];
var rowId = liveRows[i].RowId;
var id = liveRows[i].Id;
switch (outcome)
{
case HistorianWriteOutcome.Ack:
DeleteRow(conn, tx, rowId);
// Deleted, not marked delivered. The replication engine carries the
// delete as a tombstone, so the peer drops its copy and cannot
// re-deliver the row after a failover.
await DeleteRowAsync(tx, id, ct).ConfigureAwait(false);
rowsLeavingQueue++;
break;
case HistorianWriteOutcome.PermanentFail:
DeadLetterRow(conn, tx, rowId, $"permanent fail at {_clock():O}");
await DeadLetterRowAsync(tx, id, $"permanent fail at {_clock():O}", ct).ConfigureAwait(false);
rowsLeavingQueue++;
break;
case HistorianWriteOutcome.RetryPlease:
// finding 002: cap retries so a perpetually-RetryPlease (poison)
// row cannot retry forever at the 60s backoff floor. The incoming
// AttemptCount is the count BEFORE this attempt; +1 accounts for the
// attempt count is the count BEFORE this attempt; +1 accounts for the
// bump this drain represents. At the cap, dead-letter instead of
// bumping — and count it as leaving the live queue like PermanentFail.
if (liveRows[i].AttemptCount + 1 >= _maxAttempts)
{
DeadLetterRow(conn, tx, rowId, $"max attempts ({_maxAttempts}) exceeded");
await DeadLetterRowAsync(tx, id, $"max attempts ({_maxAttempts}) exceeded", ct).ConfigureAwait(false);
rowsLeavingQueue++;
}
else
{
BumpAttempt(conn, tx, rowId, "retry-please");
await BumpAttemptAsync(tx, id, "retry-please", ct).ConfigureAwait(false);
}
break;
}
}
tx.Commit();
await tx.CommitAsync(ct).ConfigureAwait(false);
}
// Ack-deleted + PermanentFail-dead-lettered rows both leave the
// non-dead-lettered queue, as do RetryPlease rows that hit the max-attempts
@@ -464,22 +485,66 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
}
finally
{
_drainGate.Release();
_drainGateLock.Release();
}
}
/// <summary>
/// Consults the injected gate, logging only when the decision changes.
/// </summary>
/// <remarks>
/// Logged at all because the closed state is indistinguishable from a healthy idle queue
/// from the outside, and one way to reach it is a redundancy snapshot that never names this
/// node — the documented identity-mismatch shape. Without this line, that misconfiguration
/// would present as alarm history quietly ceasing on both nodes of a pair. Logged on the
/// transition rather than per tick because the drain runs every few seconds.
/// </remarks>
/// <returns><c>true</c> when this tick may proceed.</returns>
private bool EvaluateDrainGate()
{
bool open;
try
{
open = _drainGate();
}
catch (Exception ex)
{
// A throwing gate must not wedge the queue silently, and must not be treated as
// permission either: skip this tick, say why, and re-ask on the next one.
_logger.Error(ex, "Historian drain gate threw; skipping this tick");
return false;
}
bool? previous;
lock (_statusLock)
{
previous = _lastGateDecision;
_lastGateDecision = open;
}
if (previous == open) return open;
if (open)
_logger.Information("Historian drain resumed — this node now services the alarm queue");
else
_logger.Information(
"Historian drain suspended — this node is not the Primary. Queued alarm events are retained and will drain from whichever node holds the Primary role");
return open;
}
/// <inheritdoc />
public HistorianSinkStatus GetStatus()
{
if (_disposed) throw new ObjectDisposedException(nameof(SqliteStoreAndForwardSink));
if (_disposed) throw new ObjectDisposedException(nameof(LocalDbStoreAndForwardSink));
var queued = Interlocked.Read(ref _queuedRowCount);
if (queued < 0) queued = 0;
long deadlettered;
using (var conn = OpenConnection())
using (var conn = _db.CreateConnection())
using (var cmd = conn.CreateCommand())
{
cmd.CommandText = "SELECT COUNT(*) FROM Queue WHERE DeadLettered = 1";
cmd.CommandText = "SELECT COUNT(*) FROM alarm_sf_events WHERE dead_lettered = 1";
deadlettered = (long)(cmd.ExecuteScalar() ?? 0L);
}
@@ -510,10 +575,13 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
/// <returns>The number of rows revived from the dead-letter state.</returns>
public int RetryDeadLettered()
{
if (_disposed) throw new ObjectDisposedException(nameof(SqliteStoreAndForwardSink));
using var conn = OpenConnection();
if (_disposed) throw new ObjectDisposedException(nameof(LocalDbStoreAndForwardSink));
// Through a LocalDb connection, not a private one: the capture triggers are what make this
// revival replicate to the peer, and they only exist on the registered table.
using var conn = _db.CreateConnection();
using var cmd = conn.CreateCommand();
cmd.CommandText = "UPDATE Queue SET DeadLettered = 0, AttemptCount = 0, LastError = NULL WHERE DeadLettered = 1";
cmd.CommandText =
"UPDATE alarm_sf_events SET dead_lettered = 0, attempt_count = 0, last_error = NULL WHERE dead_lettered = 1";
var revived = cmd.ExecuteNonQuery();
// Dead-lettered rows rejoin the non-dead-lettered queue — keep the in-memory
// counter aligned.
@@ -523,29 +591,31 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
/// <summary>
/// One queued row paired with its deserialized event. <see cref="Event"/> is
/// <c>null</c> when the row's <c>PayloadJson</c> is corrupt or un-deserializable —
/// the <see cref="RowId"/> always stays bound to its own row so outcomes can
/// <c>null</c> when the row's <c>payload_json</c> is corrupt or un-deserializable —
/// the <see cref="Id"/> always stays bound to its own row so outcomes can
/// never be mapped to the wrong row.
/// </summary>
private readonly record struct QueueRow(long RowId, AlarmHistorianEvent? Event, long AttemptCount);
private readonly record struct QueueRow(string Id, AlarmHistorianEvent? Event, long AttemptCount);
private List<QueueRow> ReadBatch(SqliteConnection conn)
private async Task<List<QueueRow>> ReadBatchAsync(CancellationToken ct)
{
var rows = new List<QueueRow>();
using var cmd = conn.CreateCommand();
cmd.CommandText = """
SELECT RowId, PayloadJson, AttemptCount FROM Queue
WHERE DeadLettered = 0
ORDER BY RowId ASC
LIMIT $limit
""";
cmd.Parameters.AddWithValue("$limit", _batchSize);
using var reader = cmd.ExecuteReader();
while (reader.Read())
// Ordered by enqueue time rather than insertion order: a hashed TEXT primary key carries
// no sequence the way the legacy AUTOINCREMENT rowid did. Round-trip ("O") timestamps sort
// lexicographically in chronological order; id makes the ordering total.
var raw = await _db.QueryAsync(
"""
SELECT id, payload_json, attempt_count FROM alarm_sf_events
WHERE dead_lettered = 0
ORDER BY enqueued_at_utc ASC, id ASC
LIMIT @Limit
""",
r => (Id: r.GetString(0), Payload: r.GetString(1), AttemptCount: r.GetInt64(2)),
new { Limit = _batchSize },
ct).ConfigureAwait(false);
var rows = new List<QueueRow>(raw.Count);
foreach (var (id, payload, attemptCount) in raw)
{
var rowId = reader.GetInt64(0);
var payload = reader.GetString(1);
var attemptCount = reader.GetInt64(2);
AlarmHistorianEvent? evt;
try
{
@@ -556,77 +626,58 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
// Malformed JSON — carry a null event so the caller dead-letters this row.
evt = null;
}
rows.Add(new QueueRow(rowId, evt, attemptCount));
rows.Add(new QueueRow(id, evt, attemptCount));
}
return rows;
}
private static void DeleteRow(SqliteConnection conn, SqliteTransaction tx, long rowId)
{
using var cmd = conn.CreateCommand();
cmd.Transaction = tx;
cmd.CommandText = "DELETE FROM Queue WHERE RowId = $id";
cmd.Parameters.AddWithValue("$id", rowId);
cmd.ExecuteNonQuery();
}
private static Task DeleteRowAsync(ILocalDbTransaction tx, string id, CancellationToken ct) =>
tx.ExecuteAsync("DELETE FROM alarm_sf_events WHERE id = @Id", new { Id = id }, ct);
private void DeadLetterRow(SqliteConnection conn, SqliteTransaction tx, long rowId, string reason)
{
using var cmd = conn.CreateCommand();
cmd.Transaction = tx;
cmd.CommandText = """
UPDATE Queue SET DeadLettered = 1, LastAttemptUtc = $now, LastError = $err, AttemptCount = AttemptCount + 1
WHERE RowId = $id
""";
cmd.Parameters.AddWithValue("$now", _clock().ToString("O"));
cmd.Parameters.AddWithValue("$err", reason);
cmd.Parameters.AddWithValue("$id", rowId);
cmd.ExecuteNonQuery();
}
private Task DeadLetterRowAsync(ILocalDbTransaction tx, string id, string reason, CancellationToken ct) =>
tx.ExecuteAsync(
"""
UPDATE alarm_sf_events
SET dead_lettered = 1, last_attempt_utc = @Now, last_error = @Err,
attempt_count = attempt_count + 1
WHERE id = @Id
""",
new { Now = _clock().ToString("O"), Err = reason, Id = id },
ct);
private void BumpAttempt(SqliteConnection conn, SqliteTransaction tx, long rowId, string reason)
{
using var cmd = conn.CreateCommand();
cmd.Transaction = tx;
cmd.CommandText = """
UPDATE Queue SET LastAttemptUtc = $now, LastError = $err, AttemptCount = AttemptCount + 1
WHERE RowId = $id
""";
cmd.Parameters.AddWithValue("$now", _clock().ToString("O"));
cmd.Parameters.AddWithValue("$err", reason);
cmd.Parameters.AddWithValue("$id", rowId);
cmd.ExecuteNonQuery();
}
private Task BumpAttemptAsync(ILocalDbTransaction tx, string id, string reason, CancellationToken ct) =>
tx.ExecuteAsync(
"""
UPDATE alarm_sf_events
SET last_attempt_utc = @Now, last_error = @Err, attempt_count = attempt_count + 1
WHERE id = @Id
""",
new { Now = _clock().ToString("O"), Err = reason, Id = id },
ct);
// Async variant used by EnqueueAsync.
private async Task EnforceCapacityAsync(SqliteConnection conn, CancellationToken ct)
private async Task EnforceCapacityAsync(CancellationToken ct)
{
Interlocked.Increment(ref _capacityProbeCount);
long count;
using (var cmd = conn.CreateCommand())
{
cmd.CommandText = "SELECT COUNT(*) FROM Queue WHERE DeadLettered = 0";
count = (long)(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false) ?? 0L);
}
var count = await CountAsync("dead_lettered = 0", ct).ConfigureAwait(false);
// Resync the in-memory counter while we have a fresh number.
Interlocked.Exchange(ref _queuedRowCount, count);
if (count < _capacity) return;
var toEvict = count - _capacity + 1;
using (var cmd = conn.CreateCommand())
{
cmd.CommandText = """
DELETE FROM Queue
WHERE RowId IN (
SELECT RowId FROM Queue
WHERE DeadLettered = 0
ORDER BY RowId ASC
LIMIT $n
)
""";
cmd.Parameters.AddWithValue("$n", toEvict);
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
}
await _db.ExecuteAsync(
"""
DELETE FROM alarm_sf_events
WHERE id IN (
SELECT id FROM alarm_sf_events
WHERE dead_lettered = 0
ORDER BY enqueued_at_utc ASC, id ASC
LIMIT @N
)
""",
new { N = toEvict },
ct).ConfigureAwait(false);
Interlocked.Add(ref _queuedRowCount, -toEvict);
long lifetimeEvicted;
lock (_statusLock) { _evictedCount += toEvict; lifetimeEvicted = _evictedCount; }
@@ -635,40 +686,21 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
_capacity, toEvict, lifetimeEvicted);
}
private void PurgeAgedDeadLetters(SqliteConnection conn)
private async Task PurgeAgedDeadLettersAsync(CancellationToken ct)
{
var cutoff = (_clock() - _deadLetterRetention).ToString("O");
using var cmd = conn.CreateCommand();
cmd.CommandText = """
DELETE FROM Queue
WHERE DeadLettered = 1 AND LastAttemptUtc IS NOT NULL AND LastAttemptUtc < $cutoff
""";
cmd.Parameters.AddWithValue("$cutoff", cutoff);
var purged = cmd.ExecuteNonQuery();
var purged = await _db.ExecuteAsync(
"""
DELETE FROM alarm_sf_events
WHERE dead_lettered = 1 AND last_attempt_utc IS NOT NULL AND last_attempt_utc < @Cutoff
""",
new { Cutoff = cutoff },
ct).ConfigureAwait(false);
if (purged > 0)
_logger.Information("Purged {Count} dead-lettered row(s) past retention window", purged);
}
private void InitializeSchema()
{
using var conn = OpenConnection();
using var cmd = conn.CreateCommand();
cmd.CommandText = """
CREATE TABLE IF NOT EXISTS Queue (
RowId INTEGER PRIMARY KEY AUTOINCREMENT,
AlarmId TEXT NOT NULL,
EnqueuedUtc TEXT NOT NULL,
PayloadJson TEXT NOT NULL,
AttemptCount INTEGER NOT NULL DEFAULT 0,
LastAttemptUtc TEXT NULL,
LastError TEXT NULL,
DeadLettered INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS IX_Queue_Drain ON Queue (DeadLettered, RowId);
""";
cmd.ExecuteNonQuery();
}
private void BumpBackoff() => _backoffIndex = Math.Min(_backoffIndex + 1, BackoffLadder.Length - 1);
private void ResetBackoff() => _backoffIndex = 0;
@@ -676,12 +708,16 @@ public sealed class SqliteStoreAndForwardSink : IAlarmHistorianSink, IDisposable
public TimeSpan CurrentBackoff => BackoffLadder[_backoffIndex];
/// <summary>Disposes the sink and releases all held resources including the drain timer and the writer.</summary>
/// <remarks>
/// The <see cref="ILocalDb"/> is NOT disposed here — it is a shared singleton owned by the
/// host and used by the deployment cache too.
/// </remarks>
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_drainTimer?.Dispose();
_drainGate.Dispose();
_drainGateLock.Dispose();
if (_writer is IDisposable writerDisposable) writerDisposable.Dispose();
}
}
src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.csproj
+6
View File
@@ -20,6 +20,12 @@
-->
<PackageReference Include="SQLitePCLRaw.bundle_e_sqlite3"/>
<PackageReference Include="Serilog"/>
<!--
The store-and-forward buffer lives in the node's consolidated LocalDb file so it
replicates to the redundant pair peer. Only the ILocalDb SQL surface is used here; the
Host owns the DI registration and the replication package.
-->
<PackageReference Include="ZB.MOM.WW.LocalDb"/>
</ItemGroup>
<ItemGroup>