LocalDb adoption Phase 1 + 2: consolidate the site database, delete the bespoke replicators #23
@@ -0,0 +1,348 @@
|
||||
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>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[Collection("LocalDbSitePairConvergence")]
|
||||
public sealed class LocalDbSitePairConvergenceTests : IAsyncLifetime
|
||||
{
|
||||
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)
|
||||
=> db.ExecuteAsync(
|
||||
"""
|
||||
INSERT INTO OperationTracking (
|
||||
TrackedOperationId, Kind, TargetSummary, Status, RetryCount,
|
||||
CreatedAtUtc, UpdatedAtUtc)
|
||||
VALUES (@id, 'ApiCallCached', @target, @status, 0, @now, @now)
|
||||
ON CONFLICT(TrackedOperationId) DO UPDATE SET
|
||||
Status = excluded.Status,
|
||||
TargetSummary = excluded.TargetSummary,
|
||||
UpdatedAtUtc = excluded.UpdatedAtUtc;
|
||||
""",
|
||||
new { id, status, target, now = DateTime.UtcNow.ToString("o") });
|
||||
|
||||
private static Task WriteEventAsync(ILocalDb db, string id, string message)
|
||||
=> db.ExecuteAsync(
|
||||
"""
|
||||
INSERT INTO site_events (id, timestamp, event_type, severity, source, message)
|
||||
VALUES (@id, @ts, 'script', 'Info', 'convergence-test', @message);
|
||||
""",
|
||||
new { id, ts = DateTimeOffset.UtcNow.ToString("o"), message });
|
||||
|
||||
private static async Task<string?> ReadTrackingStatusAsync(ILocalDb db, string id)
|
||||
{
|
||||
var rows = await db.QueryAsync(
|
||||
"SELECT Status FROM OperationTracking WHERE TrackedOperationId = @id",
|
||||
static r => r.GetString(0), new { id });
|
||||
return rows.Count == 0 ? null : rows[0];
|
||||
}
|
||||
|
||||
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]
|
||||
public async Task TrackingRow_WrittenOnA_BecomesReadableOnB()
|
||||
{
|
||||
await WriteTrackingAsync(A, "op-a-1", "Submitted", "ERP.GetOrder");
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () => await ReadTrackingStatusAsync(B, "op-a-1") == "Submitted",
|
||||
"the tracking row written on node A to appear on node B");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task TrackingRow_WrittenOnB_BecomesReadableOnA()
|
||||
{
|
||||
// Replication is bidirectional even though only A dials: proving the passive node's
|
||||
// writes flow back is what makes a failover in EITHER direction safe.
|
||||
await WriteTrackingAsync(B, "op-b-1", "Submitted", "ERP.GetOrder");
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () => await ReadTrackingStatusAsync(A, "op-b-1") == "Submitted",
|
||||
"the tracking row written on node B to appear on node A");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SameOperation_UpdatedOnBothNodes_ConvergesToOneWinner()
|
||||
{
|
||||
// Last-writer-wins on the primary key. The specific winner is not asserted — that is
|
||||
// the HLC's business — but the two nodes MUST agree, and must agree on a value one of
|
||||
// them actually wrote rather than a merge of both.
|
||||
await WriteTrackingAsync(A, "op-conflict", "Submitted", "ERP.GetOrder");
|
||||
await WaitUntilAsync(
|
||||
async () => await ReadTrackingStatusAsync(B, "op-conflict") is not null,
|
||||
"the conflicting row to exist on both nodes before it is updated");
|
||||
|
||||
await WriteTrackingAsync(A, "op-conflict", "Delivered", "ERP.GetOrder");
|
||||
await WriteTrackingAsync(B, "op-conflict", "Parked", "ERP.GetOrder");
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () =>
|
||||
{
|
||||
var a = await ReadTrackingStatusAsync(A, "op-conflict");
|
||||
var b = await ReadTrackingStatusAsync(B, "op-conflict");
|
||||
return a is not null && a == b;
|
||||
},
|
||||
"both nodes to converge on one status for the contended operation");
|
||||
|
||||
var winner = await ReadTrackingStatusAsync(A, "op-conflict");
|
||||
Assert.Contains(winner, new[] { "Delivered", "Parked" });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EventsLoggedOnBothNodes_ConvergeToTheUnion_WithNoIdCollisions()
|
||||
{
|
||||
// The reason site_events moved to GUID ids. Under the old autoincrement scheme both
|
||||
// nodes would independently mint id=1, id=2, ... and last-writer-wins on the primary
|
||||
// key would silently DESTROY one node's events instead of merging them. The union
|
||||
// count is the assertion that catches that.
|
||||
var idsFromA = Enumerable.Range(0, 5).Select(_ => Guid.NewGuid().ToString("N")).ToList();
|
||||
var idsFromB = Enumerable.Range(0, 5).Select(_ => Guid.NewGuid().ToString("N")).ToList();
|
||||
|
||||
foreach (var id in idsFromA) await WriteEventAsync(A, id, "from A");
|
||||
foreach (var id in idsFromB) await WriteEventAsync(B, id, "from B");
|
||||
|
||||
var expected = idsFromA.Concat(idsFromB).OrderBy(x => x, StringComparer.Ordinal).ToList();
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () => (await ReadEventIdsAsync(A)).SequenceEqual(expected)
|
||||
&& (await ReadEventIdsAsync(B)).SequenceEqual(expected),
|
||||
"both nodes to hold the union of all 10 events");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PeerOffline_ThenRejoins_CatchesUpOnEverythingItMissed()
|
||||
{
|
||||
// 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;
|
||||
|
||||
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();
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () =>
|
||||
{
|
||||
var events = await ReadEventIdsAsync(B);
|
||||
return idsDuringOutage.All(events.Contains)
|
||||
&& await ReadTrackingStatusAsync(B, "op-during-outage") == "Delivered";
|
||||
},
|
||||
"node B to catch up on everything written while it was offline");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user