feat(central): one central cache-invalidation topic; role-mapping and external-definition caches follow peer edits (ScadaBridge#100 item 3, ScadaBridge#83)
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
This commit is contained in:
Joseph Doherty
@@ -1,9 +1,8 @@
|
||||
using Akka.Actor;
|
||||
using Akka.Cluster.Tools.PublishSubscribe;
|
||||
using Akka.Event;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
||||
using ZB.MOM.WW.Secrets;
|
||||
using ZB.MOM.WW.Secrets.Abstractions;
|
||||
@@ -26,9 +25,9 @@ namespace ZB.MOM.WW.ScadaBridge.Host;
|
||||
/// <para>
|
||||
/// <b>Why the peer needs telling.</b> Both central nodes cache the key in their own
|
||||
/// <see cref="SitePskProvider"/> (and their own resolver cache), and the write only ran on one of
|
||||
/// them. The peer is told over the central <c>DistributedPubSub</c> mediator on
|
||||
/// <see cref="Topic"/> — the same mediator the peer health replication uses. A node receives its
|
||||
/// own broadcast too; applying it twice is harmless, so there is no origin filtering.
|
||||
/// them. The peer is told with a <see cref="SitePskInvalidated"/> on the central cache-invalidation
|
||||
/// topic (<see cref="CentralCacheInvalidationBus"/>, ScadaBridge#100 item 3, which generalised
|
||||
/// the #96 <c>site-psk-invalidation</c> topic); this class is also the peer's handler for it.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Writes made outside this process (the <c>secret</c> CLI, a direct SQL write) never reach this
|
||||
@@ -36,23 +35,22 @@ namespace ZB.MOM.WW.ScadaBridge.Host;
|
||||
/// <c>ControlPlaneCredentials.WithSiteCredentials</c>).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator
|
||||
public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator, ICentralCacheInvalidationHandler
|
||||
{
|
||||
/// <summary>The <c>DistributedPubSub</c> topic the central nodes exchange PSK invalidations on.</summary>
|
||||
public const string Topic = "site-psk-invalidation";
|
||||
|
||||
private readonly ISecretCacheInvalidator _inner;
|
||||
private readonly ISitePskProvider _pskProvider;
|
||||
private readonly ICentralCacheInvalidationPublisher? _publisher;
|
||||
private readonly ILogger<SitePskRotationInvalidator> _logger;
|
||||
private volatile IActorRef? _mediator;
|
||||
|
||||
/// <summary>Creates the invalidator.</summary>
|
||||
/// <param name="inner">The library's own invalidator (the secret resolver's cache).</param>
|
||||
/// <param name="pskProvider">Central's per-site PSK cache.</param>
|
||||
/// <param name="publisher">The central cache-invalidation publisher; null keeps evictions local.</param>
|
||||
/// <param name="logger">Logger.</param>
|
||||
public SitePskRotationInvalidator(
|
||||
ISecretCacheInvalidator inner,
|
||||
ISitePskProvider pskProvider,
|
||||
ICentralCacheInvalidationPublisher? publisher,
|
||||
ILogger<SitePskRotationInvalidator> logger)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(inner);
|
||||
@@ -60,16 +58,10 @@ public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
_inner = inner;
|
||||
_pskProvider = pskProvider;
|
||||
_publisher = publisher;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Supplies the <c>DistributedPubSub</c> mediator once the actor system exists. Until then
|
||||
/// (and on a host without one) evictions stay local.
|
||||
/// </summary>
|
||||
/// <param name="mediator">The mediator to publish peer invalidations to.</param>
|
||||
public void SetMediator(IActorRef mediator) => _mediator = mediator;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Invalidate(SecretName name)
|
||||
{
|
||||
@@ -85,12 +77,21 @@ public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator
|
||||
_logger.LogInformation(
|
||||
"gRPC preshared key secret '{SecretName}' was written; dropped the cached key for site "
|
||||
+ "{SiteId} and notified the peer central node.", name.Value, siteId);
|
||||
_mediator?.Tell(new Publish(Topic, new SitePskInvalidated(siteId)));
|
||||
_publisher?.Publish(new SitePskInvalidated(siteId));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Apply(ICentralCacheInvalidation invalidation)
|
||||
{
|
||||
if (invalidation is SitePskInvalidated psk)
|
||||
{
|
||||
ApplyPeerInvalidation(psk.SiteId);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Applies an invalidation received from a central node (the peer, or this node's own
|
||||
/// broadcast): evicts the resolver's cached secret and the PSK cache entry. Never re-publishes.
|
||||
/// Applies an invalidation received from the peer central node: evicts the resolver's cached
|
||||
/// secret and the PSK cache entry. Never re-publishes.
|
||||
/// </summary>
|
||||
/// <param name="siteId">The site whose key was rotated.</param>
|
||||
public void ApplyPeerInvalidation(string siteId)
|
||||
@@ -107,51 +108,22 @@ public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator
|
||||
|
||||
/// <summary>
|
||||
/// Peer message: the gRPC preshared key of <paramref name="SiteId"/> was rotated on a central node.
|
||||
/// Rides the central cache-invalidation topic; kept in this namespace because the #96
|
||||
/// <c>site-psk-invalidation</c> topic carries it bare by this type name (see
|
||||
/// <see cref="CentralCacheInvalidationBus"/>).
|
||||
/// </summary>
|
||||
/// <param name="SiteId">The site whose key changed.</param>
|
||||
public sealed record SitePskInvalidated(string SiteId);
|
||||
|
||||
/// <summary>
|
||||
/// Central actor that subscribes to <see cref="SitePskRotationInvalidator.Topic"/> and applies
|
||||
/// each <see cref="SitePskInvalidated"/> to this node's caches.
|
||||
/// </summary>
|
||||
public sealed class SitePskInvalidationSubscriber : ReceiveActor
|
||||
{
|
||||
private readonly SitePskRotationInvalidator _invalidator;
|
||||
|
||||
/// <summary>Creates the subscriber.</summary>
|
||||
/// <param name="invalidator">This node's invalidator.</param>
|
||||
public SitePskInvalidationSubscriber(SitePskRotationInvalidator invalidator)
|
||||
{
|
||||
_invalidator = invalidator;
|
||||
Receive<SitePskInvalidated>(m => _invalidator.ApplyPeerInvalidation(m.SiteId));
|
||||
Receive<SubscribeAck>(_ => { });
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void PreStart()
|
||||
{
|
||||
try
|
||||
{
|
||||
DistributedPubSub.Get(Context.System).Mediator.Tell(
|
||||
new Subscribe(SitePskRotationInvalidator.Topic, Self));
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Non-clustered host (TestKit): peer invalidation is simply unavailable.
|
||||
Context.GetLogger().Debug("DistributedPubSub not available — peer PSK invalidation disabled: {0}", ex.Message);
|
||||
}
|
||||
}
|
||||
}
|
||||
public sealed record SitePskInvalidated(string SiteId) : ICentralCacheInvalidation;
|
||||
|
||||
/// <summary>DI wiring for <see cref="SitePskRotationInvalidator"/>.</summary>
|
||||
public static class SitePskRotationServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Replaces the secrets library's <see cref="ISecretCacheInvalidator"/> with
|
||||
/// <see cref="SitePskRotationInvalidator"/> wrapping the library's resolver. Central only;
|
||||
/// call after <c>AddScadaBridgeSecrets</c> and after <see cref="ISitePskProvider"/> is
|
||||
/// registered.
|
||||
/// <see cref="SitePskRotationInvalidator"/> wrapping the library's resolver, and registers it
|
||||
/// as a central cache-invalidation handler. Central only; call after
|
||||
/// <c>AddScadaBridgeSecrets</c>, after <see cref="ISitePskProvider"/> is registered and after
|
||||
/// <see cref="CentralCacheInvalidationServiceCollectionExtensions.AddCentralCacheInvalidation"/>.
|
||||
/// </summary>
|
||||
/// <param name="services">The service collection.</param>
|
||||
/// <returns>The same collection.</returns>
|
||||
@@ -160,9 +132,12 @@ public static class SitePskRotationServiceCollectionExtensions
|
||||
services.TryAddSingleton(sp => new SitePskRotationInvalidator(
|
||||
sp.GetRequiredService<DefaultSecretResolver>(),
|
||||
sp.GetRequiredService<ISitePskProvider>(),
|
||||
sp.GetService<ICentralCacheInvalidationPublisher>(),
|
||||
sp.GetRequiredService<ILogger<SitePskRotationInvalidator>>()));
|
||||
services.Replace(ServiceDescriptor.Singleton<ISecretCacheInvalidator>(
|
||||
sp => sp.GetRequiredService<SitePskRotationInvalidator>()));
|
||||
services.AddSingleton<ICentralCacheInvalidationHandler>(
|
||||
sp => sp.GetRequiredService<SitePskRotationInvalidator>());
|
||||
return services;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user