test(localdb): two-node convergence for config tables + sf_messages
Four scenarios over the real site pair, driven through the REAL SiteStorageService rather than hand-written SQL. The sibling suites had to hand-write SQL because they were specifications written BEFORE the cutover; this suite runs after it, so it can drive the shipped writers — which is what makes the cascade scenario meaningful. - DeployedConfigRow_ConvergesToB_ColumnForColumn: the existing suite asserts config_json only, so a capture that dropped or defaulted any other column would still pass. deployed_at matters most — SiteReconciliationActor's guarded write compares it. - RemovingAnInstance_ConvergesAllThreeCascadeTables: the plan's flagged highest-risk case. RemoveDeployedConfigAsync deletes from three tables in one transaction, the schema has no foreign keys, and CDC ships three independent per-table streams. A dropped delete leaves a permanently stale override or alarm row on the standby, invisible until the instance name is redeployed. Carries a never-removed control instance, without which "the cascade converged" is indistinguishable from "node B lost these tables entirely". - ANativeAlarmBurst_Converges_AndTheOplogDrains: convergence alone would pass if entries replicated but were never acked, and an oplog that only grows trips the caps into a snapshot resync. - RowsWrittenOnBWhileItsListenerIsDown_SurviveTheRejoin: the union-survives property is per-table, and an unregistered table is silently local-only rather than an error, so it is re-proved on the tables the N1 scenario does not touch. Non-vacuity verified as the plan requires: with the eight Phase 2 RegisterReplicated calls commented out in SiteLocalDbSetup, all four go red (4 failed / 0 passed); restored, 20/20 pass across the three LocalDb suites. Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,193 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Phase 2 convergence driven through the REAL <see cref="SiteStorageService"/>, over a real
|
||||
/// site pair, rather than through hand-written SQL.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The sibling suites (<see cref="LocalDbConfigConvergenceTests"/>,
|
||||
/// <see cref="LocalDbStoreAndForwardConvergenceTests"/>) were written as specifications
|
||||
/// BEFORE the cutover, so they necessarily use hand-written SQL — the production writers were
|
||||
/// still going through the bespoke replicator at the time. This suite runs after the cutover
|
||||
/// and therefore drives the production writer, which is what makes the cascade scenario below
|
||||
/// meaningful: it is the shipped multi-statement transaction under test, not a re-creation
|
||||
/// of it.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <see cref="SiteStorageService"/> is constructed directly on the fixture's
|
||||
/// <see cref="ILocalDb"/> — the same instance the host registers — so its writes flow through
|
||||
/// the same CDC triggers.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[Collection("LocalDbSitePairConvergence")]
|
||||
public sealed class LocalDbPhase2ConvergenceTests : LocalDbSitePairHarness
|
||||
{
|
||||
private SiteStorageService Storage(ILocalDb db) =>
|
||||
new(db, NullLogger<SiteStorageService>.Instance);
|
||||
|
||||
// ---- read helpers -----------------------------------------------------------------
|
||||
|
||||
private static Task<long> CountAsync(ILocalDb db, string table, string instance) =>
|
||||
ScalarAsync(db, $"SELECT COUNT(*) FROM {table} WHERE instance_unique_name = @instance",
|
||||
new { instance });
|
||||
|
||||
private static async Task<long> ScalarAsync(ILocalDb db, string sql, object? args = null)
|
||||
{
|
||||
var rows = await db.QueryAsync(sql, static r => r.GetInt64(0), args);
|
||||
return rows[0];
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The whole <c>deployed_configurations</c> row as one comparable string, so a scenario
|
||||
/// can assert every column replicated rather than just the payload it happened to check.
|
||||
/// </summary>
|
||||
private static async Task<string?> ReadWholeConfigRowAsync(ILocalDb db, string instance)
|
||||
{
|
||||
var rows = await db.QueryAsync(
|
||||
"""
|
||||
SELECT instance_unique_name || '|' || config_json || '|' || deployment_id || '|' ||
|
||||
revision_hash || '|' || is_enabled || '|' || deployed_at
|
||||
FROM deployed_configurations WHERE instance_unique_name = @instance
|
||||
""",
|
||||
static r => r.GetString(0), new { instance });
|
||||
return rows.Count == 0 ? null : rows[0];
|
||||
}
|
||||
|
||||
/// <summary>Unacked oplog backlog. Drains to zero once the peer acknowledges.</summary>
|
||||
private static Task<long> OplogDepthAsync(ILocalDb db) =>
|
||||
ScalarAsync(db,
|
||||
"SELECT COUNT(*) FROM __localdb_oplog WHERE seq > " +
|
||||
"(SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)");
|
||||
|
||||
// ---- scenarios --------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task DeployedConfigRow_ConvergesToB_ColumnForColumn()
|
||||
{
|
||||
// LocalDbConfigConvergenceTests asserts config_json converges. That is the payload
|
||||
// only: a capture that dropped, defaulted or reordered any other column would still
|
||||
// pass it. deployed_at is the one that matters most — SiteReconciliationActor's
|
||||
// guarded write compares it, so a node whose replicated copy carried a different
|
||||
// timestamp would make different staleness decisions from its peer.
|
||||
await Storage(A).StoreDeployedConfigAsync(
|
||||
"inst-columns", """{"setpoint":7}""", "dep-col-1", "hash-col-1", isEnabled: true);
|
||||
|
||||
var onA = await ReadWholeConfigRowAsync(A, "inst-columns");
|
||||
Assert.NotNull(onA);
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () => await ReadWholeConfigRowAsync(B, "inst-columns") == onA,
|
||||
"node B's deployed_configurations row to match node A's in EVERY column");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RemovingAnInstance_ConvergesAllThreeCascadeTables()
|
||||
{
|
||||
// The plan flagged this as the most likely real defect in Phase 2, and the reasoning
|
||||
// is sound: RemoveDeployedConfigAsync deletes from three tables in one transaction,
|
||||
// the schema has NO foreign keys, and CDC captures the three deletes as independent
|
||||
// per-table streams that LWW may apply to the peer in any order. Nothing in the
|
||||
// system re-derives an orphan, so a dropped delete leaves a permanently stale
|
||||
// static_attribute_overrides or native_alarm_state row on the standby — invisible
|
||||
// until that instance name is redeployed and picks up ghost overrides.
|
||||
var storageA = Storage(A);
|
||||
|
||||
// A second instance that is never removed. Without it this test cannot distinguish
|
||||
// "the cascade converged" from "node B lost these tables entirely" — a registration
|
||||
// bug that wiped B would satisfy the absence assertions on its own.
|
||||
foreach (var instance in new[] { "inst-cascade", "inst-survivor" })
|
||||
{
|
||||
await storageA.StoreDeployedConfigAsync(
|
||||
instance, """{"v":1}""", $"dep-{instance}", $"hash-{instance}", isEnabled: true);
|
||||
await storageA.SetStaticOverrideAsync(instance, "Setpoint", "42");
|
||||
await storageA.SetStaticOverrideAsync(instance, "Mode", "Auto");
|
||||
await storageA.UpsertNativeAlarmAsync(
|
||||
instance, "Area1.Line1.Alarm", "ref-1", """{"active":true}""", DateTimeOffset.UtcNow);
|
||||
}
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () =>
|
||||
await CountAsync(B, "deployed_configurations", "inst-cascade") == 1 &&
|
||||
await CountAsync(B, "static_attribute_overrides", "inst-cascade") == 2 &&
|
||||
await CountAsync(B, "native_alarm_state", "inst-cascade") == 1,
|
||||
"all three tables to reach node B before the instance is removed");
|
||||
|
||||
await storageA.RemoveDeployedConfigAsync("inst-cascade");
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () =>
|
||||
await CountAsync(B, "deployed_configurations", "inst-cascade") == 0 &&
|
||||
await CountAsync(B, "static_attribute_overrides", "inst-cascade") == 0 &&
|
||||
await CountAsync(B, "native_alarm_state", "inst-cascade") == 0,
|
||||
"the 3-table cascade delete to converge on node B with NO orphans left behind");
|
||||
|
||||
// The control instance is untouched on both nodes: the cascade is scoped, not a wipe.
|
||||
Assert.Equal(1, await CountAsync(B, "deployed_configurations", "inst-survivor"));
|
||||
Assert.Equal(2, await CountAsync(B, "static_attribute_overrides", "inst-survivor"));
|
||||
Assert.Equal(1, await CountAsync(B, "native_alarm_state", "inst-survivor"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ANativeAlarmBurst_Converges_AndTheOplogDrains()
|
||||
{
|
||||
// native_alarm_state is the highest-rate replicated table (Task 1 found it bounded by
|
||||
// per-SourceReference coalescing on a 100 ms flush, correcting the plan's D4). This
|
||||
// asserts the burst converges AND that the oplog goes back to empty afterwards —
|
||||
// convergence alone would still pass if entries replicated but were never acked, and
|
||||
// an oplog that only grows eventually trips the caps and forces a snapshot resync.
|
||||
const int burst = 200;
|
||||
var storageA = Storage(A);
|
||||
var at = DateTimeOffset.UtcNow;
|
||||
|
||||
for (var i = 0; i < burst; i++)
|
||||
{
|
||||
await storageA.UpsertNativeAlarmAsync(
|
||||
"inst-burst", "Area1.Line1.Alarm", $"ref-{i}", $$"""{"seq":{{i}}}""", at);
|
||||
}
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () => await CountAsync(B, "native_alarm_state", "inst-burst") == burst,
|
||||
$"all {burst} alarm rows to reach node B");
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () => await OplogDepthAsync(A) == 0,
|
||||
"node A's oplog to drain once node B acknowledges the burst");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RowsWrittenOnBWhileItsListenerIsDown_SurviveTheRejoin()
|
||||
{
|
||||
// Companion to LocalDbConfigConvergenceTests' N1 scenario, which covers the same
|
||||
// rejoin for sf_messages and deployed_configurations. This one exercises the tables
|
||||
// that scenario does not touch, because "the union survives" is a per-table property:
|
||||
// it holds only if each table is actually registered, and an unregistered table is
|
||||
// silently local-only rather than an error.
|
||||
await Storage(A).StoreSharedScriptAsync("script-before", "return 1;", null, null);
|
||||
await WaitUntilAsync(
|
||||
async () => await ScalarAsync(B,
|
||||
"SELECT COUNT(*) FROM shared_scripts WHERE name = 'script-before'") == 1,
|
||||
"the pre-outage script to reach node B");
|
||||
|
||||
await StopPassiveAsync();
|
||||
|
||||
// Each node writes to a different Phase 2 table while they cannot see each other.
|
||||
await Storage(A).StoreExternalSystemAsync(
|
||||
"sys-from-a", "http://a.invalid", "None", null, null);
|
||||
await Storage(B).SetStaticOverrideAsync("inst-offline", "WrittenWhileDown", "yes");
|
||||
|
||||
await RestartPairAsync();
|
||||
|
||||
await WaitUntilAsync(
|
||||
async () =>
|
||||
await ScalarAsync(A,
|
||||
"SELECT COUNT(*) FROM static_attribute_overrides WHERE attribute_name = 'WrittenWhileDown'") == 1 &&
|
||||
await ScalarAsync(B,
|
||||
"SELECT COUNT(*) FROM external_systems WHERE name = 'sys-from-a'") == 1,
|
||||
"both nodes to hold the union of what each wrote while partitioned");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user