feat(localdb): one-time alarm-historian.db migrator
Copies the pre-consolidation store-and-forward queue into the consolidated
database on first boot, then renames the legacy file aside. Rows are in that
file precisely because the historian could not be reached, so dropping them
on upgrade would discard exactly the alarm audit trail the queue exists to
protect.
Runs last in OnReady, after every RegisterReplicated call. It is the only
thing in OnReady that writes rows, and capture is trigger-based: a migration
that ran before registration would recover the backlog locally and never
replicate a line of it, silently and permanently.
Ids are derived from the payload rather than the plan's mig-{node}-{legacyId}
scheme. Node-prefixing solves the collision the legacy AUTOINCREMENT key
would cause -- node A's row 7 and node B's row 7 are different alarms -- but
it preserves a duplication that should be collapsed instead. A warm pair's
two legacy files OVERLAP: HistorianAdapterActor default-writes while its
redundancy role is unknown, so both nodes accepted the same transitions
during every boot window. Prefixed ids would carry those duplicates into the
merged buffer forever; equal-payload ids converge them. The same property
makes a crash between commit and rename harmless under INSERT OR IGNORE.
OnReady now takes IConfiguration rather than offering an overload that skips
the migration. A wiring mistake that silently discarded a node's undelivered
alarm history is not a mistake worth making possible.
The copy is restricted to the columns the legacy table actually has. Naming
a column an older build never wrote throws "no such column", which would
discard every row in the table rather than the one field.
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -37,10 +37,11 @@
|
||||
{
|
||||
"id": 4,
|
||||
"subject": "Task 4: One-time alarm-historian.db legacy migrator",
|
||||
"status": "pending",
|
||||
"status": "completed",
|
||||
"blockedBy": [
|
||||
3
|
||||
]
|
||||
],
|
||||
"note": "AlarmSfLegacyMigrator runs LAST in OnReady (which now takes IConfiguration - no skip-migration overload exists). DEVIATION D-6: ids are the payload hash (AlarmSfSchema.DeriveId, lifted out of the sink) rather than mig-{node}-{legacyId} - a warm pair's two legacy files OVERLAP, and node-prefixing would carry that duplication forward forever. DEVIATION: tests live in Host.IntegrationTests; the plan's Host.Tests project does not exist."
|
||||
},
|
||||
{
|
||||
"id": 5,
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using Microsoft.Data.Sqlite;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
@@ -35,6 +37,32 @@ public static class AlarmSfSchema
|
||||
/// <summary>The single primary-key column.</summary>
|
||||
public const string IdColumn = "id";
|
||||
|
||||
/// <summary>
|
||||
/// Derives a row's primary key from its serialized payload.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Deterministic rather than a fresh GUID, so that the same event arriving twice
|
||||
/// converges to one row under last-writer-wins instead of duplicating. That happens for
|
||||
/// real in two places: <c>HistorianAdapterActor</c> default-writes while its redundancy
|
||||
/// role is unknown, so both nodes of a pair accept the same fanned transition in the
|
||||
/// window before the first snapshot; and both nodes independently migrate their own
|
||||
/// legacy queue file, which for a warm pair holds overlapping history.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Two genuinely distinct events cannot collide. <c>AlarmHistorianEvent</c> carries a
|
||||
/// full-precision timestamp alongside the alarm id, transition kind, message and user,
|
||||
/// so an equal hash means an equal event.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="payloadJson">The serialized <c>AlarmHistorianEvent</c>.</param>
|
||||
/// <returns>The row's primary key.</returns>
|
||||
public static string DeriveId(string payloadJson)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(payloadJson);
|
||||
return Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(payloadJson)));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates the buffer table if it does not already exist. Idempotent.
|
||||
/// </summary>
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Serilog;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
@@ -213,21 +211,6 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl
|
||||
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)
|
||||
{
|
||||
@@ -249,7 +232,7 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl
|
||||
""",
|
||||
new
|
||||
{
|
||||
Id = DeriveId(payload),
|
||||
Id = AlarmSfSchema.DeriveId(payload),
|
||||
AlarmId = evt.AlarmId,
|
||||
EnqueuedAtUtc = _clock().ToString("O"),
|
||||
PayloadJson = payload,
|
||||
|
||||
@@ -0,0 +1,217 @@
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
||||
|
||||
/// <summary>
|
||||
/// One-time copy of the pre-consolidation <c>alarm-historian.db</c> store-and-forward queue
|
||||
/// into the consolidated <c>ZB.MOM.WW.LocalDb</c> database.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>What is at stake.</b> Rows sit in that file precisely because the historian could not
|
||||
/// be reached. Discarding them on upgrade would throw away exactly the alarm audit trail
|
||||
/// the store-and-forward queue exists to protect — so the copy tolerates an older column
|
||||
/// set rather than bailing out, and refuses to rename a file it did not actually read.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Runs after <c>RegisterReplicated</c>, deliberately.</b> Capture is trigger-based, so
|
||||
/// rows inserted before registration never enter the oplog and never reach the peer —
|
||||
/// silently, and permanently, because nothing revisits history. Migrating after
|
||||
/// registration means the recovered backlog replicates like any other write.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Ids come from the payload, not from the legacy row id.</b> The legacy primary key was
|
||||
/// <c>RowId INTEGER PRIMARY KEY AUTOINCREMENT</c>, which cannot survive into a replicated
|
||||
/// table: each node allocated its own sequence, so node A's row 7 and node B's row 7 are
|
||||
/// different alarms that would silently overwrite one another under last-writer-wins.
|
||||
/// Hashing the payload (<see cref="AlarmSfSchema.DeriveId"/>) solves that and one more
|
||||
/// problem besides — a warm pair's two legacy files <i>overlap</i>, because
|
||||
/// <c>HistorianAdapterActor</c> default-writes while its redundancy role is unknown and
|
||||
/// both nodes therefore accepted the same transitions during every boot window. A
|
||||
/// node-prefixed id would carry that duplication into the merged buffer forever; an
|
||||
/// equal-payload id collapses it. It is also what makes a re-run harmless under
|
||||
/// <c>INSERT OR IGNORE</c>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>All-or-nothing.</b> The copy runs in one transaction and the legacy file is renamed
|
||||
/// to <c><name>.migrated</c> only after the commit. A failure throws out of
|
||||
/// <c>OnReady</c>, failing host startup with the legacy file untouched — no half-migrated
|
||||
/// state to reason about, and the operator still holds the original. A later boot sees the
|
||||
/// renamed file and no-ops.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static class AlarmSfLegacyMigrator
|
||||
{
|
||||
private const string MigratedSuffix = ".migrated";
|
||||
|
||||
/// <summary>The legacy queue table.</summary>
|
||||
private const string LegacyTable = "Queue";
|
||||
|
||||
/// <summary>
|
||||
/// The configuration key that used to carry the standalone queue file's path. Read as a raw
|
||||
/// key because the corresponding <c>AlarmHistorianOptions</c> property was removed with the
|
||||
/// bespoke file management it configured.
|
||||
/// </summary>
|
||||
public const string LegacyPathKey = "AlarmHistorian:DatabasePath";
|
||||
|
||||
/// <summary>The pre-removal default for <see cref="LegacyPathKey"/>.</summary>
|
||||
private const string DefaultLegacyPath = "alarm-historian.db";
|
||||
|
||||
/// <summary>
|
||||
/// The legacy column whose absence means this is not a queue file we can read. Without the
|
||||
/// payload there is no event and no id to derive from one.
|
||||
/// </summary>
|
||||
private const string RequiredColumn = "PayloadJson";
|
||||
|
||||
/// <summary>
|
||||
/// Every legacy column, mapped to its consolidated counterpart. Columns absent from an
|
||||
/// older file are dropped from the copy rather than failing it — see
|
||||
/// <see cref="PresentColumns"/>.
|
||||
/// </summary>
|
||||
private static readonly (string Legacy, string Current)[] ColumnMap =
|
||||
[
|
||||
("AlarmId", "alarm_id"),
|
||||
("EnqueuedUtc", "enqueued_at_utc"),
|
||||
("PayloadJson", "payload_json"),
|
||||
("AttemptCount", "attempt_count"),
|
||||
("LastAttemptUtc", "last_attempt_utc"),
|
||||
("LastError", "last_error"),
|
||||
("DeadLettered", "dead_lettered"),
|
||||
];
|
||||
|
||||
/// <summary>
|
||||
/// Copies any legacy alarm queue into <paramref name="db"/>, then renames the legacy file.
|
||||
/// </summary>
|
||||
/// <param name="db">The consolidated database, with <c>alarm_sf_events</c> already registered.</param>
|
||||
/// <param name="config">Configuration supplying the legacy queue's path.</param>
|
||||
public static void Migrate(ILocalDb db, IConfiguration config)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(db);
|
||||
ArgumentNullException.ThrowIfNull(config);
|
||||
|
||||
var legacyPath = ResolveLegacyPath(config);
|
||||
if (!ShouldMigrate(legacyPath)) return;
|
||||
|
||||
using (var legacy = OpenLegacyReadOnly(legacyPath))
|
||||
{
|
||||
var present = PresentColumns(legacy);
|
||||
|
||||
// An absent table probes as zero columns, so this one guard covers both "no queue
|
||||
// table" and "a shape we do not recognise". Returning without renaming leaves the file
|
||||
// for an operator to inspect.
|
||||
if (!present.Contains(RequiredColumn)) return;
|
||||
|
||||
using var connection = db.CreateConnection();
|
||||
using var transaction = connection.BeginTransaction();
|
||||
CopyRows(legacy, connection, transaction, present);
|
||||
transaction.Commit();
|
||||
}
|
||||
|
||||
// Only after the commit. A crash between the two leaves the file in place and the next
|
||||
// boot copies it again, which the payload-derived ids make harmless.
|
||||
File.Move(legacyPath, legacyPath + MigratedSuffix, overwrite: true);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves the legacy queue path from the removed configuration key, falling back to the
|
||||
/// code default it used to carry.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Relative paths resolve against the process working directory — not the LocalDb
|
||||
/// directory — because that is where the old code actually put them.
|
||||
/// </remarks>
|
||||
/// <param name="config">The application configuration.</param>
|
||||
/// <returns>An absolute path, or empty when there is nothing durable to migrate from.</returns>
|
||||
internal static string ResolveLegacyPath(IConfiguration config)
|
||||
{
|
||||
var path = config[LegacyPathKey] ?? DefaultLegacyPath;
|
||||
|
||||
// In-memory and URI-form sources (test / dev configurations) have nothing durable behind
|
||||
// them, and Path.GetFullPath on them would produce nonsense.
|
||||
if (string.IsNullOrWhiteSpace(path) ||
|
||||
path.Equals(":memory:", StringComparison.OrdinalIgnoreCase) ||
|
||||
path.StartsWith("file:", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
return string.Empty;
|
||||
}
|
||||
|
||||
return Path.GetFullPath(path);
|
||||
}
|
||||
|
||||
/// <summary>Whether there is a legacy file worth opening.</summary>
|
||||
private static bool ShouldMigrate(string legacyPath) =>
|
||||
!string.IsNullOrEmpty(legacyPath)
|
||||
&& File.Exists(legacyPath)
|
||||
&& !File.Exists(legacyPath + MigratedSuffix);
|
||||
|
||||
/// <summary>Opens the legacy file read-only, so a failed migration cannot damage it.</summary>
|
||||
private static SqliteConnection OpenLegacyReadOnly(string path)
|
||||
{
|
||||
var connection = new SqliteConnection(new SqliteConnectionStringBuilder
|
||||
{
|
||||
DataSource = path,
|
||||
Mode = SqliteOpenMode.ReadOnly,
|
||||
}.ToString());
|
||||
connection.Open();
|
||||
return connection;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Which of the columns we know about the legacy table actually has.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// A file written by an older build predates some columns, and naming a missing one in the
|
||||
/// SELECT throws "no such column" — which discards every row in the table rather than the
|
||||
/// one field. Intersecting first means an old file migrates its data and simply leaves the
|
||||
/// newer columns at their schema defaults.
|
||||
/// </remarks>
|
||||
private static HashSet<string> PresentColumns(SqliteConnection legacy)
|
||||
{
|
||||
var present = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
using var cmd = legacy.CreateCommand();
|
||||
cmd.CommandText = $"SELECT name FROM pragma_table_info('{LegacyTable}')";
|
||||
using var reader = cmd.ExecuteReader();
|
||||
while (reader.Read())
|
||||
present.Add(reader.GetString(0));
|
||||
|
||||
return present;
|
||||
}
|
||||
|
||||
/// <summary>Copies every legacy row into the consolidated table, keyed by payload hash.</summary>
|
||||
private static void CopyRows(
|
||||
SqliteConnection legacy,
|
||||
SqliteConnection target,
|
||||
SqliteTransaction transaction,
|
||||
HashSet<string> present)
|
||||
{
|
||||
var columns = ColumnMap.Where(c => present.Contains(c.Legacy)).ToArray();
|
||||
var payloadOrdinal = Array.FindIndex(columns, c => c.Legacy == RequiredColumn);
|
||||
|
||||
using var read = legacy.CreateCommand();
|
||||
read.CommandText =
|
||||
$"SELECT {string.Join(", ", columns.Select(c => c.Legacy))} FROM {LegacyTable}";
|
||||
|
||||
using var reader = read.ExecuteReader();
|
||||
while (reader.Read())
|
||||
{
|
||||
using var insert = target.CreateCommand();
|
||||
insert.Transaction = transaction;
|
||||
insert.CommandText = $"""
|
||||
INSERT OR IGNORE INTO {AlarmSfSchema.EventsTable}
|
||||
({AlarmSfSchema.IdColumn}, {string.Join(", ", columns.Select(c => c.Current))})
|
||||
VALUES
|
||||
($id, {string.Join(", ", columns.Select((_, i) => $"$c{i}"))})
|
||||
""";
|
||||
|
||||
insert.Parameters.AddWithValue("$id", AlarmSfSchema.DeriveId(reader.GetString(payloadOrdinal)));
|
||||
for (var i = 0; i < columns.Length; i++)
|
||||
insert.Parameters.AddWithValue($"$c{i}", reader.GetValue(i) ?? DBNull.Value);
|
||||
|
||||
insert.ExecuteNonQuery();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -88,7 +88,7 @@ public static class LocalDbRegistration
|
||||
ArgumentNullException.ThrowIfNull(services);
|
||||
ArgumentNullException.ThrowIfNull(configuration);
|
||||
|
||||
services.AddZbLocalDb(configuration, LocalDbSetup.OnReady);
|
||||
services.AddZbLocalDb(configuration, db => LocalDbSetup.OnReady(db, configuration));
|
||||
services.AddZbLocalDbReplication(configuration);
|
||||
|
||||
// The consumer-facing seam. WithOtOpcUaRuntimeActors resolves this optionally and threads it
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
|
||||
@@ -27,8 +28,8 @@ public static class LocalDbSetup
|
||||
/// <c>RegisterReplicated</c> is what installs the three AFTER triggers that capture
|
||||
/// changes into the oplog. Any row written before that call is never captured, so it
|
||||
/// never reaches the peer — silently, and permanently, because nothing ever revisits
|
||||
/// history. The Phase-2 legacy alarm migrator therefore runs <i>after</i> every
|
||||
/// registration, not alongside the DDL.
|
||||
/// history. <see cref="AlarmSfLegacyMigrator"/> therefore runs <b>last</b>, after every
|
||||
/// registration — it writes rows, and they must be captured like any other write.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The alarm buffer's tables are created unconditionally, regardless of whether this
|
||||
@@ -39,9 +40,15 @@ public static class LocalDbSetup
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="db">The freshly constructed local database.</param>
|
||||
public static void OnReady(ILocalDb db)
|
||||
/// <param name="configuration">
|
||||
/// Application configuration, read by the legacy migrator for the pre-consolidation queue's
|
||||
/// path. Required rather than optional: an overload that silently skipped the migration
|
||||
/// would be one wiring mistake away from discarding a node's undelivered alarm history.
|
||||
/// </param>
|
||||
public static void OnReady(ILocalDb db, IConfiguration configuration)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(db);
|
||||
ArgumentNullException.ThrowIfNull(configuration);
|
||||
|
||||
// CreateConnection() hands back an already-open, pragma-configured connection carrying the
|
||||
// zb_hlc_next() UDF the capture triggers need. Calling Open() on it would throw.
|
||||
@@ -54,5 +61,8 @@ public static class LocalDbSetup
|
||||
db.RegisterReplicated(DeploymentCacheSchema.ArtifactsTable);
|
||||
db.RegisterReplicated(DeploymentCacheSchema.PointerTable);
|
||||
db.RegisterReplicated(AlarmSfSchema.EventsTable);
|
||||
|
||||
// LAST, and only here. This is the one call in OnReady that writes rows.
|
||||
AlarmSfLegacyMigrator.Migrate(db, configuration);
|
||||
}
|
||||
}
|
||||
|
||||
+333
@@ -0,0 +1,333 @@
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb;
|
||||
|
||||
/// <summary>
|
||||
/// The one-time copy of the pre-consolidation <c>alarm-historian.db</c> queue into the
|
||||
/// consolidated LocalDb.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// What is being protected is undelivered alarm history: rows that are in the legacy file
|
||||
/// precisely because the historian could not be reached. Dropping them on upgrade would
|
||||
/// discard exactly the audit trail the store-and-forward queue exists to preserve.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Run against a real temp-file <see cref="ILocalDb"/> built through the production
|
||||
/// <see cref="LocalDbSetup.OnReady"/>, so the registration order the migrator depends on is
|
||||
/// the one the host actually runs — a hand-written schema here would prove only that the
|
||||
/// test agrees with itself.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class AlarmSfLegacyMigratorTests : IDisposable
|
||||
{
|
||||
private readonly string _dbPath =
|
||||
Path.Combine(Path.GetTempPath(), $"otopcua-alarmsf-mig-{Guid.NewGuid():N}.db");
|
||||
|
||||
private readonly string _legacyPath =
|
||||
Path.Combine(Path.GetTempPath(), $"otopcua-alarmsf-legacy-{Guid.NewGuid():N}.db");
|
||||
|
||||
private ServiceProvider? _provider;
|
||||
|
||||
// ---- the cases ---------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task Every_legacy_row_lands_in_the_consolidated_table()
|
||||
{
|
||||
SeedLegacy(
|
||||
(1, "eq/a", Payload("eq/a"), 0, dead: 0),
|
||||
(2, "eq/b", Payload("eq/b"), 3, dead: 0),
|
||||
(3, "eq/c", Payload("eq/c"), 9, dead: 1));
|
||||
|
||||
var db = BuildDb();
|
||||
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
|
||||
|
||||
(await CountAsync(db)).ShouldBe(3);
|
||||
// Attempt counts and the dead-letter flag must survive: a migrated row that lost its
|
||||
// attempt count would restart its retry budget, and one that lost its dead-letter flag
|
||||
// would be re-sent to a historian that already refused it permanently.
|
||||
var rows = await db.QueryAsync(
|
||||
"SELECT alarm_id, attempt_count, dead_lettered FROM alarm_sf_events ORDER BY alarm_id",
|
||||
r => (AlarmId: r.GetString(0), Attempts: r.GetInt64(1), Dead: r.GetInt64(2)),
|
||||
parameters: null,
|
||||
TestContext.Current.CancellationToken);
|
||||
|
||||
rows.ShouldBe([("eq/a", 0L, 0L), ("eq/b", 3L, 0L), ("eq/c", 9L, 1L)]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Migrated_rows_enter_the_oplog()
|
||||
{
|
||||
// THE ordering pin. Capture is trigger-based, so rows copied in before
|
||||
// RegisterReplicated would never enter the oplog and would never reach the peer —
|
||||
// silently, and permanently. A migrator invoked too early passes every other case here.
|
||||
SeedLegacy((1, "eq/a", Payload("eq/a"), 0, dead: 0));
|
||||
|
||||
var db = BuildDb();
|
||||
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
|
||||
|
||||
var oplog = await db.QueryAsync(
|
||||
"SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_sf_events'",
|
||||
r => r.GetInt64(0),
|
||||
parameters: null,
|
||||
TestContext.Current.CancellationToken);
|
||||
|
||||
oplog[0].ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_second_run_is_a_no_op()
|
||||
{
|
||||
SeedLegacy((1, "eq/a", Payload("eq/a"), 0, dead: 0));
|
||||
var db = BuildDb();
|
||||
var config = ConfigWithLegacyPath();
|
||||
|
||||
AlarmSfLegacyMigrator.Migrate(db, config);
|
||||
AlarmSfLegacyMigrator.Migrate(db, config);
|
||||
|
||||
(await CountAsync(db)).ShouldBe(1);
|
||||
File.Exists(_legacyPath).ShouldBeFalse("the legacy file is renamed once the copy commits");
|
||||
File.Exists(_legacyPath + ".migrated").ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Re_migrating_the_same_rows_cannot_duplicate_them()
|
||||
{
|
||||
// The sidecar rename is the normal guard, but it is not the only one that has to hold:
|
||||
// a crash between commit and rename leaves the file in place, and the next boot copies it
|
||||
// again. Ids are derived from the payload, so the re-copy collides with itself and is
|
||||
// ignored rather than doubling the queue.
|
||||
SeedLegacy((1, "eq/a", Payload("eq/a"), 0, dead: 0));
|
||||
var db = BuildDb();
|
||||
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
|
||||
|
||||
File.Move(_legacyPath + ".migrated", _legacyPath); // simulate the crash-before-rename
|
||||
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
|
||||
|
||||
(await CountAsync(db)).ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_older_legacy_file_missing_a_column_still_migrates()
|
||||
{
|
||||
// A file written by an older build predates LastError. Naming a missing column in the
|
||||
// SELECT throws "no such column", which would discard every row in the table rather than
|
||||
// the one field. Intersecting the column list first means the data migrates and the
|
||||
// absent column keeps its schema default.
|
||||
SeedLegacyWithoutLastError((1, "eq/a", Payload("eq/a")));
|
||||
|
||||
var db = BuildDb();
|
||||
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
|
||||
|
||||
(await CountAsync(db)).ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_failed_copy_leaves_the_legacy_file_untouched()
|
||||
{
|
||||
// The legacy table exists but its payload column is missing, so the required-column guard
|
||||
// rejects the file. Nothing is copied and nothing is renamed: the operator still has the
|
||||
// original to recover from, which is the whole point of renaming only after commit.
|
||||
SeedUnrecognisedLegacy();
|
||||
|
||||
var db = BuildDb();
|
||||
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
|
||||
|
||||
(await CountAsync(db)).ShouldBe(0);
|
||||
File.Exists(_legacyPath).ShouldBeTrue("an un-copied file must not be renamed away");
|
||||
File.Exists(_legacyPath + ".migrated").ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_missing_legacy_file_is_a_clean_no_op()
|
||||
{
|
||||
// The expected case on a fresh install and on any node that never enabled the historian.
|
||||
var db = BuildDb();
|
||||
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
|
||||
|
||||
(await CountAsync(db)).ShouldBe(0);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(":memory:")]
|
||||
[InlineData("file:thing?mode=memory")]
|
||||
[InlineData("")]
|
||||
public async Task Non_file_legacy_sources_are_no_ops(string configured)
|
||||
{
|
||||
var db = BuildDb();
|
||||
AlarmSfLegacyMigrator.Migrate(db, Config(configured));
|
||||
|
||||
(await CountAsync(db)).ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Two_nodes_migrating_the_same_event_converge_on_one_row()
|
||||
{
|
||||
// A warm pair's two legacy files overlap: HistorianAdapterActor default-writes while the
|
||||
// redundancy role is unknown, so both nodes accepted the same transitions during every
|
||||
// boot window. Node-prefixed ids would carry that duplication forward into the merged
|
||||
// buffer permanently; payload-derived ids collapse it.
|
||||
var shared = Payload("eq/shared");
|
||||
SeedLegacy((1, "eq/shared", shared, 0, dead: 0));
|
||||
|
||||
var db = BuildDb();
|
||||
AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
|
||||
|
||||
// The peer's file: a different legacy RowId for the very same event.
|
||||
var peerLegacy = _legacyPath + ".peer";
|
||||
SeedLegacyAt(peerLegacy, (77, "eq/shared", shared, 0, dead: 0));
|
||||
AlarmSfLegacyMigrator.Migrate(db, Config(peerLegacy));
|
||||
|
||||
(await CountAsync(db)).ShouldBe(1);
|
||||
|
||||
try { File.Delete(peerLegacy + ".migrated"); } catch { /* best effort */ }
|
||||
}
|
||||
|
||||
// ---- fixture -----------------------------------------------------------------------------
|
||||
|
||||
private ILocalDb BuildDb()
|
||||
{
|
||||
var configuration = new ConfigurationBuilder()
|
||||
.AddInMemoryCollection(new Dictionary<string, string?> { ["LocalDb:Path"] = _dbPath })
|
||||
.Build();
|
||||
|
||||
_provider = new ServiceCollection()
|
||||
.AddZbLocalDb(configuration, db => LocalDbSetup.OnReady(db, configuration))
|
||||
.BuildServiceProvider();
|
||||
|
||||
return _provider.GetRequiredService<ILocalDb>();
|
||||
}
|
||||
|
||||
private IConfiguration ConfigWithLegacyPath() => Config(_legacyPath);
|
||||
|
||||
private static IConfiguration Config(string legacyPath) =>
|
||||
new ConfigurationBuilder()
|
||||
.AddInMemoryCollection(new Dictionary<string, string?>
|
||||
{
|
||||
["AlarmHistorian:DatabasePath"] = legacyPath,
|
||||
})
|
||||
.Build();
|
||||
|
||||
private static async Task<long> CountAsync(ILocalDb db)
|
||||
{
|
||||
var rows = await db.QueryAsync(
|
||||
"SELECT COUNT(*) FROM alarm_sf_events",
|
||||
r => r.GetInt64(0),
|
||||
parameters: null,
|
||||
TestContext.Current.CancellationToken);
|
||||
return rows[0];
|
||||
}
|
||||
|
||||
/// <summary>A payload shaped like a serialized <c>AlarmHistorianEvent</c>.</summary>
|
||||
private static string Payload(string alarmId) =>
|
||||
$$"""{"AlarmId":"{{alarmId}}","EquipmentPath":"line/eq","AlarmName":"hi","TimestampUtc":"2026-07-20T00:00:00Z"}""";
|
||||
|
||||
private void SeedLegacy(params (long RowId, string AlarmId, string Payload, long Attempts, long dead)[] rows) =>
|
||||
SeedLegacyAt(_legacyPath, rows);
|
||||
|
||||
/// <summary>Creates a legacy queue file with the full pre-cutover schema and the given rows.</summary>
|
||||
private static void SeedLegacyAt(
|
||||
string path, params (long RowId, string AlarmId, string Payload, long Attempts, long dead)[] rows)
|
||||
{
|
||||
using var conn = Open(path);
|
||||
Exec(conn, """
|
||||
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
|
||||
);
|
||||
""");
|
||||
|
||||
foreach (var row in rows)
|
||||
{
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = """
|
||||
INSERT INTO Queue (RowId, AlarmId, EnqueuedUtc, PayloadJson, AttemptCount, DeadLettered)
|
||||
VALUES ($id, $alarm, '2026-07-20T00:00:00.0000000Z', $payload, $attempts, $dead)
|
||||
""";
|
||||
cmd.Parameters.AddWithValue("$id", row.RowId);
|
||||
cmd.Parameters.AddWithValue("$alarm", row.AlarmId);
|
||||
cmd.Parameters.AddWithValue("$payload", row.Payload);
|
||||
cmd.Parameters.AddWithValue("$attempts", row.Attempts);
|
||||
cmd.Parameters.AddWithValue("$dead", row.dead);
|
||||
cmd.ExecuteNonQuery();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A legacy file from an older build, predating the <c>LastError</c> column.</summary>
|
||||
private void SeedLegacyWithoutLastError(params (long RowId, string AlarmId, string Payload)[] rows)
|
||||
{
|
||||
using var conn = Open(_legacyPath);
|
||||
Exec(conn, """
|
||||
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,
|
||||
DeadLettered INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
""");
|
||||
|
||||
foreach (var row in rows)
|
||||
{
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = """
|
||||
INSERT INTO Queue (RowId, AlarmId, EnqueuedUtc, PayloadJson)
|
||||
VALUES ($id, $alarm, '2026-07-20T00:00:00.0000000Z', $payload)
|
||||
""";
|
||||
cmd.Parameters.AddWithValue("$id", row.RowId);
|
||||
cmd.Parameters.AddWithValue("$alarm", row.AlarmId);
|
||||
cmd.Parameters.AddWithValue("$payload", row.Payload);
|
||||
cmd.ExecuteNonQuery();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A file with a <c>Queue</c> table this build cannot read — no payload column.</summary>
|
||||
private void SeedUnrecognisedLegacy()
|
||||
{
|
||||
using var conn = Open(_legacyPath);
|
||||
Exec(conn, "CREATE TABLE Queue (RowId INTEGER PRIMARY KEY AUTOINCREMENT, Something TEXT);");
|
||||
Exec(conn, "INSERT INTO Queue (Something) VALUES ('x');");
|
||||
}
|
||||
|
||||
private static SqliteConnection Open(string path)
|
||||
{
|
||||
var conn = new SqliteConnection(new SqliteConnectionStringBuilder { DataSource = path }.ToString());
|
||||
conn.Open();
|
||||
return conn;
|
||||
}
|
||||
|
||||
private static void Exec(SqliteConnection conn, string sql)
|
||||
{
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = sql;
|
||||
cmd.ExecuteNonQuery();
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
_provider?.Dispose();
|
||||
SqliteConnection.ClearAllPools();
|
||||
|
||||
foreach (var baseName in new[] { _dbPath, _legacyPath, _legacyPath + ".migrated" })
|
||||
{
|
||||
foreach (var suffix in new[] { "", "-wal", "-shm" })
|
||||
{
|
||||
try { File.Delete(baseName + suffix); } catch { /* best effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -265,7 +265,7 @@ public sealed class LocalDbPairHarness : IAsyncDisposable
|
||||
.Build();
|
||||
|
||||
return new ServiceCollection()
|
||||
.AddZbLocalDb(config, LocalDbSetup.OnReady)
|
||||
.AddZbLocalDb(config, db => LocalDbSetup.OnReady(db, config))
|
||||
.BuildServiceProvider();
|
||||
}
|
||||
|
||||
|
||||
@@ -39,7 +39,7 @@ public sealed class LocalDbSetupTests : IDisposable
|
||||
.Build();
|
||||
|
||||
_provider = new ServiceCollection()
|
||||
.AddZbLocalDb(configuration, LocalDbSetup.OnReady)
|
||||
.AddZbLocalDb(configuration, db => LocalDbSetup.OnReady(db, configuration))
|
||||
.BuildServiceProvider();
|
||||
|
||||
return _provider.GetRequiredService<ILocalDb>();
|
||||
|
||||
Reference in New Issue
Block a user