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

test(localdb): port store-and-forward replication intents as CDC convergence specs

Browse Source 
Task 10. Specifications first, deletion second: the bespoke ReplicationService
(explicit Add/Remove/Park/Requeue over Akka) dies at Task 14, so its behaviour
is restated here as outcomes the CDC replacement must still deliver. Written
in terms of ROWS, not operations — under CDC there is no Add or Park message to
observe, only a row that must end up right on both nodes.

Not ported: ReplicationOperations_AreDispatchedInIssueOrder. It asserts the
mechanism (inline fire-and-forget dispatch), and CDC capture is asynchronous
and batched by construction. Its portable content is the ordering OUTCOME —
add-then-remove must never converge to present — which is a test here, with
that reasoning recorded in the file so it does not read as an accidental drop.

DEVIATION: extracted the Phase 1 fixture into LocalDbSitePairHarness rather
than duplicating ~150 lines. Phase 1's tests now derive from it and still pass
unchanged. The harness registers the Phase 2 tables itself, since production
OnReady does not until Task 14; that method is marked for deletion at the
cutover, and the 8-table list is written literally so a cutover registering the
wrong set fails these tests instead of agreeing with itself.

Non-vacuity verified by unregistering sf_messages: 6 of 7 failed. The 7th —
the ordering test — PASSED, because an absent row is also what a pair that
replicates nothing looks like. Fixed with a control row that must converge in
the same window, so the absence is evidence rather than silence.

