Joseph Doherty
365 lines
17 KiB
C#
365 lines
17 KiB
C#
using System.Net;
|
|
using Grpc.Net.Client;
|
|
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 Microsoft.Extensions.Options;
|
|
using ZB.MOM.WW.LocalDb.Internal;
|
|
using ZB.MOM.WW.LocalDb.Replication;
|
|
using ZB.MOM.WW.LocalDb.Replication.Internal;
|
|
|
|
namespace ZB.MOM.WW.LocalDb.Tests.Convergence;
|
|
|
|
/// <summary>
|
|
/// Serializes the convergence test classes against EACH OTHER (no shared class fixture — each test
|
|
/// still builds its own <see cref="ConvergenceFixture"/>): both classes stand up real Kestrel
|
|
/// listeners plus four SQLite files per test, and running the two heavy classes concurrently under
|
|
/// CI contention is a flakiness risk. The rest of the assembly still parallelizes normally.
|
|
/// </summary>
|
|
[CollectionDefinition("Convergence")]
|
|
public sealed class ConvergenceCollection;
|
|
|
|
/// <summary>
|
|
/// Two FULL replication stacks over a REAL loopback gRPC transport (Kestrel h2c on 127.0.0.1),
|
|
/// proving end-to-end bidirectional convergence. Node A is the initiator (client:
|
|
/// <see cref="SyncBackgroundService"/> dials the peer through an injected channel factory); node B
|
|
/// is passive (server: hosts <c>MapZbLocalDbSync</c>). Both build the identical
|
|
/// <see cref="SyncSession"/> stack (OplogStore + LwwApplier + snapshot hooks) via
|
|
/// <see cref="SyncSessionFactory"/>.
|
|
///
|
|
/// A real loopback socket (not an in-memory TestServer) is deliberate: KillTransport disposes the
|
|
/// server and CLOSES the socket, which faults the initiator's active stream PROMPTLY — an in-memory
|
|
/// TestServer does not model a connection drop and leaves the client hanging for tens of seconds.
|
|
///
|
|
/// The two <see cref="SqliteLocalDb"/> instances are OWNED BY THE FIXTURE and registered into the
|
|
/// DI containers as pre-constructed singletons — MS.DI does not dispose externally-`new`ed instances,
|
|
/// so a container/host tear-down (KillTransport) leaves the databases intact and writable. That is
|
|
/// what lets a test accumulate writes on both nodes while the transport is down.
|
|
/// </summary>
|
|
public sealed class ConvergenceFixture : IAsyncDisposable
|
|
{
|
|
private static readonly TimeSpan DefaultConvergeTimeout = TimeSpan.FromSeconds(30);
|
|
|
|
private readonly IConfiguration _initiatorConfig;
|
|
private readonly IConfiguration _passiveConfig;
|
|
private readonly string _pathA;
|
|
private readonly string _pathB;
|
|
|
|
private readonly SqliteLocalDb _dbA;
|
|
private readonly SqliteLocalDb _dbB;
|
|
|
|
private ServiceProvider? _clientProvider;
|
|
private SyncBackgroundService? _bg;
|
|
private IHost? _serverHost;
|
|
|
|
// Read fresh by the client's channel factory on every (re)connect: null => transport is down and
|
|
// the factory throws, driving the initiator's reconnect backoff until RestartTransport refreshes it.
|
|
private volatile string? _serverAddress;
|
|
|
|
static ConvergenceFixture() =>
|
|
// Grpc.Net.Client dials the loopback server over HTTP/2 cleartext (h2c).
|
|
AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);
|
|
|
|
private ReplicationOptions InitiatorOptions =>
|
|
_clientProvider!.GetRequiredService<IOptions<ReplicationOptions>>().Value;
|
|
|
|
/// <param name="tuneInitiator">Mutates node A's <c>LocalDb:Replication:*</c> config (e.g. MaxBatchSize, MaxOplogRows).</param>
|
|
/// <param name="tunePassive">Mutates node B's config.</param>
|
|
public ConvergenceFixture(
|
|
Action<Dictionary<string, string?>>? tuneInitiator = null,
|
|
Action<Dictionary<string, string?>>? tunePassive = null)
|
|
{
|
|
_pathA = Path.Combine(Path.GetTempPath(), "convA-" + Guid.NewGuid() + ".db");
|
|
_pathB = Path.Combine(Path.GetTempPath(), "convB-" + Guid.NewGuid() + ".db");
|
|
_initiatorConfig = BuildConfig(initiator: true, tuneInitiator);
|
|
_passiveConfig = BuildConfig(initiator: false, tunePassive);
|
|
|
|
_dbA = NewDb(_pathA);
|
|
_dbB = NewDb(_pathB);
|
|
}
|
|
|
|
/// <summary>The initiator (client) node's database.</summary>
|
|
public ILocalDb A => _dbA;
|
|
|
|
/// <summary>The passive (server) node's database.</summary>
|
|
public ILocalDb B => _dbB;
|
|
|
|
/// <summary>Initiator connection attempts so far (1 = initial dial; each reconnect increments it).</summary>
|
|
public int InitiatorConnectionAttempts => _bg?.ConnectionAttempts ?? 0;
|
|
|
|
// ---- lifecycle controls -----------------------------------------------------------------
|
|
|
|
/// <summary>Brings up the passive server, wires the initiator's channel factory to it, and starts the initiator pump.</summary>
|
|
public async Task StartAsync()
|
|
{
|
|
(_serverHost, _serverAddress) = await BuildServerHostAsync();
|
|
|
|
var services = new ServiceCollection();
|
|
services.AddLogging();
|
|
// Pre-constructed instance: the container will NOT dispose _dbA (verified MS.DI behavior).
|
|
services.AddSingleton<ILocalDb>(_dbA);
|
|
services.AddZbLocalDbReplication(_initiatorConfig);
|
|
_clientProvider = services.BuildServiceProvider();
|
|
|
|
_bg = _clientProvider.GetServices<IHostedService>().OfType<SyncBackgroundService>().Single();
|
|
_bg.ChannelFactory = () =>
|
|
{
|
|
var address = _serverAddress
|
|
?? throw new InvalidOperationException("transport down (server disposed)");
|
|
return GrpcChannel.ForAddress(address);
|
|
};
|
|
await _bg.StartAsync(CancellationToken.None);
|
|
}
|
|
|
|
/// <summary>Graceful stop of both the initiator pump and the passive server.</summary>
|
|
public async Task StopAsync()
|
|
{
|
|
if (_bg is not null)
|
|
try { await _bg.StopAsync(CancellationToken.None); } catch { /* teardown */ }
|
|
await DisposeServerHostAsync();
|
|
}
|
|
|
|
/// <summary>Closes the server socket so the initiator's in-flight stream faults (transport loss). The databases survive.</summary>
|
|
public async Task KillTransportAsync()
|
|
{
|
|
_serverAddress = null;
|
|
await DisposeServerHostAsync();
|
|
}
|
|
|
|
/// <summary>Stands up a fresh server on a new loopback port over the SAME databases; the initiator reconnects on its next backoff tick.</summary>
|
|
public async Task RestartTransportAsync() =>
|
|
(_serverHost, _serverAddress) = await BuildServerHostAsync();
|
|
|
|
// ---- convergence assertion --------------------------------------------------------------
|
|
|
|
/// <summary>
|
|
/// Polls until both oplogs are fully acked/pruned (empty) AND both nodes' <c>orders</c> +
|
|
/// <c>__localdb_row_version</c> dumps (ordered by pk) match; on timeout, dumps both states.
|
|
///
|
|
/// Row-version comparison excludes <c>tombstone_utc</c>: that column is a LOCAL retention clock
|
|
/// (the delete trigger stamps the originating wall-clock; <see cref="LwwApplier"/> stamps the
|
|
/// receiving node's apply-time), so it diverges by design and is not part of the converged
|
|
/// logical state. The convergence-defining columns (hlc, node_id, is_tombstone) are compared.
|
|
/// </summary>
|
|
public Task AssertConvergedAsync(TimeSpan? timeout = null) =>
|
|
PollUntilConvergedAsync(requireEmptyOplog: true, timeout);
|
|
|
|
/// <summary>
|
|
/// Polls until only the DATA has converged (orders + row_version identical), tolerating a
|
|
/// non-empty oplog. Use this to observe convergence immediately after a snapshot resync: the
|
|
/// sender's snapshot-covered tail oplog rows are delivered by the snapshot, not delta-acked, so
|
|
/// they are not pruned until a SUBSEQUENT delta carries the peer ack past them.
|
|
/// </summary>
|
|
public Task AssertDataConvergedAsync(TimeSpan? timeout = null) =>
|
|
PollUntilConvergedAsync(requireEmptyOplog: false, timeout);
|
|
|
|
private async Task PollUntilConvergedAsync(bool requireEmptyOplog, TimeSpan? timeout)
|
|
{
|
|
var deadline = DateTime.UtcNow + (timeout ?? DefaultConvergeTimeout);
|
|
while (DateTime.UtcNow < deadline)
|
|
{
|
|
if (await IsConvergedAsync(requireEmptyOplog))
|
|
return;
|
|
await Task.Delay(25);
|
|
}
|
|
|
|
if (await IsConvergedAsync(requireEmptyOplog))
|
|
return;
|
|
|
|
Assert.Fail(await DumpStateAsync());
|
|
}
|
|
|
|
private async Task<bool> IsConvergedAsync(bool requireEmptyOplog)
|
|
{
|
|
if (requireEmptyOplog && (await OplogCountAsync(_dbA) != 0 || await OplogCountAsync(_dbB) != 0))
|
|
return false;
|
|
if (await DumpOrdersAsync(_dbA) != await DumpOrdersAsync(_dbB))
|
|
return false;
|
|
return await DumpRowVersionAsync(_dbA) == await DumpRowVersionAsync(_dbB);
|
|
}
|
|
|
|
private async Task<string> DumpStateAsync()
|
|
{
|
|
var sb = new System.Text.StringBuilder();
|
|
sb.AppendLine("Convergence NOT reached within timeout.");
|
|
sb.AppendLine($"--- A oplog rows: {await OplogCountAsync(_dbA)} B oplog rows: {await OplogCountAsync(_dbB)}");
|
|
sb.AppendLine("--- A.orders ---").AppendLine(await DumpOrdersAsync(_dbA));
|
|
sb.AppendLine("--- B.orders ---").AppendLine(await DumpOrdersAsync(_dbB));
|
|
sb.AppendLine("--- A.row_version ---").AppendLine(await DumpRowVersionAsync(_dbA));
|
|
sb.AppendLine("--- B.row_version ---").AppendLine(await DumpRowVersionAsync(_dbB));
|
|
sb.AppendLine("--- A.dead_letter ---").AppendLine(await DumpDeadLettersAsync(_dbA));
|
|
sb.AppendLine("--- B.dead_letter ---").AppendLine(await DumpDeadLettersAsync(_dbB));
|
|
return sb.ToString();
|
|
}
|
|
|
|
// Dead-lettered entries are a prime suspect when convergence stalls (a poison row silently
|
|
// diverts instead of applying), so the timeout dump surfaces the count + a small sample.
|
|
private static async Task<string> DumpDeadLettersAsync(ILocalDb db)
|
|
{
|
|
var count = (await db.QueryAsync(
|
|
"SELECT COUNT(*) FROM __localdb_dead_letter", static r => r.GetInt64(0)))[0];
|
|
if (count == 0)
|
|
return "count: 0";
|
|
var sample = await db.QueryAsync(
|
|
"SELECT table_name, pk_json, hlc, node_id, error FROM __localdb_dead_letter ORDER BY id LIMIT 5",
|
|
static r => $"{r.GetString(0)}|{r.GetString(1)}|{r.GetInt64(2)}|{r.GetString(3)}|{r.GetString(4)}");
|
|
return $"count: {count} (first {sample.Count})\n" + string.Join("\n", sample);
|
|
}
|
|
|
|
// ---- test helpers -----------------------------------------------------------------------
|
|
|
|
/// <summary>
|
|
/// Inserts or updates one row through the normal consumer API (fires the ai/au capture triggers).
|
|
/// Uses a plain UPDATE-else-INSERT inside one transaction; the transaction makes the
|
|
/// update-or-insert atomic against a concurrent replicated apply of the same pk (no
|
|
/// INSERT/UNIQUE race on orders). <c>INSERT ... ON CONFLICT DO UPDATE</c> is also supported by
|
|
/// the capture triggers — covered by <c>Upsert_OnConflictDoUpdate_CapturesCorrectly</c>.
|
|
/// </summary>
|
|
public static async Task UpsertAsync(ILocalDb db, long id, string sku, long qty)
|
|
{
|
|
await using var tx = await db.BeginTransactionAsync();
|
|
var updated = await tx.ExecuteAsync(
|
|
"UPDATE orders SET sku = @sku, qty = @qty WHERE id = @id", new { id, sku, qty });
|
|
if (updated == 0)
|
|
await tx.ExecuteAsync(
|
|
"INSERT INTO orders (id, sku, qty) VALUES (@id, @sku, @qty)", new { id, sku, qty });
|
|
await tx.CommitAsync();
|
|
}
|
|
|
|
/// <summary>Deletes one row through the normal consumer API (fires the delete capture trigger).</summary>
|
|
public static Task DeleteAsync(ILocalDb db, long id) =>
|
|
db.ExecuteAsync("DELETE FROM orders WHERE id = @id", new { id });
|
|
|
|
public static async Task<int> CountOrdersAsync(ILocalDb db)
|
|
{
|
|
var rows = await db.QueryAsync("SELECT COUNT(*) FROM orders", static r => r.GetInt64(0));
|
|
return (int)rows[0];
|
|
}
|
|
|
|
/// <summary>
|
|
/// Enforces the oplog backlog caps on node A (prunes + flags needs_snapshot when over the cap),
|
|
/// mirroring <c>SnapshotResyncTests</c>. The running engine has no automatic caps enforcer, so a
|
|
/// test that needs a pruned horizon triggers it here on a store over the same database.
|
|
/// </summary>
|
|
public Task<bool> EnforceCapsAAsync() =>
|
|
new OplogStore(_dbA, InitiatorOptions).EnforceCapsAsync();
|
|
|
|
// ---- internals --------------------------------------------------------------------------
|
|
|
|
private static SqliteLocalDb NewDb(string path)
|
|
{
|
|
var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
|
|
using (var conn = db.CreateConnection())
|
|
using (var cmd = conn.CreateCommand())
|
|
{
|
|
cmd.CommandText = "CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)";
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
db.RegisterReplicated("orders");
|
|
return db;
|
|
}
|
|
|
|
private async Task<(IHost Host, string Address)> BuildServerHostAsync()
|
|
{
|
|
var host = await new HostBuilder()
|
|
.ConfigureWebHost(web =>
|
|
{
|
|
web.UseKestrel(o =>
|
|
// Ephemeral loopback port, h2c (HTTP/2 cleartext) so gRPC works without TLS.
|
|
o.Listen(IPAddress.Loopback, 0, listen => listen.Protocols = HttpProtocols.Http2));
|
|
web.ConfigureServices(services =>
|
|
{
|
|
services.AddRouting();
|
|
services.AddGrpc();
|
|
services.AddLogging();
|
|
// Pre-constructed instance: host tear-down (KillTransport) will NOT dispose _dbB.
|
|
services.AddSingleton<ILocalDb>(_dbB);
|
|
services.AddZbLocalDbReplication(_passiveConfig);
|
|
});
|
|
web.Configure(app =>
|
|
{
|
|
app.UseRouting();
|
|
app.UseEndpoints(e => e.MapZbLocalDbSync());
|
|
});
|
|
})
|
|
.StartAsync();
|
|
|
|
var address = host.Services.GetRequiredService<IServer>()
|
|
.Features.Get<IServerAddressesFeature>()!.Addresses.Single();
|
|
return (host, address);
|
|
}
|
|
|
|
private async Task DisposeServerHostAsync()
|
|
{
|
|
var host = _serverHost;
|
|
_serverHost = null;
|
|
if (host is null)
|
|
return;
|
|
try { await host.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ }
|
|
host.Dispose();
|
|
}
|
|
|
|
private static async Task<long> OplogCountAsync(ILocalDb db)
|
|
{
|
|
var rows = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", static r => r.GetInt64(0));
|
|
return rows[0];
|
|
}
|
|
|
|
private static async Task<string> DumpOrdersAsync(ILocalDb db)
|
|
{
|
|
var rows = await db.QueryAsync(
|
|
"SELECT id, sku, qty FROM orders ORDER BY id",
|
|
static r => $"{r.GetInt64(0)}|{(r.IsDBNull(1) ? "<null>" : r.GetString(1))}|{(r.IsDBNull(2) ? "<null>" : r.GetInt64(2).ToString())}");
|
|
return string.Join("\n", rows);
|
|
}
|
|
|
|
private static async Task<string> DumpRowVersionAsync(ILocalDb db)
|
|
{
|
|
var rows = await db.QueryAsync(
|
|
"SELECT table_name, pk_json, hlc, node_id, is_tombstone FROM __localdb_row_version " +
|
|
"ORDER BY table_name, pk_json",
|
|
static r => $"{r.GetString(0)}|{r.GetString(1)}|{r.GetInt64(2)}|{r.GetString(3)}|{r.GetInt64(4)}");
|
|
return string.Join("\n", rows);
|
|
}
|
|
|
|
private static IConfiguration BuildConfig(bool initiator, Action<Dictionary<string, string?>>? tune)
|
|
{
|
|
var dict = new Dictionary<string, string?>
|
|
{
|
|
["LocalDb:Replication:FlushInterval"] = "00:00:00.050",
|
|
// Bound the initiator's reconnect backoff so recovery after a KillTransport/RestartTransport
|
|
// cycle is deterministic (the default 60s max lets the doubling backoff grow enough to
|
|
// occasionally overrun the 30s converge poll).
|
|
["LocalDb:Replication:ReconnectBackoffMax"] = "00:00:02",
|
|
};
|
|
if (initiator)
|
|
dict["LocalDb:Replication:PeerAddress"] = "http://localhost";
|
|
tune?.Invoke(dict);
|
|
return new ConfigurationBuilder().AddInMemoryCollection(dict).Build();
|
|
}
|
|
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
if (_bg is not null)
|
|
try { await _bg.StopAsync(CancellationToken.None); } catch { /* teardown */ }
|
|
await DisposeServerHostAsync();
|
|
if (_clientProvider is not null)
|
|
await _clientProvider.DisposeAsync();
|
|
|
|
_dbA.Dispose();
|
|
_dbB.Dispose();
|
|
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
|
|
foreach (var p in new[] { _pathA, _pathB })
|
|
{
|
|
if (File.Exists(p)) File.Delete(p);
|
|
if (File.Exists(p + "-wal")) File.Delete(p + "-wal");
|
|
if (File.Exists(p + "-shm")) File.Delete(p + "-shm");
|
|
}
|
|
}
|
|
}
|