From 037798b3679bbf693170fdbc1f0b5a2fb068bf77 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 20 Jul 2026 04:20:05 -0400 Subject: [PATCH] feat(localdb)!: replicate site config + sf_messages via CDC, delete the bespoke replicators MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tasks 14, 15 and 16, landed as ONE commit. PLAN DEFECT: these three tasks cannot compile separately. SiteReplicationActor takes a ReplicationService and calls ReplaceAllAsync (Task 14 deletes both); DeploymentManagerActor Tells message types declared in ReplicationMessages.cs (Task 15 deletes it); AkkaHostedService constructs the actor (Task 16). Any ordering leaves a broken intermediate. Combining them also strengthens the invariant Task 14 already stated for itself — the two mechanisms never both run, and never neither. Registered 8 tables in SiteLocalDbSetup.OnReady: sf_messages plus the 7 site config tables. notification_lists and smtp_configurations are deliberately NOT registered — permanently empty by design, so registering them would open a standing replication channel whose only historical payload was plaintext SMTP passwords. Migrate stays the LAST call in OnReady, after all registrations, so migrated rows enter the oplog through live capture triggers. Deleted: SiteReplicationActor, ReplicationMessages.cs, ReplicationService, StoreAndForwardStorage.ReplaceAllAsync, and 6 test files. ReplaceAllAsync is not merely unused but unsafe to keep: a mass DELETE on a now-replicated table would be captured and shipped to the peer. Kept ActiveNodeEvaluator (delivery gate + heartbeat still need it) with its doc corrected, and activeNodeCheck in AkkaHostedService (SiteCommunicationActor). The positional-argument hazard the plan flagged was real: removing DeploymentManagerActor's optional IActorRef? replicationActor shifted 6 trailing optionals, and 4 test call sites bound the wrong arguments with no compile error at some positions. Converted them to named arguments where possible — Props.Create builds an expression tree, which rejects out-of-position named args, so the rest are padded positionally with a comment saying why. The Task 7 'not yet registered' test was INVERTED rather than deleted, and is exact in both directions: too few means a table silently stops replicating, too many means the SMTP tables leak. Added a separate security-named test for those two, and a composite-PK test (LWW keys on the full PK, so a truncated key set would collapse distinct rows). The convergence suites now get their registrations from the real OnReady — their temporary harness registration is deleted, so they prove the cutover rather than agreeing with themselves. Verified: build 0 warnings; SiteRuntime 512, StoreAndForward 130, Host 330, AuditLog 355, ExternalSystemGateway 142, HealthMonitoring 97, LocalDb integration 16 — all pass, 0 failures. Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts --- .../ClusterState/ActiveNodeEvaluator.cs | 19 +- .../Actors/AkkaHostedService.cs | 37 +- .../SiteLocalDbSetup.cs | 33 +- .../Actors/DeploymentManagerActor.cs | 24 +- .../Actors/SiteReplicationActor.cs | 707 ------------------ .../Messages/ReplicationMessages.cs | 44 -- .../ReplicationService.cs | 194 ----- .../ServiceCollectionExtensions.cs | 9 - .../StoreAndForwardService.cs | 32 +- .../StoreAndForwardStorage.cs | 35 +- .../CompositionRootTests.cs | 1 - .../SiteLocalDbWiringTests.cs | 59 +- .../Cluster/SfBufferResyncPredicateTests.cs | 122 --- .../LocalDbSitePairHarness.cs | 34 +- .../Actors/DeploymentManagerActorTests.cs | 34 +- .../DeploymentManagerCertReconcileTests.cs | 3 +- .../DeploymentManagerLoggerFactoryTests.cs | 9 +- .../Actors/DeploymentManagerRedeployTests.cs | 7 +- .../Actors/SiteReplicationActorTests.cs | 568 -------------- .../ResyncWireSerializationPinTests.cs | 95 --- .../CachedCallAttemptEmissionTests.cs | 2 - .../ReplicationServiceTests.cs | 236 ------ .../ReplicationWireSerializationPinTests.cs | 82 -- .../StoreAndForwardReplicationTests.cs | 266 ------- .../StoreAndForwardSiteEventTests.cs | 2 +- .../StoreAndForwardStorageTests.cs | 19 +- 26 files changed, 160 insertions(+), 2513 deletions(-) delete mode 100644 src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs delete mode 100644 src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Messages/ReplicationMessages.cs delete mode 100644 src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs delete mode 100644 tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs delete mode 100644 tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs delete mode 100644 tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs delete mode 100644 tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs delete mode 100644 tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationWireSerializationPinTests.cs delete mode 100644 tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs index 57d7b425..43ba93a7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs @@ -10,11 +10,20 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.ClusterState; /// once the original first node restarts and rejoins; every product-level active/standby /// decision must use this evaluator, never cluster.State.Leader. /// -/// Lives in Communication (not Host) so BOTH SiteCommunicationActor and -/// SiteReplicationActor can default to it — Host cannot be referenced from either. -/// The Host's ClusterActivityEvaluator.SelfIsOldest delegates here, so the S&F -/// delivery gate (IClusterNodeProvider.SelfIsPrimary), the resync authority checks, -/// and the heartbeat IsActive stamp all share one implementation. +/// Lives in Communication (not Host) so SiteCommunicationActor can default to it — +/// Host cannot be referenced from there. The Host's +/// ClusterActivityEvaluator.SelfIsOldest delegates here, so the S&F delivery gate +/// (IClusterNodeProvider.SelfIsPrimary) and the heartbeat IsActive stamp share one +/// implementation. +/// +/// It also backed SiteReplicationActor's resync authority checks until LocalDb +/// Phase 2 deleted that actor. Those checks existed because the bespoke resync applied a +/// destructive delete-all-then-insert-all, so running it in the wrong direction wiped a +/// live store-and-forward buffer. LocalDb's snapshot resync merges per row under +/// last-writer-wins and never deletes, so there is no destructive apply left to gate — the +/// evaluator survives for the delivery gate and the heartbeat, which still genuinely need +/// a single active node. +/// /// /// public static class ActiveNodeEvaluator diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index 2e29f03a..16b8b41c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -766,37 +766,18 @@ akka {{ var deploymentConfigFetcher = _serviceProvider.GetService(); - // Create SiteReplicationActor on every node (not a singleton) - var sfStorage = _serviceProvider.GetRequiredService(); - var replicationService = _serviceProvider.GetRequiredService(); - var replicationLogger = _serviceProvider.GetRequiredService() - .CreateLogger(); - - // 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. + // ONE active-node predicate instance governs the S&F delivery gate and the + // heartbeat IsActive stamp (SiteCommunicationActor, wired below) — review 02 + // round 2, N1. It also governed SiteReplicationActor's resync authority until + // LocalDb Phase 2 deleted that actor: the library's snapshot resync merges per row + // under last-writer-wins and never deletes, so there is no destructive apply left + // to need an authority check. Null in non-clustered test hosts: the consumers fall + // back to the shared oldest-Up evaluator, never to a leader check. var clusterNodeProvider = _serviceProvider.GetService(); Func? 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"); - - // Wire S&F replication handler to forward operations via the replication actor - replicationService.SetReplicationHandler(op => - { - replicationActor.Tell(new ReplicateStoreAndForward(op)); - return Task.CompletedTask; - }); - - _logger.LogInformation("SiteReplicationActor created and S&F replication handler wired"); - // Deployment Manager — role-scoped singleton via SingletonRegistrar // (review 01 round-2 N5): previously hand-rolled with bare PoisonPill // termination and NO PhaseClusterLeave drain, so in-flight SQLite @@ -807,7 +788,7 @@ akka {{ _actorSystem!, "deployment-manager", Props.Create(() => new DeploymentManagerActor( storage, compilationService, sharedScriptLibrary, streamManager, - siteRuntimeOptionsValue, dmLogger, dclManager, replicationActor, + siteRuntimeOptionsValue, dmLogger, dclManager, siteHealthCollector, _serviceProvider, null, deploymentConfigFetcher)), _logger, role: siteRole); var dmProxy = dm.Proxy; @@ -1053,7 +1034,7 @@ akka {{ // SetReady asserts a deliberately narrow contract. By this point the // actor system exists, SiteStreamManager.Initialize has run, and every // role actor (SiteCommunicationActor, deployment-manager singleton, - // SiteReplicationActor, the ClusterClient) has been created with ActorOf — + // the ClusterClient) has been created with ActorOf — // creation and the registration Tells are synchronous and strictly ordered. // What is NOT guaranteed is completion of each actor's PreStart or the // ClusterClient's initial-contact handshake with central: those are diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbSetup.cs b/src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbSetup.cs index 626c6d5d..beb7f311 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbSetup.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbSetup.cs @@ -48,8 +48,7 @@ public static class SiteLocalDbSetup SiteEventLogSchema.Apply(connection); // Phase 2: the site's configuration tables and the store-and-forward buffer - // now live in this file too. Created here, deliberately NOT registered — see - // below. + // now live in this file too. SiteStorageSchema.Apply(connection); StoreAndForwardSchema.Apply(connection); } @@ -60,12 +59,30 @@ public static class SiteLocalDbSetup db.RegisterReplicated("OperationTracking"); db.RegisterReplicated("site_events"); - // The Phase 2 tables are created above but NOT registered yet. The bespoke - // SiteReplicationActor / StoreAndForward ReplicationService still own replicating - // them until the Task 14 cutover deletes both and registers these in one commit. - // Registering them now would mean two independent replicators writing the same - // rows — harmless in principle (both upsert) but it would mask a defect in either - // one, which is exactly what the cutover needs to be able to see. + // Phase 2: the store-and-forward buffer and the seven site configuration tables. + // These replaced the bespoke SiteReplicationActor and StoreAndForward + // ReplicationService, which shipped hand-written Add/Remove/Park/Requeue operations + // over Akka; both were deleted in the same commit that added these lines, so the + // two mechanisms never ran at once. + // + // Both composite-PK tables are fine: RegisterReplicated orders multi-column PKs by + // ordinal. No Phase 2 table has a BLOB column, which it would reject. + db.RegisterReplicated("sf_messages"); + db.RegisterReplicated("deployed_configurations"); + db.RegisterReplicated("static_attribute_overrides"); + db.RegisterReplicated("shared_scripts"); + db.RegisterReplicated("external_systems"); + db.RegisterReplicated("database_connections"); + db.RegisterReplicated("data_connection_definitions"); + db.RegisterReplicated("native_alarm_state"); + + // notification_lists and smtp_configurations are created but deliberately NOT + // registered. They are permanently empty by design — the site-side write paths were + // removed on 2026-07-10, the legacy migrator skips them, and the active node's + // artifact apply purges them on every deploy. Registering them would open a standing + // replication channel whose only historical payload was plaintext SMTP passwords, in + // exchange for replicating nothing. Anyone adding them here should first establish + // that a site has a legitimate reason to hold SMTP credentials at all. // AFTER registration, so migrated rows enter the oplog and reach the peer like // any other write. Before it, they would be invisible to replication forever. diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs index e360dffc..56152dc2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs @@ -52,7 +52,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers /// private readonly ILoggerFactory _loggerFactory; private readonly IActorRef? _dclManager; - private readonly IActorRef? _replicationActor; private readonly ISiteHealthCollector? _healthCollector; private readonly IServiceProvider? _serviceProvider; /// @@ -166,7 +165,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers SiteRuntimeOptions options, ILogger logger, IActorRef? dclManager = null, - IActorRef? replicationActor = null, ISiteHealthCollector? healthCollector = null, IServiceProvider? serviceProvider = null, ILoggerFactory? loggerFactory = null, @@ -181,7 +179,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers _streamManager = streamManager; _options = options; _dclManager = dclManager; - _replicationActor = replicationActor; _healthCollector = healthCollector; _serviceProvider = serviceProvider; _configFetcher = configFetcher; @@ -785,15 +782,14 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers await _storage.ClearStaticOverridesAsync(instanceName); await _storage.ClearNativeAlarmsForInstanceAsync(instanceName); - // Replicate to standby node — notify-and-fetch: send only the deployment id + - // central fetch coordinates (NOT the config JSON). The standby fetches the - // config over HTTP itself, so a large config never crosses the intra-site Akka - // hop (which would silently drop on the 128 KB frame trap). When the coords are - // absent (deploy paths other than RefreshDeployment), the standby fetch is a - // no-op miss and reconciliation is the durable backstop. - _replicationActor?.Tell(new ReplicateConfigDeploy( - instanceName, command.DeploymentId, command.RevisionHash, true, - command.CentralFetchBaseUrl ?? "", command.FetchToken ?? "")); + // No explicit replication step: deployed_configurations is a replicated table, + // so the write above is captured and shipped to the peer like any other row. + // This used to be a notify-and-fetch Tell — the standby was sent the deployment + // id plus central fetch coordinates and pulled the config over HTTP itself, + // because a large config would silently drop on the intra-site Akka hop's + // 128 KB frame trap. Replication carries the row directly and has no such limit, + // so the standby no longer fetches on deploy. (SiteReconciliationActor still + // fetches at node startup when central reports gaps — a different path.) return new DeployPersistenceResult( command.DeploymentId, instanceName, true, null, sender, isRedeploy); @@ -967,7 +963,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers { if (t.IsCompletedSuccessfully) { - _replicationActor?.Tell(new ReplicateConfigSetEnabled(instanceName, false)); // Operational `deployment` event — disable succeeded. LogDeploymentEvent("Info", instanceName, $"Instance {instanceName} disabled"); } @@ -1001,7 +996,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers Task.Run(async () => { await _storage.SetInstanceEnabledAsync(instanceName, true); - _replicationActor?.Tell(new ReplicateConfigSetEnabled(instanceName, true)); var configs = await _storage.GetAllDeployedConfigsAsync(); var config = configs.FirstOrDefault(c => c.InstanceUniqueName == instanceName); return new EnableResult(command, config, null, sender); @@ -1099,7 +1093,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers { if (t.IsCompletedSuccessfully) { - _replicationActor?.Tell(new ReplicateConfigRemove(instanceName)); // Operational `deployment` event — delete succeeded. LogDeploymentEvent("Info", instanceName, $"Instance {instanceName} deleted"); } @@ -1949,7 +1942,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers // central-only and is never stored on a site (see the purge above). // Replicate artifacts to standby node - _replicationActor?.Tell(new ReplicateArtifacts(command)); return new ArtifactDeploymentResponse( command.DeploymentId, "", true, null, DateTimeOffset.UtcNow); diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs deleted file mode 100644 index 63066d2d..00000000 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ /dev/null @@ -1,707 +0,0 @@ -using Akka.Actor; -using Akka.Cluster; -using Akka.Event; -using Microsoft.Extensions.Logging; -using ZB.MOM.WW.ScadaBridge.Commons.Observability; -using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment; -using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages; -using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; -using ZB.MOM.WW.ScadaBridge.StoreAndForward; - -namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; - -/// -/// Runs on every site node (not a singleton). Handles both config and S&F replication -/// between site cluster peers. -/// -/// Outbound: receives local replication requests and forwards to peer via ActorSelection. -/// Inbound: receives replicated operations from peer and applies to local SQLite. -/// Uses fire-and-forget (Tell) — no ack wait per design. -/// -public class SiteReplicationActor : ReceiveActor, IWithTimers -{ - private readonly SiteStorageService _storage; - private readonly StoreAndForwardStorage _sfStorage; - private readonly ReplicationService _replicationService; - private readonly IDeploymentConfigFetcher? _configFetcher; - private readonly string _siteRole; - private readonly ILogger _logger; - private readonly Cluster _cluster; - private readonly Func _isActive; - private readonly int _configFetchRetryCount; - private readonly TimeSpan _configFetchRetryDelay; - private Address? _peerAddress; - - /// Akka timer scheduler injected by the framework via . - public ITimerScheduler Timers { get; set; } = null!; - - // ── Chunked-resync assembly (standby side; actor-thread only) ── - private string? _assemblingResyncId; - private int _assemblingTotalChunks; - private bool _assemblingTruncated; - private readonly Dictionary> _assemblingChunks = new(); - private const string ResyncAssemblyTimerKey = "sf-resync-assembly-timeout"; - - /// 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. - private readonly TimeSpan _resyncAssemblyTimeout; - - // ── Resync delivery confirmation (active side; actor-thread only) ── - private string? _pendingAckResyncId; - private const string ResyncAckTimerKey = "sf-resync-ack-timeout"; - - /// How long the active node waits for the standby's - /// before warning + counting the resync as unacknowledged (lost chunks / dead peer). Ctor - /// test seam; production default 60 s. - private readonly TimeSpan _resyncAckTimeout; - - /// - /// Maximum rows an active node returns in a single anti-entropy resync snapshot. - /// A standby whose buffer exceeded this (Truncated snapshot) resyncs the oldest - /// 10 000 rows; further divergence drains naturally as the active node delivers. - /// - private const int MaxResyncRows = 10_000; - - /// - /// Estimated per-chunk payload budget for a resync snapshot. Akka remoting's default - /// maximum-frame-size is 128 000 bytes and BuildHocon sets no override, - /// so the monolithic is silently undeliverable for any - /// realistic backlog (review 02 round 2, N2). 64 000 bytes leaves ≈50% headroom for - /// the JSON envelope, CLR type manifests, and the non-payload columns. - /// - internal const int MaxResyncChunkBytes = 64_000; - - /// Row cap per resync chunk (bounds a chunk even when every row is tiny). - internal const int MaxResyncChunkRows = 200; - - /// - /// 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). - /// - internal static List> ChunkForRemoting( - IReadOnlyList rows, int maxChunkBytes, int maxChunkRows) - { - var chunks = new List>(); - var current = new List(); - 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(); - currentBytes = 0; - } - current.Add(row); - currentBytes += estimate; - } - if (current.Count > 0) chunks.Add(current); - return chunks; - } - - /// - /// Initializes a new and registers Akka message handlers. - /// - /// Service for accessing local site storage. - /// Store-and-forward SQLite storage for replication of buffered messages. - /// Service providing replication transport logic. - /// Akka cluster role used to identify peer nodes to replicate to. - /// Logger instance. - /// - /// Fetches a deployed instance's config JSON from central over HTTP. Used by the - /// notify-and-fetch standby apply path (): the peer - /// replicates only the deployment id, and the standby fetches the config itself so a large - /// config never crosses the intra-site Akka hop. Null on nodes/tests without a fetcher. - /// - /// - /// Active-node check that gates the buffer-resync roles (a standby requests a - /// resync, the active node answers). 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 - /// (). - /// - /// Site runtime options, including the config-fetch retry count; production defaults apply when null. - /// Delay between config-fetch retry attempts; defaults to 2 seconds when null. - public SiteReplicationActor( - SiteStorageService storage, - StoreAndForwardStorage sfStorage, - ReplicationService replicationService, - string siteRole, - ILogger logger, - IDeploymentConfigFetcher? configFetcher = null, - Func? isActiveOverride = null, - SiteRuntimeOptions? options = null, - TimeSpan? configFetchRetryDelay = null, - TimeSpan? resyncAssemblyTimeout = null, - TimeSpan? resyncAckTimeout = null) - { - _storage = storage; - _sfStorage = sfStorage; - _replicationService = replicationService; - _configFetcher = configFetcher; - _siteRole = siteRole; - _logger = logger; - _cluster = Cluster.Get(Context.System); - _isActive = isActiveOverride ?? DefaultIsActive; - _resyncAssemblyTimeout = resyncAssemblyTimeout ?? TimeSpan.FromSeconds(30); - _resyncAckTimeout = resyncAckTimeout ?? TimeSpan.FromSeconds(60); - // UA2: bound the standby's replicated-config fetch retries. At least one - // attempt always runs; the fixed inter-attempt delay is a test seam - // (production default 2 s). - _configFetchRetryCount = Math.Max(1, options?.ConfigFetchRetryCount ?? 1); - _configFetchRetryDelay = configFetchRetryDelay ?? TimeSpan.FromSeconds(2); - - // Cluster member events - Receive(HandleMemberUp); - Receive(HandleMemberRemoved); - Receive(HandleCurrentClusterState); - - // Outbound — forward to peer - Receive(msg => SendToPeer(new ApplyConfigDeploy( - msg.InstanceName, msg.DeploymentId, msg.RevisionHash, msg.IsEnabled, - msg.CentralFetchBaseUrl, msg.FetchToken))); - Receive(msg => SendToPeer(new ApplyConfigRemove(msg.InstanceName))); - Receive(msg => SendToPeer(new ApplyConfigSetEnabled( - msg.InstanceName, msg.IsEnabled))); - Receive(msg => SendToPeer(new ApplyArtifacts(msg.Command))); - Receive(msg => SendToPeer(new ApplyStoreAndForward(msg.Operation))); - - // Inbound — apply from peer - Receive(HandleApplyConfigDeploy); - Receive(HandleApplyConfigRemove); - Receive(HandleApplyConfigSetEnabled); - Receive(HandleApplyArtifacts); - Receive(HandleApplyStoreAndForward); - - // Anti-entropy — full S&F buffer resync on peer (re)join - Receive(HandleRequestSfBufferResync); - Receive(HandleSfResyncSnapshotLoaded); - Receive(HandleSfBufferSnapshotChunk); - Receive(HandleResyncAssemblyTimedOut); - Receive(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(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); - }); - Receive(HandleSfBufferSnapshot); // legacy monolithic handler — retained for rolling compat - } - - /// - protected override void PreStart() - { - base.PreStart(); - _cluster.Subscribe(Self, ClusterEvent.SubscriptionInitialStateMode.InitialStateAsSnapshot, - typeof(ClusterEvent.MemberUp), - typeof(ClusterEvent.MemberRemoved)); - _logger.LogInformation("SiteReplicationActor started, subscribing to cluster events for role {Role}", _siteRole); - } - - /// - protected override void PostStop() - { - _cluster.Unsubscribe(Self); - base.PostStop(); - } - - private void HandleCurrentClusterState(ClusterEvent.CurrentClusterState state) - { - foreach (var member in state.Members) - { - if (member.Status == MemberStatus.Up) - TryTrackPeer(member); - } - } - - private void HandleMemberUp(ClusterEvent.MemberUp evt) - { - TryTrackPeer(evt.Member); - } - - private void HandleMemberRemoved(ClusterEvent.MemberRemoved evt) - { - if (evt.Member.Address.Equals(_peerAddress)) - { - _logger.LogInformation("Peer node removed: {Address}", _peerAddress); - _peerAddress = null; - } - } - - private void TryTrackPeer(Member member) - { - // Must have our site role, and must not be self - if (member.HasRole(_siteRole) && !member.Address.Equals(_cluster.SelfAddress)) - { - _peerAddress = member.Address; - _logger.LogInformation("Peer node tracked: {Address}", _peerAddress); - OnPeerTracked(); - } - } - - /// - /// Side-effect run whenever a peer is (re)tracked. A standby requests a - /// full S&F buffer snapshot for anti-entropy resync — this closes the "a - /// standby down for an hour rejoins and diverges forever" gap: it may have missed - /// replicated Add/Remove/Park ops while it was gone. The active node never - /// requests. so tests can drive it without a - /// real two-node cluster. - /// - protected virtual void OnPeerTracked() - { - if (!SafeIsActive()) - { - SendToPeer(new RequestSfBufferResync()); - } - } - - /// - /// 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. - /// - private bool DefaultIsActive() => - Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(_cluster, _siteRole); - - /// - /// Evaluates the active-node check, treating a throwing check as standby - /// (safe-by-default: a standby never delivers or answers resyncs). - /// - private bool SafeIsActive() - { - try - { - return _isActive(); - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Active-node check threw; treating node as standby"); - return false; - } - } - - /// - /// Forwards a replication message to the tracked peer node's site-replication actor - /// (fire-and-forget, dropped when no peer is tracked). - /// so tests can intercept the peer send without standing up a real two-node cluster. - /// - /// The replication message to forward to the peer. - protected virtual void SendToPeer(object message) - { - if (_peerAddress == null) - { - // A dropped op is a lost delta — surface it at Warning + a metric so a - // never-tracked peer (a dead standby) is visible, not silent (arch - // review 02). In single-node dev the peer is legitimately absent; the - // per-op warning is rate-tolerable (accepted per review). - ScadaBridgeTelemetry.RecordReplicationFailure(); - _logger.LogWarning("No peer available, dropping replication message {Type}", message.GetType().Name); - return; - } - - var path = new RootActorPath(_peerAddress) / "user" / "site-replication"; - Context.ActorSelection(path).Tell(message); - } - - // ── Inbound handlers ── - - private void HandleApplyConfigDeploy(ApplyConfigDeploy msg) - { - if (string.IsNullOrEmpty(msg.CentralFetchBaseUrl)) - { - // The direct DeployInstanceCommand cross-cluster wire path was retired. - // This guard is a defensive fallback: skip quietly rather than calling FetchAsync("") - // and logging a spurious error. Reconciliation backstops any missed writes. - _logger.LogDebug( - "No fetch coords for {Instance} (deployment {DeploymentId}) — skipping replicated fetch; T18 reconciliation is the backstop", - msg.InstanceName, msg.DeploymentId); - return; - } - - if (_configFetcher is null) - { - _logger.LogWarning( - "No config fetcher available; cannot apply replicated config for {Instance} (deployment {DeploymentId}) — reconciliation will backstop", - msg.InstanceName, msg.DeploymentId); - return; - } - - _logger.LogInformation( - "Replicating config for {Instance} (deployment {DeploymentId}) — fetching from central", - msg.InstanceName, msg.DeploymentId); - - // Notify-and-fetch: the peer sent only the id, so the standby fetches the config - // itself (off-thread; best-effort fire-and-forget, matching the no-ack replication - // model). The guarded write only overwrites a strictly-older local row. The fetch - // is retried up to ConfigFetchRetryCount times with a fixed delay (UA2) — a transient - // central hiccup no longer defers to the slower reconciliation backstop, which still - // covers a total failure after the last attempt. - _ = FetchWithRetryAsync(); - return; - - async Task FetchWithRetryAsync() - { - for (var attempt = 1; attempt <= _configFetchRetryCount; attempt++) - { - try - { - // Non-null: the outer method returns early when _configFetcher is null. - var json = await _configFetcher!.FetchAsync( - msg.CentralFetchBaseUrl, msg.DeploymentId, msg.FetchToken, CancellationToken.None); - await _storage.StoreDeployedConfigIfNewerAsync( - msg.InstanceName, json, msg.DeploymentId, msg.RevisionHash, msg.IsEnabled); - return; - } - catch (DeploymentConfigFetchException fex) when (fex.IsSuperseded) - { - // A superseded/expired fetch never heals by retrying — a newer deploy - // will replicate its own id. - _logger.LogInformation( - "Skip replicated config for {Instance}: superseded/expired (a newer deploy will replicate)", - msg.InstanceName); - return; - } - catch (Exception ex) - { - if (attempt < _configFetchRetryCount) - { - _logger.LogWarning(ex, - "Replicated config fetch attempt {Attempt}/{Max} failed for {Instance} (deployment {DeploymentId}) — retrying in {Delay}", - attempt, _configFetchRetryCount, msg.InstanceName, msg.DeploymentId, _configFetchRetryDelay); - await Task.Delay(_configFetchRetryDelay); - } - else - { - _logger.LogError(ex, - "Replicated config fetch failed after {Attempts} attempt(s) for {Instance} (deployment {DeploymentId}) — reconciliation will backstop", - _configFetchRetryCount, msg.InstanceName, msg.DeploymentId); - } - } - } - } - } - - private void HandleApplyConfigRemove(ApplyConfigRemove msg) - { - _logger.LogInformation("Applying replicated config remove for {Instance}", msg.InstanceName); - _storage.RemoveDeployedConfigAsync(msg.InstanceName) - .ContinueWith(t => - { - if (t.IsFaulted) - _logger.LogError(t.Exception, "Failed to apply replicated remove for {Instance}", msg.InstanceName); - }); - } - - private void HandleApplyConfigSetEnabled(ApplyConfigSetEnabled msg) - { - _logger.LogInformation("Applying replicated set-enabled={Enabled} for {Instance}", msg.IsEnabled, msg.InstanceName); - _storage.SetInstanceEnabledAsync(msg.InstanceName, msg.IsEnabled) - .ContinueWith(t => - { - if (t.IsFaulted) - _logger.LogError(t.Exception, "Failed to apply replicated set-enabled for {Instance}", msg.InstanceName); - }); - } - - private void HandleApplyArtifacts(ApplyArtifacts msg) - { - var command = msg.Command; - _logger.LogInformation("Applying replicated artifacts, deploymentId={DeploymentId}", command.DeploymentId); - - Task.Run(async () => - { - try - { - if (command.SharedScripts != null) - foreach (var s in command.SharedScripts) - await _storage.StoreSharedScriptAsync(s.Name, s.Code, s.ParameterDefinitions, s.ReturnDefinition); - - if (command.ExternalSystems != null) - foreach (var es in command.ExternalSystems) - await _storage.StoreExternalSystemAsync(es.Name, es.EndpointUrl, es.AuthType, es.AuthConfiguration, es.MethodDefinitionsJson, es.TimeoutSeconds); - - if (command.DatabaseConnections != null) - foreach (var db in command.DatabaseConnections) - await _storage.StoreDatabaseConnectionAsync(db.Name, db.ConnectionString, db.MaxRetries, db.RetryDelay); - - // Notification lists and SMTP - // configuration are central-only and are never persisted on a site. - // Mirror the primary apply path: purge any pre-fix rows (including the - // plaintext SMTP password) instead of writing the command's - // (now-always-null) NotificationLists/SmtpConfigurations. - await _storage.PurgeCentralOnlyNotificationConfigAsync(); - - if (command.DataConnections != null) - foreach (var dc in command.DataConnections) - await _storage.StoreDataConnectionDefinitionAsync(dc.Name, dc.Protocol, dc.PrimaryConfigurationJson, dc.BackupConfigurationJson, dc.FailoverRetryCount); - } - catch (Exception ex) - { - _logger.LogError(ex, "Failed to apply replicated artifacts"); - } - }); - } - - private void HandleApplyStoreAndForward(ApplyStoreAndForward msg) - { - _logger.LogDebug("Applying replicated S&F operation {OpType} for message {Id}", - msg.Operation.OperationType, msg.Operation.MessageId); - - _replicationService.ApplyReplicatedOperationAsync(msg.Operation, _sfStorage) - .ContinueWith(t => - { - if (t.IsFaulted) - _logger.LogError(t.Exception, "Failed to apply replicated S&F operation {Id}", msg.Operation.MessageId); - }); - } - - /// - /// Active-node side of the anti-entropy resync: answers a standby's - /// with a sequence of byte-budgeted - /// s (up to oldest rows). - /// A non-active node ignores the request — only the authoritative node may answer. - /// The snapshot is piped back to Self so chunking + ack bookkeeping stays actor-safe. - /// - private void HandleRequestSfBufferResync(RequestSfBufferResync msg) - { - if (!SafeIsActive()) - { - _logger.LogDebug("Ignoring S&F buffer resync request — this node is not active"); - return; - } - - var replyTo = Sender; - _sfStorage.GetAllMessagesAsync(MaxResyncRows).PipeTo( - Self, - failure: ex => new Status.Failure(ex), - success: result => new SfResyncSnapshotLoaded(replyTo, result.Messages, result.Truncated)); - } - - /// - /// Active-node continuation: the resync snapshot finished loading; chunk it to fit the - /// remoting frame and send the sequenced chunks to the requester (all sharing one - /// resyncId). Task 7 arms the ack-timeout here. - /// - private void HandleSfResyncSnapshotLoaded(SfResyncSnapshotLoaded msg) - { - var chunks = ChunkForRemoting(msg.Messages, MaxResyncChunkBytes, MaxResyncChunkRows); - if (chunks.Count == 0) chunks.Add(new List()); // 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); - // Arm the ack window: absence of an SfBufferResyncAck within it surfaces as a - // Warning + counter (the silent-loss mode N2 flagged). Single-outstanding-resync - // bookkeeping: a new request supersedes by overwriting the id and restarting the timer. - _pendingAckResyncId = resyncId; - Timers.StartSingleTimer(ResyncAckTimerKey, new ResyncAckTimedOut(resyncId), _resyncAckTimeout); - } - - /// - /// Standby-node side of the anti-entropy resync: replaces the local buffer - /// wholesale with the active node's snapshot. Combined with the upsert-based - /// replicated applies (arch review 02), any replicated op that lands - /// after this resync merges cleanly onto the resynced state. An active node - /// ignores a snapshot — it is the source of truth, never a resync target. - /// - private void HandleSfBufferSnapshot(SfBufferSnapshot msg) - { - if (SafeIsActive()) - { - _logger.LogDebug("Ignoring S&F buffer snapshot — this node is active"); - return; - } - - if (msg.Truncated) - { - _logger.LogWarning( - "S&F buffer resync snapshot truncated at {Cap} rows; divergence beyond the cap drains naturally", - MaxResyncRows); - } - - _logger.LogInformation( - "Applying S&F buffer resync snapshot ({Count} rows), replacing local buffer", msg.Messages.Count); - - 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"); - }); - } - - /// - /// Standby-node side of the chunked anti-entropy resync: accumulates the chunks of one - /// ResyncId, and once all have arrived, assembles them in sequence order and - /// replaces the local buffer atomically, then acks. A new ResyncId discards any - /// stale partial assembly (review 02 round 2, N2). An active node ignores chunks. - /// - 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(); - } - - /// Internal: the resync snapshot finished loading; chunk and send to the requester. - internal sealed record SfResyncSnapshotLoaded( - IActorRef ReplyTo, List Messages, bool Truncated); - - /// Internal: a partial chunk assembly for exceeded its window. - internal sealed record ResyncAssemblyTimedOut(string ResyncId); - - /// Internal: the active node's ack window for expired. - internal sealed record ResyncAckTimedOut(string ResyncId); -} - -/// -/// Standby→active: request a full S&F buffer snapshot for anti-entropy resync -/// (sent when a standby (re)tracks a peer). Crosses Akka remoting between the two -/// site nodes; the POCO rides the default serializer. -/// -public sealed record RequestSfBufferResync; - -/// -/// Active→standby: full-buffer snapshot. is true when -/// the active node's buffer exceeded MaxResyncRows (the standby logs a Warning — -/// divergence beyond the cap drains naturally as the active node delivers). Crosses -/// Akka remoting; the message list rides the default serializer. -/// -public sealed record SfBufferSnapshot(List Messages, bool Truncated); - -/// -/// Active→standby: one sequenced chunk of a full-buffer anti-entropy snapshot -/// (review 02 round 2, N2 — the monolithic exceeds Akka -/// remoting's default 128 000-byte frame for any realistic backlog). All chunks of one -/// resync share ; is 1-based up to -/// . Additive message — the legacy monolithic snapshot -/// handler is retained for rolling upgrades. Crosses intra-site Akka remoting (NOT -/// ClusterClient — ClusterClientContractLockTests is intentionally not involved). -/// -public sealed record SfBufferSnapshotChunk( - string ResyncId, int Sequence, int TotalChunks, - List Messages, bool Truncated); - -/// -/// Standby→active: delivery confirmation — the standby assembled all chunks of -/// and applied them atomically ( -/// rows installed). Absence within the ack window is surfaced by the active node -/// (Warning + counter) — the silent-loss mode N2 flagged. -/// -public sealed record SfBufferResyncAck(string ResyncId, int RowCount); diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Messages/ReplicationMessages.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Messages/ReplicationMessages.cs deleted file mode 100644 index b937cc0d..00000000 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Messages/ReplicationMessages.cs +++ /dev/null @@ -1,44 +0,0 @@ -using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts; -using ZB.MOM.WW.ScadaBridge.StoreAndForward; - -namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages; - -// Outbound messages — sent by local DeploymentManagerActor/S&F service -// to the local SiteReplicationActor for forwarding to the peer node. - -/// Outbound: tell the peer to fetch+apply a deployed instance config by id (notify-and-fetch; no inline config). -public record ReplicateConfigDeploy( - string InstanceName, string DeploymentId, string RevisionHash, bool IsEnabled, - string CentralFetchBaseUrl, string FetchToken); - -/// Outbound: replicate removal of a deployed instance config to the peer node. -public record ReplicateConfigRemove(string InstanceName); - -/// Outbound: replicate an instance enabled/disabled flag change to the peer node. -public record ReplicateConfigSetEnabled(string InstanceName, bool IsEnabled); - -/// Outbound: replicate a system-wide artifact deployment (shared scripts, external systems, etc.) to the peer node. -public record ReplicateArtifacts(DeployArtifactsCommand Command); - -/// Outbound: replicate a store-and-forward buffer mutation (enqueue/dequeue/park/etc.) to the peer node. -public record ReplicateStoreAndForward(ReplicationOperation Operation); - -// Inbound messages — received from the peer's SiteReplicationActor -// and applied to local SQLite storage. - -/// Inbound: peer-replicated config deploy — the standby fetches the config by id and writes it (guarded). -public record ApplyConfigDeploy( - string InstanceName, string DeploymentId, string RevisionHash, bool IsEnabled, - string CentralFetchBaseUrl, string FetchToken); - -/// Inbound: apply peer-replicated removal of a deployed instance config to local SQLite. -public record ApplyConfigRemove(string InstanceName); - -/// Inbound: apply a peer-replicated instance enabled/disabled flag change to local SQLite. -public record ApplyConfigSetEnabled(string InstanceName, bool IsEnabled); - -/// Inbound: apply a peer-replicated system-wide artifact deployment to local SQLite. -public record ApplyArtifacts(DeployArtifactsCommand Command); - -/// Inbound: apply a peer-replicated store-and-forward buffer mutation to the local buffer. -public record ApplyStoreAndForward(ReplicationOperation Operation); diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs deleted file mode 100644 index 60f2b162..00000000 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs +++ /dev/null @@ -1,194 +0,0 @@ -using Microsoft.Extensions.Logging; -using ZB.MOM.WW.ScadaBridge.Commons.Observability; -using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; - -namespace ZB.MOM.WW.ScadaBridge.StoreAndForward; - -/// -/// Async replication of buffer operations to standby node. -/// -/// - Forwards add/remove/park operations to standby via a replication handler. -/// - No ack wait (fire-and-forget per design). -/// - Standby applies operations to its own SQLite. -/// - On failover, standby resumes delivery from its replicated state. -/// -public class ReplicationService -{ - private readonly StoreAndForwardOptions _options; - private readonly ILogger _logger; - private Func? _replicationHandler; - - /// Initializes a new instance of . - /// Store-and-forward configuration options. - /// Logger instance. - public ReplicationService( - StoreAndForwardOptions options, - ILogger logger) - { - _options = options; - _logger = logger; - } - - /// - /// Sets the handler for forwarding replication operations to the standby node. - /// Typically wraps Akka Tell to the standby's replication actor. - /// - /// The async delegate that forwards each replication operation to the standby. - public void SetReplicationHandler(Func handler) - { - _replicationHandler = handler; - } - - /// - /// Replicates an enqueue operation to standby (fire-and-forget). - /// - /// The message that was enqueued on the active node. - public void ReplicateEnqueue(StoreAndForwardMessage message) - { - if (!_options.ReplicationEnabled || _replicationHandler == null) return; - - FireAndForget(new ReplicationOperation( - ReplicationOperationType.Add, - message.Id, - message)); - } - - /// - /// Replicates a remove operation to standby (fire-and-forget). - /// - /// The identifier of the message to remove from the standby buffer. - public void ReplicateRemove(string messageId) - { - if (!_options.ReplicationEnabled || _replicationHandler == null) return; - - FireAndForget(new ReplicationOperation( - ReplicationOperationType.Remove, - messageId, - null)); - } - - /// - /// Replicates a park operation to standby (fire-and-forget). - /// - /// The message that was parked on the active node. - public void ReplicatePark(StoreAndForwardMessage message) - { - if (!_options.ReplicationEnabled || _replicationHandler == null) return; - - FireAndForget(new ReplicationOperation( - ReplicationOperationType.Park, - message.Id, - message)); - } - - /// - /// Replicates an operator-initiated requeue (a parked - /// message moved back to the pending queue) to standby (fire-and-forget). The - /// carried message reflects the active node's post-requeue state (Pending, - /// retry_count = 0) so the standby's copy can be brought into sync. - /// - /// The message in its post-requeue (Pending, retry_count=0) state. - public void ReplicateRequeue(StoreAndForwardMessage message) - { - if (!_options.ReplicationEnabled || _replicationHandler == null) return; - - FireAndForget(new ReplicationOperation( - ReplicationOperationType.Requeue, - message.Id, - message)); - } - - /// - /// Applies a replicated operation received from the active node. - /// Used by the standby node to keep its SQLite in sync. - /// - /// Add/Park/Requeue are applied as upserts (), - /// not blind INSERT/UPDATE: the full message rides in every one of those operations, - /// so a Park/Requeue whose original Add was lost (fire-and-forget replication is - /// best-effort) self-heals by materialising the row, and a duplicate Add (e.g. - /// re-issued by an anti-entropy resync) applies newest-wins instead of throwing a - /// primary-key violation. Remove is a plain delete. - /// - /// The replication operation to apply. - /// The standby node's store-and-forward storage to update. - /// A task representing the asynchronous apply operation. - public async Task ApplyReplicatedOperationAsync( - ReplicationOperation operation, - StoreAndForwardStorage storage) - { - switch (operation.OperationType) - { - case ReplicationOperationType.Add when operation.Message != null: - await storage.UpsertMessageAsync(operation.Message); - break; - - case ReplicationOperationType.Remove: - await storage.RemoveMessageAsync(operation.MessageId); - break; - - case ReplicationOperationType.Park when operation.Message != null: - operation.Message.Status = StoreAndForwardMessageStatus.Parked; - await storage.UpsertMessageAsync(operation.Message); - break; - - case ReplicationOperationType.Requeue when operation.Message != null: - operation.Message.Status = StoreAndForwardMessageStatus.Pending; - operation.Message.RetryCount = 0; - await storage.UpsertMessageAsync(operation.Message); - break; - } - } - - private void FireAndForget(ReplicationOperation operation) - { - // Invoked inline, NOT via Task.Run: the handler is a non-blocking Akka - // Tell, and thread-pool hand-off destroyed Add/Remove ordering for the - // same message id (arch review 02). Inline invocation preserves issue - // order; Akka's per-sender/receiver guarantee preserves it on the wire. - try - { - var task = _replicationHandler!.Invoke(operation); - if (!task.IsCompletedSuccessfully) - { - task.ContinueWith(t => - { - ScadaBridgeTelemetry.RecordReplicationFailure(); - _logger.LogWarning(t.Exception, - "Replication of {OpType} for message {MessageId} failed (best-effort); standby buffer may be diverging", - operation.OperationType, operation.MessageId); - }, - TaskContinuationOptions.OnlyOnFaulted); - } - } - catch (Exception ex) - { - ScadaBridgeTelemetry.RecordReplicationFailure(); - _logger.LogWarning(ex, - "Replication of {OpType} for message {MessageId} failed (best-effort); standby buffer may be diverging", - operation.OperationType, operation.MessageId); - } - } -} - -/// -/// Represents a buffer operation to be replicated to standby. -/// -public record ReplicationOperation( - ReplicationOperationType OperationType, - string MessageId, - StoreAndForwardMessage? Message); - -/// -/// Types of buffer operations that are replicated. -/// -public enum ReplicationOperationType -{ - Add, - Remove, - Park, - /// - /// An operator moved a parked message back to the pending - /// queue. The standby resets its matching row to Pending with retry_count = 0. - /// - Requeue -} diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ServiceCollectionExtensions.cs index a5c2e513..314b5522 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ServiceCollectionExtensions.cs @@ -28,7 +28,6 @@ public static class ServiceCollectionExtensions var storage = sp.GetRequiredService(); var options = sp.GetRequiredService>().Value; var logger = sp.GetRequiredService>(); - var replication = sp.GetRequiredService(); // Wire the cached-call lifecycle // observer + site identity through DI so the S&F retry loop emits // per-attempt + terminal telemetry under the same TrackedOperationId @@ -53,19 +52,11 @@ public static class ServiceCollectionExtensions storage, options, logger, - replication, cachedCallObserver, siteId, siteEventLogger); }); - services.AddSingleton(sp => - { - var options = sp.GetRequiredService>().Value; - var logger = sp.GetRequiredService>(); - return new ReplicationService(options, logger); - }); - return services; } diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs index 2b8dda17..53a6ad2f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs @@ -36,7 +36,6 @@ public class StoreAndForwardService { private readonly StoreAndForwardStorage _storage; private readonly StoreAndForwardOptions _options; - private readonly ReplicationService? _replication; private readonly ILogger _logger; /// /// Site-side observer notified @@ -107,7 +106,7 @@ public class StoreAndForwardService /// null when no sweep is currently running. Captured when the timer /// callback starts a sweep so can wait for it to /// finish before the host disposes downstream dependencies - /// (, ) that the sweep is + /// () that the sweep is /// still touching. Written from the timer thread and from /// , so reads are synchronised via the /// APIs. @@ -227,7 +226,6 @@ public class StoreAndForwardService /// The storage backend for buffered messages. /// Configuration options. /// Logger instance. - /// Optional replication service for standby synchronization. /// Optional observer for cached call lifecycle events. /// The site identifier this service belongs to. /// @@ -240,7 +238,6 @@ public class StoreAndForwardService StoreAndForwardStorage storage, StoreAndForwardOptions options, ILogger logger, - ReplicationService? replication = null, ICachedCallLifecycleObserver? cachedCallObserver = null, string siteId = "", ISiteEventLogger? siteEventLogger = null) @@ -248,7 +245,6 @@ public class StoreAndForwardService _storage = storage; _options = options; _logger = logger; - _replication = replication; _cachedCallObserver = cachedCallObserver; _siteId = string.IsNullOrWhiteSpace(siteId) ? UnknownSiteSentinel : siteId; _siteEventLogger = siteEventLogger; @@ -651,7 +647,6 @@ public class StoreAndForwardService private async Task BufferAsync(StoreAndForwardMessage message) { await _storage.EnqueueAsync(message); - _replication?.ReplicateEnqueue(message); Interlocked.Increment(ref _bufferedCount); } @@ -802,7 +797,6 @@ public class StoreAndForwardService if (success) { await _storage.RemoveMessageAsync(message.Id); - _replication?.ReplicateRemove(message.Id); Interlocked.Decrement(ref _bufferedCount); RaiseActivity("Delivered", message.Category, $"Delivered to {message.Target} after {message.RetryCount} retries"); @@ -833,7 +827,6 @@ public class StoreAndForwardService return RetryOutcome.Skipped; } Interlocked.Decrement(ref _bufferedCount); - _replication?.ReplicatePark(message); RaiseActivity("Parked", message.Category, $"Permanent failure for {message.Target}: handler returned false"); @@ -869,7 +862,6 @@ public class StoreAndForwardService return RetryOutcome.Skipped; } Interlocked.Decrement(ref _bufferedCount); - _replication?.ReplicatePark(message); RaiseActivity("Parked", message.Category, $"Max retries ({message.MaxRetries}) reached for {message.Target}"); _logger.LogWarning( @@ -1077,17 +1069,12 @@ public class StoreAndForwardService /// /// Retries a parked message (moves back to pending queue). /// - /// An operator requeue is a buffer state change and is - /// replicated to the standby (as a ) - /// so a failover preserves the operator's retry intent. + /// An operator requeue is a buffer state change, and the peer picks it up because + /// sf_messages is a replicated table: the row update is captured and shipped + /// like any other write, so a failover preserves the operator's retry intent. This + /// used to be an explicit Requeue operation sent to the standby. /// The activity-log entry carries the message's true /// category rather than a hard-coded one. - /// The parked row is captured before the local - /// requeue write rather than re-read after it, so a concurrent - /// RemoveMessageAsync or DiscardParkedMessageAsync running - /// between the two storage calls cannot leave the standby in Parked - /// while the active node has already requeued — we always have the row in - /// hand for the Requeue replication. /// /// The identifier of the message to retry. /// True if successfully retried, false otherwise. @@ -1117,7 +1104,6 @@ public class StoreAndForwardService captured.RetryCount = 0; captured.LastError = null; captured.LastAttemptAt = null; - _replication?.ReplicateRequeue(captured); RaiseActivity("Retry", captured.Category, $"Parked message {messageId} moved back to queue"); @@ -1127,9 +1113,10 @@ public class StoreAndForwardService /// /// Permanently discards a parked message. /// - /// An operator discard is a buffer removal and is replicated - /// to the standby (as a ) so the - /// discarded message does not reappear after a failover. + /// An operator discard is a buffer removal. The delete is captured as a tombstone on + /// the replicated sf_messages table and carries the later HLC, so the discarded + /// message cannot reappear after a failover regardless of arrival order. This used to + /// be an explicit Remove operation sent to the standby. /// The activity-log entry carries the message's true /// category rather than a hard-coded one. /// @@ -1143,7 +1130,6 @@ public class StoreAndForwardService var success = await _storage.DiscardParkedMessageAsync(messageId); if (success) { - _replication?.ReplicateRemove(messageId); RaiseActivity("Discard", message?.Category ?? StoreAndForwardCategory.ExternalSystem, $"Parked message {messageId} discarded"); } diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs index fbe87cc6..54859796 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs @@ -65,7 +65,7 @@ public class StoreAndForwardStorage /// /// INSERT statement for a full message row. Shared by - /// and ; bind with + /// and ; bind with /// so the column list and the parameters never drift apart. /// private const string InsertMessageSql = @" @@ -151,39 +151,6 @@ public class StoreAndForwardStorage return (rows.Take(limit).ToList(), truncated); } - /// - /// Atomically replaces the entire buffer with in a - /// single transaction (delete-all then insert-all). Standby-side anti-entropy - /// apply: a peer-join resync overwrites the standby's divergent copy with the - /// active node's authoritative snapshot. Never call on an active node — - /// it discards every in-flight row. - /// - /// The full buffer snapshot to install. - /// A task that represents the asynchronous operation. - public async Task ReplaceAllAsync(IReadOnlyList messages) - { - await using var connection = OpenConnection(); - await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync(); - - await using (var deleteCmd = connection.CreateCommand()) - { - deleteCmd.Transaction = transaction; - deleteCmd.CommandText = "DELETE FROM sf_messages"; - await deleteCmd.ExecuteNonQueryAsync(); - } - - foreach (var message in messages) - { - await using var insertCmd = connection.CreateCommand(); - insertCmd.Transaction = transaction; - insertCmd.CommandText = InsertMessageSql; - BindMessageParameters(insertCmd, message); - await insertCmd.ExecuteNonQueryAsync(); - } - - await transaction.CommitAsync(); - } - /// /// Inserts a message, or updates every mutable column in place if a row with the /// same id already exists (ON CONFLICT(id) DO UPDATE). Used by the standby diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CompositionRootTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CompositionRootTests.cs index 488f186f..59d4bf31 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CompositionRootTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CompositionRootTests.cs @@ -471,7 +471,6 @@ public class SiteCompositionRootTests : IDisposable new object[] { typeof(IDataConnectionFactory) }, new object[] { typeof(StoreAndForwardStorage) }, new object[] { typeof(StoreAndForwardService) }, - new object[] { typeof(ReplicationService) }, new object[] { typeof(ISiteEventLogger) }, new object[] { typeof(IEventLogQueryService) }, new object[] { typeof(ISiteIdentityProvider) }, diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbWiringTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbWiringTests.cs index f9e8e3eb..4340a148 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbWiringTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbWiringTests.cs @@ -243,22 +243,65 @@ public class SiteLocalDbWiringTests : IDisposable } [Fact] - public void Site_LocalDb_DoesNotYetRegisterThePhase2Tables() + public void Site_LocalDb_RegistersExactlyTheTenReplicatedTables() { - // The "not yet" is the assertion that matters, and it is why this is an EXACTLY - // check rather than a Contains. Until the Task 14 cutover, the bespoke - // SiteReplicationActor and the StoreAndForward ReplicationService still own these - // tables. Registering them early would run two replicators over the same rows and - // hide a defect in either one behind the other's writes. + // The Phase 2 cutover. This is an EXACTLY check rather than a Contains in both + // directions, and both directions are load-bearing. // - // When Task 14 lands, this test does not get deleted — it gets inverted. + // Too few: the table silently stops replicating. Storage still works, every other + // test still passes, and the pair just quietly diverges — the failure mode Phase 1 + // existed to end. + // + // Too many: notification_lists and smtp_configurations must NOT be here. They are + // permanently empty by design, and registering them would open a standing + // replication channel whose only historical payload was plaintext SMTP passwords. + // A Contains-based test would never catch that. var db = _host.Services.GetRequiredService(); Assert.Equal( - ["OperationTracking", "site_events"], + [ + // Ordinal order: '_' (0x5F) sorts before 'b' (0x62), so + // data_connection_definitions precedes database_connections. + "OperationTracking", "data_connection_definitions", "database_connections", + "deployed_configurations", "external_systems", "native_alarm_state", + "sf_messages", "shared_scripts", "site_events", "static_attribute_overrides", + ], db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal).ToArray()); } + [Fact] + public void Site_LocalDb_DoesNotReplicateTheCentralOnlyNotificationTables() + { + // Stated separately from the exact-set test above because this one is a security + // property, not a wiring property, and deserves to fail with its own name. The + // tables exist (SiteStorageSchema creates them) — they simply must never be + // captured. See SiteLocalDbSetup.OnReady for the rationale. + var db = _host.Services.GetRequiredService(); + + Assert.DoesNotContain("notification_lists", db.ReplicatedTables.Keys); + Assert.DoesNotContain("smtp_configurations", db.ReplicatedTables.Keys); + } + + [Fact] + public void Site_LocalDb_ReplicatedPhase2TablesHaveTheirExpectedPrimaryKeys() + { + // RegisterReplicated refuses a table with no explicit PK, so reaching these + // assertions proves the DDL ran before registration. The composite keys are the + // interesting ones: LWW conflict resolution keys on the FULL PK, so a wrong or + // truncated key set would silently collapse distinct rows into one. + var db = _host.Services.GetRequiredService(); + + Assert.Equal(["id"], db.ReplicatedTables["sf_messages"].PkColumns); + Assert.Equal( + ["instance_unique_name"], db.ReplicatedTables["deployed_configurations"].PkColumns); + Assert.Equal( + ["instance_unique_name", "attribute_name"], + db.ReplicatedTables["static_attribute_overrides"].PkColumns); + Assert.Equal( + ["instance_unique_name", "source_canonical_name", "source_reference"], + db.ReplicatedTables["native_alarm_state"].PkColumns); + } + private static HashSet TableNames(ILocalDb db) { using var connection = db.CreateConnection(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs deleted file mode 100644 index 9e9f3cb2..00000000 --- a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs +++ /dev/null @@ -1,122 +0,0 @@ -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; -using ZB.MOM.WW.ScadaBridge.TestSupport; - -namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster; - -/// -/// 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. -/// -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); - - var fixture = await TwoNodeClusterFixture.StartAsync( - role: "site-int", portA: portHigh, portB: portLow); - - // The S&F stores AND SiteStorageService take an ILocalDb (LocalDb has no in-memory - // mode), so each node gets its own temp-file local databases. They are disposed - // AFTER the cluster is shut down — the actors hold connections while the systems - // are alive — and only then are the files (plus their WAL sidecars) deleted. - var localDbs = new List(); - - try - { - // Real S&F storage + replication actor per node, production default predicate - // (no isActiveOverride) — the exact wiring under test. - var (storageOldest, _, sfDbOldest, siteDbOldest) = - await CreateReplicationActorAsync(fixture.NodeA, "oldest"); - localDbs.Add(sfDbOldest); - localDbs.Add(siteDbOldest); - var (storageJoiner, _, sfDbJoiner, siteDbJoiner) = - await CreateReplicationActorAsync(fixture.NodeB, "joiner"); - localDbs.Add(sfDbJoiner); - localDbs.Add(siteDbJoiner); - - // 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")); - } - finally - { - await fixture.DisposeAsync(); - - foreach (var localDb in localDbs) - { - var path = localDb.Path; - localDb.Dispose(); - TestLocalDb.DeleteFiles(path); - } - } - } - - private static async Task<( - StoreAndForwardStorage Storage, IActorRef Actor, TestLocalDb SfLocalDb, TestLocalDb SiteLocalDb)> - CreateReplicationActorAsync(ActorSystem node, string tag) - { - var sfLocalDb = TestLocalDb.CreateTemp($"sf-resync-{tag}"); - var sfStorage = new StoreAndForwardStorage(sfLocalDb.Db, - NullLogger.Instance); - await sfStorage.InitializeAsync(); - var siteLocalDb = TestLocalDb.CreateTemp($"site-resync-{tag}"); - var siteStorage = new SiteStorageService(siteLocalDb.Db, - NullLogger.Instance); - var replicationService = new ReplicationService( - new StoreAndForwardOptions(), NullLogger.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.Instance, null, null, null, null)), - "site-replication"); - return (sfStorage, actor, sfLocalDb, siteLocalDb); - } - - 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> 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); - } -} diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairHarness.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairHarness.cs index dc37eefb..0bc73e80 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairHarness.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairHarness.cs @@ -129,40 +129,14 @@ public abstract class LocalDbSitePairHarness : IAsyncLifetime .Build(); return new ServiceCollection() - .AddZbLocalDb(config, db => - { - SiteLocalDbSetup.OnReady(db, config); - RegisterPhase2TablesUntilCutover(db); - }) + .AddZbLocalDb(config, db => SiteLocalDbSetup.OnReady(db, config)) .BuildServiceProvider(); } /// - /// Registers the Phase 2 tables for capture, which production OnReady does not do - /// until the Task 14 cutover. - /// - /// - /// Delete this method at Task 14, along with its call above. Until the cutover the - /// bespoke SiteReplicationActor and ReplicationService still own these - /// tables, so OnReady deliberately leaves them unregistered — but the convergence - /// specifications the cutover has to satisfy need capture triggers to mean anything. The - /// tables are empty at this point, so registering here captures nothing retroactively; - /// the ordering guarantee under test is unaffected. - /// - /// After Task 14 these registrations come from OnReady itself, and leaving this - /// method behind would mask a cutover that forgot one — the whole point of these tests. - /// - /// - private static void RegisterPhase2TablesUntilCutover(ILocalDb db) - { - foreach (var table in Phase2ReplicatedTables) - db.RegisterReplicated(table); - } - - /// - /// The eight tables Task 14 registers. Listed literally rather than derived from - /// production code, so that a cutover which registers the wrong set fails these tests - /// instead of agreeing with itself. + /// The eight tables the Phase 2 cutover registers. Listed literally rather than derived + /// from production code, so a registration that drifts fails these tests instead of + /// agreeing with itself. /// /// /// notification_lists and smtp_configurations are absent by design. They diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerActorTests.cs index f898add5..4f448aa0 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerActorTests.cs @@ -67,12 +67,14 @@ public class DeploymentManagerActorTests : TestKit, IDisposable null, // no stream manager in tests options, NullLogger.Instance, - null, - null, - null, - serviceProvider, - null, - configFetcher))); + // Named from here on. These trailing parameters are all optional and several + // share a type, so a positional list silently binds the wrong argument when the + // signature changes — which is exactly what removing replicationActor did. + dclManager: null, + healthCollector: null, + serviceProvider: serviceProvider, + loggerFactory: null, + configFetcher: configFetcher))); } private static string MakeConfigJson(string instanceName) @@ -301,8 +303,11 @@ public class DeploymentManagerActorTests : TestKit, IDisposable var dm = ActorOf(Props.Create(() => new DeploymentManagerActor( _storage, _compilationService, _sharedScriptLibrary, null, new SiteRuntimeOptions(), NullLogger.Instance, - null, null, null, null, null, null, - TimeSpan.FromMilliseconds(200), loader))); + // dclManager, healthCollector, serviceProvider, loggerFactory, configFetcher. + // Props.Create builds an expression tree, which rejects named arguments that + // are out of position, so the optional tail has to be padded positionally. + null, null, null, null, null, + startupLoadRetryInterval: TimeSpan.FromMilliseconds(200), configLoader: loader))); AwaitAssert(() => { @@ -892,13 +897,14 @@ public class DeploymentManagerActorTests : TestKit, IDisposable // Security cleanup. notification_lists and smtp_configurations can hold plaintext // SMTP passwords written by a pre-2026-07-10 build, and the ACTIVE node's artifact // apply is what clears them (DeploymentManagerActor.HandleDeployArtifacts). The - // standby's copy of this call lives in SiteReplicationActor and dies with it at - // Task 15, which makes this call site the ONLY remaining one. + // standby used to hold a second copy of this call in SiteReplicationActor; LocalDb + // Phase 2 deleted that actor, so this is now the ONLY call site that keeps the + // tables empty. // - // Nothing pins it today: ArtifactStorageTests covers the storage method, not the - // actor's call to it, so Task 16's edits to this actor could drop the call and every - // suite would stay green. That is precisely the kind of silent security regression - // this test exists to prevent — verified red-first by commenting out the call. + // ArtifactStorageTests covers the storage method, not the actor's call to it, so + // without this test the call could be dropped and every suite would stay green. + // That is precisely the kind of silent security regression it exists to prevent — + // verified red-first by commenting out the call. await SeedCentralOnlyRowsAsync(); Assert.Equal(1, await RowCountAsync("notification_lists")); Assert.Equal(1, await RowCountAsync("smtp_configurations")); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerCertReconcileTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerCertReconcileTests.cs index 7ea67d09..11c6a0b4 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerCertReconcileTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerCertReconcileTests.cs @@ -80,8 +80,7 @@ akka { private IActorRef CreateDeploymentManager() => ActorOf(Props.Create(() => new DeploymentManagerActor( _storage, _compilationService, _sharedScriptLibrary, - null, new SiteRuntimeOptions(), NullLogger.Instance, - null, null, null, null, null, null, null, null))); + null, new SiteRuntimeOptions(), NullLogger.Instance))); [Fact] public void SiteNodeJoined_PushesLocalTrustedCertsToJoinedNode() diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerLoggerFactoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerLoggerFactoryTests.cs index 087b7f0d..fcf6da18 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerLoggerFactoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerLoggerFactoryTests.cs @@ -103,11 +103,10 @@ public class DeploymentManagerLoggerFactoryTests : TestKit, IDisposable null, new SiteRuntimeOptions { StartupBatchSize = 100, StartupBatchDelayMs = 5 }, NullLogger.Instance, - null, - null, - null, - null, - loggerFactory))); + // dclManager, healthCollector, serviceProvider — padded positionally because + // Props.Create is an expression tree and rejects out-of-position named args. + null, null, null, + loggerFactory: loggerFactory))); // Allow async startup (load configs + staggered creation). await Task.Delay(2000); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerRedeployTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerRedeployTests.cs index a853d061..663243a3 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerRedeployTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DeploymentManagerRedeployTests.cs @@ -61,10 +61,11 @@ public class DeploymentManagerRedeployTests : TestKit, IDisposable null, new SiteRuntimeOptions(), NullLogger.Instance, + // dclManager — padded positionally because Props.Create is an expression tree + // and rejects out-of-position named args. null, - null, - healthCollector, - serviceProvider))); + healthCollector: healthCollector, + serviceProvider: serviceProvider))); } /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs deleted file mode 100644 index 0dae7d19..00000000 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs +++ /dev/null @@ -1,568 +0,0 @@ -using System.Collections.Concurrent; -using System.Diagnostics.Metrics; -using Akka.Actor; -using Akka.TestKit.Xunit2; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Logging.Abstractions; -using ZB.MOM.WW.ScadaBridge.SiteRuntime; -using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; -using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment; -using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages; -using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; -using ZB.MOM.WW.ScadaBridge.StoreAndForward; -using ZB.MOM.WW.ScadaBridge.Commons.Observability; -using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; -using ZB.MOM.WW.ScadaBridge.TestSupport; - -namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors; - -/// -/// Tests for 's notify-and-fetch config replication: -/// the active node now replicates an id-only (no inline -/// config JSON — killing the intra-site 128 KB frame trap), and the standby fetches the -/// config from central over HTTP and writes it with the older-write guard. -/// -public class SiteReplicationActorTests : TestKit, IDisposable -{ - // Cluster provider is required because SiteReplicationActor calls Cluster.Get in its ctor - // and subscribes to cluster events in PreStart. We use the in-memory TestTransport (not - // dot-netty) so no real socket is bound and no DNS lookup happens — the actor only needs - // the cluster extension to load; these tests never form a real two-node cluster. - private const string ClusterConfig = @" -akka { - actor { provider = cluster } - remote { - enabled-transports = [""akka.remote.test""] - test { - transport-class = ""Akka.Remote.Transport.TestTransport, Akka.Remote"" - applied-adapters = [] - registry-key = site-repl-test - local-address = ""test://site-repl@localhost:1"" - maximum-payload-bytes = 128000b - scheme-identifier = test - } - } - cluster { roles = [""site-test""] } - loglevel = WARNING -}"; - - private const string SiteRole = "site-test"; - - private readonly SiteStorageService _storage; - private readonly TestLocalDb _siteLocalDb; - private readonly TestLocalDb _sfLocalDb; - private readonly StoreAndForwardStorage _sfStorage; - private readonly ReplicationService _replicationService; - private readonly string _sfDbFile; - - public SiteReplicationActorTests() : base(ClusterConfig, "site-repl") - { - _sfDbFile = Path.Combine(Path.GetTempPath(), $"site-repl-sf-{Guid.NewGuid():N}.db"); - - // SiteStorageService takes an ILocalDb now; LocalDb has no in-memory mode, so the - // site store gets its own temp-file database alongside the S&F one. - _siteLocalDb = TestLocalDb.CreateTemp("site-repl-test"); - _storage = new SiteStorageService( - _siteLocalDb.Db, NullLogger.Instance); - _storage.InitializeAsync().GetAwaiter().GetResult(); - - _sfLocalDb = TestLocalDb.Create(_sfDbFile); - _sfStorage = new StoreAndForwardStorage( - _sfLocalDb.Db, NullLogger.Instance); - _sfStorage.InitializeAsync().GetAwaiter().GetResult(); - - _replicationService = new ReplicationService( - new StoreAndForwardOptions(), NullLogger.Instance); - } - - void IDisposable.Dispose() - { - Shutdown(); - // The master connection anchors the WAL — dispose before deleting. - var siteDbPath = _siteLocalDb.Path; - _siteLocalDb.Dispose(); - _sfLocalDb.Dispose(); - TestLocalDb.DeleteFiles(siteDbPath); - TestLocalDb.DeleteFiles(_sfDbFile); - } - - private IActorRef CreateReplicationActor(IDeploymentConfigFetcher fetcher) => - ActorOf(Props.Create(() => new SiteReplicationActor( - _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, fetcher))); - - private IActorRef CreateReplicationActor( - IDeploymentConfigFetcher fetcher, SiteRuntimeOptions options, TimeSpan retryDelay) => - ActorOf(Props.Create(() => new SiteReplicationActor( - _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, fetcher, null, options, retryDelay))); - - [Fact] - public async Task ReplicatedFetch_RetriesUpToConfigFetchRetryCount() - { - // The first two fetches fail transiently; the third succeeds. With - // ConfigFetchRetryCount = 3 the standby must retry to the third attempt and - // then guarded-write the fetched config (a short retry delay keeps the test fast). - var attempts = 0; - var fetcher = new FakeConfigFetcher(_ => - Interlocked.Increment(ref attempts) < 3 - ? Task.FromException(new InvalidOperationException("central hiccup")) - : Task.FromResult("{\"instanceUniqueName\":\"RetryPump\"}")); - var actor = CreateReplicationActor( - fetcher, new SiteRuntimeOptions { ConfigFetchRetryCount = 3 }, - TimeSpan.FromMilliseconds(50)); - - actor.Tell(new ApplyConfigDeploy( - "RetryPump", "dep-r1", "sha256:r1", true, - "http://central:9000", "tok-r1")); - - await AwaitAssertAsync(async () => - { - Assert.Equal(3, Volatile.Read(ref attempts)); - var configs = await _storage.GetAllDeployedConfigsAsync(); - Assert.Single(configs, c => c.InstanceUniqueName == "RetryPump"); - }, TimeSpan.FromSeconds(10)); - } - - [Fact] - public async Task ApplyConfigDeploy_StandbyFetchesConfigAndGuardedWrites() - { - // The standby receives an id-only ApplyConfigDeploy; it fetches the config from - // central using the message's coords, then guarded-writes the fetched config. - const string configJson = "{\"instanceUniqueName\":\"Pump1\"}"; - var fetcher = new FakeConfigFetcher(_ => Task.FromResult(configJson)); - var actor = CreateReplicationActor(fetcher); - - actor.Tell(new ApplyConfigDeploy( - "Pump1", "dep-100", "sha256:abc", true, - "http://central:9000", "tok-xyz")); - - // The continuation runs off-thread; await the guarded write landing. - await AwaitAssertAsync(async () => - { - var configs = await _storage.GetAllDeployedConfigsAsync(); - var row = Assert.Single(configs, c => c.InstanceUniqueName == "Pump1"); - Assert.Equal(configJson, row.ConfigJson); - Assert.Equal("dep-100", row.DeploymentId); - Assert.Equal("sha256:abc", row.RevisionHash); - Assert.True(row.IsEnabled); - }, TimeSpan.FromSeconds(5)); - - // The fetcher was called with the message's coords. - var call = Assert.Single(fetcher.Calls); - Assert.Equal("http://central:9000", call.BaseUrl); - Assert.Equal("dep-100", call.DeploymentId); - Assert.Equal("tok-xyz", call.Token); - } - - [Fact] - public async Task ApplyConfigDeploy_Superseded404_SkipsWriteAndActorSurvives() - { - // A 404 (superseded/expired) surfaces as DeploymentConfigFetchException{IsSuperseded}. - // The standby must skip the write, observe the exception (no crash), and stay alive. - var fetcher = new FakeConfigFetcher(_ => - Task.FromException( - new DeploymentConfigFetchException("expired", isSuperseded: true))); - var actor = CreateReplicationActor(fetcher); - - actor.Tell(new ApplyConfigDeploy( - "GonePump", "dep-stale", "sha256:gone", true, - "http://central:9000", "tok-stale")); - - // The fetch was attempted... - await AwaitAssertAsync(() => - { - Assert.Single(fetcher.Calls); - return Task.CompletedTask; - }, TimeSpan.FromSeconds(5)); - - // ...the actor did not crash (no Terminated to its watcher within the window)... - Watch(actor); - ExpectNoMsg(TimeSpan.FromMilliseconds(500)); - - // ...and nothing was written for the superseded instance. - var configs = await _storage.GetAllDeployedConfigsAsync(); - Assert.DoesNotContain(configs, c => c.InstanceUniqueName == "GonePump"); - } - - [Fact] - public async Task ApplyConfigDeploy_EmptyFetchCoords_SkipsFetchAndWrite() - { - // The direct DeployInstanceCommand cross-cluster wire path was retired in Task 14. - // This tests the defensive guard: if empty coords arrive, the actor must skip quietly - // — no FetchAsync("") call, no write — rather than erroring. - var fetcher = new FakeConfigFetcher(_ => Task.FromResult("never")); - var actor = CreateReplicationActor(fetcher); - - actor.Tell(new ApplyConfigDeploy( - "NoCoordsPump", "dep-direct", "sha256:nc", true, - CentralFetchBaseUrl: "", FetchToken: "")); - - // Give any (erroneous) async continuation time to run, then prove neither happened. - Watch(actor); - ExpectNoMsg(TimeSpan.FromMilliseconds(500)); - Assert.Empty(fetcher.Calls); - var configs = await _storage.GetAllDeployedConfigsAsync(); - Assert.DoesNotContain(configs, c => c.InstanceUniqueName == "NoCoordsPump"); - } - - [Fact] - public void ReplicateConfigDeploy_MapsToIdOnlyApplyConfigDeploy_ForPeer() - { - // The outbound mapping must forward an id-only ApplyConfigDeploy carrying the fetch - // coords (and NO inline config) to the peer. - var probe = CreateTestProbe(); - var fetcher = new FakeConfigFetcher(_ => Task.FromResult("unused")); - var actor = ActorOf(Props.Create(() => new ProbeForwardingReplicationActor( - _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, fetcher, probe.Ref))); - - actor.Tell(new ReplicateConfigDeploy( - "Pump2", "dep-200", "sha256:def", false, - "http://central:9000", "tok-abc")); - - var applied = probe.ExpectMsg(TimeSpan.FromSeconds(3)); - Assert.Equal("Pump2", applied.InstanceName); - Assert.Equal("dep-200", applied.DeploymentId); - Assert.Equal("sha256:def", applied.RevisionHash); - Assert.False(applied.IsEnabled); - Assert.Equal("http://central:9000", applied.CentralFetchBaseUrl); - Assert.Equal("tok-abc", applied.FetchToken); - } - - // ── Task 21: peer-join S&F buffer resync (anti-entropy) ── - - [Fact] - public void StandbyTrackingPeer_SendsResyncRequest() - { - var probe = CreateTestProbe(); - var actor = ActorOf(Props.Create(() => new ResyncTestActor( - _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, probe.Ref, () => false))); - - actor.Tell(new TriggerPeerTracked()); // stands in for TryTrackPeer's MemberUp path - - probe.ExpectMsg(TimeSpan.FromSeconds(3)); - } - - [Fact] - public void ActiveTrackingPeer_DoesNotRequestResync() - { - var probe = CreateTestProbe(); - var actor = ActorOf(Props.Create(() => new ResyncTestActor( - _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, probe.Ref, () => true))); - - actor.Tell(new TriggerPeerTracked()); - - probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // active node never requests a resync - } - - [Fact] - public async Task ActiveNode_AnswersResyncRequest_WithChunkedSnapshot() - { - // Post-R2-T5 the active node answers with byte-budgeted SfBufferSnapshotChunk(s) - // (a single small row rides one chunk) rather than the monolithic SfBufferSnapshot. - await _sfStorage.EnqueueAsync(NewSfMessage("m1")); - var probe = CreateTestProbe(); - var actor = ActorOf(Props.Create(() => new ResyncTestActor( - _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, probe.Ref, () => true))); - - actor.Tell(new RequestSfBufferResync(), TestActor); - - var chunk = ExpectMsg(TimeSpan.FromSeconds(3)); - Assert.Equal(1, chunk.TotalChunks); - Assert.Equal(1, chunk.Sequence); - Assert.Single(chunk.Messages); - Assert.False(chunk.Truncated); - } - - [Fact] - public async Task StandbyNode_AppliesSnapshot_ReplacingItsBuffer() - { - await _sfStorage.EnqueueAsync(NewSfMessage("stale")); - var probe = CreateTestProbe(); - var actor = ActorOf(Props.Create(() => new ResyncTestActor( - _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, probe.Ref, () => false))); - - actor.Tell(new SfBufferSnapshot(new List { NewSfMessage("fresh") }, false)); - - await AwaitAssertAsync(async () => - { - Assert.Null(await _sfStorage.GetMessageByIdAsync("stale")); - Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh")); - }, TimeSpan.FromSeconds(5)); - } - - // ── 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 - { 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(TimeSpan.FromSeconds(5)); - var rest = Enumerable.Range(1, first.TotalChunks - 1) - .Select(_ => ExpectMsg(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)); - } - - // ── R2 T6: standby chunk assembly + atomic apply + ack ── - - [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 { NewMessage("f1") }, false), TestActor); - actor.Tell(new SfBufferSnapshotChunk(resyncId, 2, 2, - new List { NewMessage("f2") }, false), TestActor); - - var ack = ExpectMsg(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 { NewMessage("orphan") }, false), TestActor); - actor.Tell(new SfBufferSnapshotChunk("new", 1, 1, - new List { NewMessage("fresh") }, false), TestActor); - - ExpectMsg(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 { NewMessage("x") }, false), TestActor); - ExpectNoMsg(TimeSpan.FromMilliseconds(300)); - } - - // ── R2 T7: active-side resync ack confirmation + telemetry ── - // - // NOTE (deviation from plan): the actor logs via Microsoft ILogger (NullLogger in - // tests), NOT Akka's EventStream, so the plan's EventFilter.Warning assertions can - // never observe these warnings. We observe the two OTel counters via a MeterListener - // instead — the equivalent, and stronger, observable signal. - - [Fact] - public async Task ActiveNode_ReceivingAck_CountsResyncCompleted() - { - long completed = 0; - using var listener = ListenCounter("scadabridge.store_and_forward.resync.completed", - m => Interlocked.Add(ref completed, m)); - - await _sfStorage.EnqueueAsync(NewMessage("m1")); - var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromSeconds(30)); - actor.Tell(new RequestSfBufferResync(), TestActor); - var chunk = ExpectMsg(TimeSpan.FromSeconds(5)); - - actor.Tell(new SfBufferResyncAck(chunk.ResyncId, 1), TestActor); - - await AwaitAssertAsync(() => - { - Assert.True(Interlocked.Read(ref completed) >= 1); // ack recorded the completion - return Task.CompletedTask; - }, TimeSpan.FromSeconds(5)); - } - - [Fact] - public async Task ActiveNode_MissingAck_WarnsAfterAckTimeout() - { - long ackMissing = 0; - using var listener = ListenCounter("scadabridge.store_and_forward.resync.ack_missing", - m => Interlocked.Add(ref ackMissing, m)); - - await _sfStorage.EnqueueAsync(NewMessage("m1")); - var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromMilliseconds(200)); - actor.Tell(new RequestSfBufferResync(), TestActor); - ExpectMsg(TimeSpan.FromSeconds(5)); - // No ack is sent → the ack window expires and the resync is counted unacknowledged. - - await AwaitAssertAsync(() => - { - Assert.True(Interlocked.Read(ref ackMissing) >= 1); - return Task.CompletedTask; - }, TimeSpan.FromSeconds(5)); - } - - /// Attaches a to a single ScadaBridge counter by name, - /// forwarding each recorded increment to . - private static MeterListener ListenCounter(string instrumentName, Action onMeasurement) - { - var listener = new MeterListener(); - listener.InstrumentPublished = (inst, l) => - { - if (inst.Meter.Name == ScadaBridgeTelemetry.MeterName && inst.Name == instrumentName) - l.EnableMeasurementEvents(inst); - }; - listener.SetMeasurementEventCallback((_, m, _, _) => onMeasurement(m)); - listener.Start(); - return listener; - } - - private static StoreAndForwardMessage NewSfMessage(string id) => new() - { - Id = id, - Category = StoreAndForwardCategory.ExternalSystem, - Target = "t", - PayloadJson = "{}", - RetryCount = 0, - MaxRetries = 50, - RetryIntervalMs = 30000, - CreatedAt = DateTimeOffset.UtcNow, - Status = StoreAndForwardMessageStatus.Pending, - }; - - /// - /// Builds a resync-test message with a settable payload (additive to - /// — the chunker sizes on PayloadJson length). - /// - private static StoreAndForwardMessage NewMessage(string id, string payloadJson = "{}") => new() - { - Id = id, - Category = StoreAndForwardCategory.ExternalSystem, - Target = "t", - PayloadJson = payloadJson, - RetryCount = 0, - MaxRetries = 50, - RetryIntervalMs = 30000, - CreatedAt = DateTimeOffset.UtcNow, - Status = StoreAndForwardMessageStatus.Pending, - }; - - /// Constructs a with the given active-node check - /// (the resync chunk/ack tests Tell to and expect from ). - /// is the active-side ack window seam (T7). - private IActorRef CreateResyncActor(Func isActive, TimeSpan? ackTimeout = null) => - ActorOf(Props.Create(() => new ResyncTestActor( - _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, CreateTestProbe().Ref, isActive, ackTimeout))); - - /// Test message: drives directly, - /// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer). - private sealed record TriggerPeerTracked; - - /// - /// Test subclass for the resync tests: captures peer sends to a probe, injects the - /// active-node check, and exposes via a test message. - /// - private sealed class ResyncTestActor : SiteReplicationActor - { - private readonly IActorRef _peerProbe; - - public ResyncTestActor( - SiteStorageService storage, StoreAndForwardStorage sfStorage, - ReplicationService replicationService, string siteRole, - ILogger logger, IActorRef peerProbe, Func isActive, - TimeSpan? ackTimeout = null) - : base(storage, sfStorage, replicationService, siteRole, logger, - configFetcher: null, isActiveOverride: isActive, resyncAckTimeout: ackTimeout) - { - _peerProbe = peerProbe; - Receive(_ => OnPeerTracked()); - } - - protected override void SendToPeer(object message) => _peerProbe.Tell(message, Self); - } - - /// - /// Test subclass exposing the peer send: is - /// overridden to forward to a probe so the outbound mapping can be asserted without a real - /// two-node cluster (a single-node TestKit has no peer address, so the real send is dropped). - /// - private sealed class ProbeForwardingReplicationActor : SiteReplicationActor - { - private readonly IActorRef _peerProbe; - - public ProbeForwardingReplicationActor( - SiteStorageService storage, StoreAndForwardStorage sfStorage, - ReplicationService replicationService, string siteRole, - ILogger logger, IDeploymentConfigFetcher configFetcher, - IActorRef peerProbe) - : base(storage, sfStorage, replicationService, siteRole, logger, configFetcher) - => _peerProbe = peerProbe; - - protected override void SendToPeer(object message) => _peerProbe.Tell(message, Self); - } - - /// - /// In-test fake : runs a per-deploymentId behavior - /// (return config JSON or throw, as a Task — mirroring the real async HTTP fetcher) and - /// records every call's coords thread-safely (the continuation runs on a pool thread). - /// - private sealed class FakeConfigFetcher : IDeploymentConfigFetcher - { - private readonly Func> _behavior; - public ConcurrentQueue<(string BaseUrl, string DeploymentId, string Token)> Calls { get; } = new(); - - public FakeConfigFetcher(Func> behavior) => _behavior = behavior; - - public async Task FetchAsync( - string centralFetchBaseUrl, string deploymentId, string token, CancellationToken ct) - { - Calls.Enqueue((centralFetchBaseUrl, deploymentId, token)); - await Task.Yield(); - return await _behavior(deploymentId); - } - } -} diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs deleted file mode 100644 index 43560a14..00000000 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs +++ /dev/null @@ -1,95 +0,0 @@ -using Akka.TestKit.Xunit2; -using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; -using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; -using ZB.MOM.WW.ScadaBridge.StoreAndForward; - -namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests; - -/// -/// Characterization pin for the chunked anti-entropy resync contract (review 02 round 2, -/// N2). (active→standby) and -/// (standby→active) ride intra-site Akka remoting on the default reflective-JSON wire -/// format. A rename/move, dropped setter, or non-default-constructible message would -/// silently break resync across a rolling upgrade and only surface as a divergent buffer -/// after a failover. These pin round-trip fidelity and type identity. (These messages are -/// NOT ClusterClient traffic, so they are intentionally absent from ClusterClientContractLockTests.) -/// -public class ResyncWireSerializationPinTests : TestKit -{ - private T RoundTrip(T message) - { - var serialization = Sys.Serialization; - var serializer = serialization.FindSerializerFor(message); - var bytes = serializer.ToBinary(message); - return (T)serialization.Deserialize(bytes, serializer.Identifier, message!.GetType()); - } - - private static StoreAndForwardMessage FullMessage() => new() - { - Id = Guid.NewGuid().ToString("N"), - Category = StoreAndForwardCategory.Notification, - Target = "Operators", - PayloadJson = "{\"notificationId\":\"abc\"}", - RetryCount = 4, - MaxRetries = 0, - RetryIntervalMs = 30000, - CreatedAt = DateTimeOffset.UtcNow, - LastAttemptAt = DateTimeOffset.UtcNow, - Status = StoreAndForwardMessageStatus.Parked, - LastError = "central rejected", - OriginInstanceName = "Plant.Pump3", - ExecutionId = Guid.NewGuid(), - SourceScript = "ScriptActor:MonitorSpeed", - ParentExecutionId = Guid.NewGuid(), - }; - - [Fact] - public void SfBufferSnapshotChunk_WithFullMessage_RoundTripsOnTheWire() - { - var message = FullMessage(); - var original = new SfBufferSnapshotChunk( - "resync-1", 2, 5, new List { message }, Truncated: true); - - var back = RoundTrip(original); - - Assert.Equal(original.ResyncId, back.ResyncId); - Assert.Equal(original.Sequence, back.Sequence); - Assert.Equal(original.TotalChunks, back.TotalChunks); - Assert.Equal(original.Truncated, back.Truncated); - var m = Assert.Single(back.Messages); - Assert.Equal(message.Id, m.Id); - Assert.Equal(message.Category, m.Category); - Assert.Equal(message.Target, m.Target); - Assert.Equal(message.PayloadJson, m.PayloadJson); - Assert.Equal(message.RetryCount, m.RetryCount); - Assert.Equal(message.MaxRetries, m.MaxRetries); - Assert.Equal(message.RetryIntervalMs, m.RetryIntervalMs); - Assert.Equal(message.CreatedAt, m.CreatedAt); - Assert.Equal(message.LastAttemptAt, m.LastAttemptAt); - Assert.Equal(message.Status, m.Status); - Assert.Equal(message.LastError, m.LastError); - Assert.Equal(message.OriginInstanceName, m.OriginInstanceName); - Assert.Equal(message.ExecutionId, m.ExecutionId); - Assert.Equal(message.SourceScript, m.SourceScript); - Assert.Equal(message.ParentExecutionId, m.ParentExecutionId); - } - - [Fact] - public void SfBufferResyncAck_RoundTripsOnTheWire() - { - var original = new SfBufferResyncAck("resync-1", 42); - - var back = RoundTrip(original); - - Assert.Equal(original.ResyncId, back.ResyncId); - Assert.Equal(original.RowCount, back.RowCount); - } - - // Type-identity pins: the reflective-JSON wire embeds CLR type manifests, so a - // rename/move of either type silently breaks resync across a rolling upgrade. - [Theory] - [InlineData(typeof(SfBufferSnapshotChunk), "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.SfBufferSnapshotChunk")] - [InlineData(typeof(SfBufferResyncAck), "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.SfBufferResyncAck")] - public void ResyncContract_TypeIdentity_IsPinned(Type type, string expectedFullName) => - Assert.Equal(expectedFullName, type.FullName); -} diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs index 4bf217e7..f01746fa 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs @@ -45,7 +45,6 @@ public class CachedCallAttemptEmissionTests : IAsyncLifetime, IDisposable _storage, _options, NullLogger.Instance, - replication: null, cachedCallObserver: _observer, siteId: "site-77"); } @@ -490,7 +489,6 @@ public class CachedCallAttemptEmissionTests : IAsyncLifetime, IDisposable RetryTimerInterval = TimeSpan.FromHours(1), // timer never fires in-test }, NullLogger.Instance, - replication: null, cachedCallObserver: observer, siteId: "site-77"); diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs deleted file mode 100644 index 19fa4b2b..00000000 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs +++ /dev/null @@ -1,236 +0,0 @@ -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Logging.Abstractions; -using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; -using ZB.MOM.WW.ScadaBridge.TestSupport; - -namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests; - -/// -/// WP-11: Tests for async replication to standby. -/// -public class ReplicationServiceTests : IAsyncLifetime, IDisposable -{ - private readonly TestLocalDb _localDb; - private readonly StoreAndForwardStorage _storage; - private readonly ReplicationService _replicationService; - - public ReplicationServiceTests() - { - _localDb = TestLocalDb.CreateTemp("RepTests"); - - _storage = new StoreAndForwardStorage(_localDb.Db, NullLogger.Instance); - - var options = new StoreAndForwardOptions { ReplicationEnabled = true }; - _replicationService = new ReplicationService( - options, NullLogger.Instance); - } - - public async Task InitializeAsync() => await _storage.InitializeAsync(); - - public Task DisposeAsync() => Task.CompletedTask; - - public void Dispose() - { - var path = _localDb.Path; - _localDb.Dispose(); - TestLocalDb.DeleteFiles(path); - } - - [Fact] - public void ReplicateEnqueue_NoHandler_DoesNotThrow() - { - var msg = CreateMessage("rep1"); - _replicationService.ReplicateEnqueue(msg); - } - - [Fact] - public async Task ReplicateEnqueue_WithHandler_ForwardsOperation() - { - ReplicationOperation? captured = null; - _replicationService.SetReplicationHandler(op => - { - captured = op; - return Task.CompletedTask; - }); - - var msg = CreateMessage("rep2"); - _replicationService.ReplicateEnqueue(msg); - - await Task.Delay(200); - - Assert.NotNull(captured); - Assert.Equal(ReplicationOperationType.Add, captured!.OperationType); - Assert.Equal("rep2", captured.MessageId); - } - - [Fact] - public async Task ReplicateRemove_WithHandler_ForwardsRemoveOperation() - { - ReplicationOperation? captured = null; - _replicationService.SetReplicationHandler(op => - { - captured = op; - return Task.CompletedTask; - }); - - _replicationService.ReplicateRemove("rep3"); - - await Task.Delay(200); - - Assert.NotNull(captured); - Assert.Equal(ReplicationOperationType.Remove, captured!.OperationType); - Assert.Equal("rep3", captured.MessageId); - } - - [Fact] - public async Task ReplicatePark_WithHandler_ForwardsParkOperation() - { - ReplicationOperation? captured = null; - _replicationService.SetReplicationHandler(op => - { - captured = op; - return Task.CompletedTask; - }); - - var msg = CreateMessage("rep4"); - _replicationService.ReplicatePark(msg); - - await Task.Delay(200); - - Assert.NotNull(captured); - Assert.Equal(ReplicationOperationType.Park, captured!.OperationType); - } - - [Fact] - public async Task ApplyReplicatedOperationAsync_Add_EnqueuesMessage() - { - var msg = CreateMessage("apply1"); - var operation = new ReplicationOperation(ReplicationOperationType.Add, "apply1", msg); - - await _replicationService.ApplyReplicatedOperationAsync(operation, _storage); - - var retrieved = await _storage.GetMessageByIdAsync("apply1"); - Assert.NotNull(retrieved); - } - - [Fact] - public async Task ApplyReplicatedOperationAsync_Remove_DeletesMessage() - { - var msg = CreateMessage("apply2"); - await _storage.EnqueueAsync(msg); - - var operation = new ReplicationOperation(ReplicationOperationType.Remove, "apply2", null); - await _replicationService.ApplyReplicatedOperationAsync(operation, _storage); - - var retrieved = await _storage.GetMessageByIdAsync("apply2"); - Assert.Null(retrieved); - } - - [Fact] - public async Task ApplyReplicatedOperationAsync_Park_UpdatesStatus() - { - var msg = CreateMessage("apply3"); - await _storage.EnqueueAsync(msg); - - var operation = new ReplicationOperation(ReplicationOperationType.Park, "apply3", msg); - await _replicationService.ApplyReplicatedOperationAsync(operation, _storage); - - var retrieved = await _storage.GetMessageByIdAsync("apply3"); - Assert.NotNull(retrieved); - Assert.Equal(StoreAndForwardMessageStatus.Parked, retrieved!.Status); - } - - [Fact] - public void ReplicateEnqueue_WhenReplicationDisabled_DoesNothing() - { - var options = new StoreAndForwardOptions { ReplicationEnabled = false }; - var service = new ReplicationService(options, NullLogger.Instance); - - bool handlerCalled = false; - service.SetReplicationHandler(_ => { handlerCalled = true; return Task.CompletedTask; }); - - service.ReplicateEnqueue(CreateMessage("disabled1")); - - Assert.False(handlerCalled); - } - - [Fact] - public async Task ReplicateEnqueue_HandlerThrows_DoesNotPropagateException() - { - _replicationService.SetReplicationHandler(_ => - throw new InvalidOperationException("standby down")); - - _replicationService.ReplicateEnqueue(CreateMessage("err1")); - - await Task.Delay(200); - // No exception -- fire-and-forget, best-effort - } - - // ── Task 10 (arch review 02): ordered, observable replication dispatch ── - - [Fact] - public void ReplicationOperations_AreDispatchedInIssueOrder() - { - var seen = new List<(ReplicationOperationType, string)>(); - _replicationService.SetReplicationHandler(op => - { - seen.Add((op.OperationType, op.MessageId)); - return Task.CompletedTask; - }); - - for (var i = 0; i < 200; i++) - { - _replicationService.ReplicateEnqueue(CreateMessage($"m{i}")); - _replicationService.ReplicateRemove($"m{i}"); - } - - // Inline dispatch: by the time the calls return, every op was handed to the - // handler, Add strictly before Remove per id. Pre-fix (Task.Run) this was - // racy in both count and order. - Assert.Equal(400, seen.Count); - for (var i = 0; i < 200; i++) - { - Assert.Equal((ReplicationOperationType.Add, $"m{i}"), seen[2 * i]); - Assert.Equal((ReplicationOperationType.Remove, $"m{i}"), seen[2 * i + 1]); - } - } - - [Fact] - public void ReplicationHandlerThrow_IsSwallowed_AndLoggedAtWarning() - { - var logger = new CapturingLogger(); - var service = new ReplicationService(new StoreAndForwardOptions(), logger); - service.SetReplicationHandler(_ => throw new InvalidOperationException("peer gone")); - - service.ReplicateRemove("m1"); // must not throw - - Assert.Contains(logger.Entries, e => e.Level == LogLevel.Warning); - } - - private static StoreAndForwardMessage CreateMessage(string id) - { - return new StoreAndForwardMessage - { - Id = id, - Category = StoreAndForwardCategory.ExternalSystem, - Target = "target", - PayloadJson = "{}", - RetryCount = 0, - MaxRetries = 50, - RetryIntervalMs = 30000, - CreatedAt = DateTimeOffset.UtcNow, - Status = StoreAndForwardMessageStatus.Pending - }; - } - - /// Minimal in-memory logger that records level + rendered message. - private sealed class CapturingLogger : ILogger - { - public List<(LogLevel Level, string Message)> Entries { get; } = new(); - public IDisposable? BeginScope(TState state) where TState : notnull => null; - public bool IsEnabled(LogLevel logLevel) => true; - public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, - Func formatter) - => Entries.Add((logLevel, formatter(state, exception))); - } -} diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationWireSerializationPinTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationWireSerializationPinTests.cs deleted file mode 100644 index f1d194f0..00000000 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationWireSerializationPinTests.cs +++ /dev/null @@ -1,82 +0,0 @@ -using Akka.TestKit.Xunit2; -using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; - -namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests; - -/// -/// Characterization pin for the intra-cluster S&F replication contract. A -/// (carrying a full ) -/// is Told from the active node to the standby node's replication actor and rides -/// the Akka default reflective-JSON wire format. Nothing tested that it round-trips -/// or that its type identity is stable — a rename/move, a dropped setter, or a -/// non-default-constructible message would silently break standby buffer sync and -/// only surface as a divergent buffer after a failover. -/// -/// The serializer swap (proto / explicit bindings) is deferred; until then this pin -/// is the guardrail. -/// -public class ReplicationWireSerializationPinTests : TestKit -{ - private T RoundTrip(T message) - { - var serialization = Sys.Serialization; - var serializer = serialization.FindSerializerFor(message); - var bytes = serializer.ToBinary(message); - return (T)serialization.Deserialize(bytes, serializer.Identifier, message!.GetType()); - } - - [Fact] - public void ReplicationOperation_WithFullMessage_RoundTripsOnTheWire() - { - var message = new StoreAndForwardMessage - { - Id = Guid.NewGuid().ToString("N"), - Category = StoreAndForwardCategory.Notification, - Target = "Operators", - PayloadJson = "{\"notificationId\":\"abc\"}", - RetryCount = 4, - MaxRetries = 0, - RetryIntervalMs = 30000, - CreatedAt = DateTimeOffset.UtcNow, - LastAttemptAt = DateTimeOffset.UtcNow, - Status = StoreAndForwardMessageStatus.Parked, - LastError = "central rejected", - OriginInstanceName = "Plant.Pump3", - ExecutionId = Guid.NewGuid(), - SourceScript = "ScriptActor:MonitorSpeed", - ParentExecutionId = Guid.NewGuid(), - }; - var original = new ReplicationOperation(ReplicationOperationType.Park, message.Id, message); - - var back = RoundTrip(original); - - Assert.Equal(original.OperationType, back.OperationType); - Assert.Equal(original.MessageId, back.MessageId); - Assert.NotNull(back.Message); - var m = back.Message!; - Assert.Equal(message.Id, m.Id); - Assert.Equal(message.Category, m.Category); - Assert.Equal(message.Target, m.Target); - Assert.Equal(message.PayloadJson, m.PayloadJson); - Assert.Equal(message.RetryCount, m.RetryCount); - Assert.Equal(message.MaxRetries, m.MaxRetries); - Assert.Equal(message.RetryIntervalMs, m.RetryIntervalMs); - Assert.Equal(message.CreatedAt, m.CreatedAt); - Assert.Equal(message.LastAttemptAt, m.LastAttemptAt); - Assert.Equal(message.Status, m.Status); - Assert.Equal(message.LastError, m.LastError); - Assert.Equal(message.OriginInstanceName, m.OriginInstanceName); - Assert.Equal(message.ExecutionId, m.ExecutionId); - Assert.Equal(message.SourceScript, m.SourceScript); - Assert.Equal(message.ParentExecutionId, m.ParentExecutionId); - } - - // Type-identity pins: the reflective-JSON wire embeds CLR type manifests, so a - // rename/move of either type silently breaks standby replication across a - // rolling upgrade. If one fails, you are making a wire-breaking change. - [Theory] - [InlineData(typeof(ReplicationOperation), "ZB.MOM.WW.ScadaBridge.StoreAndForward.ReplicationOperation")] - [InlineData(typeof(StoreAndForwardMessage), "ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardMessage")] - public void ReplicationContract_TypeIdentity_IsPinned(Type type, string expectedFullName) => - Assert.Equal(expectedFullName, type.FullName); -} diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs deleted file mode 100644 index 8080ce78..00000000 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs +++ /dev/null @@ -1,266 +0,0 @@ -using Microsoft.Extensions.Logging.Abstractions; -using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; -using ZB.MOM.WW.ScadaBridge.TestSupport; - -namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests; - -/// -/// StoreAndForward-001: the active node must forward every buffer operation -/// (add / remove / park) to the standby via the ReplicationService, so a -/// failover does not lose the buffer. -/// -public class StoreAndForwardReplicationTests : IAsyncLifetime, IDisposable -{ - private readonly TestLocalDb _localDb; - private readonly StoreAndForwardStorage _storage; - private readonly StoreAndForwardService _service; - private readonly List _replicated = new(); - - public StoreAndForwardReplicationTests() - { - _localDb = TestLocalDb.CreateTemp("ReplTests"); - - _storage = new StoreAndForwardStorage(_localDb.Db, NullLogger.Instance); - - var options = new StoreAndForwardOptions - { - DefaultRetryInterval = TimeSpan.Zero, - DefaultMaxRetries = 1, - RetryTimerInterval = TimeSpan.FromMinutes(10), - ReplicationEnabled = true, - }; - - var replication = new ReplicationService(options, NullLogger.Instance); - replication.SetReplicationHandler(op => - { - lock (_replicated) _replicated.Add(op); - return Task.CompletedTask; - }); - - _service = new StoreAndForwardService( - _storage, options, NullLogger.Instance, replication); - } - - public async Task InitializeAsync() => await _storage.InitializeAsync(); - public Task DisposeAsync() => Task.CompletedTask; - public void Dispose() - { - var path = _localDb.Path; - _localDb.Dispose(); - TestLocalDb.DeleteFiles(path); - } - - /// Replication is fire-and-forget (Task.Run); poll until the expected ops arrive. - private async Task> WaitForReplicationAsync(int count) - { - for (var i = 0; i < 100; i++) - { - lock (_replicated) - if (_replicated.Count >= count) return _replicated.ToList(); - await Task.Delay(20); - } - lock (_replicated) return _replicated.ToList(); - } - - [Fact] - public async Task BufferingAMessage_ReplicatesAnAddOperation() - { - // No handler registered → message is buffered → an Add is replicated. - var result = await _service.EnqueueAsync( - StoreAndForwardCategory.ExternalSystem, "api", """{}"""); - Assert.True(result.WasBuffered); - - var ops = await WaitForReplicationAsync(1); - Assert.Contains(ops, o => - o.OperationType == ReplicationOperationType.Add && o.MessageId == result.MessageId); - } - - [Fact] - public async Task SuccessfulRetry_ReplicatesARemoveOperation() - { - var calls = 0; - _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, - _ => ++calls == 1 - ? throw new HttpRequestException("transient") - : Task.FromResult(true)); - - var result = await _service.EnqueueAsync( - StoreAndForwardCategory.ExternalSystem, "api", """{}"""); - await _service.RetryPendingMessagesAsync(); - - var ops = await WaitForReplicationAsync(2); - Assert.Contains(ops, o => o.OperationType == ReplicationOperationType.Add); - Assert.Contains(ops, o => - o.OperationType == ReplicationOperationType.Remove && o.MessageId == result.MessageId); - } - - [Fact] - public async Task ParkedMessage_ReplicatesAParkOperation() - { - _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, - _ => throw new HttpRequestException("always fails")); - - var result = await _service.EnqueueAsync( - StoreAndForwardCategory.ExternalSystem, "api", """{}""", maxRetries: 1); - await _service.RetryPendingMessagesAsync(); - - var ops = await WaitForReplicationAsync(2); - Assert.Contains(ops, o => - o.OperationType == ReplicationOperationType.Park && o.MessageId == result.MessageId); - } - - /// - /// StoreAndForward-016: an operator discarding a parked message must replicate - /// a Remove so the standby's copy is also deleted (otherwise the discarded - /// message reappears in the parked list after a failover). - /// - [Fact] - public async Task DiscardingAParkedMessage_ReplicatesARemoveOperation() - { - _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, - _ => throw new HttpRequestException("always fails")); - - var result = await _service.EnqueueAsync( - StoreAndForwardCategory.ExternalSystem, "api", """{}""", maxRetries: 1); - await _service.RetryPendingMessagesAsync(); // -> parked - await WaitForReplicationAsync(2); - - var discarded = await _service.DiscardParkedMessageAsync(result.MessageId); - Assert.True(discarded); - - var ops = await WaitForReplicationAsync(3); - Assert.Contains(ops, o => - o.OperationType == ReplicationOperationType.Remove && o.MessageId == result.MessageId); - } - - /// - /// StoreAndForward-016: an operator retrying a parked message must replicate a - /// Requeue so the standby's copy moves back to Pending (otherwise it stays - /// Parked on the standby and the operator's retry is lost across a failover). - /// - [Fact] - public async Task RetryingAParkedMessage_ReplicatesARequeueOperation() - { - _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, - _ => throw new HttpRequestException("always fails")); - - var result = await _service.EnqueueAsync( - StoreAndForwardCategory.ExternalSystem, "api", """{}""", maxRetries: 1); - await _service.RetryPendingMessagesAsync(); // -> parked - await WaitForReplicationAsync(2); - - var retried = await _service.RetryParkedMessageAsync(result.MessageId); - Assert.True(retried); - - var ops = await WaitForReplicationAsync(3); - var requeue = ops.SingleOrDefault(o => - o.OperationType == ReplicationOperationType.Requeue && o.MessageId == result.MessageId); - Assert.NotNull(requeue); - Assert.NotNull(requeue!.Message); - Assert.Equal(StoreAndForwardMessageStatus.Pending, requeue.Message!.Status); - } - - /// - /// StoreAndForward-016: the standby applies a Requeue by moving its row back to - /// Pending with retry_count = 0, mirroring the active node's local state. - /// - [Fact] - public async Task ApplyReplicatedOperation_Requeue_MovesStandbyRowBackToPending() - { - var replication = new ReplicationService( - new StoreAndForwardOptions { ReplicationEnabled = true }, - NullLogger.Instance); - - var parked = new StoreAndForwardMessage - { - Id = "requeue1", - Category = StoreAndForwardCategory.ExternalSystem, - Target = "api", - PayloadJson = "{}", - RetryCount = 5, - MaxRetries = 1, - RetryIntervalMs = 0, - CreatedAt = DateTimeOffset.UtcNow, - Status = StoreAndForwardMessageStatus.Parked, - }; - await _storage.EnqueueAsync(parked); - - var requeued = new StoreAndForwardMessage - { - Id = parked.Id, - Category = parked.Category, - Target = parked.Target, - PayloadJson = parked.PayloadJson, - RetryCount = 0, - MaxRetries = parked.MaxRetries, - RetryIntervalMs = parked.RetryIntervalMs, - CreatedAt = parked.CreatedAt, - Status = StoreAndForwardMessageStatus.Pending, - }; - await replication.ApplyReplicatedOperationAsync( - new ReplicationOperation(ReplicationOperationType.Requeue, parked.Id, requeued), - _storage); - - var row = await _storage.GetMessageByIdAsync(parked.Id); - Assert.NotNull(row); - Assert.Equal(StoreAndForwardMessageStatus.Pending, row!.Status); - Assert.Equal(0, row.RetryCount); - } - - private static ReplicationService NewReplicationService() => - new(new StoreAndForwardOptions { ReplicationEnabled = true }, - NullLogger.Instance); - - private static StoreAndForwardMessage NewMessage(string id) => new() - { - Id = id, - Category = StoreAndForwardCategory.ExternalSystem, - Target = "api", - PayloadJson = "{}", - RetryCount = 0, - MaxRetries = 1, - RetryIntervalMs = 0, - CreatedAt = DateTimeOffset.UtcNow, - Status = StoreAndForwardMessageStatus.Pending, - }; - - /// - /// A Park (or Requeue) whose original Add was lost — e.g. the Add's - /// fire-and-forget replication dropped — must still materialise the row on the - /// standby. The full message rides in the operation, so an upsert self-heals; - /// a blind UPDATE would affect 0 rows and the row would be gone forever. - /// - [Fact] - public async Task ApplyReplicatedPark_WhenAddWasLost_MaterializesTheParkedRow() - { - var service = NewReplicationService(); - var msg = NewMessage("lost-add-1"); // never Added on this (standby) storage - msg.Status = StoreAndForwardMessageStatus.Parked; - - await service.ApplyReplicatedOperationAsync( - new ReplicationOperation(ReplicationOperationType.Park, msg.Id, msg), _storage); - - var row = await _storage.GetMessageByIdAsync("lost-add-1"); - Assert.NotNull(row); // pre-fix: blind UPDATE affected 0 rows, row is gone forever - Assert.Equal(StoreAndForwardMessageStatus.Parked, row!.Status); - } - - /// - /// A duplicate Add (e.g. after Task 21's anti-entropy resync re-issues an Add - /// the standby already holds) must not violate the PK — the upsert applies - /// newest-wins instead of throwing. - /// - [Fact] - public async Task ApplyReplicatedAdd_Twice_IsIdempotent_NewestWins() - { - var service = NewReplicationService(); - var msg = NewMessage("dup-add"); - await service.ApplyReplicatedOperationAsync( - new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage); - msg.RetryCount = 3; - await service.ApplyReplicatedOperationAsync( // pre-fix: SqliteException PK violation - new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage); - - Assert.Equal(3, (await _storage.GetMessageByIdAsync("dup-add"))!.RetryCount); - } -} diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardSiteEventTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardSiteEventTests.cs index 52144ee7..b21acfd8 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardSiteEventTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardSiteEventTests.cs @@ -54,7 +54,7 @@ public class StoreAndForwardSiteEventTests : IAsyncLifetime, IDisposable _service = new StoreAndForwardService( _storage, _options, NullLogger.Instance, - replication: null, cachedCallObserver: null, siteId: "site-a", + cachedCallObserver: null, siteId: "site-a", siteEventLogger: _siteLog); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs index 668a3007..18ce27c2 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs @@ -769,16 +769,15 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable Assert.True(truncated); // a third row exists beyond the limit } - [Fact] - public async Task ReplaceAll_SwapsTheEntireBuffer_Atomically() - { - await _storage.EnqueueAsync(NewMsg("stale")); - await _storage.ReplaceAllAsync(new[] { NewMsg("fresh-1"), NewMsg("fresh-2") }); - - Assert.Null(await _storage.GetMessageByIdAsync("stale")); - Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-1")); - Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-2")); - } + // ReplaceAll_SwapsTheEntireBuffer_Atomically was DELETED with ReplaceAllAsync in + // LocalDb Phase 2, and deliberately not replaced. It asserted a destructive + // delete-all-then-insert-all, which was the standby's anti-entropy apply: a resync + // overwrote the standby's copy with the active node's snapshot. sf_messages is now a + // replicated table, so a mass DELETE would be CAPTURED and shipped to the peer — the + // method is not merely unused, it is unsafe to keep. LocalDb's own snapshot resync + // merges per row under last-writer-wins and never deletes, which is what + // LocalDbConfigConvergenceTests.ANodeWithNewerLocalRows_KeepsThem_WhenAPeerSnapshotArrives + // now covers. // ── Task 23: oldest-parked-age health signal ──