diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index 0d5daed6..8ecc5d2d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -769,7 +769,7 @@ akka {{ var replicationActor = _actorSystem!.ActorOf( Props.Create(() => new SiteReplicationActor( storage, sfStorage, replicationService, siteRole, replicationLogger, - deploymentConfigFetcher)), + deploymentConfigFetcher, null, siteRuntimeOptionsValue, null)), "site-replication"); // Wire S&F replication handler to forward operations via the replication actor diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs index af6ee5c4..fe8a3a64 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs @@ -28,6 +28,8 @@ public class SiteReplicationActor : ReceiveActor private readonly ILogger _logger; private readonly Cluster _cluster; private readonly Func _isActive; + private readonly int _configFetchRetryCount; + private readonly TimeSpan _configFetchRetryDelay; private Address? _peerAddress; /// @@ -64,7 +66,9 @@ public class SiteReplicationActor : ReceiveActor string siteRole, ILogger logger, IDeploymentConfigFetcher? configFetcher = null, - Func? isActiveOverride = null) + Func? isActiveOverride = null, + SiteRuntimeOptions? options = null, + TimeSpan? configFetchRetryDelay = null) { _storage = storage; _sfStorage = sfStorage; @@ -74,6 +78,11 @@ public class SiteReplicationActor : ReceiveActor _logger = logger; _cluster = Cluster.Get(Context.System); _isActive = isActiveOverride ?? DefaultIsActive; + // UA2: bound the standby's replicated-config fetch retries. At least one + // attempt always runs; the fixed inter-attempt delay is a test seam + // (production default 2 s). + _configFetchRetryCount = Math.Max(1, options?.ConfigFetchRetryCount ?? 1); + _configFetchRetryDelay = configFetchRetryDelay ?? TimeSpan.FromSeconds(2); // Cluster member events Receive(HandleMemberUp); @@ -254,43 +263,53 @@ public class SiteReplicationActor : ReceiveActor // 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. A single - // fetch attempt — reconciliation is the durable backstop for a lost fetch. - _configFetcher.FetchAsync(msg.CentralFetchBaseUrl, msg.DeploymentId, msg.FetchToken, CancellationToken.None) - .ContinueWith(async t => + // 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 { - if (t.IsCompletedSuccessfully) - { - await _storage.StoreDeployedConfigIfNewerAsync( - msg.InstanceName, t.Result, msg.DeploymentId, msg.RevisionHash, msg.IsEnabled); - return; - } - - var ex = t.Exception?.GetBaseException(); - if (ex is DeploymentConfigFetchException { IsSuperseded: true }) - _logger.LogInformation( - "Skip replicated config for {Instance}: superseded/expired (a newer deploy will replicate)", - msg.InstanceName); - else if (t.IsCanceled) - _logger.LogWarning( - "Replicated config fetch cancelled for {Instance} (deployment {DeploymentId})", - msg.InstanceName, msg.DeploymentId); - else - _logger.LogError(ex, - "Replicated config fetch failed for {Instance} (deployment {DeploymentId})", - msg.InstanceName, msg.DeploymentId); + // 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 (Exception writeEx) + catch (DeploymentConfigFetchException fex) when (fex.IsSuperseded) { - // Guarded-write failure is best-effort; observe + log so nothing faults silently. - _logger.LogError(writeEx, - "Failed to write replicated config for {Instance} (deployment {DeploymentId})", - msg.InstanceName, msg.DeploymentId); + // 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; } - }) - .Unwrap(); + 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) diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs index a05e65d2..8606fea5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs @@ -61,8 +61,9 @@ public class SiteRuntimeOptions public int ConfigFetchTimeoutSeconds { get; set; } = 30; /// - /// Bounded retry count for the standby's best-effort replicated-config fetch. - /// Reserved — consumed by the standby replication fetch in a later task; not yet wired. + /// Bounded attempt count (including the first) for the standby's replicated-config + /// fetch; a 2 s fixed delay separates attempts and superseded fetches never retry. + /// Consumed by SiteReplicationActor.HandleApplyConfigDeploy (UA2). Default: 3. /// public int ConfigFetchRetryCount { get; set; } = 3; diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs index 2fb8910b..7dccdddd 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs @@ -3,6 +3,7 @@ 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; @@ -79,6 +80,39 @@ akka { _storage, _sfStorage, _replicationService, SiteRole, NullLogger.Instance, fetcher))); + private IActorRef CreateReplicationActor( + IDeploymentConfigFetcher fetcher, SiteRuntimeOptions options, TimeSpan retryDelay) => + ActorOf(Props.Create(() => new SiteReplicationActor( + _storage, _sfStorage, _replicationService, SiteRole, + NullLogger.Instance, fetcher, null, options, retryDelay))); + + [Fact] + public async Task ReplicatedFetch_RetriesUpToConfigFetchRetryCount() + { + // The first two fetches fail transiently; the third succeeds. With + // ConfigFetchRetryCount = 3 the standby must retry to the third attempt and + // then guarded-write the fetched config (a short retry delay keeps the test fast). + var attempts = 0; + var fetcher = new FakeConfigFetcher(_ => + Interlocked.Increment(ref attempts) < 3 + ? Task.FromException(new InvalidOperationException("central hiccup")) + : Task.FromResult("{\"instanceUniqueName\":\"RetryPump\"}")); + var actor = CreateReplicationActor( + fetcher, new SiteRuntimeOptions { ConfigFetchRetryCount = 3 }, + TimeSpan.FromMilliseconds(50)); + + actor.Tell(new ApplyConfigDeploy( + "RetryPump", "dep-r1", "sha256:r1", true, + "http://central:9000", "tok-r1")); + + await AwaitAssertAsync(async () => + { + Assert.Equal(3, Volatile.Read(ref attempts)); + var configs = await _storage.GetAllDeployedConfigsAsync(); + Assert.Single(configs, c => c.InstanceUniqueName == "RetryPump"); + }, TimeSpan.FromSeconds(10)); + } + [Fact] public async Task ApplyConfigDeploy_StandbyFetchesConfigAndGuardedWrites() {