feat: replace site registration with database-driven site addressing
Central now resolves site Akka remoting addresses from the Sites DB table (NodeAAddress/NodeBAddress) instead of relying on runtime RegisterSite messages. Eliminates the race condition where sites starting before central had their registration dead-lettered. Addresses are cached in CentralCommunicationActor with 60s periodic refresh and on-demand refresh when sites are added/edited/deleted via UI or CLI.
This commit is contained in:
Joseph Doherty
@@ -1,5 +1,9 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using NSubstitute;
|
||||
using ScadaLink.Commons.Entities.Sites;
|
||||
using ScadaLink.Commons.Interfaces.Repositories;
|
||||
using ScadaLink.Commons.Messages.Communication;
|
||||
using ScadaLink.Commons.Messages.Deployment;
|
||||
using ScadaLink.Commons.Messages.DebugView;
|
||||
@@ -9,97 +13,158 @@ using ScadaLink.Communication.Actors;
|
||||
namespace ScadaLink.Communication.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// WP-4: Tests for CentralCommunicationActor message routing.
|
||||
/// WP-5: Tests for connection failure and failover handling.
|
||||
/// Tests for CentralCommunicationActor with database-driven site addressing.
|
||||
/// WP-4: Message routing via site address cache loaded from DB.
|
||||
/// WP-5: Connection failure and failover handling.
|
||||
/// </summary>
|
||||
public class CentralCommunicationActorTests : TestKit
|
||||
{
|
||||
public CentralCommunicationActorTests()
|
||||
: base(@"akka.loglevel = DEBUG")
|
||||
public CentralCommunicationActorTests() : base(@"akka.loglevel = DEBUG") { }
|
||||
|
||||
private (IActorRef actor, ISiteRepository mockRepo) CreateActorWithMockRepo(IEnumerable<Site>? sites = null)
|
||||
{
|
||||
var mockRepo = Substitute.For<ISiteRepository>();
|
||||
mockRepo.GetAllSitesAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(sites?.ToList() ?? new List<Site>());
|
||||
|
||||
var services = new ServiceCollection();
|
||||
services.AddScoped(_ => mockRepo);
|
||||
var sp = services.BuildServiceProvider();
|
||||
|
||||
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp)));
|
||||
return (actor, mockRepo);
|
||||
}
|
||||
|
||||
private Site CreateSite(string identifier, string? nodeAPath, string? nodeBPath = null) =>
|
||||
new("Test Site", identifier) { NodeAAddress = nodeAPath, NodeBAddress = nodeBPath };
|
||||
|
||||
[Fact]
|
||||
public void RegisterSite_AllowsMessageRouting()
|
||||
public void DatabaseDrivenRouting_RoutesToConfiguredSite()
|
||||
{
|
||||
var centralActor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor()));
|
||||
|
||||
// Register a site pointing to the test probe
|
||||
var probe = CreateTestProbe();
|
||||
centralActor.Tell(new RegisterSite("site1", probe.Ref.Path.ToString()));
|
||||
var site = CreateSite("site1", probe.Ref.Path.ToString());
|
||||
var (actor, _) = CreateActorWithMockRepo(new[] { site });
|
||||
|
||||
// Send explicit refresh and wait for async DB load + PipeTo
|
||||
actor.Tell(new RefreshSiteAddresses());
|
||||
Thread.Sleep(1000);
|
||||
|
||||
// Send a message to the site
|
||||
var command = new DeployInstanceCommand(
|
||||
"dep1", "inst1", "hash1", "{}", "admin", DateTimeOffset.UtcNow);
|
||||
centralActor.Tell(new SiteEnvelope("site1", command));
|
||||
actor.Tell(new SiteEnvelope("site1", command));
|
||||
|
||||
// The probe should receive the inner message (not the envelope)
|
||||
probe.ExpectMsg<DeployInstanceCommand>(msg => msg.DeploymentId == "dep1");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void UnregisteredSite_MessageIsDropped()
|
||||
public void UnconfiguredSite_MessageIsDropped()
|
||||
{
|
||||
var centralActor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor()));
|
||||
var (actor, _) = CreateActorWithMockRepo();
|
||||
|
||||
actor.Tell(new RefreshSiteAddresses());
|
||||
Thread.Sleep(1000);
|
||||
|
||||
var command = new DeployInstanceCommand(
|
||||
"dep1", "inst1", "hash1", "{}", "admin", DateTimeOffset.UtcNow);
|
||||
centralActor.Tell(new SiteEnvelope("unknown-site", command));
|
||||
actor.Tell(new SiteEnvelope("unknown-site", command));
|
||||
|
||||
// No crash, no response — the ask will timeout on the caller side
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ConnectionLost_DebugStreamsKilled()
|
||||
{
|
||||
var centralActor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor()));
|
||||
var siteProbe = CreateTestProbe();
|
||||
var site = CreateSite("site1", siteProbe.Ref.Path.ToString());
|
||||
var (actor, _) = CreateActorWithMockRepo(new[] { site });
|
||||
|
||||
// Register site
|
||||
centralActor.Tell(new RegisterSite("site1", siteProbe.Ref.Path.ToString()));
|
||||
actor.Tell(new RefreshSiteAddresses());
|
||||
Thread.Sleep(1000);
|
||||
|
||||
// Subscribe to debug view (this tracks the subscription)
|
||||
// Subscribe to debug view (tracks the subscription)
|
||||
var subscriberProbe = CreateTestProbe();
|
||||
var subRequest = new SubscribeDebugViewRequest("inst1", "corr-123");
|
||||
centralActor.Tell(new SiteEnvelope("site1", subRequest), subscriberProbe.Ref);
|
||||
actor.Tell(new SiteEnvelope("site1", subRequest), subscriberProbe.Ref);
|
||||
|
||||
// Simulate site disconnection
|
||||
centralActor.Tell(new ConnectionStateChanged("site1", false, DateTimeOffset.UtcNow));
|
||||
actor.Tell(new ConnectionStateChanged("site1", false, DateTimeOffset.UtcNow));
|
||||
|
||||
// The subscriber should receive a DebugStreamTerminated notification
|
||||
subscriberProbe.ExpectMsg<DebugStreamTerminated>(
|
||||
msg => msg.SiteId == "site1" && msg.CorrelationId == "corr-123");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ConnectionLost_SiteSelectionRemoved()
|
||||
{
|
||||
var centralActor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor()));
|
||||
var siteProbe = CreateTestProbe();
|
||||
|
||||
centralActor.Tell(new RegisterSite("site1", siteProbe.Ref.Path.ToString()));
|
||||
|
||||
// Disconnect
|
||||
centralActor.Tell(new ConnectionStateChanged("site1", false, DateTimeOffset.UtcNow));
|
||||
|
||||
// Sending a message to the disconnected site should be dropped
|
||||
centralActor.Tell(new SiteEnvelope("site1",
|
||||
new DeployInstanceCommand("dep2", "inst2", "hash2", "{}", "admin", DateTimeOffset.UtcNow)));
|
||||
|
||||
siteProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Heartbeat_ForwardedToParent()
|
||||
{
|
||||
// Actor still needs IServiceProvider even though this test doesn't use routing
|
||||
var mockRepo = Substitute.For<ISiteRepository>();
|
||||
mockRepo.GetAllSitesAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<Site>());
|
||||
|
||||
var services = new ServiceCollection();
|
||||
services.AddScoped(_ => mockRepo);
|
||||
var sp = services.BuildServiceProvider();
|
||||
|
||||
var parentProbe = CreateTestProbe();
|
||||
var centralActor = parentProbe.ChildActorOf(
|
||||
Props.Create(() => new CentralCommunicationActor()));
|
||||
Props.Create(() => new CentralCommunicationActor(sp)));
|
||||
|
||||
var heartbeat = new HeartbeatMessage("site1", "host1", true, DateTimeOffset.UtcNow);
|
||||
centralActor.Tell(heartbeat);
|
||||
|
||||
parentProbe.ExpectMsg<HeartbeatMessage>(msg => msg.SiteId == "site1");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void RefreshSiteAddresses_UpdatesCache()
|
||||
{
|
||||
var probe1 = CreateTestProbe();
|
||||
var probe2 = CreateTestProbe();
|
||||
|
||||
var site1 = CreateSite("site1", probe1.Ref.Path.ToString());
|
||||
var (actor, mockRepo) = CreateActorWithMockRepo(new[] { site1 });
|
||||
|
||||
// Wait for initial load, then send explicit refresh
|
||||
actor.Tell(new RefreshSiteAddresses());
|
||||
Thread.Sleep(1000);
|
||||
|
||||
// Verify routing to site1 works
|
||||
var cmd1 = new DeployInstanceCommand(
|
||||
"dep1", "inst1", "hash1", "{}", "admin", DateTimeOffset.UtcNow);
|
||||
actor.Tell(new SiteEnvelope("site1", cmd1));
|
||||
probe1.ExpectMsg<DeployInstanceCommand>(msg => msg.DeploymentId == "dep1");
|
||||
|
||||
// Update mock repo to return both sites
|
||||
var site2 = CreateSite("site2", probe2.Ref.Path.ToString());
|
||||
mockRepo.GetAllSitesAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<Site> { site1, site2 });
|
||||
|
||||
// Refresh again
|
||||
actor.Tell(new RefreshSiteAddresses());
|
||||
Thread.Sleep(1000);
|
||||
|
||||
// Verify routing to site2 now works
|
||||
var cmd2 = new DeployInstanceCommand(
|
||||
"dep2", "inst2", "hash2", "{}", "admin", DateTimeOffset.UtcNow);
|
||||
actor.Tell(new SiteEnvelope("site2", cmd2));
|
||||
probe2.ExpectMsg<DeployInstanceCommand>(msg => msg.DeploymentId == "dep2");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NodeBFallback_WhenNodeANotConfigured()
|
||||
{
|
||||
var probe = CreateTestProbe();
|
||||
var site = CreateSite("site1", null, probe.Ref.Path.ToString());
|
||||
var (actor, _) = CreateActorWithMockRepo(new[] { site });
|
||||
|
||||
actor.Tell(new RefreshSiteAddresses());
|
||||
Thread.Sleep(1000);
|
||||
|
||||
var command = new DeployInstanceCommand(
|
||||
"dep1", "inst1", "hash1", "{}", "admin", DateTimeOffset.UtcNow);
|
||||
actor.Tell(new SiteEnvelope("site1", command));
|
||||
|
||||
probe.ExpectMsg<DeployInstanceCommand>(msg => msg.DeploymentId == "dep1");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
<PackageReference Include="Akka.TestKit.Xunit2" Version="1.5.62" />
|
||||
<PackageReference Include="coverlet.collector" Version="6.0.4" />
|
||||
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
|
||||
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.5" />
|
||||
<PackageReference Include="NSubstitute" Version="5.3.0" />
|
||||
<PackageReference Include="xunit" Version="2.9.3" />
|
||||
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.4" />
|
||||
|
||||
Reference in New Issue
Block a user