Joseph Doherty
5bbd7689fa
Re-ran all 8 domain reviews at HEAD8c888f13against theb910f5ebbaseline: every round-1 finding source-verified (168 fixed, 0 regressions, 0 false claims); 56 new findings (1 Critical / 4 High / 15 Medium / 36 Low), concentrated in post-baseline code (anti-entropy resync, KPI rollup backfill, live alarm stream) and seams the fixes exposed. Headliners: S&F resync predicate inversion can wipe the delivering node's buffer (02-N1 Critical); resync snapshot exceeds the Akka remoting frame size (02-N2); failover drill kills the one node keep-oldest can't survive (01-N1); unbounded rollup backfill per failover (04-R1); live production API key in untracked test.txt (08-NF1). Adds PLAN-R2-01..08 + .tasks.json manifests and the Round-2 board, P0 list, cross-plan mutexes, and wave order in 00-MASTER-TRACKER.
1341 lines
88 KiB
Markdown
1341 lines
88 KiB
Markdown
# PLAN-R2-02 — Communication & Store-and-Forward Round-2 Fix Implementation Plan
|
||
|
||
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans to implement this plan task-by-task.
|
||
|
||
**Goal:** Fix every NEW finding in `archreview/02-communication-store-and-forward.md` (round 2, dated 2026-07-12, on `main` @ `8c888f13`) — the Critical resync predicate inversion that lets a routine rolling restart wipe the delivering node's live S&F buffer (N1, which also closes the round-1 partial: the leader-vs-oldest helper was never swapped into `SiteCommunicationActor`/`SiteReplicationActor`), the High frame-size-undeliverable resync snapshot (N2), the `_sweepTask` clobber that defeats the shutdown drain (N3), live-alarm delta coalescing (N6), and the Low hygiene items (N4, N5, N7, N8, N9).
|
||
|
||
**Architecture:** The fixes stay inside the existing seams. A single oldest-Up active-node predicate (`ActiveNodeEvaluator`, new in Communication, delegated to by the Host's `ClusterActivityEvaluator`) replaces both surviving leader+Up defaults, and the Host wires the *same* `IClusterNodeProvider.SelfIsPrimary` delegate that already gates the S&F delivery sweep into `SiteReplicationActor` and `SiteCommunicationActor` — one definition of "active" governs delivery, resync-request, resync-answer, and heartbeat `IsActive`. The anti-entropy snapshot becomes a chunked, byte-budgeted, ack-confirmed protocol (additive messages; the legacy monolithic `SfBufferSnapshot` handler is retained for rolling compat). The sweep handoff publishes `_sweepTask` only when the CAS is won. The live alarm aggregator gains a publish-coalescing timer, a queued re-seed, a stream-generation stamp, and instance-injected test seams; `SiteStreamGrpcClientFactory.RemoveSiteAsync` gets its production caller on site delete; single-endpoint sites can go live.
|
||
|
||
**Tech Stack:** C#/.NET 10, Akka.NET (TestKit.Xunit2, Cluster, ClusterClient), Microsoft.Data.Sqlite, Grpc.Net, xUnit + NSubstitute. Build: `dotnet build ZB.MOM.WW.ScadaBridge.slnx`. Test per-project: `dotnet test tests/<project>` — targeted `--filter` runs only, never the full suite mid-plan.
|
||
|
||
## Cross-plan MUTEX (binding)
|
||
|
||
**PLAN-R2-07 also touches `SiteAlarmLiveCacheService.cs` (sticky `IsLive`) and `AlarmSummary.razor`.** Serialization rule: **Task 10 (delta coalescing), Task 12 (aggregator ctor seams), and Task 15 (single-endpoint resolve) must NOT run concurrently with PLAN-R2-07's sticky-IsLive / AlarmSummary tasks.** Tasks 10/12/15 edit `SiteAlarmAggregatorActor.cs` + the `Props.Create` call and `ResolveSiteAsync` inside `SiteAlarmLiveCacheService.cs` — the exact file R2-07's sticky-IsLive task rewrites, and Task 10 changes the publish cadence that sticky-IsLive semantics observe. Whichever plan starts its live-alarm slice first finishes it (through commit) before the other begins; coordinate via the master tracker. All other tasks in this plan are file-disjoint from R2-07. `AlarmSummary.razor` is never edited here (page-side fixes belong to PLAN-R2-07).
|
||
|
||
## Findings Coverage
|
||
|
||
| Report finding | Severity | Task(s) |
|
||
|---|---|---|
|
||
| N1: Resync authority uses *leader* predicate while delivery uses *oldest-Up* — rolling restart of the lower-address node wipes the delivering node's live buffer | Critical | 1, 2, 3, 4 (2 is the mandated failing-first divergence repro; 3 includes the belt-and-braces apply-time re-check) |
|
||
| Round-1 partial: oldest-Up helper shipped (PLAN-01) but never swapped into `SiteCommunicationActor.DefaultIsActiveCheck` / `SiteReplicationActor.DefaultIsActive` | Low → materially worse | **Subsumed by N1's fix** — 1 (shared evaluator), 3 (`SiteReplicationActor`), 4 (`SiteCommunicationActor`) |
|
||
| N2: Resync snapshot rides ONE Akka remoting message — exceeds the 128 000-byte default frame for any realistic backlog, silently undeliverable, no retry, no telemetry | High | 5, 6, 7 (chunked byte-budgeted protocol + standby assembly + ack-based delivery confirmation + telemetry) |
|
||
| N3: `_sweepTask` clobbered by every timer tick / `TriggerSweep` — `StopAsync` awaits a no-op while the real sweep is mid-delivery | Medium | 8 |
|
||
| N4: `SweepBatchLimit` / `SweepTargetParallelism` missing from the eager options validator | Low | 9 |
|
||
| N5: Resync snapshot vs in-flight replication op race leaves orphan rows on the standby | Low | 6 (explicit doc comment on the chunk-apply path so nobody "fixes" it into something worse — code change deliberately none, per report) |
|
||
| N6: No live-delta coalescing — every alarm delta copies the whole site cache and fans out to every viewer circuit | Medium | 10 (**MUTEX with PLAN-R2-07** — see rule above) |
|
||
| N7.1: Reconnect re-seed skipped when a reconcile fan-out is already in flight (stale up to 60 s) | Low | 11 |
|
||
| N7.2: No stream-generation stamp on `GrpcAlarmStreamError` — a late error from the previous stream burns retry budget | Low | 11 |
|
||
| N7.3: `ReconnectDelay`/`StabilityWindow` are process-global mutable statics | Low | 12 |
|
||
| N8: `RemoveSiteAsync` has no production caller — deleted site's channels leak until restart | Low | 13 |
|
||
| N8 (second half): "lives only on the active central node" is aspirational — standby browsing starts a second aggregator | Low | 14 (**documented acceptance** — read-only, bounded, linger-reaped; doc claim corrected rather than gating `SetActorSystem`, matching the report's "harmless but contradicting the doc" framing) |
|
||
| N9: A site with only one configured gRPC endpoint can never go live | Low | 15 |
|
||
| Alarm Summary page-side fixes (poll/live interplay, render path) | — | **Not this plan** — owned by PLAN-R2-07 (referenced here only for the MUTEX rule) |
|
||
|
||
---
|
||
|
||
### Task 1: Shared oldest-Up active-node evaluator, reachable from Communication and SiteRuntime
|
||
|
||
**Classification:** standard
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 2, 8, 9, 10, 13, 14
|
||
**Files:**
|
||
- Create: `src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs`
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs` (lines 20-32 — `SelfIsOldest` delegates to the new evaluator)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs` (create)
|
||
|
||
**Design decision — new home in Communication, Host delegates:** PLAN-01's `ClusterActivityEvaluator` (`Host/Health/ClusterActivityEvaluator.cs:14`) declares itself "THE single definition of active node", but it lives in Host, which `Communication` and `SiteRuntime` cannot reference (Host references them). That is *why* the two leader+Up defaults were never swapped (round-1 partial). The logic moves to `Communication` (referenced by both `SiteRuntime` — see `ZB.MOM.WW.ScadaBridge.SiteRuntime.csproj:39` — and Host); Host's evaluator keeps its public API and becomes a one-line delegate, so `AkkaClusterNodeProvider.SelfIsPrimary` (`Host/Health/AkkaClusterNodeProvider.cs:29-38`) and every existing Host test keep working against the *same single implementation*.
|
||
|
||
1. Write the failing test (cluster-provider harness copied from `tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ClusterActivityEvaluatorTests.cs` — single-node self-seeding cluster on a free port):
|
||
|
||
```csharp
|
||
// tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs
|
||
using System.Net;
|
||
using System.Net.Sockets;
|
||
using Akka.Actor;
|
||
using Akka.Configuration;
|
||
using Xunit;
|
||
using ZB.MOM.WW.ScadaBridge.Communication.ClusterState;
|
||
|
||
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
|
||
|
||
public class ActiveNodeEvaluatorTests : IAsyncLifetime
|
||
{
|
||
private ActorSystem? _system;
|
||
|
||
public async Task InitializeAsync()
|
||
{
|
||
var port = FreePort();
|
||
var config = ConfigurationFactory.ParseString($@"
|
||
akka {{
|
||
actor.provider = cluster
|
||
remote.dot-netty.tcp {{ hostname = ""127.0.0.1"", port = {port} }}
|
||
cluster {{
|
||
seed-nodes = [""akka.tcp://ane-test@127.0.0.1:{port}""]
|
||
roles = [""site-x""]
|
||
min-nr-of-members = 1
|
||
}}
|
||
}}");
|
||
_system = ActorSystem.Create("ane-test", config);
|
||
var cluster = Akka.Cluster.Cluster.Get(_system);
|
||
var deadline = DateTime.UtcNow.AddSeconds(20);
|
||
while (cluster.SelfMember.Status != Akka.Cluster.MemberStatus.Up && DateTime.UtcNow < deadline)
|
||
await Task.Delay(100);
|
||
}
|
||
|
||
public async Task DisposeAsync() { if (_system != null) await _system.Terminate(); }
|
||
|
||
[Fact]
|
||
public void SelfIsOldestUp_SoleUpMember_ReturnsTrue()
|
||
{
|
||
var cluster = Akka.Cluster.Cluster.Get(_system!);
|
||
Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(cluster));
|
||
Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(cluster, "site-x"));
|
||
}
|
||
|
||
[Fact]
|
||
public void SelfIsOldestUp_RoleNotHeld_ReturnsFalse()
|
||
{
|
||
var cluster = Akka.Cluster.Cluster.Get(_system!);
|
||
Assert.False(ActiveNodeEvaluator.SelfIsOldestUp(cluster, "role-nonexistent"));
|
||
}
|
||
|
||
private static int FreePort()
|
||
{
|
||
using var l = new TcpListener(IPAddress.Loopback, 0);
|
||
l.Start();
|
||
return ((IPEndPoint)l.LocalEndpoint).Port;
|
||
}
|
||
}
|
||
```
|
||
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter ActiveNodeEvaluatorTests` — expect FAIL (compile error: `ActiveNodeEvaluator` does not exist).
|
||
3. Implement `src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs` (verbatim move of the oldest-Up logic from `ClusterActivityEvaluator.SelfIsOldest`):
|
||
|
||
```csharp
|
||
using Akka.Cluster;
|
||
|
||
namespace ZB.MOM.WW.ScadaBridge.Communication.ClusterState;
|
||
|
||
/// <summary>
|
||
/// THE single definition of "active node" (review 01 [High]; review 02 round 2 [Critical] N1):
|
||
/// a node is active when it is the OLDEST Up member (optionally within a role scope) — i.e.
|
||
/// the member the ClusterSingletonManager places singletons on. Cluster LEADERSHIP (lowest
|
||
/// address) is an Akka-internal concept that diverges from singleton placement permanently
|
||
/// once the original first node restarts and rejoins; every product-level active/standby
|
||
/// decision must use this evaluator, never <c>cluster.State.Leader</c>.
|
||
/// <para>
|
||
/// Lives in Communication (not Host) so BOTH <c>SiteCommunicationActor</c> and
|
||
/// <c>SiteReplicationActor</c> can default to it — Host cannot be referenced from either.
|
||
/// The Host's <c>ClusterActivityEvaluator.SelfIsOldest</c> delegates here, so the S&F
|
||
/// delivery gate (<c>IClusterNodeProvider.SelfIsPrimary</c>), the resync authority checks,
|
||
/// and the heartbeat IsActive stamp all share one implementation.
|
||
/// </para>
|
||
/// </summary>
|
||
public static class ActiveNodeEvaluator
|
||
{
|
||
/// <summary>True when self is Up and no other Up member (in the role scope) is older.</summary>
|
||
/// <param name="cluster">The Akka cluster to evaluate.</param>
|
||
/// <param name="role">Optional role scope; when set, only members with this role are considered.</param>
|
||
/// <returns><c>true</c> when self is Up and the oldest Up member in the role scope.</returns>
|
||
public static bool SelfIsOldestUp(Cluster cluster, string? role = null)
|
||
{
|
||
var self = cluster.SelfMember;
|
||
if (self.Status != MemberStatus.Up)
|
||
return false;
|
||
if (role != null && !self.HasRole(role))
|
||
return false;
|
||
|
||
return cluster.State.Members
|
||
.Where(m => m.Status == MemberStatus.Up)
|
||
.Where(m => role == null || m.HasRole(role))
|
||
.All(m => m.UniqueAddress.Equals(self.UniqueAddress) || self.IsOlderThan(m));
|
||
}
|
||
}
|
||
```
|
||
|
||
And in `Host/Health/ClusterActivityEvaluator.cs`, replace the `SelfIsOldest` body (keep `OldestUpMember` unchanged):
|
||
|
||
```csharp
|
||
/// <summary>True when self is Up and no other Up member (in the role scope) is older.
|
||
/// Delegates to the shared <see cref="Communication.ClusterState.ActiveNodeEvaluator"/> —
|
||
/// one implementation for the delivery gate, the resync authority checks, and the
|
||
/// heartbeat IsActive stamp (review 02 round 2, N1).</summary>
|
||
public static bool SelfIsOldest(Cluster cluster, string? role = null) =>
|
||
Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(cluster, role);
|
||
```
|
||
|
||
(Full namespace: `ZB.MOM.WW.ScadaBridge.Communication.ClusterState` — add a `using` alias if the shorthand does not resolve.)
|
||
4. Run the filter — expect PASS. Regression: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Host.Tests --filter ClusterActivityEvaluatorTests` — expect PASS (delegation preserves behavior).
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs && git commit -m "fix(comm): shared oldest-Up ActiveNodeEvaluator in Communication; Host evaluator delegates (plan R2-02 T1)"`
|
||
|
||
### Task 2: Two-node divergence repro — leader≠oldest wipes the delivering node's buffer (failing first)
|
||
|
||
**Classification:** high-risk (real two-node cluster; the mandated N1 regression repro)
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 1, 8, 9, 10, 13, 14, 15
|
||
**Files:**
|
||
- Modify: `tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs` (`StartAsync` gains optional explicit ports — additive)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs` (create)
|
||
|
||
**This test is written BEFORE the fix and must FAIL on unfixed code** — it reproduces the report's N1 sequence. The report narrates it as a rolling restart of the lower-address node; the essence is *older-but-higher-address vs younger-but-lower-address*, which the test produces directly (deterministically, no crash/restart needed) by starting the **higher**-port node first: the first-started (higher-address) node is **oldest** (owns delivery), the second-started (lower-address) node is **leader**. Pre-fix, `SiteReplicationActor.DefaultIsActive` (`SiteReplicationActor.cs:190-198`, leader+Up) then makes the *delivering* node request a resync and apply the joiner's stale/empty snapshot via `ReplaceAllAsync` — the buffer wipe. Post-fix (Task 3), the oldest node answers and the joiner applies: the buffered row survives on the oldest node AND appears on the joiner.
|
||
|
||
1. Add the additive fixture parameters (`TwoNodeClusterFixture.StartAsync`, currently `StartAsync(string role = "Central", TimeSpan? stableAfter = null)`):
|
||
|
||
```csharp
|
||
public static async Task<TwoNodeClusterFixture> StartAsync(
|
||
string role = "Central", TimeSpan? stableAfter = null,
|
||
int? portA = null, int? portB = null)
|
||
{
|
||
var f = new TwoNodeClusterFixture();
|
||
f.PortA = portA ?? GetFreeTcpPort();
|
||
f.PortB = portB ?? GetFreeTcpPort();
|
||
// ... rest unchanged
|
||
```
|
||
|
||
2. Write the failing test:
|
||
|
||
```csharp
|
||
// tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs
|
||
using Akka.Actor;
|
||
using Microsoft.Extensions.Logging.Abstractions;
|
||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
|
||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
|
||
using ZB.MOM.WW.ScadaBridge.StoreAndForward;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||
|
||
namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster;
|
||
|
||
/// <summary>
|
||
/// N1 regression (review 02 round 2, Critical): the resync authority must use the same
|
||
/// oldest-Up predicate as the S&F delivery gate. Divergence scenario = the delivering node
|
||
/// is OLDEST but not LEADER (leader = lowest address), the exact state a rolling restart of
|
||
/// the lower-address node produces. Pre-fix the delivering node requests a resync from the
|
||
/// stale peer and ReplaceAllAsync wipes its live buffer.
|
||
/// </summary>
|
||
public class SfBufferResyncPredicateTests
|
||
{
|
||
[Fact]
|
||
public async Task OldestButNotLeaderNode_KeepsItsBuffer_AndSeedsTheJoiner()
|
||
{
|
||
// Two explicit ports, deliberately assigned so the FIRST-started (oldest,
|
||
// delivering) node has the HIGHER address → the second node is cluster leader.
|
||
var p1 = TwoNodeClusterFixture.GetFreeTcpPort();
|
||
var p2 = TwoNodeClusterFixture.GetFreeTcpPort();
|
||
var (portHigh, portLow) = p1 > p2 ? (p1, p2) : (p2, p1);
|
||
|
||
await using var fixture = await TwoNodeClusterFixture.StartAsync(
|
||
role: "site-int", portA: portHigh, portB: portLow);
|
||
|
||
// Real S&F storage + replication actor per node, production default predicate
|
||
// (no isActiveOverride) — the exact wiring under test.
|
||
var (storageOldest, _) = await CreateReplicationActorAsync(fixture.NodeA, "oldest");
|
||
var (storageJoiner, _) = await CreateReplicationActorAsync(fixture.NodeB, "joiner");
|
||
|
||
// The delivering (oldest) node has a live buffered row the standby never saw.
|
||
await storageOldest.EnqueueAsync(NewMessage("live-row"));
|
||
|
||
// Trigger peer (re)tracking on both sides: each actor got InitialStateAsSnapshot
|
||
// in PreStart, but the enqueue raced it — re-deliver via a fresh MemberUp is not
|
||
// needed; OnPeerTracked already fired on join. The resync exchange is async:
|
||
// wait until the JOINER holds the row (proves the snapshot flowed oldest→joiner,
|
||
// the correct direction). Pre-fix this times out (the joiner, as leader, never
|
||
// requests) AND the oldest node's row is deleted by the stale wipe.
|
||
await AwaitAsync(async () => await storageJoiner.GetMessageByIdAsync("live-row") != null,
|
||
TimeSpan.FromSeconds(20),
|
||
"joiner never received the resync snapshot (resync ran in the wrong direction)");
|
||
|
||
// And the delivering node's buffer is untouched — the N1 wipe assertion.
|
||
Assert.NotNull(await storageOldest.GetMessageByIdAsync("live-row"));
|
||
}
|
||
|
||
private static async Task<(StoreAndForwardStorage Storage, IActorRef Actor)> CreateReplicationActorAsync(
|
||
ActorSystem node, string tag)
|
||
{
|
||
var sfDb = Path.Combine(Path.GetTempPath(), $"sf-resync-{tag}-{Guid.NewGuid():N}.db");
|
||
var siteDb = Path.Combine(Path.GetTempPath(), $"site-resync-{tag}-{Guid.NewGuid():N}.db");
|
||
var sfStorage = new StoreAndForwardStorage($"Data Source={sfDb}",
|
||
NullLogger<StoreAndForwardStorage>.Instance);
|
||
await sfStorage.InitializeAsync();
|
||
var siteStorage = new SiteStorageService($"Data Source={siteDb}",
|
||
NullLogger<SiteStorageService>.Instance);
|
||
var replicationService = new ReplicationService(
|
||
new StoreAndForwardOptions(), NullLogger<ReplicationService>.Instance);
|
||
// Name MUST be "site-replication" — SendToPeer targets /user/site-replication.
|
||
var actor = node.ActorOf(Props.Create(() => new SiteReplicationActor(
|
||
siteStorage, sfStorage, replicationService, "site-int",
|
||
NullLogger<SiteReplicationActor>.Instance, null, null, null, null)),
|
||
"site-replication");
|
||
return (sfStorage, actor);
|
||
}
|
||
|
||
private static StoreAndForwardMessage NewMessage(string id) => new()
|
||
{
|
||
Id = id,
|
||
Category = StoreAndForwardCategory.Notification,
|
||
Target = "central",
|
||
PayloadJson = "{}",
|
||
CreatedAt = DateTimeOffset.UtcNow,
|
||
Status = StoreAndForwardMessageStatus.Pending,
|
||
MaxRetries = 0,
|
||
};
|
||
|
||
private static async Task AwaitAsync(Func<Task<bool>> condition, TimeSpan timeout, string why)
|
||
{
|
||
var deadline = DateTime.UtcNow + timeout;
|
||
while (DateTime.UtcNow < deadline)
|
||
{
|
||
if (await condition()) return;
|
||
await Task.Delay(250);
|
||
}
|
||
throw new TimeoutException(why);
|
||
}
|
||
}
|
||
```
|
||
|
||
(Adapt member names to the real `SiteStorageService`/`SiteReplicationActor` ctor order — see `SiteReplicationActor.cs:64-73` and the harness in `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs:48-70`. `GetFreeTcpPort` may need its access raised to `public`/`internal` on the fixture — additive.)
|
||
3. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.IntegrationTests --filter SfBufferResyncPredicateTests` — **expect FAIL on unfixed code** (either the timeout — resync ran leader-ward — or the final `Assert.NotNull` after the wipe). This failure is the deliverable of this task; do NOT fix it here.
|
||
4. Commit (the red test rides with the fix task's history; commit it now so Task 3's diff shows red→green): `git add tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster && git commit -m "test(saf): failing two-node repro — leader-vs-oldest resync divergence wipes delivering node buffer (plan R2-02 T2)"`
|
||
|
||
### Task 3: Swap `SiteReplicationActor` to the shared oldest-Up predicate + Host wires the delivery-gate delegate
|
||
|
||
**Classification:** high-risk (resync authority semantics — the N1 fix)
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 8, 9, 10, 13
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs` (`DefaultIsActive` lines 190-198; ctor doc lines 56-61; `HandleSfBufferSnapshot` lines 420-444 belt-and-braces)
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs` (`RegisterSiteActorsAsync`: replication actor creation lines 775-779; provider resolve lines 878-882 hoisted)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs` (Task 2's red test goes green), `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs` (regression)
|
||
|
||
Depends on Tasks 1 (evaluator) and 2 (red repro). TDD note (mirrors round-1 Task 4's pattern): this task writes no *new* test — its failing test is Task 2's repro, plus the existing resync suite as regression.
|
||
|
||
1. Replace `DefaultIsActive` (`SiteReplicationActor.cs:190-198`) and its doc:
|
||
|
||
```csharp
|
||
/// <summary>
|
||
/// Repo-standard active-node check: this node is active when it is the OLDEST Up
|
||
/// member carrying the site role — the same oldest-Up semantics as the S&F delivery
|
||
/// gate (IClusterNodeProvider.SelfIsPrimary → ClusterActivityEvaluator → shared
|
||
/// ActiveNodeEvaluator). NEVER the cluster leader: leadership is lowest-address and
|
||
/// diverges from singleton/delivery placement permanently after the lower-address
|
||
/// node restarts — the divergence that made the delivering node wipe its own live
|
||
/// buffer via a wrong-direction resync (review 02 round 2, N1 Critical). Any other
|
||
/// state reports standby — safe-by-default.
|
||
/// </summary>
|
||
private bool DefaultIsActive() =>
|
||
Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(_cluster, _siteRole);
|
||
```
|
||
|
||
Update the ctor `isActiveOverride` doc (lines 56-61): the "swap point for plan 01's shared helper" sentence becomes "production wiring passes the Host's `IClusterNodeProvider.SelfIsPrimary` delegate (the same instance gating the S&F delivery sweep); null falls back to the shared oldest-Up evaluator."
|
||
2. Belt-and-braces in `HandleSfBufferSnapshot` (report's explicit suggestion): re-evaluate the predicate at *apply* time, inside the async continuation, so an active-flip between receive and apply can never run `ReplaceAllAsync` on a node that just became the delivering node:
|
||
|
||
```csharp
|
||
Task.Run(async () =>
|
||
{
|
||
// Belt-and-braces (N1): re-check at apply time. ReplaceAllAsync discards
|
||
// every in-flight row (StoreAndForwardStorage.cs "Never call on an active
|
||
// node"); a flip between message receipt and this point must abort.
|
||
if (SafeIsActive())
|
||
{
|
||
_logger.LogWarning(
|
||
"Discarding S&F buffer resync snapshot: this node became active before apply");
|
||
return;
|
||
}
|
||
await _sfStorage.ReplaceAllAsync(msg.Messages);
|
||
})
|
||
.ContinueWith(t =>
|
||
{
|
||
if (t.IsFaulted)
|
||
_logger.LogError(t.Exception, "Failed to apply S&F buffer resync snapshot");
|
||
});
|
||
```
|
||
|
||
3. Host wiring (`AkkaHostedService.RegisterSiteActorsAsync`): hoist the `clusterNodeProvider` resolve (currently at line 878, inside the S&F block) to just above the replication-actor creation at line 775, and pass the SAME delegate:
|
||
|
||
```csharp
|
||
// ONE active-node predicate instance governs the S&F delivery gate, the resync
|
||
// authority checks (SiteReplicationActor), and the heartbeat IsActive stamp
|
||
// (SiteCommunicationActor, wired below) — review 02 round 2, N1. Null in
|
||
// non-clustered test hosts: the actors fall back to the shared oldest-Up
|
||
// evaluator, never to a leader check.
|
||
var clusterNodeProvider = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.IClusterNodeProvider>();
|
||
Func<bool>? activeNodeCheck = clusterNodeProvider != null
|
||
? () => clusterNodeProvider.SelfIsPrimary
|
||
: null;
|
||
|
||
var replicationActor = _actorSystem!.ActorOf(
|
||
Props.Create(() => new SiteReplicationActor(
|
||
storage, sfStorage, replicationService, siteRole, replicationLogger,
|
||
deploymentConfigFetcher, activeNodeCheck, siteRuntimeOptionsValue, null)),
|
||
"site-replication");
|
||
```
|
||
|
||
Then at line ~878 delete the duplicate resolve and keep `storeAndForwardService.SetDeliveryGate(() => clusterNodeProvider.SelfIsPrimary);` guarded by the hoisted variable.
|
||
4. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.IntegrationTests --filter SfBufferResyncPredicateTests` — **Task 2's repro now PASSES** (the oldest node keeps `live-row` and seeds the joiner). Regression: `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter SiteReplicationActorTests` (the resync tests inject `isActiveOverride`, so they are predicate-agnostic — expect PASS) and `dotnet test tests/ZB.MOM.WW.ScadaBridge.Host.Tests --filter ClusterActivity`.
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs && git commit -m "fix(saf): resync authority uses the shared oldest-Up predicate + delivery-gate delegate; apply-time re-check guard (plan R2-02 T3)"`
|
||
|
||
### Task 4: Swap `SiteCommunicationActor.DefaultIsActiveCheck` + Host wiring + doc sync
|
||
|
||
**Classification:** small (predicate swap + wiring; divergence coverage is Tasks 1-3)
|
||
**Estimated implement time:** ~4 min
|
||
**Parallelizable with:** 8, 9, 10, 13, 15
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs` (`DefaultIsActiveCheck` lines 508-526; ctor doc lines 68-74)
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs` (`siteCommActor` creation lines 822-827 — pass `activeNodeCheck`)
|
||
- Modify: `docs/requirements/Component-Communication.md` (any "leader" wording on the heartbeat IsActive stamp), `docs/requirements/Component-StoreAndForward.md` line 83 ("repo-standard leader+Up active-node predicate" → oldest-Up; the fuller resync rewrite lands in Task 7)
|
||
|
||
Depends on Tasks 1 and 3 (shared `AkkaHostedService.cs` — serialized).
|
||
|
||
1. Replace `DefaultIsActiveCheck` (`SiteCommunicationActor.cs:517-526`):
|
||
|
||
```csharp
|
||
/// <summary>
|
||
/// Default active-node check used when no override is supplied: oldest-Up member
|
||
/// semantics via the shared <see cref="ClusterState.ActiveNodeEvaluator"/> — the
|
||
/// same predicate as the S&F delivery gate and the replication resync authority
|
||
/// (review 02 round 2, N1). Unscoped by role: a site cluster's members all carry
|
||
/// the site role, so role scoping is a no-op here; production wiring passes the
|
||
/// Host's role-scoped IClusterNodeProvider delegate anyway. Any other state
|
||
/// (still joining, leaving) reports standby — safe-by-default.
|
||
/// </summary>
|
||
private bool DefaultIsActiveCheck() =>
|
||
ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(Cluster.Get(Context.System));
|
||
```
|
||
|
||
Update the ctor doc (lines 68-74): "…<c>null</c> uses the shared oldest-Up evaluator (production wiring passes the Host's singleton-host delegate)…".
|
||
2. Host wiring (line 822-827) — pass the Task 3 delegate as the 4th ctor arg:
|
||
|
||
```csharp
|
||
var siteCommActor = _actorSystem.ActorOf(
|
||
Props.Create(() => new SiteCommunicationActor(
|
||
_nodeOptions.SiteId!,
|
||
_communicationOptions,
|
||
dmProxy,
|
||
activeNodeCheck)),
|
||
"site-communication");
|
||
```
|
||
|
||
3. Docs: grep both component docs for `leader` in the heartbeat/resync context and align to "oldest-Up member (singleton-host semantics, shared `ActiveNodeEvaluator`)". Do NOT rewrite the full resync paragraph yet (Task 7 owns it — avoid a double edit).
|
||
4. Build + regression: `dotnet build ZB.MOM.WW.ScadaBridge.slnx`, then `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter SiteCommunicationActor` (existing tests inject `isActiveCheck` stubs — expect PASS).
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs docs/requirements/Component-Communication.md docs/requirements/Component-StoreAndForward.md && git commit -m "fix(comm): heartbeat IsActive uses the shared oldest-Up predicate; Host wires one delegate everywhere (plan R2-02 T4)"`
|
||
|
||
### Task 5: Chunked resync protocol — additive messages, byte-budgeted chunker, active-side chunked answer
|
||
|
||
**Classification:** high-risk (cluster protocol; frame-size correctness — the N2 fix, part 1)
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 8, 9, 10, 13, 14, 15
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs` (new records; `HandleRequestSfBufferResync` lines 398-411; new chunker + constants near `MaxResyncRows` line 40)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs` (append), `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs` (create)
|
||
|
||
Depends on Task 3 (same file). **Message-contract rules (additive-only):** `SfBufferSnapshotChunk` and `SfBufferResyncAck` are NEW records; the legacy `SfBufferSnapshot` record AND its standby handler are **retained** so a rolling upgrade keeps working (old active → monolithic snapshot to new standby: handled; new active → chunks to old standby: dead-lettered, divergence self-heals at the next post-upgrade peer-track — same exposure as today's silently-dropped oversized frame, now bounded to the upgrade window and logged). These messages ride **intra-site Akka remoting, not ClusterClient**, so `ClusterClientContractLockTests` (PLAN-08 T9, `tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Messages/ClusterClientContractLockTests.cs`) is deliberately untouched — the wire pins land in a new `ResyncWireSerializationPinTests` following the `ReplicationWireSerializationPinTests` pattern instead.
|
||
|
||
**Sizing rationale:** Akka remoting's default `maximum-frame-size` is 128 000 bytes and `BuildHocon` does not raise it. Budget `MaxResyncChunkBytes = 64_000` estimated payload per chunk (≈50% headroom for the JSON envelope, type manifests, and non-payload columns) with a `MaxResyncChunkRows = 200` row cap. A single row whose payload alone exceeds the budget ships as its own chunk with a Warning (it may still exceed the frame — pathological, logged, no longer silent for everything else). The active node still materializes at most `MaxResyncRows` (10 000) rows in memory — unchanged from today and bounded; accepted (report: "reconsider" — the cap already bounds it).
|
||
|
||
1. Failing tests (append to `SiteReplicationActorTests.cs`; reuse `NewMessage`/`ResyncTestActor` helpers):
|
||
|
||
```csharp
|
||
// ── R2 T5: chunked resync answer ──
|
||
|
||
[Fact]
|
||
public void ChunkForRemoting_SplitsByByteBudget_PreservingOrderAndSequence()
|
||
{
|
||
var rows = Enumerable.Range(0, 10)
|
||
.Select(i => NewMessage($"m{i}", payloadJson: new string('x', 20_000)))
|
||
.ToList();
|
||
|
||
var chunks = SiteReplicationActor.ChunkForRemoting(rows, maxChunkBytes: 64_000, maxChunkRows: 200);
|
||
|
||
Assert.True(chunks.Count > 1); // 10 × 20 KB cannot ride one 64 KB chunk
|
||
Assert.Equal(rows.Select(r => r.Id), chunks.SelectMany(c => c).Select(r => r.Id)); // order preserved
|
||
Assert.All(chunks, c => Assert.True(
|
||
c.Sum(r => r.PayloadJson.Length) <= 64_000 || c.Count == 1)); // budget honored (oversized row isolated)
|
||
}
|
||
|
||
[Fact]
|
||
public void ChunkForRemoting_RowCapHonored_AndSingleOversizedRowIsolated()
|
||
{
|
||
var many = Enumerable.Range(0, 500).Select(i => NewMessage($"s{i}", payloadJson: "{}")).ToList();
|
||
Assert.All(SiteReplicationActor.ChunkForRemoting(many, 64_000, 200), c => Assert.True(c.Count <= 200));
|
||
|
||
var oversized = new List<StoreAndForwardMessage>
|
||
{ NewMessage("big", payloadJson: new string('y', 100_000)), NewMessage("small", payloadJson: "{}") };
|
||
var chunks = SiteReplicationActor.ChunkForRemoting(oversized, 64_000, 200);
|
||
Assert.Equal(2, chunks.Count); // the oversized row rides alone
|
||
}
|
||
|
||
[Fact]
|
||
public async Task ActiveNode_AnswersResyncRequest_WithSequencedChunks_SharingOneResyncId()
|
||
{
|
||
for (var i = 0; i < 3; i++)
|
||
await _sfStorage.EnqueueAsync(NewMessage($"c{i}", payloadJson: new string('z', 30_000)));
|
||
var actor = CreateResyncActor(isActive: () => true);
|
||
|
||
actor.Tell(new RequestSfBufferResync(), TestActor);
|
||
|
||
var first = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
|
||
var rest = Enumerable.Range(1, first.TotalChunks - 1)
|
||
.Select(_ => ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5)))
|
||
.Prepend(first)
|
||
.ToList();
|
||
|
||
Assert.True(first.TotalChunks > 1);
|
||
Assert.All(rest, c => Assert.Equal(first.ResyncId, c.ResyncId));
|
||
Assert.Equal(Enumerable.Range(1, first.TotalChunks), rest.Select(c => c.Sequence));
|
||
Assert.Equal(3, rest.Sum(c => c.Messages.Count));
|
||
}
|
||
```
|
||
|
||
(`NewMessage` may need an optional `payloadJson` parameter — additive to the existing helper. `CreateResyncActor` = the existing `ResyncTestActor` construction.)
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "ChunkForRemoting|WithSequencedChunks"` — expect FAIL (compile: chunker/records missing).
|
||
3. Implement in `SiteReplicationActor.cs`:
|
||
- Constants next to `MaxResyncRows` (line 40): `internal const int MaxResyncChunkBytes = 64_000;` and `internal const int MaxResyncChunkRows = 200;` with the sizing-rationale doc above (cite the 128 000-byte default frame and that `BuildHocon` sets no override).
|
||
- Pure chunker (internal static, unit-testable):
|
||
|
||
```csharp
|
||
/// <summary>
|
||
/// Splits a resync snapshot into chunks that fit Akka remoting's default
|
||
/// 128 000-byte frame (review 02 round 2, N2): rows accumulate until the estimated
|
||
/// payload budget or the row cap is hit. Estimation is payload-dominated
|
||
/// (payload_json length + 512 bytes fixed overhead per row); a single row whose
|
||
/// payload exceeds the budget ships alone (Warning at the call site). Order is
|
||
/// preserved (oldest-first, matching GetAllMessagesAsync).
|
||
/// </summary>
|
||
internal static List<List<StoreAndForwardMessage>> ChunkForRemoting(
|
||
IReadOnlyList<StoreAndForwardMessage> rows, int maxChunkBytes, int maxChunkRows)
|
||
{
|
||
var chunks = new List<List<StoreAndForwardMessage>>();
|
||
var current = new List<StoreAndForwardMessage>();
|
||
var currentBytes = 0;
|
||
foreach (var row in rows)
|
||
{
|
||
var estimate = (row.PayloadJson?.Length ?? 0) + 512;
|
||
if (current.Count > 0 && (currentBytes + estimate > maxChunkBytes || current.Count >= maxChunkRows))
|
||
{
|
||
chunks.Add(current);
|
||
current = new List<StoreAndForwardMessage>();
|
||
currentBytes = 0;
|
||
}
|
||
current.Add(row);
|
||
currentBytes += estimate;
|
||
}
|
||
if (current.Count > 0) chunks.Add(current);
|
||
return chunks;
|
||
}
|
||
```
|
||
|
||
- `HandleRequestSfBufferResync` (lines 398-411): keep the authority check; pipe the snapshot back to **Self** (chunking on the actor thread keeps `SendToPeer`/ack bookkeeping actor-safe), then send sequenced chunks to the requester:
|
||
|
||
```csharp
|
||
var replyTo = Sender;
|
||
_sfStorage.GetAllMessagesAsync(MaxResyncRows).PipeTo(
|
||
Self,
|
||
failure: ex => new Status.Failure(ex),
|
||
success: result => new SfResyncSnapshotLoaded(replyTo, result.Messages, result.Truncated));
|
||
```
|
||
|
||
with a new internal record + handler:
|
||
|
||
```csharp
|
||
/// <summary>Internal: the resync snapshot finished loading; chunk and send to the requester.</summary>
|
||
internal sealed record SfResyncSnapshotLoaded(
|
||
IActorRef ReplyTo, List<StoreAndForwardMessage> Messages, bool Truncated);
|
||
```
|
||
|
||
```csharp
|
||
private void HandleSfResyncSnapshotLoaded(SfResyncSnapshotLoaded msg)
|
||
{
|
||
var chunks = ChunkForRemoting(msg.Messages, MaxResyncChunkBytes, MaxResyncChunkRows);
|
||
if (chunks.Count == 0) chunks.Add(new List<StoreAndForwardMessage>()); // empty buffer still resyncs (clears the standby)
|
||
var resyncId = Guid.NewGuid().ToString("N");
|
||
for (var i = 0; i < chunks.Count; i++)
|
||
{
|
||
if (chunks[i].Count == 1 && (chunks[i][0].PayloadJson?.Length ?? 0) + 512 > MaxResyncChunkBytes)
|
||
_logger.LogWarning(
|
||
"Resync row {Id} alone exceeds the chunk budget ({Bytes}B payload); sending solo — it may exceed the remoting frame",
|
||
chunks[i][0].Id, chunks[i][0].PayloadJson?.Length ?? 0);
|
||
msg.ReplyTo.Tell(new SfBufferSnapshotChunk(resyncId, i + 1, chunks.Count, chunks[i], msg.Truncated), Self);
|
||
}
|
||
_logger.LogInformation(
|
||
"Answered S&F resync request with {Rows} row(s) in {Chunks} chunk(s), resyncId={ResyncId}",
|
||
msg.Messages.Count, chunks.Count, resyncId);
|
||
// Task 7 registers the pending-ack entry here.
|
||
}
|
||
```
|
||
|
||
- New public records at the bottom of the file (next to `SfBufferSnapshot`, which is KEPT verbatim for rolling compat):
|
||
|
||
```csharp
|
||
/// <summary>
|
||
/// Active→standby: one sequenced chunk of a full-buffer anti-entropy snapshot
|
||
/// (review 02 round 2, N2 — the monolithic <see cref="SfBufferSnapshot"/> exceeds Akka
|
||
/// remoting's default 128 000-byte frame for any realistic backlog). All chunks of one
|
||
/// resync share <paramref name="ResyncId"/>; <paramref name="Sequence"/> is 1-based up to
|
||
/// <paramref name="TotalChunks"/>. Additive message — the legacy monolithic snapshot
|
||
/// handler is retained for rolling upgrades. Crosses intra-site Akka remoting (NOT
|
||
/// ClusterClient — ClusterClientContractLockTests is intentionally not involved).
|
||
/// </summary>
|
||
public sealed record SfBufferSnapshotChunk(
|
||
string ResyncId, int Sequence, int TotalChunks,
|
||
List<StoreAndForwardMessage> Messages, bool Truncated);
|
||
|
||
/// <summary>
|
||
/// Standby→active: delivery confirmation — the standby assembled all chunks of
|
||
/// <paramref name="ResyncId"/> and applied them atomically (<paramref name="RowCount"/>
|
||
/// rows installed). Absence within the ack window is surfaced by the active node
|
||
/// (Warning + counter) — the silent-loss mode N2 flagged.
|
||
/// </summary>
|
||
public sealed record SfBufferResyncAck(string ResyncId, int RowCount);
|
||
```
|
||
|
||
- Register `Receive<SfResyncSnapshotLoaded>(HandleSfResyncSnapshotLoaded);` in the ctor next to the other resync receives (line 111-113).
|
||
4. Create `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs` following `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationWireSerializationPinTests.cs`: round-trip `SfBufferSnapshotChunk` (fully-populated `StoreAndForwardMessage` incl. `ExecutionId`/`ParentExecutionId`) and `SfBufferResyncAck` through `Sys.Serialization`, plus type-identity pins (`ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.SfBufferSnapshotChunk`, `...SfBufferResyncAck`) — a rename/move is an intra-site wire break during rolling upgrades.
|
||
5. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "ChunkForRemoting|WithSequencedChunks|ResyncWireSerializationPin"` — expect PASS. Then the full resync group: `--filter SiteReplicationActorTests`.
|
||
6. Commit: `git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests && git commit -m "fix(saf): chunk the resync snapshot to fit Akka remoting frames (additive protocol, byte-budgeted) (plan R2-02 T5)"`
|
||
|
||
### Task 6: Standby-side chunk assembly, atomic apply, ack + the N5 race comment
|
||
|
||
**Classification:** high-risk (cluster protocol; buffer-replace correctness — the N2 fix, part 2)
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 8, 9, 10, 13, 14, 15
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs` (new assembly state + `Receive<SfBufferSnapshotChunk>`; N5 doc comment on the apply path)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs` (append)
|
||
|
||
Depends on Task 5 (same file, chunk records).
|
||
|
||
1. Failing tests:
|
||
|
||
```csharp
|
||
[Fact]
|
||
public async Task StandbyNode_AssemblesChunks_AppliesOnce_AndAcks()
|
||
{
|
||
await _sfStorage.EnqueueAsync(NewMessage("stale"));
|
||
var actor = CreateResyncActor(isActive: () => false);
|
||
var resyncId = "r1";
|
||
|
||
actor.Tell(new SfBufferSnapshotChunk(resyncId, 1, 2,
|
||
new List<StoreAndForwardMessage> { NewMessage("f1") }, false), TestActor);
|
||
actor.Tell(new SfBufferSnapshotChunk(resyncId, 2, 2,
|
||
new List<StoreAndForwardMessage> { NewMessage("f2") }, false), TestActor);
|
||
|
||
var ack = ExpectMsg<SfBufferResyncAck>(TimeSpan.FromSeconds(5));
|
||
Assert.Equal(resyncId, ack.ResyncId);
|
||
Assert.Equal(2, ack.RowCount);
|
||
await AwaitAssertAsync(async () =>
|
||
{
|
||
Assert.Null(await _sfStorage.GetMessageByIdAsync("stale")); // replaced wholesale
|
||
Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f1"));
|
||
Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f2"));
|
||
});
|
||
}
|
||
|
||
[Fact]
|
||
public async Task StandbyNode_NewResyncId_DiscardsStalePartialAssembly()
|
||
{
|
||
var actor = CreateResyncActor(isActive: () => false);
|
||
actor.Tell(new SfBufferSnapshotChunk("old", 1, 2,
|
||
new List<StoreAndForwardMessage> { NewMessage("orphan") }, false), TestActor);
|
||
actor.Tell(new SfBufferSnapshotChunk("new", 1, 1,
|
||
new List<StoreAndForwardMessage> { NewMessage("fresh") }, false), TestActor);
|
||
|
||
ExpectMsg<SfBufferResyncAck>(TimeSpan.FromSeconds(5)); // "new" completed
|
||
await AwaitAssertAsync(async () =>
|
||
{
|
||
Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh"));
|
||
Assert.Null(await _sfStorage.GetMessageByIdAsync("orphan")); // stale partial never applied
|
||
});
|
||
}
|
||
|
||
[Fact]
|
||
public void ActiveNode_IgnoresChunks_NeverAcks()
|
||
{
|
||
var actor = CreateResyncActor(isActive: () => true);
|
||
actor.Tell(new SfBufferSnapshotChunk("r", 1, 1,
|
||
new List<StoreAndForwardMessage> { NewMessage("x") }, false), TestActor);
|
||
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||
}
|
||
```
|
||
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "AssemblesChunks|DiscardsStalePartialAssembly|IgnoresChunks_NeverAcks"` — expect FAIL (no chunk handler; chunks dead-letter).
|
||
3. Implement — actor gains `IWithTimers` (`public ITimerScheduler Timers { get; set; } = null!;`) plus assembly state and the handler:
|
||
|
||
```csharp
|
||
// ── Chunked-resync assembly (standby side; actor-thread only) ──
|
||
private string? _assemblingResyncId;
|
||
private int _assemblingTotalChunks;
|
||
private bool _assemblingTruncated;
|
||
private readonly Dictionary<int, List<StoreAndForwardMessage>> _assemblingChunks = new();
|
||
private const string ResyncAssemblyTimerKey = "sf-resync-assembly-timeout";
|
||
|
||
/// <summary>How long a partial chunk assembly may wait for its missing chunks before
|
||
/// being discarded (a lost chunk = lost resync; the next peer-track retries). Ctor
|
||
/// test seam; production default 30 s.</summary>
|
||
private readonly TimeSpan _resyncAssemblyTimeout;
|
||
```
|
||
|
||
(ctor gains `TimeSpan? resyncAssemblyTimeout = null` → `_resyncAssemblyTimeout = resyncAssemblyTimeout ?? TimeSpan.FromSeconds(30);` — additive optional parameter, existing callers unchanged.)
|
||
|
||
```csharp
|
||
private void HandleSfBufferSnapshotChunk(SfBufferSnapshotChunk msg)
|
||
{
|
||
if (SafeIsActive())
|
||
{
|
||
_logger.LogDebug("Ignoring S&F resync chunk — this node is active");
|
||
return;
|
||
}
|
||
|
||
if (_assemblingResyncId != msg.ResyncId)
|
||
{
|
||
if (_assemblingResyncId != null)
|
||
_logger.LogWarning(
|
||
"Discarding partial S&F resync assembly {Old} ({Have}/{Want} chunks): a new resync {New} superseded it",
|
||
_assemblingResyncId, _assemblingChunks.Count, _assemblingTotalChunks, msg.ResyncId);
|
||
_assemblingResyncId = msg.ResyncId;
|
||
_assemblingTotalChunks = msg.TotalChunks;
|
||
_assemblingTruncated = msg.Truncated;
|
||
_assemblingChunks.Clear();
|
||
}
|
||
|
||
_assemblingChunks[msg.Sequence] = msg.Messages;
|
||
Timers.StartSingleTimer(ResyncAssemblyTimerKey, new ResyncAssemblyTimedOut(msg.ResyncId), _resyncAssemblyTimeout);
|
||
|
||
if (_assemblingChunks.Count < _assemblingTotalChunks)
|
||
return;
|
||
|
||
// Complete: assemble in sequence order and apply atomically.
|
||
var assembled = Enumerable.Range(1, _assemblingTotalChunks)
|
||
.SelectMany(seq => _assemblingChunks[seq])
|
||
.ToList();
|
||
var resyncId = _assemblingResyncId!;
|
||
var truncated = _assemblingTruncated;
|
||
_assemblingResyncId = null;
|
||
_assemblingChunks.Clear();
|
||
Timers.Cancel(ResyncAssemblyTimerKey);
|
||
|
||
if (truncated)
|
||
_logger.LogWarning(
|
||
"S&F buffer resync snapshot truncated at {Cap} rows; divergence beyond the cap drains naturally",
|
||
MaxResyncRows);
|
||
_logger.LogInformation(
|
||
"Applying chunked S&F resync {ResyncId} ({Count} rows), replacing local buffer", resyncId, assembled.Count);
|
||
|
||
// KNOWN, ACCEPTED race (review 02 round 2, N5 — do NOT "fix" this into something
|
||
// worse): a replicated Remove sent after the active node read its snapshot but
|
||
// before the snapshot's chunks is ordered BEFORE them on the wire (same
|
||
// sender/receiver pair), so this apply can re-add the removed row → an orphan
|
||
// Pending row that a later failover re-delivers ONCE. Bounded, self-correcting at
|
||
// the next resync, and inherent to no-ack replication; a delivered-side dedup or
|
||
// op-sequencing scheme would cost far more than one rare duplicate.
|
||
var replyTo = Sender;
|
||
Task.Run(async () =>
|
||
{
|
||
if (SafeIsActive()) // belt-and-braces, mirrors the monolithic path (T3)
|
||
{
|
||
_logger.LogWarning("Discarding chunked S&F resync {ResyncId}: node became active before apply", resyncId);
|
||
return;
|
||
}
|
||
await _sfStorage.ReplaceAllAsync(assembled);
|
||
replyTo.Tell(new SfBufferResyncAck(resyncId, assembled.Count));
|
||
})
|
||
.ContinueWith(t =>
|
||
{
|
||
if (t.IsFaulted)
|
||
_logger.LogError(t.Exception, "Failed to apply chunked S&F resync {ResyncId}", resyncId);
|
||
});
|
||
}
|
||
|
||
private void HandleResyncAssemblyTimedOut(ResyncAssemblyTimedOut msg)
|
||
{
|
||
if (_assemblingResyncId != msg.ResyncId) return; // superseded already
|
||
_logger.LogWarning(
|
||
"S&F resync assembly {ResyncId} timed out with {Have}/{Want} chunks — discarding partial (a lost chunk; the next peer-track retries)",
|
||
msg.ResyncId, _assemblingChunks.Count, _assemblingTotalChunks);
|
||
ScadaBridgeTelemetry.RecordReplicationFailure();
|
||
_assemblingResyncId = null;
|
||
_assemblingChunks.Clear();
|
||
}
|
||
```
|
||
|
||
Plus `internal sealed record ResyncAssemblyTimedOut(string ResyncId);` and ctor registrations `Receive<SfBufferSnapshotChunk>(HandleSfBufferSnapshotChunk);` / `Receive<ResyncAssemblyTimedOut>(HandleResyncAssemblyTimedOut);` next to the legacy `Receive<SfBufferSnapshot>` (line 113 — which stays, for rolling compat). `Sender` inside `HandleSfBufferSnapshotChunk` is the active node's replication actor (chunks are Tell'd with `Self` as sender in T5), so the ack lands where Task 7 listens.
|
||
4. Run the filter — expect PASS. Full group: `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter SiteReplicationActorTests`, and re-run the Task 2 integration repro (`--filter SfBufferResyncPredicateTests`) — it must still pass with the exchange now chunked.
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs && git commit -m "fix(saf): standby assembles chunked resync, applies atomically with re-check, acks; N5 race documented (plan R2-02 T6)"`
|
||
|
||
### Task 7: Resync delivery confirmation on the active node + telemetry + doc rewrite
|
||
|
||
**Classification:** standard
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 8, 9, 10, 13, 15
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs` (ack tracking in `HandleSfResyncSnapshotLoaded` + `Receive<SfBufferResyncAck>`), `src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs` (two counters, next to `RecordReplicationFailure` at line 108)
|
||
- Modify: `docs/requirements/Component-StoreAndForward.md` (line 83 — full resync-paragraph rewrite)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs` (append)
|
||
|
||
Depends on Task 6 (same file). N2's "no telemetry counts it / nothing retries until the next peer-track" — the ack closes the silent-loss mode: a resync that was answered but never applied is now a Warning + counter instead of nothing.
|
||
|
||
1. Failing tests:
|
||
|
||
```csharp
|
||
[Fact]
|
||
public async Task ActiveNode_ReceivingAck_CountsResyncCompleted()
|
||
{
|
||
await _sfStorage.EnqueueAsync(NewMessage("m1"));
|
||
var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromSeconds(30));
|
||
actor.Tell(new RequestSfBufferResync(), TestActor);
|
||
var chunk = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
|
||
|
||
actor.Tell(new SfBufferResyncAck(chunk.ResyncId, 1), TestActor);
|
||
// Observable: the ack-timeout no longer fires (next test proves the inverse);
|
||
// completing must not warn. EventFilter pins it:
|
||
// (arrange the EventFilter *around* the ack Tell)
|
||
}
|
||
|
||
[Fact]
|
||
public async Task ActiveNode_MissingAck_WarnsAfterAckTimeout()
|
||
{
|
||
await _sfStorage.EnqueueAsync(NewMessage("m1"));
|
||
var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromMilliseconds(200));
|
||
|
||
await EventFilter.Warning(contains: "was never acknowledged").ExpectOneAsync(async () =>
|
||
{
|
||
actor.Tell(new RequestSfBufferResync(), TestActor);
|
||
ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
|
||
await Task.Delay(500); // let the ack timer fire
|
||
});
|
||
}
|
||
```
|
||
|
||
(`CreateResyncActor` gains an `ackTimeout` pass-through to the new ctor seam.)
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "CountsResyncCompleted|WarnsAfterAckTimeout"` — expect FAIL (no ack handling; ctor param missing).
|
||
3. Implement:
|
||
- `ScadaBridgeTelemetry` (alongside `RecordReplicationFailure`, line 108):
|
||
|
||
```csharp
|
||
private static readonly Counter<long> _sfResyncCompleted =
|
||
Meter.CreateCounter<long>("scadabridge.store_and_forward.resync.completed", unit: "1",
|
||
description: "S&F anti-entropy resyncs the standby acknowledged as applied");
|
||
/// <summary>Records one acknowledged (applied) S&F buffer resync.</summary>
|
||
public static void RecordSfResyncCompleted() => _sfResyncCompleted.Add(1);
|
||
|
||
private static readonly Counter<long> _sfResyncAckMissing =
|
||
Meter.CreateCounter<long>("scadabridge.store_and_forward.resync.ack_missing", unit: "1",
|
||
description: "S&F resyncs answered by the active node but never acknowledged by the standby within the ack window (lost chunks / dead peer)");
|
||
/// <summary>Records one resync whose ack window expired.</summary>
|
||
public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1);
|
||
```
|
||
|
||
- `SiteReplicationActor`: ctor seam `TimeSpan? resyncAckTimeout = null` → `_resyncAckTimeout` (default 60 s); in `HandleSfResyncSnapshotLoaded` after the chunk loop: `_pendingAckResyncId = resyncId; Timers.StartSingleTimer(ResyncAckTimerKey, new ResyncAckTimedOut(resyncId), _resyncAckTimeout);` plus:
|
||
|
||
```csharp
|
||
Receive<SfBufferResyncAck>(msg =>
|
||
{
|
||
if (msg.ResyncId == _pendingAckResyncId)
|
||
{
|
||
_pendingAckResyncId = null;
|
||
Timers.Cancel(ResyncAckTimerKey);
|
||
}
|
||
ScadaBridgeTelemetry.RecordSfResyncCompleted();
|
||
_logger.LogInformation("S&F resync {ResyncId} acknowledged by standby: {Rows} row(s) applied",
|
||
msg.ResyncId, msg.RowCount);
|
||
});
|
||
Receive<ResyncAckTimedOut>(msg =>
|
||
{
|
||
if (msg.ResyncId != _pendingAckResyncId) return;
|
||
_pendingAckResyncId = null;
|
||
ScadaBridgeTelemetry.RecordSfResyncAckMissing();
|
||
_logger.LogWarning(
|
||
"S&F resync {ResyncId} was never acknowledged within {Window} — snapshot chunks may have been lost (frame drop / dead peer); the next peer-track retries",
|
||
msg.ResyncId, _resyncAckTimeout);
|
||
});
|
||
```
|
||
|
||
(single-outstanding-resync bookkeeping is sufficient: a new request supersedes — overwrite `_pendingAckResyncId` and restart the timer.)
|
||
- Doc rewrite — `Component-StoreAndForward.md:83`, replace the resync paragraph: chunked `SfBufferSnapshotChunk` protocol (byte-budgeted for Akka remoting's 128 KB default frame), standby assembly + atomic `ReplaceAllAsync` + `SfBufferResyncAck` confirmation, ack-missing Warning + counters, the shared **oldest-Up** predicate on both roles (same predicate as the delivery gate — cite N1), the retained legacy monolithic handler for rolling upgrades, and the accepted N5 orphan-row race.
|
||
4. Run the filter + `--filter SiteReplicationActorTests` — expect PASS.
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs docs/requirements/Component-StoreAndForward.md tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs && git commit -m "feat(saf): resync ack confirmation + completed/ack-missing telemetry; doc resync rewrite (plan R2-02 T7)"`
|
||
|
||
### Task 8: Publish `_sweepTask` only when the sweep CAS is won (unclobber the shutdown drain)
|
||
|
||
**Classification:** high-risk (shutdown/concurrency semantics)
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 1, 2, 3, 4, 5, 6, 7, 9, 10, 13, 14, 15
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs` (timer callback line 396-400; `TriggerSweep` lines 664-668; `RetryPendingMessagesAsync` lines 674-744 split)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs` (append)
|
||
|
||
N3: both the timer callback and `TriggerSweep` unconditionally `Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync())`; while a sweep is in flight the new call no-ops via the `_retryInProgress` CAS and returns a **completed** task that overwrites the drain handle — `StopAsync` (line 422-431) then awaits the no-op and shuts down under the real sweep. Fix: the caller performs the CAS and publishes the task only on a win.
|
||
|
||
1. Failing tests (append; `CreateService(retryTimerInterval)` helper exists at line 53 of the test file):
|
||
|
||
```csharp
|
||
[Fact]
|
||
public async Task TriggerSweep_WhileSweepInFlight_DoesNotClobberTheDrainHandle()
|
||
{
|
||
var service = CreateService(retryTimerInterval: TimeSpan.FromHours(1));
|
||
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, async _ =>
|
||
{
|
||
entered.TrySetResult();
|
||
await release.Task;
|
||
return true;
|
||
});
|
||
await service.StartAsync();
|
||
try
|
||
{
|
||
await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}",
|
||
attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);
|
||
|
||
service.TriggerSweep(); // real sweep, blocked in the handler
|
||
await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
|
||
service.TriggerSweep(); // redundant kick — pre-fix clobbers _sweepTask
|
||
|
||
var handle = service.CurrentSweepTaskForTest;
|
||
Assert.NotNull(handle);
|
||
Assert.False(handle!.IsCompleted); // pre-fix: true (a completed no-op replaced the real sweep)
|
||
}
|
||
finally
|
||
{
|
||
release.TrySetResult();
|
||
await service.StopAsync();
|
||
}
|
||
}
|
||
|
||
[Fact]
|
||
public async Task StopAsync_WaitsForTheRealInFlightSweep_EvenAfterARedundantTrigger()
|
||
{
|
||
var service = CreateService(retryTimerInterval: TimeSpan.FromHours(1));
|
||
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, async _ =>
|
||
{
|
||
entered.TrySetResult();
|
||
await release.Task;
|
||
return true;
|
||
});
|
||
await service.StartAsync();
|
||
await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}",
|
||
attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);
|
||
|
||
service.TriggerSweep();
|
||
await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
|
||
service.TriggerSweep(); // the clobbering kick
|
||
|
||
var stop = service.StopAsync();
|
||
await Task.Delay(300);
|
||
Assert.False(stop.IsCompleted); // pre-fix: StopAsync already returned (awaited the no-op)
|
||
|
||
release.TrySetResult();
|
||
await stop.WaitAsync(TimeSpan.FromSeconds(5)); // drains the real sweep promptly once released
|
||
}
|
||
```
|
||
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "DoesNotClobberTheDrainHandle|WaitsForTheRealInFlightSweep"` — expect FAIL (`CurrentSweepTaskForTest` missing; then the pre-fix clobber assertions).
|
||
3. Implement in `StoreAndForwardService`:
|
||
|
||
```csharp
|
||
/// <summary>The current drain handle — test seam for the N3 clobber regression.</summary>
|
||
internal Task? CurrentSweepTaskForTest => Volatile.Read(ref _sweepTask);
|
||
|
||
/// <summary>
|
||
/// Starts a sweep IF none is in flight, publishing the new task into
|
||
/// <see cref="_sweepTask"/> only when this call wins the <see cref="_retryInProgress"/>
|
||
/// CAS. A kick that loses the CAS returns without touching <see cref="_sweepTask"/> —
|
||
/// pre-fix it overwrote the drain handle with an instantly-completed no-op, so
|
||
/// <see cref="StopAsync"/> proceeded with disposal under a still-running sweep
|
||
/// (review 02 round 2, N3) — defeating the drain exactly when a sweep outlives a tick.
|
||
/// </summary>
|
||
private void KickSweep()
|
||
{
|
||
if (Interlocked.CompareExchange(ref _retryInProgress, 1, 0) != 0)
|
||
return;
|
||
Volatile.Write(ref _sweepTask, RunSweepOwnedAsync());
|
||
}
|
||
```
|
||
|
||
- `RetryPendingMessagesAsync` keeps its signature for the existing direct-call tests, but becomes the CAS wrapper: `internal async Task RetryPendingMessagesAsync() { if (Interlocked.CompareExchange(ref _retryInProgress, 1, 0) != 0) return; await RunSweepOwnedAsync(); }` — and the existing body (lines 680-743, including the delivery gate, lanes, catch, and the `finally { Interlocked.Exchange(ref _retryInProgress, 0); }`) moves verbatim into `private async Task RunSweepOwnedAsync()` with the entry CAS removed (ownership is the caller's).
|
||
- Timer callback (line 397): `_ => KickSweep()`.
|
||
- `TriggerSweep` (line 664-668): `if (_retryTimer == null) return; KickSweep();` (doc comment updated: "publishes into `_sweepTask` only when the CAS is won — see `KickSweep`").
|
||
4. Run the filter, then the whole project's sweep group: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "RetrySweep_|Enqueue_DeferToSweep|StopAsync"` — expect PASS (existing sweep/defer tests unchanged: `RetryPendingMessagesAsync` still self-CASes for direct calls).
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs && git commit -m "fix(saf): publish _sweepTask only on CAS win so StopAsync drains the real in-flight sweep (plan R2-02 T8)"`
|
||
|
||
### Task 9: Eager validation for `SweepBatchLimit` / `SweepTargetParallelism`
|
||
|
||
**Classification:** trivial
|
||
**Estimated implement time:** ~3 min
|
||
**Parallelizable with:** 1, 2, 3, 4, 5, 6, 7, 8, 10, 11, 13, 14, 15
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs` (lines 17-34)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs` (append)
|
||
|
||
N4: a negative `SweepBatchLimit` silently means "unlimited (legacy)" (`StoreAndForwardStorage.cs:393` checks `> 0`); a non-positive `SweepTargetParallelism` is silently clamped to 1 (`StoreAndForwardService.cs:715`). Both get fail-fast messages per the §1.5 eager-validation convention (PLAN-08).
|
||
|
||
1. Failing tests (append, following the file's existing valid/invalid pattern):
|
||
|
||
```csharp
|
||
[Theory]
|
||
[InlineData(-1)]
|
||
[InlineData(-500)]
|
||
public void Validate_NegativeSweepBatchLimit_Fails(int limit)
|
||
{
|
||
var options = ValidOptions();
|
||
options.SweepBatchLimit = limit;
|
||
var result = Validate(options);
|
||
Assert.True(result.Failed);
|
||
Assert.Contains("SweepBatchLimit", result.FailureMessage);
|
||
}
|
||
|
||
[Fact]
|
||
public void Validate_ZeroSweepBatchLimit_IsValidUnlimitedLegacy()
|
||
{
|
||
var options = ValidOptions();
|
||
options.SweepBatchLimit = 0;
|
||
Assert.True(Validate(options).Succeeded);
|
||
}
|
||
|
||
[Theory]
|
||
[InlineData(0)]
|
||
[InlineData(-4)]
|
||
public void Validate_NonPositiveSweepTargetParallelism_Fails(int parallelism)
|
||
{
|
||
var options = ValidOptions();
|
||
options.SweepTargetParallelism = parallelism;
|
||
var result = Validate(options);
|
||
Assert.True(result.Failed);
|
||
Assert.Contains("SweepTargetParallelism", result.FailureMessage);
|
||
}
|
||
```
|
||
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "SweepBatchLimit_Fails|ZeroSweepBatchLimit|SweepTargetParallelism_Fails"` — expect FAIL (validator passes everything).
|
||
3. Implement (append inside `Validate`, after the `DefaultMaxRetries` rule at line 31-33):
|
||
|
||
```csharp
|
||
builder.RequireThat(options.SweepBatchLimit >= 0,
|
||
$"ScadaBridge:StoreAndForward:SweepBatchLimit must be >= 0 " +
|
||
$"(was {options.SweepBatchLimit}); it bounds due rows per retry sweep — 0 means unlimited (legacy).");
|
||
|
||
builder.RequireThat(options.SweepTargetParallelism >= 1,
|
||
$"ScadaBridge:StoreAndForward:SweepTargetParallelism must be >= 1 " +
|
||
$"(was {options.SweepTargetParallelism}); it caps concurrent (category,target) sweep lanes — 1 means serial.");
|
||
```
|
||
|
||
4. Run the filter + `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter StoreAndForwardOptionsValidator` — expect PASS.
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs && git commit -m "chore(saf): eagerly validate SweepBatchLimit/SweepTargetParallelism (plan R2-02 T9)"`
|
||
|
||
### Task 10: Coalesce live-alarm delta publishes (bound the per-circuit fan-out)
|
||
|
||
**Classification:** standard (perf; correctness unchanged — last write wins). **MUTEX with PLAN-R2-07's sticky-IsLive task — see the rule at the top; do not start while R2-07's live-alarm slice is in flight.**
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 1, 2, 3, 4, 5, 6, 7, 8, 9, 13, 14
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs` (`HandleLiveDelta` lines 347-368; `OnSeedCompleted` line 303; ctor + new timer), `src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs` (after line 124), `src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs` (after line 87), `src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs` (`Props.Create` lines 228-236 — pass the new option)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs` (append), `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs` (append)
|
||
|
||
N6: `HandleLiveDelta` → `Publish()` materializes `_cache.Values.ToList()` **per delta** and `OnPublish` fans out to every viewer circuit per delta — O(cacheSize) copies and a render per circuit per transition during an alarm storm. Fix: dirty-flag + single-shot coalescing timer (default 250 ms; `TimeSpan.Zero` restores per-delta publish). Seed/reconcile completion still publishes immediately (already batched).
|
||
|
||
1. Failing tests (append; reuse `SeedStub`/`PublishSink`/`MockSiteAlarmStreamClient` and the existing `CreateAggregator`-style construction):
|
||
|
||
```csharp
|
||
[Fact]
|
||
public void DeltaBurst_IsCoalesced_IntoFewPublishes_ContainingEveryRow()
|
||
{
|
||
var (actor, seed, sink, client) = CreateAggregator(publishCoalesce: TimeSpan.FromMilliseconds(150));
|
||
seed.CompleteNext(); // empty initial seed
|
||
AwaitAssert(() => Assert.Equal(1, sink.Count)); // seed publish, immediate
|
||
|
||
var sub = client.Subs.Single();
|
||
for (var i = 0; i < 50; i++)
|
||
sub.OnAlarm(Alarm($"A{i}", DateTimeOffset.UtcNow)); // 50 distinct keys, one burst
|
||
|
||
AwaitAssert(() =>
|
||
{
|
||
Assert.Equal(50, sink.Latest!.Count); // nothing lost
|
||
Assert.InRange(sink.Count, 2, 4); // pre-fix: 51 publishes (1 seed + 1 per delta)
|
||
});
|
||
}
|
||
|
||
[Fact]
|
||
public void ZeroCoalesce_PreservesLegacyPerDeltaPublish()
|
||
{
|
||
var (actor, seed, sink, client) = CreateAggregator(publishCoalesce: TimeSpan.Zero);
|
||
seed.CompleteNext();
|
||
AwaitAssert(() => Assert.Equal(1, sink.Count));
|
||
|
||
var sub = client.Subs.Single();
|
||
sub.OnAlarm(Alarm("A1", DateTimeOffset.UtcNow));
|
||
sub.OnAlarm(Alarm("A2", DateTimeOffset.UtcNow));
|
||
AwaitAssert(() => Assert.Equal(3, sink.Count)); // 1 seed + 1 per delta
|
||
}
|
||
```
|
||
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "DeltaBurst_IsCoalesced|ZeroCoalesce_Preserves"` — expect FAIL (ctor lacks the parameter; then 51 publishes).
|
||
3. Implement:
|
||
- `CommunicationOptions` (after `LiveAlarmCacheMaxSubscribersPerSite`, line 124):
|
||
|
||
```csharp
|
||
/// <summary>
|
||
/// Publish-coalescing window for live alarm deltas: an applied delta marks the cache
|
||
/// dirty and one publish (fresh snapshot + per-viewer onChanged fan-out) fires after
|
||
/// this window, batching an alarm storm into ~4 publishes/second instead of one per
|
||
/// transition (review 02 round 2, N6). Zero = publish per delta (legacy). Seed and
|
||
/// reconcile publishes are always immediate. Default 250 ms.
|
||
/// </summary>
|
||
public TimeSpan LiveAlarmCachePublishCoalesce { get; set; } = TimeSpan.FromMilliseconds(250);
|
||
```
|
||
|
||
- `CommunicationOptionsValidator` (after line 87): `builder.RequireThat(options.LiveAlarmCachePublishCoalesce >= TimeSpan.Zero, $"Communication:LiveAlarmCachePublishCoalesce must be zero or a positive duration (was {options.LiveAlarmCachePublishCoalesce}).");` — plus a validator test in `CommunicationOptionsValidatorTests` (negative fails, zero succeeds).
|
||
- `SiteAlarmAggregatorActor`: ctor gains `TimeSpan publishCoalesce` (expression trees reject optional args — every `Props.Create`/test call site passes it explicitly); fields `private readonly TimeSpan _publishCoalesce; private bool _publishPending;` and `private const string PublishTimerKey = "alarm-publish-coalesce";`; `HandleLiveDelta`'s pass-through branch (line 366-367) becomes `if (ApplyDelta(delta, requireStrictlyNewer: false)) SchedulePublish();` with:
|
||
|
||
```csharp
|
||
/// <summary>
|
||
/// Coalesced publish: with a positive window, the first dirtying delta arms a
|
||
/// single-shot timer and further deltas ride the same tick — one snapshot copy and
|
||
/// one viewer fan-out per window instead of per transition (N6). Zero = legacy
|
||
/// immediate publish. Last write wins, so batching never changes final state.
|
||
/// </summary>
|
||
private void SchedulePublish()
|
||
{
|
||
if (_publishCoalesce <= TimeSpan.Zero) { Publish(); return; }
|
||
if (_publishPending) return;
|
||
_publishPending = true;
|
||
Timers.StartSingleTimer(PublishTimerKey, new PublishCoalesced(), _publishCoalesce);
|
||
}
|
||
```
|
||
|
||
ctor registration `Receive<PublishCoalesced>(_ => { _publishPending = false; if (!_stopped) Publish(); });` and record `internal sealed record PublishCoalesced;`. In `OnSeedCompleted` (and `OnSeedFailed`'s publish branch), before the immediate `Publish()`: `Timers.Cancel(PublishTimerKey); _publishPending = false;` (the seed snapshot already carries the buffered deltas).
|
||
- `SiteAlarmLiveCacheService` Props (line 228-236): append `_options.LiveAlarmCachePublishCoalesce` as the new last argument.
|
||
4. Run the filter, then the aggregator + service groups: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "SiteAlarmAggregatorActor|SiteAlarmLiveCache|CommunicationOptionsValidator"` — fix any existing per-delta-publish assertions by constructing with `TimeSpan.Zero` (legacy behavior preserved for them).
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication tests/ZB.MOM.WW.ScadaBridge.Communication.Tests && git commit -m "perf(comm): coalesce live-alarm delta publishes (250ms window, 0 = legacy) (plan R2-02 T10)"`
|
||
|
||
### Task 11: Aggregator lifecycle — queued re-seed after reconnect + stream-generation stamp
|
||
|
||
**Classification:** standard
|
||
**Estimated implement time:** ~5 min
|
||
**Parallelizable with:** 1, 2, 5, 6, 7, 8, 9, 13, 14
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs` (`StartFanout` lines 245-274; `OnSeedCompleted`/`OnSeedFailed`; `OpenGrpcStream` lines 407-433; `GrpcAlarmStreamError` record line 524 + handler lines 164-168)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs` (append)
|
||
|
||
Depends on Task 10 (same files). N7.1: the post-failover `StartFanout(isInitial: false)` (line 472) no-ops if a reconcile fan-out is in flight — deltas missed between stream death and that snapshot's read-time stay stale up to 60 s despite the "never silently serve stale" comment. N7.2: a late error raced out of the *previous* cancelled stream is indistinguishable from a current-stream failure and burns retry budget / double-flips.
|
||
|
||
1. Failing tests:
|
||
|
||
```csharp
|
||
[Fact]
|
||
public void ReseedRequestedDuringInFlightFanout_IsQueued_NotDropped()
|
||
{
|
||
var (actor, seed, sink, client) = CreateAggregator(publishCoalesce: TimeSpan.Zero);
|
||
seed.CompleteNext(); // initial seed done (CallCount 1)
|
||
AwaitAssert(() => Assert.Equal(1, sink.Count));
|
||
|
||
actor.Tell(new RunReconcile()); // reconcile fan-out now in flight (CallCount 2)
|
||
AwaitAssert(() => Assert.Equal(2, seed.CallCount));
|
||
|
||
// Stream error while the reconcile is in flight → the failover re-seed must not
|
||
// be silently skipped. Pre-fix: StartFanout no-ops and CallCount stays 2 until
|
||
// the next 60s reconcile tick.
|
||
client.Subs.Last().OnError(new Exception("stream fault"));
|
||
|
||
seed.CompleteNext(); // finish the in-flight reconcile
|
||
AwaitAssert(() => Assert.Equal(3, seed.CallCount)); // queued re-seed ran immediately after
|
||
}
|
||
|
||
[Fact]
|
||
public void LateErrorFromAPreviousStreamGeneration_IsIgnored()
|
||
{
|
||
var (actor, seed, sink, client) = CreateAggregator(publishCoalesce: TimeSpan.Zero);
|
||
seed.CompleteNext();
|
||
|
||
var firstSub = client.Subs.Single();
|
||
firstSub.OnError(new Exception("real fault")); // gen 1 error → flip, reconnect
|
||
AwaitAssert(() => Assert.Equal(2, client.Subs.Count)); // gen 2 stream open
|
||
|
||
firstSub.OnError(new Exception("late zombie fault")); // stale gen-1 error races in
|
||
// Pre-fix this burns retry budget and opens a THIRD stream; post-fix it is ignored.
|
||
Task.Delay(400).Wait();
|
||
Assert.Equal(2, client.Subs.Count);
|
||
}
|
||
```
|
||
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "IsQueued_NotDropped|PreviousStreamGeneration_IsIgnored"` — expect FAIL.
|
||
3. Implement:
|
||
- Queued re-seed: field `private bool _reseedQueued;`. `StartFanout`: `if (_fanoutInFlight) { if (!isInitial) _reseedQueued = true; return; }` (comment: a failover re-seed requested mid-fan-out must run right after — the in-flight snapshot's read-time predates the stream death, N7.1). In `OnSeedCompleted` and `OnSeedFailed`, after `_fanoutInFlight = false;` (and after the flush/publish): `if (_reseedQueued) { _reseedQueued = false; StartFanout(isInitial: false); }`.
|
||
- Generation stamp: field `private int _streamGeneration;`. `OpenGrpcStream`: `var generation = ++_streamGeneration;` and the error callback becomes `ex => self.Tell(new GrpcAlarmStreamError(ex, generation))`; record becomes `internal sealed record GrpcAlarmStreamError(Exception Exception, int Generation);`; the handler (line 164-168) prepends: `if (msg.Generation != _streamGeneration) { _log.Debug("Ignoring stale gRPC error from stream generation {0} (current {1})", msg.Generation, _streamGeneration); return; }` (comment: the `RpcException(Cancelled)` filter at `SiteStreamGrpcClient.cs:256-258` covers the normal path; a genuine socket fault can beat the cancel — N7.2).
|
||
4. Run the filter + `--filter SiteAlarmAggregatorActor` — expect PASS.
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs && git commit -m "fix(comm): queue failover re-seed behind in-flight fan-out; stamp stream generation on gRPC errors (plan R2-02 T11)"`
|
||
|
||
### Task 12: Instance-injected `ReconnectDelay`/`StabilityWindow` (kill the process-global test seams)
|
||
|
||
**Classification:** small (mechanical; N7.3)
|
||
**Estimated implement time:** ~4 min
|
||
**Parallelizable with:** 1, 2, 5, 6, 7, 8, 9, 13, 14
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs` (lines 61-69 statics → ctor-injected instance fields), `src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs` (`Props.Create` — pass production defaults explicitly)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs` (ctor at lines 27-28 — construct with explicit values instead of mutating statics)
|
||
|
||
Depends on Task 11 (same files). The `internal static` mutable `ReconnectDelay`/`StabilityWindow` are process-global state parallel test classes can trample; every other tuning knob in this file is ctor-injected.
|
||
|
||
1. This is a refactor with existing coverage: the aggregator suite exercises both knobs (fast reconnect at 50 ms, long stability at 30 s). No new test — the "failing test" is the compile break that proves the statics are gone.
|
||
2. Implement: delete the static properties (lines 61-69); ctor gains `TimeSpan reconnectDelay, TimeSpan stabilityWindow` (explicit — expression trees reject optional args) stored in `private readonly TimeSpan _reconnectDelay; private readonly TimeSpan _stabilityWindow;`; replace the three usages (`OpenGrpcStream` line 419 stability timer, `HandleGrpcError` line 477 reconnect timer). `SiteAlarmLiveCacheService.Props.Create` passes `TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60)` — annotated as the former static defaults. Test harness: remove the ctor static mutations (lines 27-28) and pass `TimeSpan.FromMilliseconds(50), TimeSpan.FromSeconds(30)` through its `CreateAggregator` helper.
|
||
3. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "SiteAlarmAggregatorActor|SiteAlarmLiveCache"` — expect PASS (identical timings, now instance-scoped).
|
||
4. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs && git commit -m "chore(comm): ctor-inject aggregator reconnect/stability tuning, remove process-global statics (plan R2-02 T12)"`
|
||
|
||
### Task 13: Production caller for `RemoveSiteAsync` — dispose a deleted site's gRPC channels
|
||
|
||
**Classification:** small
|
||
**Estimated implement time:** ~4 min
|
||
**Parallelizable with:** 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs` (`HandleDeleteSite`, lines 1574-1588)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs` (append)
|
||
|
||
N8 first half: `SiteStreamGrpcClientFactory.RemoveSiteAsync` (`SiteStreamGrpcClientFactory.cs:95-102`) is documented as "call it when a site record is deleted" and has **no production caller** — a deleted site's channels persist until process restart. `ManagementActor.HandleDeleteSite` is the single site-deletion path (the CLI and Central UI both dispatch `DeleteSiteCommand` through it).
|
||
|
||
1. Failing test (harness: `ManagementActorTests` DI `ServiceCollection` + `Envelope(cmd, Roles.Administrator)` pattern):
|
||
|
||
```csharp
|
||
[Fact]
|
||
public async Task DeleteSite_DisposesTheSitesCachedGrpcChannels()
|
||
{
|
||
var siteRepo = Substitute.For<ISiteRepository>();
|
||
siteRepo.GetSiteByIdAsync(7).Returns(new Site
|
||
{ Id = 7, SiteIdentifier = "site-7", Name = "Seven" });
|
||
siteRepo.GetInstancesBySiteIdAsync(7).Returns(new List<Instance>());
|
||
_services.AddScoped(_ => siteRepo);
|
||
|
||
var factory = new TrackingGrpcFactory(); // subclass overriding CreateClient, as in SiteStreamGrpcClientFactoryTests
|
||
var cached = (TrackingClient)factory.GetOrCreate("site-7", "http://node-a:8083");
|
||
_services.AddSingleton<SiteStreamGrpcClientFactory>(factory);
|
||
|
||
var actor = CreateActor();
|
||
actor.Tell(Envelope(new DeleteSiteCommand(7), Roles.Administrator));
|
||
ExpectMsg<ManagementResponse>(r => Assert.True(r.Success));
|
||
|
||
await AwaitAssertAsync(() =>
|
||
{
|
||
Assert.True(cached.Disposed);
|
||
Assert.Null(factory.TryGet("site-7", "http://node-a:8083"));
|
||
});
|
||
}
|
||
```
|
||
|
||
(Copy the `TrackingGrpcFactory`/`TrackingClient` doubles from `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryTests.cs` — or reference them if visible.)
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests --filter DeleteSite_DisposesTheSitesCachedGrpcChannels` — expect FAIL (channels untouched).
|
||
3. Implement in `HandleDeleteSite` (after `commService?.RefreshSiteAddresses();`, line 1585):
|
||
|
||
```csharp
|
||
// Dispose the deleted site's cached gRPC channels — RemoveSiteAsync is the
|
||
// factory's designed site-deletion disposal path and previously had no
|
||
// production caller (arch review 02 round 2, N8): a deleted site's channels
|
||
// (both node endpoints) otherwise persist until process restart. Null-safe:
|
||
// test/composition roots without the factory skip it. Any live-alarm
|
||
// aggregator for the site reconciles to an empty snapshot and is reaped by
|
||
// the viewer linger (documented acceptance — see Component-Communication.md).
|
||
var grpcFactory = sp.GetService<SiteStreamGrpcClientFactory>();
|
||
if (grpcFactory is not null && site is not null)
|
||
await grpcFactory.RemoveSiteAsync(site.SiteIdentifier);
|
||
```
|
||
|
||
(`using ZB.MOM.WW.ScadaBridge.Communication.Grpc;` — ManagementService already references Communication.)
|
||
4. Run the filter + `dotnet test tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests --filter ManagementActorTests` — expect PASS.
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs && git commit -m "fix(comm): site delete disposes the site's cached gRPC channels via RemoveSiteAsync (plan R2-02 T13)"`
|
||
|
||
### Task 14: Documented acceptance — standby-node aggregators + deleted-site viewer behavior
|
||
|
||
**Classification:** trivial (docs only)
|
||
**Estimated implement time:** ~3 min
|
||
**Parallelizable with:** 1, 2, 3, 5, 6, 7, 8, 9, 10, 11, 12, 13, 15
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs` (doc lines 12-17), `docs/requirements/Component-Communication.md` (live-alarm-stream section)
|
||
|
||
Depends on Task 4 (shared `Component-Communication.md` — serialized). N8 second half: `SetActorSystem` is wired on every central node (`AkkaHostedService.cs:429-430`), so browsing the standby directly (port 9002, documented for diagnostics) starts a second, fully-functional aggregator + gRPC stream. The report calls it "harmless (read-only) but contradicting the `[PERM]` doc claim" — the fix is honest documentation, not gating (gating `SetActorSystem` behind the active check would break the diagnostic-browse path and buy nothing: the aggregator is read-only and linger-reaped).
|
||
|
||
1. `ISiteAlarmLiveCache.cs` doc (lines 12-17): replace "…lives only on the active central node…" with: "…purely in-memory, per-node, no EF entity/table/migration. In routine operation only the active node (behind Traefik) hosts viewers, so only it runs aggregators; browsing the standby node directly (diagnostic ports) starts an independent read-only aggregator there — accepted: bounded (one stream/site, viewer-capped), read-only, and torn down by the viewer linger. On a NodeA↔NodeB failover the new active node re-seeds from scratch."
|
||
2. `Component-Communication.md`, live-alarm-stream section — add an explicit "Accepted limitations (arch review 02 round 2, N8)" note covering (a) the standby-browse aggregator above, and (b) a site deleted while an Alarm Summary viewer is open: the viewer's aggregator reconciles to an empty snapshot until its viewers leave, then the linger stop reaps it; the site's gRPC channels are disposed at deletion (Task 13).
|
||
3. Verify: `dotnet build ZB.MOM.WW.ScadaBridge.slnx` (doc-comment change compiles) and grep — `grep -rn "lives only on the active central node" src/` returns nothing.
|
||
4. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs docs/requirements/Component-Communication.md && git commit -m "docs(comm): accept + document standby aggregators and deleted-site viewer behavior (plan R2-02 T14)"`
|
||
|
||
### Task 15: Single-endpoint sites can go live
|
||
|
||
**Classification:** small. **MUTEX with PLAN-R2-07 (same `SiteAlarmLiveCacheService.cs`) — see the rule at the top.**
|
||
**Estimated implement time:** ~4 min
|
||
**Parallelizable with:** 2, 4, 5, 6, 7, 8, 9, 13, 14
|
||
**Files:**
|
||
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs` (`ResolveSiteAsync`, lines 294-302)
|
||
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs` (append)
|
||
|
||
Depends on Task 12 (file-serialized chain 10 → 12 → 15 on the live-alarm files). N9: `ResolveSiteAsync` requires **both** `GrpcNodeAAddress` and `GrpcNodeBAddress`; a single-node site silently never starts an aggregator and polls forever. The aggregator works fine flipping between one address — accept a single endpoint by reusing it for both slots (the NodeA↔NodeB flip degenerates to a reconnect against the same node).
|
||
|
||
1. Failing test (follow the file's `CreateService(...)` harness — it substitutes `ISiteRepository` and injects a `CountingFactory`):
|
||
|
||
```csharp
|
||
[Fact]
|
||
public async Task SiteWithOnlyNodeAEndpoint_StartsAnAggregator_AgainstThatEndpoint()
|
||
{
|
||
var service = CreateService(TimeSpan.FromSeconds(30), out var factory,
|
||
site: new Site
|
||
{
|
||
Id = 3, SiteIdentifier = "site-3",
|
||
GrpcNodeAAddress = "http://only-node:8083",
|
||
GrpcNodeBAddress = null, // single-endpoint site — pre-fix: never goes live
|
||
});
|
||
|
||
using var sub = service.Subscribe(3, onChanged: () => { });
|
||
|
||
await AwaitAssertAsync(() =>
|
||
{
|
||
Assert.Equal(1, factory.CreatedCount); // pre-fix: 0 — ResolveSiteAsync bailed
|
||
Assert.NotNull(factory.TryGet("site-3", "http://only-node:8083"));
|
||
});
|
||
}
|
||
```
|
||
|
||
(Adapt the harness call to `CreateService`'s real signature; the essential arrangement is a site row with exactly one gRPC address.)
|
||
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter SiteWithOnlyNodeAEndpoint` — expect FAIL (no client ever created; the service logged "has no gRPC node addresses").
|
||
3. Implement — replace the both-required check (`SiteAlarmLiveCacheService.cs:294-300`):
|
||
|
||
```csharp
|
||
var nodeA = site.GrpcNodeAAddress;
|
||
var nodeB = site.GrpcNodeBAddress;
|
||
if (string.IsNullOrWhiteSpace(nodeA) && string.IsNullOrWhiteSpace(nodeB))
|
||
{
|
||
_logger.LogWarning(
|
||
"Live alarm cache: site {SiteId} ({SiteIdentifier}) has no gRPC node addresses; " +
|
||
"not starting an aggregator", siteId, site.SiteIdentifier);
|
||
return null;
|
||
}
|
||
|
||
// Single-endpoint site (review 02 round 2, N9): reuse the sole configured
|
||
// endpoint for both slots — the aggregator's NodeA↔NodeB failover flip
|
||
// degenerates to a reconnect against the same node, which is exactly the
|
||
// right behavior for a one-node site.
|
||
var grpcA = !string.IsNullOrWhiteSpace(nodeA) ? nodeA! : nodeB!;
|
||
var grpcB = !string.IsNullOrWhiteSpace(nodeB) ? nodeB! : nodeA!;
|
||
return (site.SiteIdentifier, grpcA, grpcB);
|
||
```
|
||
|
||
4. Run the filter + `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter SiteAlarmLiveCache` — expect PASS.
|
||
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs && git commit -m "fix(comm): single-gRPC-endpoint sites can go live (flip degenerates to same-node reconnect) (plan R2-02 T15)"`
|
||
|
||
---
|
||
|
||
## Dependencies on other plans
|
||
|
||
- **PLAN-R2-07 (UI/Management/Security round 2)** — owns every `AlarmSummary.razor` page-side finding and the sticky-`IsLive` change in `SiteAlarmLiveCacheService.cs`. **Binding MUTEX** (see top of plan): Tasks 10, 12, 15 must not run concurrently with R2-07's live-alarm slice. Everything else here is file-disjoint from R2-07.
|
||
- **PLAN-01 heritage** — Task 1 relocates (by delegation, API-preserving) the oldest-Up logic PLAN-01 shipped in `Host/Health/ClusterActivityEvaluator.cs`; the PLAN-01 `TwoNodeClusterFixture` is reused (additively extended) by Task 2. No live round-2 plan owns those files.
|
||
- **Contract locks** — the new `SfBufferSnapshotChunk`/`SfBufferResyncAck` messages ride intra-site Akka remoting, NOT ClusterClient, so `ClusterClientContractLockTests` (PLAN-08 T9) needs no update; the equivalent wire pins land in `ResyncWireSerializationPinTests` (Task 5). If a later change routes any resync message over ClusterClient, it MUST be added to the frozen contract-lock matrix.
|
||
|
||
## Execution order
|
||
|
||
**P0 (the Critical, strictly ordered):** 1 → 2 (red repro) → 3 (repro goes green) → 4. Do not ship a build between 2 and 3.
|
||
|
||
**Critical path (file-serialized chains):**
|
||
- `SiteReplicationActor.cs`: 3 → 5 → 6 → 7
|
||
- `AkkaHostedService.cs`: 3 → 4
|
||
- Live-alarm files (`SiteAlarmAggregatorActor.cs` / `SiteAlarmLiveCacheService.cs`): 10 → 11 → 12 → 15 (**MUTEX with PLAN-R2-07 throughout**)
|
||
- `Component-Communication.md`: 4 → 14
|
||
- `StoreAndForwardService.cs`: 8 standalone; validator: 9 standalone; `ManagementActor.cs`: 13 standalone
|
||
|
||
**P1 (the High):** 5 → 6 → 7 immediately after P0. **P2:** 8, 9, 10, 11, 12, 13, 14, 15 in any dependency-respecting order; 8/9/13/14 are safe to run in parallel with the P0/P1 chains (file-disjoint).
|
||
|
||
After each task: build the touched projects + the named per-project `--filter` command only. At plan end: `dotnet build ZB.MOM.WW.ScadaBridge.slnx`, then `dotnet test ZB.MOM.WW.ScadaBridge.slnx` (the slnx now includes CLI.Tests), and rebuild the docker cluster (`bash docker/deploy.sh`) since Host/actor runtime changed.
|