Also corrected two comments from Task 9 that claimed Task 14 makes
notification_lists/smtp_configurations replicated. It explicitly does not
register them, for the same reason the migrator skips them.

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
Joseph Doherty
2026-07-20 03:50:48 -04:00
parent 0dbfefba62
commit 2bbe66311d
5 changed files with 528 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbLegacyMigrator.cs
+10 -4
View File
@@ -83,10 +83,16 @@ public static class SiteLocalDbLegacyMigrator
/// <remarks>
/// <b><c>notification_lists</c> and <c>smtp_configurations</c> are deliberately absent.</b>
/// Both are purged on every deploy and are permanently empty by design — the site-side
/// write paths were removed on 2026-07-10. A pre-fix legacy file can still hold rows,
/// and <c>smtp_configurations.password</c> is plaintext, so migrating them would
/// resurrect plaintext SMTP passwords into a table that Task 14 makes REPLICATED. The
/// tables themselves are still created (see <c>SiteStorageSchema</c>); only their
/// write paths were removed on 2026-07-10. A pre-fix legacy file can still hold rows, and
/// <c>smtp_configurations.password</c> is plaintext.
/// <para>
/// Skipping them here is one half of a pair: the cutover also declines to register them
/// for replication, for the same reason. Migrating them would leave plaintext SMTP
/// passwords sitting in the consolidated database — one future <c>RegisterReplicated</c>
/// away from being shipped to a peer — in exchange for resurrecting config that nothing
/// reads. Keeping the tables permanently empty is what makes both decisions safe.
/// </para>
/// The tables themselves are still created (see <c>SiteStorageSchema</c>); only their
/// historical contents are left behind.
/// </remarks>
internal static readonly LegacyTable[] SiteStorageTables =
tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbLegacyMigratorTests.cs
+4 -3
View File
@@ -465,9 +465,10 @@ public class SiteLocalDbLegacyMigratorTests : IDisposable
{
// smtp_configurations.password is PLAINTEXT. Both tables are purged on every deploy
// and permanently empty by design since the site write paths were removed
// (2026-07-10), but a pre-fix legacy file can still hold rows — and Task 14 makes
// these tables REPLICATED. Migrating them would push plaintext SMTP passwords across
// a replication channel whose only historical payload was exactly that.
// (2026-07-10), but a pre-fix legacy file can still hold rows. The cutover also
// declines to REGISTER these two for replication, for the same reason — so migrating
// them would leave plaintext SMTP passwords in the consolidated database, one future
// RegisterReplicated away from being shipped to a peer, for config nothing reads.
var sitePath = Path_("scadabridge.db");
SeedLegacySiteStorage(sitePath);
var config = Config(siteDbPath: sitePath);
tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairConvergenceTests.cs
+7 -203
View File
@@ -1,202 +1,23 @@
using System.Net;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Hosting.Server;
using Microsoft.AspNetCore.Hosting.Server.Features;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.ScadaBridge.Host;
namespace ZB.MOM.WW.ScadaBridge.IntegrationTests;
/// <summary>
/// Serializes the site-pair convergence tests against each other: each one stands up a real
/// Kestrel listener plus two SQLite files, and running them concurrently under CI
/// contention is a flakiness risk.
/// </summary>
[CollectionDefinition("LocalDbSitePairConvergence")]
public sealed class LocalDbSitePairConvergenceCollection;
/// <summary>
/// Two ScadaBridge site nodes replicating the consolidated site database over a REAL
/// loopback gRPC transport, through the REAL fail-closed auth interceptor.
/// Phase 1 convergence: operation tracking and site events across a real site pair.
/// </summary>
/// <remarks>
/// <para>
/// This is the test that answers the question Phase 1 exists to answer: does a site node
/// pair actually stop losing operation-tracking and site-event state? Everything upstream
/// of it — schema helpers, DI wiring, the interceptor — can be individually green while the
/// pair still fails to converge.
/// </para>
/// <para>
/// It uses <see cref="SiteLocalDbSetup.OnReady"/>, not a hand-written schema, so the tables,
/// their primary keys, and the registration ORDER under test are the ones the host actually
/// runs. A separate schema here would prove only that the test agrees with itself.
/// </para>
/// <para>
/// Offline: no docker, no external services. Loopback Kestrel with h2c.
/// The fixture lives in <see cref="LocalDbSitePairHarness"/>, shared with the Phase 2
/// convergence suites.
/// </para>
/// </remarks>
[Collection("LocalDbSitePairConvergence")]
public sealed class LocalDbSitePairConvergenceTests : IAsyncLifetime
public sealed class LocalDbSitePairConvergenceTests : LocalDbSitePairHarness
{
private const string SharedApiKey = "site-pair-convergence-key";
private static readonly TimeSpan ConvergeTimeout = TimeSpan.FromSeconds(30);
private readonly string _pathA = Path.Combine(Path.GetTempPath(), $"sitepairA-{Guid.NewGuid():N}.db");
private readonly string _pathB = Path.Combine(Path.GetTempPath(), $"sitepairB-{Guid.NewGuid():N}.db");
// The databases are owned by the fixture, in their own providers, and registered into the
// hosts as pre-constructed instances. MS.DI does not dispose instances it did not create,
// so tearing a host down (the offline-peer scenario) leaves the databases intact and
// writable — which is exactly what lets node A accumulate writes while B is down.
private ServiceProvider _dbProviderA = null!;
private ServiceProvider _dbProviderB = null!;
private IHost? _serverHost; // node B — passive
private IHost? _initiatorHost; // node A — dials the peer
static LocalDbSitePairConvergenceTests() =>
// Grpc.Net.Client dials the loopback server over HTTP/2 cleartext.
AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);
private ILocalDb A => _dbProviderA.GetRequiredService<ILocalDb>();
private ILocalDb B => _dbProviderB.GetRequiredService<ILocalDb>();
public async Task InitializeAsync()
{
_dbProviderA = BuildDatabaseProvider(_pathA, "node-a");
_dbProviderB = BuildDatabaseProvider(_pathB, "node-b");
// Force construction (and therefore OnReady) before anything replicates.
_ = A;
_ = B;
await StartPassiveAsync();
await StartInitiatorAsync();
}
public async Task DisposeAsync()
{
await StopHostAsync(_initiatorHost);
await StopHostAsync(_serverHost);
await _dbProviderA.DisposeAsync();
await _dbProviderB.DisposeAsync();
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
foreach (var path in new[] { _pathA, _pathB })
{
foreach (var suffix in new[] { "", "-wal", "-shm" })
{
try { File.Delete(path + suffix); } catch { /* best effort */ }
}
}
}
// ---- fixture internals ------------------------------------------------------------
/// <summary>
/// A provider owning one consolidated site database, initialized through the host's own
/// <see cref="SiteLocalDbSetup.OnReady"/> — same schema, same registration order.
/// </summary>
private static ServiceProvider BuildDatabaseProvider(string path, string nodeName)
{
var config = new ConfigurationBuilder()
.AddInMemoryCollection(new Dictionary<string, string?>
{
["LocalDb:Path"] = path,
["ScadaBridge:Node:NodeName"] = nodeName,
// Point the legacy migrator at paths that do not exist, so it no-ops rather
// than picking up stray files from the test working directory.
["ScadaBridge:OperationTracking:ConnectionString"] =
$"Data Source={Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db")}",
["ScadaBridge:SiteEventLog:DatabasePath"] =
Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db"),
})
.Build();
return new ServiceCollection()
.AddZbLocalDb(config, db => SiteLocalDbSetup.OnReady(db, config))
.BuildServiceProvider();
}
private static IConfiguration ReplicationConfig(string? peerAddress)
{
var values = new Dictionary<string, string?>
{
// Tight flush + bounded reconnect backoff so convergence is observable well
// inside the poll deadline. The 60 s production default would let the doubling
// backoff overrun it after a peer outage.
["LocalDb:Replication:FlushInterval"] = "00:00:00.050",
["LocalDb:Replication:ReconnectBackoffMax"] = "00:00:02",
// Both nodes share one key — the interceptor is fail-closed, so a mismatch here
// turns every scenario below red (verified by deliberately breaking it).
["LocalDb:Replication:ApiKey"] = SharedApiKey,
};
if (peerAddress is not null)
values["LocalDb:Replication:PeerAddress"] = peerAddress;
return new ConfigurationBuilder().AddInMemoryCollection(values).Build();
}
private async Task StartPassiveAsync()
{
var config = ReplicationConfig(peerAddress: null);
_serverHost = await new HostBuilder()
.ConfigureWebHost(web =>
{
web.UseKestrel(o =>
o.Listen(IPAddress.Loopback, 0, listen => listen.Protocols = HttpProtocols.Http2));
web.ConfigureServices(services =>
{
services.AddLogging();
services.AddRouting();
// The REAL interceptor, not a stand-in. If it rejected legitimate peer
// traffic, every scenario below would fail — which is the point.
services.AddGrpc(o => o.Interceptors.Add<LocalDbSyncAuthInterceptor>());
services.AddSingleton(B);
services.AddZbLocalDbReplication(config);
});
web.Configure(app =>
{
app.UseRouting();
app.UseEndpoints(e => e.MapZbLocalDbSync());
});
})
.StartAsync();
}
private async Task StartInitiatorAsync()
{
var config = ReplicationConfig(PassiveAddress());
_initiatorHost = await new HostBuilder()
.ConfigureServices(services =>
{
services.AddLogging();
services.AddSingleton(A);
services.AddZbLocalDbReplication(config);
})
.StartAsync();
}
private string PassiveAddress()
=> _serverHost!.Services.GetRequiredService<IServer>()
.Features.Get<IServerAddressesFeature>()!.Addresses.Single();
private static async Task StopHostAsync(IHost? host)
{
if (host is null) return;
try { await host.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ }
host.Dispose();
}
// ---- data helpers -----------------------------------------------------------------
private static Task WriteTrackingAsync(ILocalDb db, string id, string status, string target)
@@ -232,19 +53,6 @@ public sealed class LocalDbSitePairConvergenceTests : IAsyncLifetime
private static Task<IReadOnlyList<string>> ReadEventIdsAsync(ILocalDb db)
=> db.QueryAsync("SELECT id FROM site_events ORDER BY id", static r => r.GetString(0));
/// <summary>Polls <paramref name="condition"/> until true or the deadline passes.</summary>
private static async Task WaitUntilAsync(Func<Task<bool>> condition, string because)
{
var deadline = DateTime.UtcNow + ConvergeTimeout;
while (DateTime.UtcNow < deadline)
{
if (await condition()) return;
await Task.Delay(50);
}
Assert.Fail($"Timed out after {ConvergeTimeout.TotalSeconds:0}s waiting for: {because}");
}
// ---- scenarios --------------------------------------------------------------------
[Fact]
@@ -322,19 +130,15 @@ public sealed class LocalDbSitePairConvergenceTests : IAsyncLifetime
{
// The failover case that motivates Phase 1: one node is down while the other keeps
// working, and nothing written during the outage may be lost.
await StopHostAsync(_serverHost);
_serverHost = null;
await StopPassiveAsync();
var idsDuringOutage = Enumerable.Range(0, 5).Select(_ => Guid.NewGuid().ToString("N")).ToList();
foreach (var id in idsDuringOutage) await WriteEventAsync(A, id, "written while B was down");
await WriteTrackingAsync(A, "op-during-outage", "Delivered", "ERP.GetOrder");
// Node B's database survived the host teardown (pre-constructed instance), so this is
// a genuine rejoin rather than a fresh node. It comes back on a NEW loopback port;
// the initiator's channel factory re-reads the peer address on each reconnect.
await StartPassiveAsync();
await StopHostAsync(_initiatorHost);
await StartInitiatorAsync();
// a genuine rejoin rather than a fresh node.
await RestartPairAsync();
await WaitUntilAsync(
async () =>
tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairHarness.cs
+286
View File
@@ -0,0 +1,286 @@
using System.Net;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Hosting.Server;
using Microsoft.AspNetCore.Hosting.Server.Features;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.ScadaBridge.Host;
namespace ZB.MOM.WW.ScadaBridge.IntegrationTests;
/// <summary>
/// Serializes the site-pair convergence tests against each other: each one stands up a real
/// Kestrel listener plus two SQLite files, and running them concurrently under CI
/// contention is a flakiness risk.
/// </summary>
[CollectionDefinition("LocalDbSitePairConvergence")]
public sealed class LocalDbSitePairConvergenceCollection;
/// <summary>
/// Two ScadaBridge site nodes replicating the consolidated site database over a REAL
/// loopback gRPC transport, through the REAL fail-closed auth interceptor.
/// </summary>
/// <remarks>
/// <para>
/// Extracted from the Phase 1 convergence tests once Phase 2 needed the same pair for the
/// store-and-forward buffer and the configuration tables. Everything here is fixture; the
/// derived classes hold only their own data helpers and scenarios.
/// </para>
/// <para>
/// It uses <see cref="SiteLocalDbSetup.OnReady"/>, not a hand-written schema, so the tables,
/// their primary keys, and the registration ORDER under test are the ones the host actually
/// runs. A separate schema here would prove only that the test agrees with itself.
/// </para>
/// <para>
/// Offline: no docker, no external services. Loopback Kestrel with h2c.
/// </para>
/// </remarks>
public abstract class LocalDbSitePairHarness : IAsyncLifetime
{
private const string SharedApiKey = "site-pair-convergence-key";
/// <summary>How long a scenario waits for the pair to agree before failing.</summary>
protected static readonly TimeSpan ConvergeTimeout = TimeSpan.FromSeconds(30);
private readonly string _pathA = Path.Combine(Path.GetTempPath(), $"sitepairA-{Guid.NewGuid():N}.db");
private readonly string _pathB = Path.Combine(Path.GetTempPath(), $"sitepairB-{Guid.NewGuid():N}.db");
// The databases are owned by the fixture, in their own providers, and registered into the
// hosts as pre-constructed instances. MS.DI does not dispose instances it did not create,
// so tearing a host down (the offline-peer scenario) leaves the databases intact and
// writable — which is exactly what lets node A accumulate writes while B is down.
private ServiceProvider _dbProviderA = null!;
private ServiceProvider _dbProviderB = null!;
private IHost? _serverHost; // node B — passive
private IHost? _initiatorHost; // node A — dials the peer
static LocalDbSitePairHarness() =>
// Grpc.Net.Client dials the loopback server over HTTP/2 cleartext.
AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);
/// <summary>Node A — the initiator, which dials the peer.</summary>
protected ILocalDb A => _dbProviderA.GetRequiredService<ILocalDb>();
/// <summary>Node B — the passive node, which listens.</summary>
protected ILocalDb B => _dbProviderB.GetRequiredService<ILocalDb>();
public async Task InitializeAsync()
{
_dbProviderA = BuildDatabaseProvider(_pathA, "node-a");
_dbProviderB = BuildDatabaseProvider(_pathB, "node-b");
// Force construction (and therefore OnReady) before anything replicates.
_ = A;
_ = B;
await StartPassiveAsync();
await StartInitiatorAsync();
}
public async Task DisposeAsync()
{
await StopHostAsync(_initiatorHost);
await StopHostAsync(_serverHost);
await _dbProviderA.DisposeAsync();
await _dbProviderB.DisposeAsync();
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
foreach (var path in new[] { _pathA, _pathB })
{
foreach (var suffix in new[] { "", "-wal", "-shm" })
{
try { File.Delete(path + suffix); } catch { /* best effort */ }
}
}
}
// ---- fixture internals ------------------------------------------------------------
/// <summary>
/// A provider owning one consolidated site database, initialized through the host's own
/// <see cref="SiteLocalDbSetup.OnReady"/> — same schema, same registration order.
/// </summary>
private static ServiceProvider BuildDatabaseProvider(string path, string nodeName)
{
var config = new ConfigurationBuilder()
.AddInMemoryCollection(new Dictionary<string, string?>
{
["LocalDb:Path"] = path,
["ScadaBridge:Node:NodeName"] = nodeName,
// Point the legacy migrators at paths that do not exist, so they no-op rather
// than picking up stray files from the test working directory. The two Phase 2
// defaults matter most: unlike the Phase 1 pair they resolve inside ./data/,
// and a migration would also RENAME whatever it found.
["ScadaBridge:OperationTracking:ConnectionString"] =
$"Data Source={Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db")}",
["ScadaBridge:SiteEventLog:DatabasePath"] =
Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db"),
["ScadaBridge:StoreAndForward:SqliteDbPath"] =
Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db"),
["ScadaBridge:Database:SiteDbPath"] =
Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db"),
})
.Build();
return new ServiceCollection()
.AddZbLocalDb(config, db =>
{
SiteLocalDbSetup.OnReady(db, config);
RegisterPhase2TablesUntilCutover(db);
})
.BuildServiceProvider();
}
/// <summary>
/// Registers the Phase 2 tables for capture, which production <c>OnReady</c> does not do
/// until the Task 14 cutover.
/// </summary>
/// <remarks>
/// <b>Delete this method at Task 14</b>, along with its call above. Until the cutover the
/// bespoke <c>SiteReplicationActor</c> and <c>ReplicationService</c> still own these
/// tables, so <c>OnReady</c> deliberately leaves them unregistered — but the convergence
/// specifications the cutover has to satisfy need capture triggers to mean anything. The
/// tables are empty at this point, so registering here captures nothing retroactively;
/// the ordering guarantee under test is unaffected.
/// <para>
/// After Task 14 these registrations come from <c>OnReady</c> itself, and leaving this
/// method behind would mask a cutover that forgot one — the whole point of these tests.
/// </para>
/// </remarks>
private static void RegisterPhase2TablesUntilCutover(ILocalDb db)
{
foreach (var table in Phase2ReplicatedTables)
db.RegisterReplicated(table);
}
/// <summary>
/// The eight tables Task 14 registers. Listed literally rather than derived from
/// production code, so that a cutover which registers the wrong set fails these tests
/// instead of agreeing with itself.
/// </summary>
/// <remarks>
/// <c>notification_lists</c> and <c>smtp_configurations</c> are absent by design. They
/// are permanently empty (no site writer since 2026-07-10, the migrator skips them, the
/// active-node purge keeps them empty), so registering them would open a standing
/// replication channel whose only historical payload was plaintext SMTP passwords.
/// </remarks>
protected static readonly string[] Phase2ReplicatedTables =
[
"sf_messages",
"deployed_configurations", "static_attribute_overrides", "shared_scripts",
"external_systems", "database_connections", "data_connection_definitions",
"native_alarm_state",
];
private static IConfiguration ReplicationConfig(string? peerAddress)
{
var values = new Dictionary<string, string?>
{
// Tight flush + bounded reconnect backoff so convergence is observable well
// inside the poll deadline. The 60 s production default would let the doubling
// backoff overrun it after a peer outage.
["LocalDb:Replication:FlushInterval"] = "00:00:00.050",
["LocalDb:Replication:ReconnectBackoffMax"] = "00:00:02",
// Both nodes share one key — the interceptor is fail-closed, so a mismatch here
// turns every scenario below red (verified by deliberately breaking it).
["LocalDb:Replication:ApiKey"] = SharedApiKey,
};
if (peerAddress is not null)
values["LocalDb:Replication:PeerAddress"] = peerAddress;
return new ConfigurationBuilder().AddInMemoryCollection(values).Build();
}
/// <summary>Starts node B, the passive listener.</summary>
protected async Task StartPassiveAsync()
{
var config = ReplicationConfig(peerAddress: null);
_serverHost = await new HostBuilder()
.ConfigureWebHost(web =>
{
web.UseKestrel(o =>
o.Listen(IPAddress.Loopback, 0, listen => listen.Protocols = HttpProtocols.Http2));
web.ConfigureServices(services =>
{
services.AddLogging();
services.AddRouting();
// The REAL interceptor, not a stand-in. If it rejected legitimate peer
// traffic, every scenario below would fail — which is the point.
services.AddGrpc(o => o.Interceptors.Add<LocalDbSyncAuthInterceptor>());
services.AddSingleton(B);
services.AddZbLocalDbReplication(config);
});
web.Configure(app =>
{
app.UseRouting();
app.UseEndpoints(e => e.MapZbLocalDbSync());
});
})
.StartAsync();
}
/// <summary>Starts node A, which dials the passive node.</summary>
protected async Task StartInitiatorAsync()
{
var config = ReplicationConfig(PassiveAddress());
_initiatorHost = await new HostBuilder()
.ConfigureServices(services =>
{
services.AddLogging();
services.AddSingleton(A);
services.AddZbLocalDbReplication(config);
})
.StartAsync();
}
/// <summary>Takes node B's listener down, leaving its database intact and writable.</summary>
protected async Task StopPassiveAsync()
{
await StopHostAsync(_serverHost);
_serverHost = null;
}
/// <summary>
/// Brings node B back on a NEW loopback port and re-dials from A. The initiator's channel
/// factory re-reads the peer address on each reconnect, so this is a genuine rejoin.
/// </summary>
protected async Task RestartPairAsync()
{
await StartPassiveAsync();
await StopHostAsync(_initiatorHost);
await StartInitiatorAsync();
}
private string PassiveAddress()
=> _serverHost!.Services.GetRequiredService<IServer>()
.Features.Get<IServerAddressesFeature>()!.Addresses.Single();
private static async Task StopHostAsync(IHost? host)
{
if (host is null) return;
try { await host.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ }
host.Dispose();
}
/// <summary>Polls <paramref name="condition"/> until true or the deadline passes.</summary>
protected static async Task WaitUntilAsync(Func<Task<bool>> condition, string because)
{
var deadline = DateTime.UtcNow + ConvergeTimeout;
while (DateTime.UtcNow < deadline)
{
if (await condition()) return;
await Task.Delay(50);
}
Assert.Fail($"Timed out after {ConvergeTimeout.TotalSeconds:0}s waiting for: {because}");
}
}
tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbStoreAndForwardConvergenceTests.cs
+221
View File
@@ -0,0 +1,221 @@
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
namespace ZB.MOM.WW.ScadaBridge.IntegrationTests;
/// <summary>
/// Phase 2 convergence: the store-and-forward buffer across a real site pair.
/// </summary>
/// <remarks>
/// <para>
/// These are <b>specifications, written before the deletion they justify</b>. The bespoke
/// <c>ReplicationService</c> (an explicit Add/Remove/Park/Requeue operation stream over Akka)
/// is deleted at Task 14 and replaced by LocalDb's trigger-based change capture. Each test
/// here is one behaviour the old mechanism provided, restated as an outcome the replacement
/// must still deliver — so the cutover has something to be judged against other than "it
/// compiles".
/// </para>
/// <para>
/// They are written in terms of <i>rows</i>, not operations, which is the whole point:
/// under CDC there is no Add or Park message to observe, only a row that must end up in the
/// right state on both nodes.
/// </para>
/// <para>
/// <b>One intent is deliberately not ported:</b>
/// <c>ReplicationServiceTests.ReplicationOperations_AreDispatchedInIssueOrder</c> — 200
/// interleaved operations dispatched synchronously, observed in strict issue order. That
/// asserts the <i>mechanism</i> (inline fire-and-forget dispatch), not an outcome, and CDC
/// capture is asynchronous and batched by construction, so no honest port exists. Its
/// portable content is the ordering <i>outcome</i>: an add followed by a remove must never
/// converge to present. That is
/// <see cref="MessageAddedThenRemoved_NeverConvergesToPresent"/>. This paragraph exists so a
/// future reader does not conclude the test was dropped by accident.
/// </para>
/// </remarks>
[Collection("LocalDbSitePairConvergence")]
public sealed class LocalDbStoreAndForwardConvergenceTests : LocalDbSitePairHarness
{
// ---- data helpers -----------------------------------------------------------------
/// <summary>Buffers a message, as <c>StoreAndForwardStorage.EnqueueAsync</c> does.</summary>
private static Task EnqueueAsync(ILocalDb db, string id, string target = "ERP.GetOrder")
=> db.ExecuteAsync(
"""
INSERT INTO sf_messages (
id, category, target, payload_json, retry_count, max_retries,
retry_interval_ms, created_at, status)
VALUES (@id, 0, @target, '{"order":1}', 0, 50, 30000, @now, @pending)
ON CONFLICT(id) DO UPDATE SET
status = excluded.status,
retry_count = excluded.retry_count;
""",
new
{
id,
target,
now = DateTime.UtcNow.ToString("o"),
pending = (int)StoreAndForwardMessageStatus.Pending,
});
/// <summary>Deletes a delivered message, as the successful-retry path does.</summary>
private static Task DeleteAsync(ILocalDb db, string id)
=> db.ExecuteAsync("DELETE FROM sf_messages WHERE id = @id;", new { id });
private static Task SetStatusAsync(ILocalDb db, string id, StoreAndForwardMessageStatus status, int retryCount)
=> db.ExecuteAsync(
"UPDATE sf_messages SET status = @status, retry_count = @retryCount WHERE id = @id;",
new { id, status = (int)status, retryCount });
private static async Task<(int Status, int RetryCount)?> ReadAsync(ILocalDb db, string id)
{
var rows = await db.QueryAsync(
"SELECT status, retry_count FROM sf_messages WHERE id = @id",
static r => (r.GetInt32(0), r.GetInt32(1)), new { id });
return rows.Count == 0 ? null : rows[0];
}
private static async Task<bool> ExistsAsync(ILocalDb db, string id)
=> await ReadAsync(db, id) is not null;
private static async Task<bool> HasStatusAsync(
ILocalDb db, string id, StoreAndForwardMessageStatus status)
=> await ReadAsync(db, id) is { } row && row.Status == (int)status;
// ---- scenarios --------------------------------------------------------------------
[Fact]
public async Task BufferedMessage_MaterialisesOnThePeer()
{
// Was: BufferingAMessage_ReplicatesAnAddOperation.
await EnqueueAsync(A, "msg-add");
await WaitUntilAsync(
() => ExistsAsync(B, "msg-add"),
"a message buffered on node A to appear on node B");
}
[Fact]
public async Task DeliveredMessage_DisappearsFromThePeer()
{
// Was: SuccessfulRetry_ReplicatesARemoveOperation. Row deletes propagate as
// tombstones under CDC — the peer must not keep re-delivering a message that node A
// already succeeded on.
await EnqueueAsync(A, "msg-remove");
await WaitUntilAsync(() => ExistsAsync(B, "msg-remove"), "the message to arrive on B first");
await DeleteAsync(A, "msg-remove");
await WaitUntilAsync(
async () => !await ExistsAsync(B, "msg-remove"),
"the delivered message to be gone from node B");
}
[Fact]
public async Task ParkedMessage_ShowsAsParkedOnThePeer()
{
// Was: ParkedMessage_ReplicatesAParkOperation.
await EnqueueAsync(A, "msg-park");
await WaitUntilAsync(() => ExistsAsync(B, "msg-park"), "the message to arrive on B first");
await SetStatusAsync(A, "msg-park", StoreAndForwardMessageStatus.Parked, retryCount: 50);
await WaitUntilAsync(
() => HasStatusAsync(B, "msg-park", StoreAndForwardMessageStatus.Parked),
"the parked status to reach node B");
}
[Fact]
public async Task RequeuedMessage_ResetsStatusAndRetryCountOnThePeer()
{
// Was: RetryingAParkedMessage_ReplicatesARequeueOperation +
// ApplyReplicatedOperation_Requeue_MovesStandbyRowBackToPending. RetryCount matters
// as much as status: a peer that took Pending but kept retry_count at max would park
// the message again on its first attempt after a failover.
await EnqueueAsync(A, "msg-requeue");
await SetStatusAsync(A, "msg-requeue", StoreAndForwardMessageStatus.Parked, retryCount: 50);
await WaitUntilAsync(
() => HasStatusAsync(B, "msg-requeue", StoreAndForwardMessageStatus.Parked),
"the message to be parked on B before it is requeued");
await SetStatusAsync(A, "msg-requeue", StoreAndForwardMessageStatus.Pending, retryCount: 0);
await WaitUntilAsync(
async () => await ReadAsync(B, "msg-requeue")
is { Status: (int)StoreAndForwardMessageStatus.Pending, RetryCount: 0 },
"node B to show the requeued message as Pending with retry_count 0");
}
[Fact]
public async Task MessageAddedThenRemoved_NeverConvergesToPresent()
{
// The portable intent of ReplicationOperations_AreDispatchedInIssueOrder (see the
// class remarks). Ordering only ever mattered because a remove overtaken by its own
// add would resurrect a delivered message and send it twice. Under LWW that
// reordering is not preventable by dispatch discipline — it is prevented because the
// tombstone carries the later HLC and therefore wins regardless of arrival order.
//
// Asserting the endpoint rather than the sequence is what makes this portable: the
// old test would fail on any async transport even when the outcome was correct.
await EnqueueAsync(A, "msg-ordering");
await DeleteAsync(A, "msg-ordering");
// A control row written in the same window. Without it this test is VACUOUS: an
// absent row is also what a pair that replicates nothing at all looks like, so it
// would pass with capture switched off entirely (observed — it was the only one of
// these seven that survived unregistering sf_messages). The control converging is
// what makes the absence of msg-ordering evidence rather than silence.
await EnqueueAsync(A, "msg-ordering-control");
await WaitUntilAsync(
() => ExistsAsync(B, "msg-ordering-control"),
"the control message to reach node B, proving the pair was replicating");
await WaitUntilAsync(
async () => !await ExistsAsync(B, "msg-ordering") && !await ExistsAsync(A, "msg-ordering"),
"the add-then-remove pair to settle as absent on both nodes");
// Absence has to persist, not just occur once: a late-arriving add would flip the row
// back and the poll above could have observed a gap it never actually converged to.
await Task.Delay(TimeSpan.FromSeconds(2));
Assert.False(await ExistsAsync(B, "msg-ordering"));
Assert.False(await ExistsAsync(A, "msg-ordering"));
Assert.True(await ExistsAsync(B, "msg-ordering-control"));
}
[Fact]
public async Task SameMessage_BufferedTwice_ConvergesToTheNewerState()
{
// Was: ApplyReplicatedAdd_Twice_IsIdempotent_NewestWins. Under CDC this is not a
// property the application has to implement — LWW on the primary key gives it — but
// it is still a property the pair must HAVE, so it is asserted rather than assumed.
await EnqueueAsync(A, "msg-twice");
await WaitUntilAsync(() => ExistsAsync(B, "msg-twice"), "the first write to reach B");
await SetStatusAsync(A, "msg-twice", StoreAndForwardMessageStatus.InFlight, retryCount: 3);
await WaitUntilAsync(
async () => await ReadAsync(B, "msg-twice")
is { Status: (int)StoreAndForwardMessageStatus.InFlight, RetryCount: 3 },
"the newer state to win on node B");
}
[Fact]
public async Task ParkArrivingWithoutItsAdd_StillMaterialisesTheRow()
{
// Was: ApplyReplicatedPark_WhenAddWasLost_MaterializesTheParkedRow. The bespoke
// replicator needed explicit upsert semantics so a lost Add did not leave a Park with
// nothing to update. The CDC equivalent is a peer that was offline for the add and
// only ever sees the row in its parked state — it must still end up with the row,
// not skip it as an update to something it never had.
await StopPassiveAsync();
await EnqueueAsync(A, "msg-park-no-add");
await SetStatusAsync(A, "msg-park-no-add", StoreAndForwardMessageStatus.Parked, retryCount: 50);
await RestartPairAsync();
await WaitUntilAsync(
() => HasStatusAsync(B, "msg-park-no-add", StoreAndForwardMessageStatus.Parked),
"node B to materialise a row it only ever saw in its parked state");
}
}