tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbConfigConvergenceTests.cs

@@ -0,0 +1,177 @@
+using ZB.MOM.WW.LocalDb;
+
+namespace ZB.MOM.WW.ScadaBridge.IntegrationTests;
+
+/// <summary>
+/// Phase 2 convergence: deployed configuration across a real site pair, plus the resync
+/// behaviour that the bespoke replicator needed a directional guard for.
+/// </summary>
+/// <remarks>
+/// Companion to <see cref="LocalDbStoreAndForwardConvergenceTests"/>: same
+/// specifications-before-deletion role, for the intents held by
+/// <c>SiteReplicationActorTests</c> and <c>SfBufferResyncPredicateTests</c> rather than by
+/// the store-and-forward <c>ReplicationService</c>.
+/// </remarks>
+[Collection("LocalDbSitePairConvergence")]
+public sealed class LocalDbConfigConvergenceTests : LocalDbSitePairHarness
+{
+    // ---- data helpers -----------------------------------------------------------------
+
+    private static Task DeployConfigAsync(ILocalDb db, string instance, string configJson, string deploymentId)
+        => db.ExecuteAsync(
+            """
+            INSERT INTO deployed_configurations (
+                instance_unique_name, config_json, deployment_id, revision_hash,
+                is_enabled, deployed_at)
+            VALUES (@instance, @configJson, @deploymentId, @deploymentId, 1, @now)
+            ON CONFLICT(instance_unique_name) DO UPDATE SET
+                config_json   = excluded.config_json,
+                deployment_id = excluded.deployment_id,
+                revision_hash = excluded.revision_hash,
+                deployed_at   = excluded.deployed_at;
+            """,
+            new { instance, configJson, deploymentId, now = DateTime.UtcNow.ToString("o") });
+
+    private static async Task<string?> ReadConfigAsync(ILocalDb db, string instance)
+    {
+        var rows = await db.QueryAsync(
+            "SELECT config_json FROM deployed_configurations WHERE instance_unique_name = @instance",
+            static r => r.GetString(0), new { instance });
+        return rows.Count == 0 ? null : rows[0];
+    }
+
+    private static Task EnqueueAsync(ILocalDb db, string id)
+        => db.ExecuteAsync(
+            """
+            INSERT INTO sf_messages (
+                id, category, target, payload_json, retry_count, max_retries,
+                retry_interval_ms, created_at, status)
+            VALUES (@id, 0, 'central', '{}', 0, 50, 30000, @now, 0);
+            """,
+            new { id, now = DateTime.UtcNow.ToString("o") });
+
+    private static async Task<bool> MessageExistsAsync(ILocalDb db, string id)
+    {
+        var rows = await db.QueryAsync(
+            "SELECT 1 FROM sf_messages WHERE id = @id", static r => r.GetInt32(0), new { id });
+        return rows.Count > 0;
+    }
+
+    // ---- scenarios --------------------------------------------------------------------
+
+    [Fact]
+    public async Task ANodeWithNewerLocalRows_KeepsThem_WhenAPeerSnapshotArrives()
+    {
+        // N1 Critical (was: SfBufferResyncPredicateTests). That test existed because the
+        // bespoke replicator's ReplaceAllAsync was a destructive DELETE-then-INSERT: a resync
+        // running in the wrong direction WIPED a live buffer, so the code needed a
+        // same-oldest-Up predicate to decide who had authority. The divergence case was a
+        // rolling restart leaving the delivering node oldest but not leader.
+        //
+        // Under LocalDb that failure mode is structurally impossible rather than guarded
+        // against: snapshot resync merges per row under LWW and never deletes. So this is
+        // deliberately NOT a directional-authority test — there is no active/standby
+        // asymmetry left to enforce. It asserts the property the guard was protecting: no
+        // node loses rows to a peer's snapshot.
+        //
+        // The setup is the wipe scenario made concrete. Each node holds a row the other has
+        // never seen, and they disagree about a third; then they resync.
+        await DeployConfigAsync(A, "shared-instance", """{"v":1}""", "dep-1");
+        await WaitUntilAsync(
+            async () => await ReadConfigAsync(B, "shared-instance") is not null,
+            "the shared row to exist on both nodes before they diverge");
+
+        await StopPassiveAsync();
+
+        // A keeps working while B is down: a brand-new buffered message and a newer version
+        // of the shared config.
+        await EnqueueAsync(A, "live-row-on-a");
+        await DeployConfigAsync(A, "shared-instance", """{"v":2}""", "dep-2");
+
+        // B is offline but its database is still writable — it also has local state the
+        // snapshot exchange must not destroy.
+        await EnqueueAsync(B, "live-row-on-b");
+
+        await RestartPairAsync();
+
+        await WaitUntilAsync(
+            async () =>
+                await MessageExistsAsync(A, "live-row-on-a") &&
+                await MessageExistsAsync(A, "live-row-on-b") &&
+                await MessageExistsAsync(B, "live-row-on-a") &&
+                await MessageExistsAsync(B, "live-row-on-b"),
+            "both nodes to hold the UNION of the rows each wrote while partitioned");
+
+        // And the contended row converges to the newer write rather than either node's
+        // snapshot flattening the other.
+        await WaitUntilAsync(
+            async () => await ReadConfigAsync(A, "shared-instance") == """{"v":2}"""
+                     && await ReadConfigAsync(B, "shared-instance") == """{"v":2}""",
+            "both nodes to converge on the newer config for the contended instance");
+    }
+
+    [Fact]
+    public async Task ConfigDeployedOnA_ConvergesToB_WithoutAnyFetch()
+    {
+        // Was: SiteReplicationActorTests #1-#5 (notify-and-fetch — the standby is told a
+        // deploy happened, then HTTP-fetches the config itself, with retries and a
+        // superseded-404 path). Under CDC the config row simply replicates, so that whole
+        // exchange is deleted at Task 15.
+        //
+        // The plan called for asserting a fetcher test double records zero invocations. That
+        // would be THEATRE here and is deliberately not done: this harness has no actor
+        // system, no central, and no IDeploymentConfigFetcher in the graph at all, so a
+        // double would record zero calls whether or not the fetch path still existed. An
+        // assertion that cannot fail is worse than none.
+        //
+        // What this test honestly proves is the positive half: the config reaches node B
+        // through replication ALONE, in a process where fetching is not merely unused but
+        // absent. The negative half — that no fetch path survives — is proved by Task 15
+        // deleting the code and the build still passing, which is a stronger check than any
+        // runtime counter.
+        //
+        // Scope note (D1): SiteReconciliationActor survives Phase 2 and legitimately fetches
+        // over HTTP at node startup when central reports gaps. "The standby never fetches,
+        // ever" would be a FALSE claim; the claim is about the deploy path only.
+        await DeployConfigAsync(A, "inst-deploy", """{"setpoint":42}""", "dep-100");
+
+        await WaitUntilAsync(
+            async () => await ReadConfigAsync(B, "inst-deploy") == """{"setpoint":42}""",
+            "the deployed config to reach node B over replication alone");
+    }
+
+    [Fact]
+    public async Task RedeployingAnInstance_ConvergesToTheNewRevision_OnBothNodes()
+    {
+        // The steady-state deploy loop, and the reason the old path needed a deployed_at
+        // guard: a standby that applied a STALE fetch would pin an instance to superseded
+        // config until the next deploy. Under LWW the newer write wins on the primary key,
+        // so the guard is not what protects this — but the outcome still has to hold.
+        await DeployConfigAsync(A, "inst-redeploy", """{"rev":"a"}""", "dep-a");
+        await WaitUntilAsync(
+            async () => await ReadConfigAsync(B, "inst-redeploy") == """{"rev":"a"}""",
+            "the first revision to reach node B");
+
+        await DeployConfigAsync(A, "inst-redeploy", """{"rev":"b"}""", "dep-b");
+
+        await WaitUntilAsync(
+            async () => await ReadConfigAsync(A, "inst-redeploy") == """{"rev":"b"}"""
+                     && await ReadConfigAsync(B, "inst-redeploy") == """{"rev":"b"}""",
+            "both nodes to converge on the redeployed revision");
+    }
+
+    [Fact]
+    public async Task ConfigDeployedOnB_ConvergesToA()
+    {
+        // Deliberately the reverse direction. The bespoke replicator was asymmetric by
+        // design — the active node pushed, the standby fetched — so "which node deployed it"
+        // was a meaningful question. Under CDC it is not, and this test is what says so.
+        // After a failover the surviving node must be able to deploy without waiting to be
+        // promoted to some authority role.
+        await DeployConfigAsync(B, "inst-from-b", """{"origin":"b"}""", "dep-b1");
+
+        await WaitUntilAsync(
+            async () => await ReadConfigAsync(A, "inst-from-b") == """{"origin":"b"}""",
+            "a config deployed on node B to reach node A");
+    }
+}