fix(saf): resync authority uses the shared oldest-Up predicate + delivery-gate delegate; apply-time re-check guard (plan R2-02 T3)
This commit is contained in:
@@ -772,10 +772,20 @@ akka {{
|
|||||||
var replicationLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
|
var replicationLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
|
||||||
.CreateLogger<SiteReplicationActor>();
|
.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.
|
||||||
|
var clusterNodeProvider = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.IClusterNodeProvider>();
|
||||||
|
Func<bool>? activeNodeCheck = clusterNodeProvider != null
|
||||||
|
? () => clusterNodeProvider.SelfIsPrimary
|
||||||
|
: null;
|
||||||
|
|
||||||
var replicationActor = _actorSystem!.ActorOf(
|
var replicationActor = _actorSystem!.ActorOf(
|
||||||
Props.Create(() => new SiteReplicationActor(
|
Props.Create(() => new SiteReplicationActor(
|
||||||
storage, sfStorage, replicationService, siteRole, replicationLogger,
|
storage, sfStorage, replicationService, siteRole, replicationLogger,
|
||||||
deploymentConfigFetcher, null, siteRuntimeOptionsValue, null)),
|
deploymentConfigFetcher, activeNodeCheck, siteRuntimeOptionsValue, null)),
|
||||||
"site-replication");
|
"site-replication");
|
||||||
|
|
||||||
// Wire S&F replication handler to forward operations via the replication actor
|
// Wire S&F replication handler to forward operations via the replication actor
|
||||||
@@ -872,10 +882,11 @@ akka {{
|
|||||||
// tick so failover resumes delivery within one RetryTimerInterval.
|
// tick so failover resumes delivery within one RetryTimerInterval.
|
||||||
// IClusterNodeProvider.SelfIsPrimary is the canonical "this node is the
|
// IClusterNodeProvider.SelfIsPrimary is the canonical "this node is the
|
||||||
// oldest Up member (singleton host)" check from the cluster-infrastructure
|
// oldest Up member (singleton host)" check from the cluster-infrastructure
|
||||||
// fix plan — the shared helper this seam was designed to accept. In a
|
// fix plan — the shared helper this seam was designed to accept. The provider
|
||||||
// non-clustered test host the provider is unregistered, so the gate stays
|
// is resolved once above (the same instance gating the resync authority and
|
||||||
// unset and the sweep is ungated (legacy behaviour, preserved).
|
// heartbeat IsActive stamp — N1). In a non-clustered test host the provider is
|
||||||
var clusterNodeProvider = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.IClusterNodeProvider>();
|
// unregistered, so the gate stays unset and the sweep is ungated (legacy
|
||||||
|
// behaviour, preserved).
|
||||||
if (clusterNodeProvider != null)
|
if (clusterNodeProvider != null)
|
||||||
{
|
{
|
||||||
storeAndForwardService.SetDeliveryGate(() => clusterNodeProvider.SelfIsPrimary);
|
storeAndForwardService.SetDeliveryGate(() => clusterNodeProvider.SelfIsPrimary);
|
||||||
|
|||||||
@@ -54,10 +54,11 @@ public class SiteReplicationActor : ReceiveActor
|
|||||||
/// config never crosses the intra-site Akka hop. Null on nodes/tests without a fetcher.
|
/// config never crosses the intra-site Akka hop. Null on nodes/tests without a fetcher.
|
||||||
/// </param>
|
/// </param>
|
||||||
/// <param name="isActiveOverride">
|
/// <param name="isActiveOverride">
|
||||||
/// Test seam for the active-node check that gates the buffer-resync roles (a
|
/// Active-node check that gates the buffer-resync roles (a standby requests a
|
||||||
/// standby requests a resync, the active node answers). Null (production) uses
|
/// resync, the active node answers). Production wiring passes the Host's
|
||||||
/// the repo-standard leader+Up check via <see cref="Cluster"/> — swap point for
|
/// <c>IClusterNodeProvider.SelfIsPrimary</c> delegate (the same instance gating the
|
||||||
/// plan 01's shared active-node helper.
|
/// S&F delivery sweep); null falls back to the shared oldest-Up evaluator
|
||||||
|
/// (<see cref="Communication.ClusterState.ActiveNodeEvaluator"/>).
|
||||||
/// </param>
|
/// </param>
|
||||||
/// <param name="options">Site runtime options, including the config-fetch retry count; production defaults apply when null.</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>
|
/// <param name="configFetchRetryDelay">Delay between config-fetch retry attempts; defaults to 2 seconds when null.</param>
|
||||||
@@ -181,21 +182,17 @@ public class SiteReplicationActor : ReceiveActor
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Repo-standard active-node check: this node is active when it is the current
|
/// Repo-standard active-node check: this node is active when it is the OLDEST Up
|
||||||
/// cluster leader AND its own <see cref="MemberStatus"/> is
|
/// member carrying the site role — the same oldest-Up semantics as the S&F delivery
|
||||||
/// <see cref="MemberStatus.Up"/>. Mirrors <c>SiteCommunicationActor.DefaultIsActiveCheck</c>
|
/// gate (IClusterNodeProvider.SelfIsPrimary → ClusterActivityEvaluator → shared
|
||||||
/// (swap point for plan 01's shared helper). Any other state reports standby —
|
/// ActiveNodeEvaluator). NEVER the cluster leader: leadership is lowest-address and
|
||||||
/// safe-by-default.
|
/// 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>
|
/// </summary>
|
||||||
private bool DefaultIsActive()
|
private bool DefaultIsActive() =>
|
||||||
{
|
Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(_cluster, _siteRole);
|
||||||
var self = _cluster.SelfMember;
|
|
||||||
if (self.Status != MemberStatus.Up)
|
|
||||||
return false;
|
|
||||||
|
|
||||||
var leader = _cluster.State.Leader;
|
|
||||||
return leader != null && leader == self.Address;
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Evaluates the active-node check, treating a throwing check as standby
|
/// Evaluates the active-node check, treating a throwing check as standby
|
||||||
@@ -435,7 +432,19 @@ public class SiteReplicationActor : ReceiveActor
|
|||||||
_logger.LogInformation(
|
_logger.LogInformation(
|
||||||
"Applying S&F buffer resync snapshot ({Count} rows), replacing local buffer", msg.Messages.Count);
|
"Applying S&F buffer resync snapshot ({Count} rows), replacing local buffer", msg.Messages.Count);
|
||||||
|
|
||||||
_sfStorage.ReplaceAllAsync(msg.Messages)
|
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 =>
|
.ContinueWith(t =>
|
||||||
{
|
{
|
||||||
if (t.IsFaulted)
|
if (t.IsFaulted)
|
||||||
|
|||||||
Reference in New Issue
Block a user