Joseph Doherty
153609ef8a
The #96 site-psk-invalidation topic was the first cross-node invalidation broadcast. It is generalised into ONE central DistributedPubSub topic, central-cache-invalidation, carrying typed messages (Commons ICentralCacheInvalidation: SitePskInvalidated, RoleMappingsInvalidated, ExternalDefinitionsInvalidated, InboundApiMethodInvalidated) in an envelope stamped with the publisher's process-unique origin id. - Host: CentralCacheInvalidationBus (ICentralCacheInvalidationPublisher, no-op until the mediator is set, never throws) and CentralCacheInvalidationSubscriber, which skips its own node's envelopes and hands the peer's to every registered ICentralCacheInvalidationHandler (a failing handler does not stop the rest). Replaces SitePskInvalidationSubscriber. - Migration of #96: SitePskRotationInvalidator publishes through the bus and is the PSK handler; SitePskInvalidated keeps its type name and is also published bare on site-psk-invalidation, which the subscriber still listens to, so a mixed-version central pair keeps exchanging PSK invalidations. Remove after one release. - Audit D1: RoleMappingCache is a handler; ManagementActor's post-commit role-mapping invalidation also publishes RoleMappingsInvalidated, so a revocation is immediate on the peer instead of after RoleMappingCacheSeconds. - Audit D3 / #83 (peer half): ExternalDefinitionLookupCache is a handler; the 9 ManagementActor handlers and the Transport bundle importer publish ExternalDefinitionsInvalidated after commit. (#83's local import half was already in.) TTLs remain the fallback for a lost broadcast or a direct SQL edit. Tests: bus and subscriber unit tests (origin filter, legacy topic), peer-node tests through the real caches in ManagementService.Tests, importer publish, composition root, and a real two-node cluster from the production HOCON (CentralCacheInvalidationTwoNodeTests) proving every message type crosses with the production serializer. Docs: Host, Communication, Security, ManagementService, ESG, and the #85 cache audit. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
81 lines
3.3 KiB
C#
81 lines
3.3 KiB
C#
using Akka.Actor;
|
|
using Akka.Cluster.Tools.PublishSubscribe;
|
|
using Akka.TestKit.Xunit2;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
|
|
|
|
/// <summary>
|
|
/// ScadaBridge#100 item 3: the central cache-invalidation bus publishes typed invalidations on
|
|
/// one topic, and the subscriber hands the peer's (never its own) to every handler.
|
|
/// </summary>
|
|
public class CentralCacheInvalidationBusTests : TestKit
|
|
{
|
|
private sealed class RecordingHandler : ICentralCacheInvalidationHandler
|
|
{
|
|
public List<ICentralCacheInvalidation> Applied { get; } = new();
|
|
|
|
public void Apply(ICentralCacheInvalidation invalidation)
|
|
{
|
|
lock (Applied) Applied.Add(invalidation);
|
|
}
|
|
}
|
|
|
|
private sealed class ThrowingHandler : ICentralCacheInvalidationHandler
|
|
{
|
|
public void Apply(ICentralCacheInvalidation invalidation) => throw new InvalidOperationException("boom");
|
|
}
|
|
|
|
private readonly CentralCacheInvalidationBus _bus = new(NullLogger<CentralCacheInvalidationBus>.Instance);
|
|
|
|
/// <summary>Publishing wraps the invalidation with this node's origin id on the shared topic.</summary>
|
|
[Fact]
|
|
public void Publish_SendsAnEnvelopeOnTheSharedTopic()
|
|
{
|
|
var mediator = CreateTestProbe();
|
|
_bus.SetMediator(mediator.Ref);
|
|
|
|
_bus.Publish(new RoleMappingsInvalidated());
|
|
|
|
var publish = mediator.ExpectMsg<Publish>();
|
|
Assert.Equal("central-cache-invalidation", publish.Topic);
|
|
var envelope = Assert.IsType<CentralCacheInvalidationEnvelope>(publish.Message);
|
|
Assert.Equal(_bus.OriginId, envelope.OriginId);
|
|
Assert.IsType<RoleMappingsInvalidated>(envelope.Invalidation);
|
|
mediator.ExpectNoMsg(TimeSpan.FromMilliseconds(100)); // only PSK also uses the legacy topic
|
|
}
|
|
|
|
/// <summary>Before the mediator exists publishing is a silent no-op.</summary>
|
|
[Fact]
|
|
public void Publish_WithoutAMediator_DoesNotThrow()
|
|
{
|
|
_bus.Publish(new ExternalDefinitionsInvalidated());
|
|
}
|
|
|
|
/// <summary>The peer's invalidation reaches every handler; this node's own broadcast is skipped.</summary>
|
|
[Fact]
|
|
public void Subscriber_AppliesPeerInvalidations_AndSkipsItsOwn()
|
|
{
|
|
var handler = new RecordingHandler();
|
|
var subscriber = Sys.ActorOf(Props.Create(() => new CentralCacheInvalidationSubscriber(
|
|
_bus, new ICentralCacheInvalidationHandler[] { new ThrowingHandler(), handler })));
|
|
|
|
subscriber.Tell(new CentralCacheInvalidationEnvelope(_bus.OriginId, new RoleMappingsInvalidated()));
|
|
subscriber.Tell(new CentralCacheInvalidationEnvelope("peer", new ExternalDefinitionsInvalidated()));
|
|
subscriber.Tell(new CentralCacheInvalidationEnvelope("peer", new InboundApiMethodInvalidated("M")));
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
lock (handler.Applied)
|
|
{
|
|
// A throwing handler ahead of it does not stop delivery.
|
|
Assert.Equal(
|
|
new ICentralCacheInvalidation[] { new ExternalDefinitionsInvalidated(), new InboundApiMethodInvalidated("M") },
|
|
handler.Applied);
|
|
}
|
|
});
|
|
}
|
|
}
|