feat(localdb)!: replicate site config + sf_messages via CDC, delete the bespoke replicators

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
This commit is contained in:
Joseph Doherty
2026-07-20 04:20:05 -04:00
parent df0c6031ba
commit 037798b367
26 changed files with 160 additions and 2513 deletions
@@ -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 <c>cluster.State.Leader</c>.
/// <para>
/// Lives in Communication (not Host) so BOTH <c>SiteCommunicationActor</c> and
/// <c>SiteReplicationActor</c> can default to it — Host cannot be referenced from either.
/// The Host's <c>ClusterActivityEvaluator.SelfIsOldest</c> delegates here, so the S&amp;F
/// delivery gate (<c>IClusterNodeProvider.SelfIsPrimary</c>), the resync authority checks,
/// and the heartbeat IsActive stamp all share one implementation.
/// Lives in Communication (not Host) so <c>SiteCommunicationActor</c> can default to it —
/// Host cannot be referenced from there. The Host's
/// <c>ClusterActivityEvaluator.SelfIsOldest</c> delegates here, so the S&amp;F delivery gate
/// (<c>IClusterNodeProvider.SelfIsPrimary</c>) and the heartbeat IsActive stamp share one
/// implementation.
/// <para>
/// It also backed <c>SiteReplicationActor</c>'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.
/// </para>
/// </para>
/// </summary>
public static class ActiveNodeEvaluator
@@ -766,37 +766,18 @@ akka {{
var deploymentConfigFetcher =
_serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment.IDeploymentConfigFetcher>();
// Create SiteReplicationActor on every node (not a singleton)
var sfStorage = _serviceProvider.GetRequiredService<StoreAndForwardStorage>();
var replicationService = _serviceProvider.GetRequiredService<ReplicationService>();
var replicationLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
.CreateLogger<SiteReplicationActor>();
// 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<ZB.MOM.WW.ScadaBridge.HealthMonitoring.IClusterNodeProvider>();
Func<bool>? activeNodeCheck = clusterNodeProvider != null
? () => clusterNodeProvider.SelfIsPrimary
: null;
var replicationActor = _actorSystem!.ActorOf(
Props.Create(() => new SiteReplicationActor(
storage, sfStorage, replicationService, siteRole, replicationLogger,
deploymentConfigFetcher, activeNodeCheck, siteRuntimeOptionsValue, null)),
"site-replication");
// 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
@@ -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.
@@ -52,7 +52,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// </summary>
private readonly ILoggerFactory _loggerFactory;
private readonly IActorRef? _dclManager;
private readonly IActorRef? _replicationActor;
private readonly ISiteHealthCollector? _healthCollector;
private readonly IServiceProvider? _serviceProvider;
/// <summary>
@@ -166,7 +165,6 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
SiteRuntimeOptions options,
ILogger<DeploymentManagerActor> 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);
@@ -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;
/// <summary>
/// Runs on every site node (not a singleton). Handles both config and S&amp;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.
/// </summary>
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<SiteReplicationActor> _logger;
private readonly Cluster _cluster;
private readonly Func<bool> _isActive;
private readonly int _configFetchRetryCount;
private readonly TimeSpan _configFetchRetryDelay;
private Address? _peerAddress;
/// <summary>Akka timer scheduler injected by the framework via <see cref="IWithTimers"/>.</summary>
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<int, List<StoreAndForwardMessage>> _assemblingChunks = new();
private const string ResyncAssemblyTimerKey = "sf-resync-assembly-timeout";
/// <summary>How long a partial chunk assembly may wait for its missing chunks before
/// being discarded (a lost chunk = lost resync; the next peer-track retries). Ctor
/// test seam; production default 30 s.</summary>
private readonly TimeSpan _resyncAssemblyTimeout;
// ── Resync delivery confirmation (active side; actor-thread only) ──
private string? _pendingAckResyncId;
private const string ResyncAckTimerKey = "sf-resync-ack-timeout";
/// <summary>How long the active node waits for the standby's <see cref="SfBufferResyncAck"/>
/// before warning + counting the resync as unacknowledged (lost chunks / dead peer). Ctor
/// test seam; production default 60 s.</summary>
private readonly TimeSpan _resyncAckTimeout;
/// <summary>
/// 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.
/// </summary>
private const int MaxResyncRows = 10_000;
/// <summary>
/// Estimated per-chunk payload budget for a resync snapshot. Akka remoting's default
/// <c>maximum-frame-size</c> is 128 000 bytes and <c>BuildHocon</c> sets no override,
/// so the monolithic <see cref="SfBufferSnapshot"/> 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.
/// </summary>
internal const int MaxResyncChunkBytes = 64_000;
/// <summary>Row cap per resync chunk (bounds a chunk even when every row is tiny).</summary>
internal const int MaxResyncChunkRows = 200;
/// <summary>
/// Splits a resync snapshot into chunks that fit Akka remoting's default
/// 128 000-byte frame (review 02 round 2, N2): rows accumulate until the estimated
/// payload budget or the row cap is hit. Estimation is payload-dominated
/// (payload_json length + 512 bytes fixed overhead per row); a single row whose
/// payload exceeds the budget ships alone (Warning at the call site). Order is
/// preserved (oldest-first, matching GetAllMessagesAsync).
/// </summary>
internal static List<List<StoreAndForwardMessage>> ChunkForRemoting(
IReadOnlyList<StoreAndForwardMessage> rows, int maxChunkBytes, int maxChunkRows)
{
var chunks = new List<List<StoreAndForwardMessage>>();
var current = new List<StoreAndForwardMessage>();
var currentBytes = 0;
foreach (var row in rows)
{
var estimate = (row.PayloadJson?.Length ?? 0) + 512;
if (current.Count > 0 && (currentBytes + estimate > maxChunkBytes || current.Count >= maxChunkRows))
{
chunks.Add(current);
current = new List<StoreAndForwardMessage>();
currentBytes = 0;
}
current.Add(row);
currentBytes += estimate;
}
if (current.Count > 0) chunks.Add(current);
return chunks;
}
/// <summary>
/// Initializes a new <see cref="SiteReplicationActor"/> and registers Akka message handlers.
/// </summary>
/// <param name="storage">Service for accessing local site storage.</param>
/// <param name="sfStorage">Store-and-forward SQLite storage for replication of buffered messages.</param>
/// <param name="replicationService">Service providing replication transport logic.</param>
/// <param name="siteRole">Akka cluster role used to identify peer nodes to replicate to.</param>
/// <param name="logger">Logger instance.</param>
/// <param name="configFetcher">
/// Fetches a deployed instance's config JSON from central over HTTP. Used by the
/// notify-and-fetch standby apply path (<see cref="HandleApplyConfigDeploy"/>): 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.
/// </param>
/// <param name="isActiveOverride">
/// Active-node check that gates the buffer-resync roles (a standby requests a
/// resync, the active node answers). Production wiring passes the Host's
/// <c>IClusterNodeProvider.SelfIsPrimary</c> delegate (the same instance gating the
/// S&amp;F delivery sweep); null falls back to the shared oldest-Up evaluator
/// (<see cref="Communication.ClusterState.ActiveNodeEvaluator"/>).
/// </param>
/// <param name="options">Site runtime options, including the config-fetch retry count; production defaults apply when null.</param>
/// <param name="configFetchRetryDelay">Delay between config-fetch retry attempts; defaults to 2 seconds when null.</param>
public SiteReplicationActor(
SiteStorageService storage,
StoreAndForwardStorage sfStorage,
ReplicationService replicationService,
string siteRole,
ILogger<SiteReplicationActor> logger,
IDeploymentConfigFetcher? configFetcher = null,
Func<bool>? 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<ClusterEvent.MemberUp>(HandleMemberUp);
Receive<ClusterEvent.MemberRemoved>(HandleMemberRemoved);
Receive<ClusterEvent.CurrentClusterState>(HandleCurrentClusterState);
// Outbound — forward to peer
Receive<ReplicateConfigDeploy>(msg => SendToPeer(new ApplyConfigDeploy(
msg.InstanceName, msg.DeploymentId, msg.RevisionHash, msg.IsEnabled,
msg.CentralFetchBaseUrl, msg.FetchToken)));
Receive<ReplicateConfigRemove>(msg => SendToPeer(new ApplyConfigRemove(msg.InstanceName)));
Receive<ReplicateConfigSetEnabled>(msg => SendToPeer(new ApplyConfigSetEnabled(
msg.InstanceName, msg.IsEnabled)));
Receive<ReplicateArtifacts>(msg => SendToPeer(new ApplyArtifacts(msg.Command)));
Receive<ReplicateStoreAndForward>(msg => SendToPeer(new ApplyStoreAndForward(msg.Operation)));
// Inbound — apply from peer
Receive<ApplyConfigDeploy>(HandleApplyConfigDeploy);
Receive<ApplyConfigRemove>(HandleApplyConfigRemove);
Receive<ApplyConfigSetEnabled>(HandleApplyConfigSetEnabled);
Receive<ApplyArtifacts>(HandleApplyArtifacts);
Receive<ApplyStoreAndForward>(HandleApplyStoreAndForward);
// Anti-entropy — full S&F buffer resync on peer (re)join
Receive<RequestSfBufferResync>(HandleRequestSfBufferResync);
Receive<SfResyncSnapshotLoaded>(HandleSfResyncSnapshotLoaded);
Receive<SfBufferSnapshotChunk>(HandleSfBufferSnapshotChunk);
Receive<ResyncAssemblyTimedOut>(HandleResyncAssemblyTimedOut);
Receive<SfBufferResyncAck>(msg =>
{
if (msg.ResyncId == _pendingAckResyncId)
{
_pendingAckResyncId = null;
Timers.Cancel(ResyncAckTimerKey);
}
ScadaBridgeTelemetry.RecordSfResyncCompleted();
_logger.LogInformation("S&F resync {ResyncId} acknowledged by standby: {Rows} row(s) applied",
msg.ResyncId, msg.RowCount);
});
Receive<ResyncAckTimedOut>(msg =>
{
if (msg.ResyncId != _pendingAckResyncId) return;
_pendingAckResyncId = null;
ScadaBridgeTelemetry.RecordSfResyncAckMissing();
_logger.LogWarning(
"S&F resync {ResyncId} was never acknowledged within {Window} — snapshot chunks may have been lost (frame drop / dead peer); the next peer-track retries",
msg.ResyncId, _resyncAckTimeout);
});
Receive<SfBufferSnapshot>(HandleSfBufferSnapshot); // legacy monolithic handler — retained for rolling compat
}
/// <inheritdoc />
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);
}
/// <inheritdoc />
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();
}
}
/// <summary>
/// Side-effect run whenever a peer is (re)tracked. A <b>standby</b> requests a
/// full S&amp;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. <see langword="protected virtual"/> so tests can drive it without a
/// real two-node cluster.
/// </summary>
protected virtual void OnPeerTracked()
{
if (!SafeIsActive())
{
SendToPeer(new RequestSfBufferResync());
}
}
/// <summary>
/// Repo-standard active-node check: this node is active when it is the OLDEST Up
/// member carrying the site role — the same oldest-Up semantics as the S&F delivery
/// gate (IClusterNodeProvider.SelfIsPrimary → ClusterActivityEvaluator → shared
/// ActiveNodeEvaluator). NEVER the cluster leader: leadership is lowest-address and
/// diverges from singleton/delivery placement permanently after the lower-address
/// node restarts — the divergence that made the delivering node wipe its own live
/// buffer via a wrong-direction resync (review 02 round 2, N1 Critical). Any other
/// state reports standby — safe-by-default.
/// </summary>
private bool DefaultIsActive() =>
Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(_cluster, _siteRole);
/// <summary>
/// Evaluates the active-node check, treating a throwing check as standby
/// (safe-by-default: a standby never delivers or answers resyncs).
/// </summary>
private bool SafeIsActive()
{
try
{
return _isActive();
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Active-node check threw; treating node as standby");
return false;
}
}
/// <summary>
/// Forwards a replication message to the tracked peer node's <c>site-replication</c> actor
/// (fire-and-forget, dropped when no peer is tracked). <see langword="protected virtual"/>
/// so tests can intercept the peer send without standing up a real two-node cluster.
/// </summary>
/// <param name="message">The replication message to forward to the peer.</param>
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);
});
}
/// <summary>
/// Active-node side of the anti-entropy resync: answers a standby's
/// <see cref="RequestSfBufferResync"/> with a sequence of byte-budgeted
/// <see cref="SfBufferSnapshotChunk"/>s (up to <see cref="MaxResyncRows"/> 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.
/// </summary>
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));
}
/// <summary>
/// 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.
/// </summary>
private void HandleSfResyncSnapshotLoaded(SfResyncSnapshotLoaded msg)
{
var chunks = ChunkForRemoting(msg.Messages, MaxResyncChunkBytes, MaxResyncChunkRows);
if (chunks.Count == 0) chunks.Add(new List<StoreAndForwardMessage>()); // empty buffer still resyncs (clears the standby)
var resyncId = Guid.NewGuid().ToString("N");
for (var i = 0; i < chunks.Count; i++)
{
if (chunks[i].Count == 1 && (chunks[i][0].PayloadJson?.Length ?? 0) + 512 > MaxResyncChunkBytes)
_logger.LogWarning(
"Resync row {Id} alone exceeds the chunk budget ({Bytes}B payload); sending solo — it may exceed the remoting frame",
chunks[i][0].Id, chunks[i][0].PayloadJson?.Length ?? 0);
msg.ReplyTo.Tell(new SfBufferSnapshotChunk(resyncId, i + 1, chunks.Count, chunks[i], msg.Truncated), Self);
}
_logger.LogInformation(
"Answered S&F resync request with {Rows} row(s) in {Chunks} chunk(s), resyncId={ResyncId}",
msg.Messages.Count, chunks.Count, resyncId);
// 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);
}
/// <summary>
/// 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.
/// </summary>
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");
});
}
/// <summary>
/// Standby-node side of the chunked anti-entropy resync: accumulates the chunks of one
/// <c>ResyncId</c>, and once all have arrived, assembles them in sequence order and
/// replaces the local buffer atomically, then acks. A new <c>ResyncId</c> discards any
/// stale partial assembly (review 02 round 2, N2). An active node ignores chunks.
/// </summary>
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();
}
/// <summary>Internal: the resync snapshot finished loading; chunk and send to the requester.</summary>
internal sealed record SfResyncSnapshotLoaded(
IActorRef ReplyTo, List<StoreAndForwardMessage> Messages, bool Truncated);
/// <summary>Internal: a partial chunk assembly for <paramref name="ResyncId"/> exceeded its window.</summary>
internal sealed record ResyncAssemblyTimedOut(string ResyncId);
/// <summary>Internal: the active node's ack window for <paramref name="ResyncId"/> expired.</summary>
internal sealed record ResyncAckTimedOut(string ResyncId);
}
/// <summary>
/// Standby→active: request a full S&amp;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.
/// </summary>
public sealed record RequestSfBufferResync;
/// <summary>
/// Active→standby: full-buffer snapshot. <paramref name="Truncated"/> is true when
/// the active node's buffer exceeded <c>MaxResyncRows</c> (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.
/// </summary>
public sealed record SfBufferSnapshot(List<StoreAndForwardMessage> Messages, bool Truncated);
/// <summary>
/// Active→standby: one sequenced chunk of a full-buffer anti-entropy snapshot
/// (review 02 round 2, N2 — the monolithic <see cref="SfBufferSnapshot"/> exceeds Akka
/// remoting's default 128 000-byte frame for any realistic backlog). All chunks of one
/// resync share <paramref name="ResyncId"/>; <paramref name="Sequence"/> is 1-based up to
/// <paramref name="TotalChunks"/>. Additive message — the legacy monolithic snapshot
/// handler is retained for rolling upgrades. Crosses intra-site Akka remoting (NOT
/// ClusterClient — ClusterClientContractLockTests is intentionally not involved).
/// </summary>
public sealed record SfBufferSnapshotChunk(
string ResyncId, int Sequence, int TotalChunks,
List<StoreAndForwardMessage> Messages, bool Truncated);
/// <summary>
/// Standby→active: delivery confirmation — the standby assembled all chunks of
/// <paramref name="ResyncId"/> and applied them atomically (<paramref name="RowCount"/>
/// rows installed). Absence within the ack window is surfaced by the active node
/// (Warning + counter) — the silent-loss mode N2 flagged.
/// </summary>
public sealed record SfBufferResyncAck(string ResyncId, int RowCount);
@@ -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.
/// <summary>Outbound: tell the peer to fetch+apply a deployed instance config by id (notify-and-fetch; no inline config).</summary>
public record ReplicateConfigDeploy(
string InstanceName, string DeploymentId, string RevisionHash, bool IsEnabled,
string CentralFetchBaseUrl, string FetchToken);
/// <summary>Outbound: replicate removal of a deployed instance config to the peer node.</summary>
public record ReplicateConfigRemove(string InstanceName);
/// <summary>Outbound: replicate an instance enabled/disabled flag change to the peer node.</summary>
public record ReplicateConfigSetEnabled(string InstanceName, bool IsEnabled);
/// <summary>Outbound: replicate a system-wide artifact deployment (shared scripts, external systems, etc.) to the peer node.</summary>
public record ReplicateArtifacts(DeployArtifactsCommand Command);
/// <summary>Outbound: replicate a store-and-forward buffer mutation (enqueue/dequeue/park/etc.) to the peer node.</summary>
public record ReplicateStoreAndForward(ReplicationOperation Operation);
// Inbound messages — received from the peer's SiteReplicationActor
// and applied to local SQLite storage.
/// <summary>Inbound: peer-replicated config deploy — the standby fetches the config by id and writes it (guarded).</summary>
public record ApplyConfigDeploy(
string InstanceName, string DeploymentId, string RevisionHash, bool IsEnabled,
string CentralFetchBaseUrl, string FetchToken);
/// <summary>Inbound: apply peer-replicated removal of a deployed instance config to local SQLite.</summary>
public record ApplyConfigRemove(string InstanceName);
/// <summary>Inbound: apply a peer-replicated instance enabled/disabled flag change to local SQLite.</summary>
public record ApplyConfigSetEnabled(string InstanceName, bool IsEnabled);
/// <summary>Inbound: apply a peer-replicated system-wide artifact deployment to local SQLite.</summary>
public record ApplyArtifacts(DeployArtifactsCommand Command);
/// <summary>Inbound: apply a peer-replicated store-and-forward buffer mutation to the local buffer.</summary>
public record ApplyStoreAndForward(ReplicationOperation Operation);
@@ -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;
/// <summary>
/// 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.
/// </summary>
public class ReplicationService
{
private readonly StoreAndForwardOptions _options;
private readonly ILogger<ReplicationService> _logger;
private Func<ReplicationOperation, Task>? _replicationHandler;
/// <summary>Initializes a new instance of <see cref="ReplicationService"/>.</summary>
/// <param name="options">Store-and-forward configuration options.</param>
/// <param name="logger">Logger instance.</param>
public ReplicationService(
StoreAndForwardOptions options,
ILogger<ReplicationService> logger)
{
_options = options;
_logger = logger;
}
/// <summary>
/// Sets the handler for forwarding replication operations to the standby node.
/// Typically wraps Akka Tell to the standby's replication actor.
/// </summary>
/// <param name="handler">The async delegate that forwards each replication operation to the standby.</param>
public void SetReplicationHandler(Func<ReplicationOperation, Task> handler)
{
_replicationHandler = handler;
}
/// <summary>
/// Replicates an enqueue operation to standby (fire-and-forget).
/// </summary>
/// <param name="message">The message that was enqueued on the active node.</param>
public void ReplicateEnqueue(StoreAndForwardMessage message)
{
if (!_options.ReplicationEnabled || _replicationHandler == null) return;
FireAndForget(new ReplicationOperation(
ReplicationOperationType.Add,
message.Id,
message));
}
/// <summary>
/// Replicates a remove operation to standby (fire-and-forget).
/// </summary>
/// <param name="messageId">The identifier of the message to remove from the standby buffer.</param>
public void ReplicateRemove(string messageId)
{
if (!_options.ReplicationEnabled || _replicationHandler == null) return;
FireAndForget(new ReplicationOperation(
ReplicationOperationType.Remove,
messageId,
null));
}
/// <summary>
/// Replicates a park operation to standby (fire-and-forget).
/// </summary>
/// <param name="message">The message that was parked on the active node.</param>
public void ReplicatePark(StoreAndForwardMessage message)
{
if (!_options.ReplicationEnabled || _replicationHandler == null) return;
FireAndForget(new ReplicationOperation(
ReplicationOperationType.Park,
message.Id,
message));
}
/// <summary>
/// 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.
/// </summary>
/// <param name="message">The message in its post-requeue (Pending, retry_count=0) state.</param>
public void ReplicateRequeue(StoreAndForwardMessage message)
{
if (!_options.ReplicationEnabled || _replicationHandler == null) return;
FireAndForget(new ReplicationOperation(
ReplicationOperationType.Requeue,
message.Id,
message));
}
/// <summary>
/// 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 <b>upserts</b> (<see cref="StoreAndForwardStorage.UpsertMessageAsync"/>),
/// 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.
/// </summary>
/// <param name="operation">The replication operation to apply.</param>
/// <param name="storage">The standby node's store-and-forward storage to update.</param>
/// <returns>A task representing the asynchronous apply operation.</returns>
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);
}
}
}
/// <summary>
/// Represents a buffer operation to be replicated to standby.
/// </summary>
public record ReplicationOperation(
ReplicationOperationType OperationType,
string MessageId,
StoreAndForwardMessage? Message);
/// <summary>
/// Types of buffer operations that are replicated.
/// </summary>
public enum ReplicationOperationType
{
Add,
Remove,
Park,
/// <summary>
/// An operator moved a parked message back to the pending
/// queue. The standby resets its matching row to Pending with retry_count = 0.
/// </summary>
Requeue
}
@@ -28,7 +28,6 @@ public static class ServiceCollectionExtensions
var storage = sp.GetRequiredService<StoreAndForwardStorage>();
var options = sp.GetRequiredService<IOptions<StoreAndForwardOptions>>().Value;
var logger = sp.GetRequiredService<ILogger<StoreAndForwardService>>();
var replication = sp.GetRequiredService<ReplicationService>();
// 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<ReplicationService>(sp =>
{
var options = sp.GetRequiredService<IOptions<StoreAndForwardOptions>>().Value;
var logger = sp.GetRequiredService<ILogger<ReplicationService>>();
return new ReplicationService(options, logger);
});
return services;
}
@@ -36,7 +36,6 @@ public class StoreAndForwardService
{
private readonly StoreAndForwardStorage _storage;
private readonly StoreAndForwardOptions _options;
private readonly ReplicationService? _replication;
private readonly ILogger<StoreAndForwardService> _logger;
/// <summary>
/// Site-side observer notified
@@ -107,7 +106,7 @@ public class StoreAndForwardService
/// <c>null</c> when no sweep is currently running. Captured when the timer
/// callback starts a sweep so <see cref="StopAsync"/> can wait for it to
/// finish before the host disposes downstream dependencies
/// (<see cref="_storage"/>, <see cref="_replication"/>) that the sweep is
/// (<see cref="_storage"/>) that the sweep is
/// still touching. Written from the timer thread and from
/// <see cref="StopAsync"/>, so reads are synchronised via the
/// <see cref="Volatile"/> APIs.
@@ -227,7 +226,6 @@ public class StoreAndForwardService
/// <param name="storage">The storage backend for buffered messages.</param>
/// <param name="options">Configuration options.</param>
/// <param name="logger">Logger instance.</param>
/// <param name="replication">Optional replication service for standby synchronization.</param>
/// <param name="cachedCallObserver">Optional observer for cached call lifecycle events.</param>
/// <param name="siteId">The site identifier this service belongs to.</param>
/// <param name="siteEventLogger">
@@ -240,7 +238,6 @@ public class StoreAndForwardService
StoreAndForwardStorage storage,
StoreAndForwardOptions options,
ILogger<StoreAndForwardService> 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
/// <summary>
/// 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 <see cref="ReplicationOperationType.Requeue"/>)
/// so a failover preserves the operator's retry intent.
/// An operator requeue is a buffer state change, and the peer picks it up because
/// <c>sf_messages</c> 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 <i>before</i> the local
/// requeue write rather than re-read after it, so a concurrent
/// <c>RemoveMessageAsync</c> or <c>DiscardParkedMessageAsync</c> running
/// between the two storage calls cannot leave the standby in <c>Parked</c>
/// while the active node has already requeued — we always have the row in
/// hand for the <c>Requeue</c> replication.
/// </summary>
/// <param name="messageId">The identifier of the message to retry.</param>
/// <returns>True if successfully retried, false otherwise.</returns>
@@ -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
/// <summary>
/// Permanently discards a parked message.
///
/// An operator discard is a buffer removal and is replicated
/// to the standby (as a <see cref="ReplicationOperationType.Remove"/>) 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 <c>sf_messages</c> 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.
/// </summary>
@@ -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");
}
@@ -65,7 +65,7 @@ public class StoreAndForwardStorage
/// <summary>
/// INSERT statement for a full message row. Shared by <see cref="EnqueueAsync"/>
/// and <see cref="ReplaceAllAsync"/>; bind with <see cref="BindMessageParameters"/>
/// and <see cref="UpsertMessageAsync"/>; bind with <see cref="BindMessageParameters"/>
/// so the column list and the parameters never drift apart.
/// </summary>
private const string InsertMessageSql = @"
@@ -151,39 +151,6 @@ public class StoreAndForwardStorage
return (rows.Take(limit).ToList(), truncated);
}
/// <summary>
/// Atomically replaces the entire buffer with <paramref name="messages"/> 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. <b>Never call on an active node</b> —
/// it discards every in-flight row.
/// </summary>
/// <param name="messages">The full buffer snapshot to install.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task ReplaceAllAsync(IReadOnlyList<StoreAndForwardMessage> 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();
}
/// <summary>
/// Inserts a message, or updates every mutable column in place if a row with the
/// same id already exists (<c>ON CONFLICT(id) DO UPDATE</c>). Used by the standby
@@ -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) },
@@ -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<ILocalDb>();
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<ILocalDb>();
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<ILocalDb>();
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<string> TableNames(ILocalDb db)
{
using var connection = db.CreateConnection();
@@ -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;
/// <summary>
/// N1 regression (review 02 round 2, Critical): the resync authority must use the same
/// oldest-Up predicate as the S&F delivery gate. Divergence scenario = the delivering node
/// is OLDEST but not LEADER (leader = lowest address), the exact state a rolling restart of
/// the lower-address node produces. Pre-fix the delivering node requests a resync from the
/// stale peer and ReplaceAllAsync wipes its live buffer.
/// </summary>
public class SfBufferResyncPredicateTests
{
[Fact]
public async Task OldestButNotLeaderNode_KeepsItsBuffer_AndSeedsTheJoiner()
{
// Two explicit ports, deliberately assigned so the FIRST-started (oldest,
// delivering) node has the HIGHER address → the second node is cluster leader.
var p1 = TwoNodeClusterFixture.GetFreeTcpPort();
var p2 = TwoNodeClusterFixture.GetFreeTcpPort();
var (portHigh, portLow) = p1 > p2 ? (p1, p2) : (p2, p1);
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<TestLocalDb>();
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<StoreAndForwardStorage>.Instance);
await sfStorage.InitializeAsync();
var siteLocalDb = TestLocalDb.CreateTemp($"site-resync-{tag}");
var siteStorage = new SiteStorageService(siteLocalDb.Db,
NullLogger<SiteStorageService>.Instance);
var replicationService = new ReplicationService(
new StoreAndForwardOptions(), NullLogger<ReplicationService>.Instance);
// Name MUST be "site-replication" — SendToPeer targets /user/site-replication.
var actor = node.ActorOf(Props.Create(() => new SiteReplicationActor(
siteStorage, sfStorage, replicationService, "site-int",
NullLogger<SiteReplicationActor>.Instance, null, null, null, null)),
"site-replication");
return (sfStorage, actor, 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<Task<bool>> condition, TimeSpan timeout, string why)
{
var deadline = DateTime.UtcNow + timeout;
while (DateTime.UtcNow < deadline)
{
if (await condition()) return;
await Task.Delay(250);
}
throw new TimeoutException(why);
}
}
@@ -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();
}
/// <summary>
/// Registers the Phase 2 tables for capture, which production <c>OnReady</c> does not do
/// until the Task 14 cutover.
/// </summary>
/// <remarks>
/// <b>Delete this method at Task 14</b>, along with its call above. Until the cutover the
/// bespoke <c>SiteReplicationActor</c> and <c>ReplicationService</c> still own these
/// tables, so <c>OnReady</c> 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.
/// <para>
/// After Task 14 these registrations come from <c>OnReady</c> itself, and leaving this
/// method behind would mask a cutover that forgot one — the whole point of these tests.
/// </para>
/// </remarks>
private static void RegisterPhase2TablesUntilCutover(ILocalDb db)
{
foreach (var table in Phase2ReplicatedTables)
db.RegisterReplicated(table);
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// <c>notification_lists</c> and <c>smtp_configurations</c> are absent by design. They
@@ -67,12 +67,14 @@ public class DeploymentManagerActorTests : TestKit, IDisposable
null, // no stream manager in tests
options,
NullLogger<DeploymentManagerActor>.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<DeploymentManagerActor>.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"));
@@ -80,8 +80,7 @@ akka {
private IActorRef CreateDeploymentManager() =>
ActorOf(Props.Create(() => new DeploymentManagerActor(
_storage, _compilationService, _sharedScriptLibrary,
null, new SiteRuntimeOptions(), NullLogger<DeploymentManagerActor>.Instance,
null, null, null, null, null, null, null, null)));
null, new SiteRuntimeOptions(), NullLogger<DeploymentManagerActor>.Instance)));
[Fact]
public void SiteNodeJoined_PushesLocalTrustedCertsToJoinedNode()
@@ -103,11 +103,10 @@ public class DeploymentManagerLoggerFactoryTests : TestKit, IDisposable
null,
new SiteRuntimeOptions { StartupBatchSize = 100, StartupBatchDelayMs = 5 },
NullLogger<DeploymentManagerActor>.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);
@@ -61,10 +61,11 @@ public class DeploymentManagerRedeployTests : TestKit, IDisposable
null,
new SiteRuntimeOptions(),
NullLogger<DeploymentManagerActor>.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)));
}
/// <summary>
@@ -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;
/// <summary>
/// Tests for <see cref="SiteReplicationActor"/>'s notify-and-fetch config replication:
/// the active node now replicates an id-only <see cref="ReplicateConfigDeploy"/> (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.
/// </summary>
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<SiteStorageService>.Instance);
_storage.InitializeAsync().GetAwaiter().GetResult();
_sfLocalDb = TestLocalDb.Create(_sfDbFile);
_sfStorage = new StoreAndForwardStorage(
_sfLocalDb.Db, NullLogger<StoreAndForwardStorage>.Instance);
_sfStorage.InitializeAsync().GetAwaiter().GetResult();
_replicationService = new ReplicationService(
new StoreAndForwardOptions(), NullLogger<ReplicationService>.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<SiteReplicationActor>.Instance, fetcher)));
private IActorRef CreateReplicationActor(
IDeploymentConfigFetcher fetcher, SiteRuntimeOptions options, TimeSpan retryDelay) =>
ActorOf(Props.Create(() => new SiteReplicationActor(
_storage, _sfStorage, _replicationService, SiteRole,
NullLogger<SiteReplicationActor>.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<string>(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<string>(
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<SiteReplicationActor>.Instance, fetcher, probe.Ref)));
actor.Tell(new ReplicateConfigDeploy(
"Pump2", "dep-200", "sha256:def", false,
"http://central:9000", "tok-abc"));
var applied = probe.ExpectMsg<ApplyConfigDeploy>(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<SiteReplicationActor>.Instance, probe.Ref, () => false)));
actor.Tell(new TriggerPeerTracked()); // stands in for TryTrackPeer's MemberUp path
probe.ExpectMsg<RequestSfBufferResync>(TimeSpan.FromSeconds(3));
}
[Fact]
public void ActiveTrackingPeer_DoesNotRequestResync()
{
var probe = CreateTestProbe();
var actor = ActorOf(Props.Create(() => new ResyncTestActor(
_storage, _sfStorage, _replicationService, SiteRole,
NullLogger<SiteReplicationActor>.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<SiteReplicationActor>.Instance, probe.Ref, () => true)));
actor.Tell(new RequestSfBufferResync(), TestActor);
var chunk = ExpectMsg<SfBufferSnapshotChunk>(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<SiteReplicationActor>.Instance, probe.Ref, () => false)));
actor.Tell(new SfBufferSnapshot(new List<StoreAndForwardMessage> { 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<StoreAndForwardMessage>
{ NewMessage("big", payloadJson: new string('y', 100_000)), NewMessage("small", payloadJson: "{}") };
var chunks = SiteReplicationActor.ChunkForRemoting(oversized, 64_000, 200);
Assert.Equal(2, chunks.Count); // the oversized row rides alone
}
[Fact]
public async Task ActiveNode_AnswersResyncRequest_WithSequencedChunks_SharingOneResyncId()
{
for (var i = 0; i < 3; i++)
await _sfStorage.EnqueueAsync(NewMessage($"c{i}", payloadJson: new string('z', 30_000)));
var actor = CreateResyncActor(isActive: () => true);
actor.Tell(new RequestSfBufferResync(), TestActor);
var first = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
var rest = Enumerable.Range(1, first.TotalChunks - 1)
.Select(_ => ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5)))
.Prepend(first)
.ToList();
Assert.True(first.TotalChunks > 1);
Assert.All(rest, c => Assert.Equal(first.ResyncId, c.ResyncId));
Assert.Equal(Enumerable.Range(1, first.TotalChunks), rest.Select(c => c.Sequence));
Assert.Equal(3, rest.Sum(c => c.Messages.Count));
}
// ── 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<StoreAndForwardMessage> { NewMessage("f1") }, false), TestActor);
actor.Tell(new SfBufferSnapshotChunk(resyncId, 2, 2,
new List<StoreAndForwardMessage> { NewMessage("f2") }, false), TestActor);
var ack = ExpectMsg<SfBufferResyncAck>(TimeSpan.FromSeconds(5));
Assert.Equal(resyncId, ack.ResyncId);
Assert.Equal(2, ack.RowCount);
await AwaitAssertAsync(async () =>
{
Assert.Null(await _sfStorage.GetMessageByIdAsync("stale")); // replaced wholesale
Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f1"));
Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f2"));
});
}
[Fact]
public async Task StandbyNode_NewResyncId_DiscardsStalePartialAssembly()
{
var actor = CreateResyncActor(isActive: () => false);
actor.Tell(new SfBufferSnapshotChunk("old", 1, 2,
new List<StoreAndForwardMessage> { NewMessage("orphan") }, false), TestActor);
actor.Tell(new SfBufferSnapshotChunk("new", 1, 1,
new List<StoreAndForwardMessage> { NewMessage("fresh") }, false), TestActor);
ExpectMsg<SfBufferResyncAck>(TimeSpan.FromSeconds(5)); // "new" completed
await AwaitAssertAsync(async () =>
{
Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh"));
Assert.Null(await _sfStorage.GetMessageByIdAsync("orphan")); // stale partial never applied
});
}
[Fact]
public void ActiveNode_IgnoresChunks_NeverAcks()
{
var actor = CreateResyncActor(isActive: () => true);
actor.Tell(new SfBufferSnapshotChunk("r", 1, 1,
new List<StoreAndForwardMessage> { NewMessage("x") }, false), TestActor);
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
// ── 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<SfBufferSnapshotChunk>(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<SfBufferSnapshotChunk>(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));
}
/// <summary>Attaches a <see cref="MeterListener"/> to a single ScadaBridge counter by name,
/// forwarding each recorded increment to <paramref name="onMeasurement"/>.</summary>
private static MeterListener ListenCounter(string instrumentName, Action<long> onMeasurement)
{
var listener = new MeterListener();
listener.InstrumentPublished = (inst, l) =>
{
if (inst.Meter.Name == ScadaBridgeTelemetry.MeterName && inst.Name == instrumentName)
l.EnableMeasurementEvents(inst);
};
listener.SetMeasurementEventCallback<long>((_, 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,
};
/// <summary>
/// Builds a resync-test message with a settable payload (additive to
/// <see cref="NewSfMessage"/> — the chunker sizes on <c>PayloadJson</c> length).
/// </summary>
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,
};
/// <summary>Constructs a <see cref="ResyncTestActor"/> with the given active-node check
/// (the resync chunk/ack tests Tell to and expect from <see cref="TestKit.TestActor"/>).
/// <paramref name="ackTimeout"/> is the active-side ack window seam (T7).</summary>
private IActorRef CreateResyncActor(Func<bool> isActive, TimeSpan? ackTimeout = null) =>
ActorOf(Props.Create(() => new ResyncTestActor(
_storage, _sfStorage, _replicationService, SiteRole,
NullLogger<SiteReplicationActor>.Instance, CreateTestProbe().Ref, isActive, ackTimeout)));
/// <summary>Test message: drives <see cref="SiteReplicationActor.OnPeerTracked"/> directly,
/// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer).</summary>
private sealed record TriggerPeerTracked;
/// <summary>
/// Test subclass for the resync tests: captures peer sends to a probe, injects the
/// active-node check, and exposes <see cref="OnPeerTracked"/> via a test message.
/// </summary>
private sealed class ResyncTestActor : SiteReplicationActor
{
private readonly IActorRef _peerProbe;
public ResyncTestActor(
SiteStorageService storage, StoreAndForwardStorage sfStorage,
ReplicationService replicationService, string siteRole,
ILogger<SiteReplicationActor> logger, IActorRef peerProbe, Func<bool> isActive,
TimeSpan? ackTimeout = null)
: base(storage, sfStorage, replicationService, siteRole, logger,
configFetcher: null, isActiveOverride: isActive, resyncAckTimeout: ackTimeout)
{
_peerProbe = peerProbe;
Receive<TriggerPeerTracked>(_ => OnPeerTracked());
}
protected override void SendToPeer(object message) => _peerProbe.Tell(message, Self);
}
/// <summary>
/// Test subclass exposing the peer send: <see cref="SiteReplicationActor.SendToPeer"/> 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).
/// </summary>
private sealed class ProbeForwardingReplicationActor : SiteReplicationActor
{
private readonly IActorRef _peerProbe;
public ProbeForwardingReplicationActor(
SiteStorageService storage, StoreAndForwardStorage sfStorage,
ReplicationService replicationService, string siteRole,
ILogger<SiteReplicationActor> logger, IDeploymentConfigFetcher configFetcher,
IActorRef peerProbe)
: base(storage, sfStorage, replicationService, siteRole, logger, configFetcher)
=> _peerProbe = peerProbe;
protected override void SendToPeer(object message) => _peerProbe.Tell(message, Self);
}
/// <summary>
/// In-test fake <see cref="IDeploymentConfigFetcher"/>: 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).
/// </summary>
private sealed class FakeConfigFetcher : IDeploymentConfigFetcher
{
private readonly Func<string, Task<string>> _behavior;
public ConcurrentQueue<(string BaseUrl, string DeploymentId, string Token)> Calls { get; } = new();
public FakeConfigFetcher(Func<string, Task<string>> behavior) => _behavior = behavior;
public async Task<string> FetchAsync(
string centralFetchBaseUrl, string deploymentId, string token, CancellationToken ct)
{
Calls.Enqueue((centralFetchBaseUrl, deploymentId, token));
await Task.Yield();
return await _behavior(deploymentId);
}
}
}
@@ -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;
/// <summary>
/// Characterization pin for the chunked anti-entropy resync contract (review 02 round 2,
/// N2). <see cref="SfBufferSnapshotChunk"/> (active→standby) and <see cref="SfBufferResyncAck"/>
/// (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.)
/// </summary>
public class ResyncWireSerializationPinTests : TestKit
{
private T RoundTrip<T>(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<StoreAndForwardMessage> { 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);
}
@@ -45,7 +45,6 @@ public class CachedCallAttemptEmissionTests : IAsyncLifetime, IDisposable
_storage,
_options,
NullLogger<StoreAndForwardService>.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<StoreAndForwardService>.Instance,
replication: null,
cachedCallObserver: observer,
siteId: "site-77");
@@ -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;
/// <summary>
/// WP-11: Tests for async replication to standby.
/// </summary>
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<StoreAndForwardStorage>.Instance);
var options = new StoreAndForwardOptions { ReplicationEnabled = true };
_replicationService = new ReplicationService(
options, NullLogger<ReplicationService>.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<ReplicationService>.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<ReplicationService>();
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
};
}
/// <summary>Minimal in-memory logger that records level + rendered message.</summary>
private sealed class CapturingLogger<T> : ILogger<T>
{
public List<(LogLevel Level, string Message)> Entries { get; } = new();
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
=> Entries.Add((logLevel, formatter(state, exception)));
}
}
@@ -1,82 +0,0 @@
using Akka.TestKit.Xunit2;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;
/// <summary>
/// Characterization pin for the intra-cluster S&amp;F replication contract. A
/// <see cref="ReplicationOperation"/> (carrying a full <see cref="StoreAndForwardMessage"/>)
/// 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.
/// </summary>
public class ReplicationWireSerializationPinTests : TestKit
{
private T RoundTrip<T>(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);
}
@@ -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;
/// <summary>
/// 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.
/// </summary>
public class StoreAndForwardReplicationTests : IAsyncLifetime, IDisposable
{
private readonly TestLocalDb _localDb;
private readonly StoreAndForwardStorage _storage;
private readonly StoreAndForwardService _service;
private readonly List<ReplicationOperation> _replicated = new();
public StoreAndForwardReplicationTests()
{
_localDb = TestLocalDb.CreateTemp("ReplTests");
_storage = new StoreAndForwardStorage(_localDb.Db, NullLogger<StoreAndForwardStorage>.Instance);
var options = new StoreAndForwardOptions
{
DefaultRetryInterval = TimeSpan.Zero,
DefaultMaxRetries = 1,
RetryTimerInterval = TimeSpan.FromMinutes(10),
ReplicationEnabled = true,
};
var replication = new ReplicationService(options, NullLogger<ReplicationService>.Instance);
replication.SetReplicationHandler(op =>
{
lock (_replicated) _replicated.Add(op);
return Task.CompletedTask;
});
_service = new StoreAndForwardService(
_storage, options, NullLogger<StoreAndForwardService>.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);
}
/// <summary>Replication is fire-and-forget (Task.Run); poll until the expected ops arrive.</summary>
private async Task<List<ReplicationOperation>> 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);
}
/// <summary>
/// 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).
/// </summary>
[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);
}
/// <summary>
/// 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).
/// </summary>
[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);
}
/// <summary>
/// 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.
/// </summary>
[Fact]
public async Task ApplyReplicatedOperation_Requeue_MovesStandbyRowBackToPending()
{
var replication = new ReplicationService(
new StoreAndForwardOptions { ReplicationEnabled = true },
NullLogger<ReplicationService>.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<ReplicationService>.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,
};
/// <summary>
/// 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.
/// </summary>
[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);
}
/// <summary>
/// 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.
/// </summary>
[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);
}
}
@@ -54,7 +54,7 @@ public class StoreAndForwardSiteEventTests : IAsyncLifetime, IDisposable
_service = new StoreAndForwardService(
_storage, _options, NullLogger<StoreAndForwardService>.Instance,
replication: null, cachedCallObserver: null, siteId: "site-a",
cachedCallObserver: null, siteId: "site-a",
siteEventLogger: _siteLog);
}
@@ -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 ──