Files
ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ExternalDefinitionCacheInvalidationTests.cs
T
Joseph Doherty 153609ef8a 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
2026-11-09 16:04:28 -05:00

159 lines
7.2 KiB
C#

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.ExternalSystems;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.ExternalSystemGateway;
using ZB.MOM.WW.ScadaBridge.Security;
namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests;
/// <summary>
/// ScadaBridge#65: the central <see cref="ExternalDefinitionLookupCache"/> used to be dropped only
/// by its TTL, so an external-system / method / database-connection edit made through the management
/// path was invisible to inbound-method calls on the same node for up to <c>DefinitionCacheTtl</c>.
/// Every such mutation now invalidates it once the change is committed.
/// </summary>
public class ExternalDefinitionCacheInvalidationTests : TestKit
{
private readonly IExternalSystemRepository _repo = Substitute.For<IExternalSystemRepository>();
// A TTL far longer than the test: only an explicit InvalidateAll can make the cache re-fetch.
private readonly ExternalDefinitionLookupCache _cache = new(TimeSpan.FromHours(1));
// The other central node's cache, reached only through the peer invalidation (audit D3 / #83).
private readonly ExternalDefinitionLookupCache _peerCache = new(TimeSpan.FromHours(1));
private readonly PeerCacheInvalidationPublisher _publisher;
private readonly ServiceCollection _services = new ServiceCollection().AddTestConfigurationTransaction();
/// <summary>Initializes the harness with one external system, one method and one DB connection.</summary>
public ExternalDefinitionCacheInvalidationTests()
{
_repo.GetExternalSystemByIdAsync(1, Arg.Any<CancellationToken>())
.Returns(_ => new ExternalSystemDefinition("Erp", "http://erp", "None") { Id = 1 });
_repo.GetExternalSystemMethodByIdAsync(2, Arg.Any<CancellationToken>())
.Returns(_ => new ExternalSystemMethod("Get", "GET", "/x") { Id = 2, ExternalSystemDefinitionId = 1 });
_repo.GetDatabaseConnectionByIdAsync(3, Arg.Any<CancellationToken>())
.Returns(_ => new DatabaseConnectionDefinition("Db", "Server=x") { Id = 3 });
_services.AddScoped(_ => _repo);
_services.AddScoped(_ => Substitute.For<IAuditService>());
_services.AddSingleton(_cache);
_publisher = new PeerCacheInvalidationPublisher(_peerCache);
_services.AddSingleton<ICentralCacheInvalidationPublisher>(_publisher);
}
private IActorRef CreateActor() =>
Sys.ActorOf(Props.Create(() => new ManagementActor(
_services.BuildServiceProvider(), NullLogger<ManagementActor>.Instance)));
private static ManagementEnvelope Admin(object command) =>
new(new AuthenticatedUser("admin", "Admin", new[] { "Designer" }, Array.Empty<string>()),
command, Guid.NewGuid().ToString("N"));
private async Task<int> PrimeAndCountFetches()
{
var fetches = 0;
Task<ExternalSystemDefinition?> Fetch()
{
fetches++;
return Task.FromResult<ExternalSystemDefinition?>(new ExternalSystemDefinition("Erp", "http://erp", "None"));
}
await _cache.GetSystemAsync("Erp", Fetch);
await _cache.GetSystemAsync("Erp", Fetch);
Assert.Equal(1, fetches); // second read was served from the cache
return fetches;
}
private async Task<bool> IsStillCached()
{
var refetched = false;
await _cache.GetSystemAsync("Erp", () =>
{
refetched = true;
return Task.FromResult<ExternalSystemDefinition?>(new ExternalSystemDefinition("Erp", "http://erp", "None"));
});
return !refetched;
}
/// <summary>Every external-system / method / database-connection mutation drops the cached definitions.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task EveryDefinitionMutation_InvalidatesTheCache()
{
object[] commands =
[
new CreateExternalSystemCommand("New", "http://n", "None", null),
new UpdateExternalSystemCommand(1, "Erp", "http://erp2", "None", null),
new DeleteExternalSystemCommand(1),
new CreateExternalSystemMethodCommand(1, "M", "GET", "/m"),
new UpdateExternalSystemMethodCommand(2, Path: "/y"),
new DeleteExternalSystemMethodCommand(2),
new CreateDatabaseConnectionDefCommand("Db2", "Server=y"),
new UpdateDatabaseConnectionDefCommand(3, "Db", "Server=z"),
new DeleteDatabaseConnectionDefCommand(3),
];
var actor = CreateActor();
foreach (var command in commands)
{
_cache.InvalidateAll();
await PrimeAndCountFetches();
Assert.True(await IsStillCached());
actor.Tell(Admin(command));
ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
Assert.False(await IsStillCached()); // dropped by the edit, so the lookup re-fetched
}
}
/// <summary>
/// Audit D3 / #83 (ScadaBridge#100 item 3): every definition edit made through this node
/// drops the peer central node's cache too, instead of leaving it to <c>DefinitionCacheTtl</c>.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task EveryDefinitionMutation_InvalidatesThePeerCache()
{
object[] commands =
[
new CreateExternalSystemCommand("New", "http://n", "None", null),
new UpdateExternalSystemCommand(1, "Erp", "http://erp2", "None", null),
new DeleteExternalSystemCommand(1),
new CreateExternalSystemMethodCommand(1, "M", "GET", "/m"),
new UpdateExternalSystemMethodCommand(2, Path: "/y"),
new DeleteExternalSystemMethodCommand(2),
new CreateDatabaseConnectionDefCommand("Db2", "Server=y"),
new UpdateDatabaseConnectionDefCommand(3, "Db", "Server=z"),
new DeleteDatabaseConnectionDefCommand(3),
];
var actor = CreateActor();
foreach (var command in commands)
{
var fetches = 0;
Task<ExternalSystemDefinition?> Fetch()
{
fetches++;
return Task.FromResult<ExternalSystemDefinition?>(new ExternalSystemDefinition("Erp", "http://erp", "None"));
}
_peerCache.InvalidateAll();
await _peerCache.GetSystemAsync("Erp", Fetch);
await _peerCache.GetSystemAsync("Erp", Fetch);
Assert.Equal(1, fetches); // positive control: the peer serves from its cache
actor.Tell(Admin(command));
ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
await _peerCache.GetSystemAsync("Erp", Fetch);
Assert.Equal(2, fetches); // the peer re-fetched
}
Assert.Equal(commands.Length, _publisher.Published.Count);
Assert.All(_publisher.Published, i => Assert.IsType<ExternalDefinitionsInvalidated>(i));
}
}