diff --git a/docs/components/Security.md b/docs/components/Security.md index f4fff3d1..d9213d54 100644 --- a/docs/components/Security.md +++ b/docs/components/Security.md @@ -163,7 +163,7 @@ if (!AuthorizationPolicies.OperationalAuditRoles.Contains(userRole)) | `JwtExpiryMinutes` | `15` | Lifetime of the bearer JWT issued by `/auth/token`. | | `IdleTimeoutMinutes` | `30` | Cookie session idle timeout; set as the cookie `ExpireTimeSpan`. | | `RoleRefreshThresholdMinutes` | `15` | Age of the role-refresh anchor after which the cookie's roles are re-mapped from the DB (LDAP-free). | -| `RoleMappingCacheSeconds` | `30` | Process-wide `RoleMappingCache` TTL (`0` disables); invalidated on the local node after mapping/scope-rule edits. | +| `RoleMappingCacheSeconds` | `30` | Process-wide `RoleMappingCache` TTL (`0` disables); invalidated after mapping/scope-rule edits on the local node and, over the central cache-invalidation topic, on the peer central node (#100). | | `RequireHttpsCookie` | `true` | When `true`, the cookie is `Secure`-only (HTTPS required). Set `false` for HTTP-only dev deployments; a warning is logged at startup. | | `CookieName` | `ZB.MOM.WW.ScadaBridge.Auth` | Authentication cookie name. Override per deployment when two ScadaBridge stacks share a hostname — browsers scope cookies by host+path but not by port. | diff --git a/docs/plans/2026-11-09-central-cache-staleness-audit.md b/docs/plans/2026-11-09-central-cache-staleness-audit.md index f7258b23..c992794d 100644 --- a/docs/plans/2026-11-09-central-cache-staleness-audit.md +++ b/docs/plans/2026-11-09-central-cache-staleness-audit.md @@ -23,12 +23,12 @@ node after a change is made through the other one. Same defect class as #83 |---|---|---|---|---| | `InboundScriptExecutor` compiled handlers (InboundAPI) | Compiled method delegates | `InvalidateMethod` on delete; change bus on import | **None** — every request revision-checks the cached handler against the resolved `ApiMethod.Script` and recompiles on mismatch | OK | | `ApiMethodCache` (InboundAPI) | `ApiMethod` rows, incl. not-found | Change bus (bundle import); **management create/update/delete — added by #85** | `InboundApi:ApiMethodCacheTtl`, default **5 s** | OK (local gap fixed, see below) | -| `ExternalDefinitionLookupCache` (ESG, central only) | External systems / methods / DB connections (found only) | 9 ManagementActor handlers (#65); bundle import (#83) | `ExternalSystemGateway:DefinitionCacheTtl`, **10 s** on central | Owner decision — #83 | -| `RoleMappingCache` (Security) — the cache at `ManagementActor` `InvalidateRoleMappingCacheAfterCommit` | LDAP group → role map + Deployer site scopes | Mapping / scope-rule commands, post-commit | `Security:RoleMappingCacheSeconds`, default **30 s** (`0` disables) | Owner decision — see D1 | +| `ExternalDefinitionLookupCache` (ESG, central only) | External systems / methods / DB connections (found only) | 9 ManagementActor handlers (#65); bundle import (#83) | Edit or import through the other node: **none** (`ExternalDefinitionsInvalidated` on `central-cache-invalidation`, #100); direct SQL: `ExternalSystemGateway:DefinitionCacheTtl`, **10 s** on central | Fixed — #100 item 3 (see D3) | +| `RoleMappingCache` (Security) — the cache at `ManagementActor` `InvalidateRoleMappingCacheAfterCommit` | LDAP group → role map + Deployer site scopes | Mapping / scope-rule commands, post-commit | **None** (`RoleMappingsInvalidated` on `central-cache-invalidation`, #100); TTL `Security:RoleMappingCacheSeconds` (**30 s**) only if the broadcast is lost | Fixed — #100 item 3 (see D1) | | `PendingUpdatesCache` (DeploymentManager, #91) | Fleet pending-updates snapshot | Deployment finished on this node | `DeploymentManager:PendingUpdatesCacheTtl`, default **60 s** | OK — display only | | `CentralCommunicationActor` site-address cache (+ `GrpcSiteTransport` channel pairs) | Site gRPC endpoints | Site create/update/delete → `RefreshSiteAddresses` | Periodic DB reload, **60 s** (hard-coded) | OK — rare edit, self-heals | | `SiteStreamGrpcClientFactory` clients | Per (site, endpoint) stream clients | Site delete → `RemoveSiteAsync` (local only) | Not stale data: a removed site's client on the peer is reclaimed by idle eviction, `Communication:GrpcClientIdleTtl` (**30 min**) × 1.5 after last use | OK — resource, not correctness | -| `SitePskProvider` (Host) | Per-site gRPC PSK from config / secret store | Site removal; `SB-GRPC-PSK-{site}` write via `/admin/secrets`; site auth rejection; `SitePsks` reload (#96) | Write via `/admin/secrets`: **none** (peer told over `DistributedPubSub` topic `site-psk-invalidation`); other writes: first rejected call after `Secrets:ResolveCacheTtl` | Fixed — #96 (see D2) | +| `SitePskProvider` (Host) | Per-site gRPC PSK from config / secret store | Site removal; `SB-GRPC-PSK-{site}` write via `/admin/secrets`; site auth rejection; `SitePsks` reload (#96) | Write via `/admin/secrets`: **none** (peer told over `DistributedPubSub` topic `central-cache-invalidation`, formerly `site-psk-invalidation`); other writes: first rejected call after `Secrets:ResolveCacheTtl` | Fixed — #96 (see D2) | | `DefaultSecretResolver` (shared `ZB.MOM.WW.Secrets`) | Decrypted secret values | Lib's own invalidator (UI writes, sync service) | `Secrets:ResolveCacheTtl`, **30 s** | OK — owned by the shared lib | | `KpiSnapshotCache` (CentralUI) | KPI query results (operational data) | — | **8 s** | OK — not config | | `SharedAlarmSummaryService` (CentralUI) | Per-site alarm summary fan-out | — | **12 s** cold-cache window | OK — not config | @@ -75,3 +75,16 @@ broadcast was added. - **D3 — `ExternalDefinitionLookupCache` peer bound (10 s).** Already tracked by #83; the recommendation there (accept and document) still stands. If D1 gets a broadcast topic, this cache and `ApiMethodCache` should join it rather than add a second mechanism. + +## Follow-up: one central cache-invalidation topic (#100 item 3, 2026-11-09) + +D1 and D3 now ride a single `DistributedPubSub` topic, `central-cache-invalidation`, with typed +messages (`ICentralCacheInvalidation`: `SitePskInvalidated`, `RoleMappingsInvalidated`, +`ExternalDefinitionsInvalidated`, `InboundApiMethodInvalidated`) instead of one topic per cache. +The #96 topic was migrated onto it: `SitePskInvalidated` travels on the shared topic, and for one +release is also published bare on `site-psk-invalidation` (still subscribed), so a mixed-version +central pair keeps working. Each envelope carries the publisher's origin id and the subscriber +skips its own. Publishers: ManagementActor post-commit (role mappings, external definitions), +the Transport bundle importer (external definitions) and the PSK invalidator. Every cache keeps its TTL / revision check as the fallback for a +lost broadcast or an out-of-band (SQL) edit. Design: Component-Host.md → *Central cache +invalidation*; tests include a real two-node cluster (`CentralCacheInvalidationTwoNodeTests`). diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 9ab6f54f..bb277a10 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -207,8 +207,9 @@ site can omit it. Central caches each resolved key, and drops the entry (Gitea #96) when: the site is removed; the secret `SB-GRPC-PSK-` is written through central's secrets write path (`/admin/secrets`) — `SitePskRotationInvalidator` wraps the secrets library's `ISecretCacheInvalidator` and also -broadcasts a `SitePskInvalidated` on the central `DistributedPubSub` topic `site-psk-invalidation` -so the peer central node drops its copy too; the site rejects a call with `PermissionDenied` / +broadcasts a `SitePskInvalidated` on the central cache-invalidation topic `central-cache-invalidation` +(ScadaBridge#100 item 3; the #96 topic `site-psk-invalidation` is kept for one release for +mixed-version pairs — see Component-Host.md) so the peer central node drops its copy too; the site rejects a call with `PermissionDenied` / `Unauthenticated` (`PskRejectionRefreshHandler`, wrapped around every per-site channel by `WithSiteCredentials`, so the next call re-reads the key — this covers keys rotated outside the process, e.g. by the `secret` CLI, after at most the resolver's `Secrets:ResolveCacheTtl`); a diff --git a/docs/requirements/Component-ExternalSystemGateway.md b/docs/requirements/Component-ExternalSystemGateway.md index 27cec1ab..d316fa93 100644 --- a/docs/requirements/Component-ExternalSystemGateway.md +++ b/docs/requirements/Component-ExternalSystemGateway.md @@ -120,7 +120,7 @@ Scripts choose between two call modes per invocation, mirroring the dual-mode da - Database connections use **standard ADO.NET connection pooling** per named connection. No custom pool management. - Pool behavior (max pool size, connection lifetime, etc.) can be tuned via connection string parameters in the database connection definition if needed. -- **Definition cache (central only).** `ExternalDefinitionLookupCache` holds resolved external-system, method and database-connection definitions for `ExternalSystemGateway:DefinitionCacheTtl` (default 0 = off in the options; the central host sets 10 s because each lookup there is a SQL Server round-trip). Only found definitions are cached, so a new definition is visible immediately; an edit/delete within one TTL. Sites keep it off — `SiteExternalSystemRepository` already serves external systems from the deploy-invalidated `ExternalSystemDefinitionCache` snapshot. +- **Definition cache (central only).** `ExternalDefinitionLookupCache` holds resolved external-system, method and database-connection definitions for `ExternalSystemGateway:DefinitionCacheTtl` (default 0 = off in the options; the central host sets 10 s because each lookup there is a SQL Server round-trip). Only found definitions are cached, so a new definition is visible immediately. A ManagementActor edit (#65) or a committed Transport bundle import (#83) drops the cache on the node that made the change and publishes `ExternalDefinitionsInvalidated` on the central cache-invalidation topic so the peer central node drops its copy too (ScadaBridge#100 item 3); an edit that reaches neither path (direct SQL) is visible within one TTL. Sites keep it off — `SiteExternalSystemRepository` already serves external systems from the deploy-invalidated `ExternalSystemDefinitionCache` snapshot. - GET/DELETE parameters are rendered into the query string with scalars via `ToString()` and collections/objects JSON-serialised (never a CLR type name). - Synchronous failures on `Database.Connection()` (e.g., unreachable server) return an error to the calling script, consistent with external system permanent failure handling. diff --git a/docs/requirements/Component-Host.md b/docs/requirements/Component-Host.md index b2ebf95d..351a6668 100644 --- a/docs/requirements/Component-Host.md +++ b/docs/requirements/Component-Host.md @@ -351,5 +351,6 @@ built)* section for where the shipped code deviates from the original design). - **Port options come from the bound `NodeOptions`.** `Program.cs` binds `ScadaBridge:Node` once and reads `GrpcPort` / `CentralGrpcPort` / `MetricsPort` from it; the 8083/8084 defaults live only in `NodeOptions`. Central's dedicated metrics listener remains an explicit *presence* check on the raw key (a defaulted read would bind a new port on every existing central node). - **Inbound method warm-up is a hosted service** (`InboundMethodWarmupService`, registered after `AkkaHostedService`), not a pre-`RunAsync` loop — see Component-InboundAPI.md. - **`SitePskProvider` distinguishes a secret-store fault from a missing secret.** A store fault logs a Warning with the exception and throws `SecretStoreUnavailableException` (not cached); only a genuinely absent/empty/tombstoned secret produces the "seed `SB-GRPC-PSK-{site}`" message. -- **`SitePskProvider`'s cache follows an in-place rotation (Gitea #96).** Central replaces the secrets library's `ISecretCacheInvalidator` with `SitePskRotationInvalidator` (`AddSitePskRotationInvalidation`), which drops the site's cached key when `SB-GRPC-PSK-{site}` is written and publishes `SitePskInvalidated` on the `site-psk-invalidation` `DistributedPubSub` topic; `AkkaHostedService.RegisterCentralActors` starts the `SitePskInvalidationSubscriber` that applies the peer's broadcasts. A site's auth rejection and a `SitePsks` reload also drop the cache — see Component-Communication.md → *Authentication*. +- **`SitePskProvider`'s cache follows an in-place rotation (Gitea #96).** Central replaces the secrets library's `ISecretCacheInvalidator` with `SitePskRotationInvalidator` (`AddSitePskRotationInvalidation`), which drops the site's cached key when `SB-GRPC-PSK-{site}` is written and publishes `SitePskInvalidated` to the peer over the central cache-invalidation topic (below). A site's auth rejection and a `SitePsks` reload also drop the cache — see Component-Communication.md → *Authentication*. +- **Central cache invalidation (ScadaBridge#100 item 3).** One `DistributedPubSub` topic, `central-cache-invalidation`, carries typed `ICentralCacheInvalidation` messages (Commons `Messages/Communication/CentralCacheInvalidations.cs`) between the two central nodes, each wrapped in a `CentralCacheInvalidationEnvelope` stamped with the publishing process's origin id. `AddCentralCacheInvalidation` registers `CentralCacheInvalidationBus` as `ICentralCacheInvalidationPublisher`; `AkkaHostedService.RegisterCentralActors` gives it the mediator and starts `CentralCacheInvalidationSubscriber`, which skips its own node's envelopes and hands the peer's to every registered `ICentralCacheInvalidationHandler`: `SitePskRotationInvalidator` (`SitePskInvalidated`, #96), `RoleMappingCache` (`RoleMappingsInvalidated`, audit D1), and `ExternalDefinitionLookupCache` (`ExternalDefinitionsInvalidated`, audit D3 / #83). Publishers invalidate locally first, post-commit, then publish; delivery is best-effort, so each cache's TTL or revision check stays the fallback. Migration: the #96 topic `site-psk-invalidation` is still published to (bare `SitePskInvalidated`) and subscribed, so a briefly mixed-version central pair keeps exchanging PSK invalidations — remove it once every central node runs this build. - Central registers `ExternalSystemGatewayOptions.DefinitionCacheTtl = 10 s` ahead of `AddExternalSystemGateway()` so an explicit config value wins; sites leave it off. diff --git a/docs/requirements/Component-ManagementService.md b/docs/requirements/Component-ManagementService.md index 2182eae1..e7e2e921 100644 --- a/docs/requirements/Component-ManagementService.md +++ b/docs/requirements/Component-ManagementService.md @@ -194,7 +194,7 @@ The two-person authorization workflow for writes through the MxAccess Gateway. B - **ListApiKeys** / **CreateApiKey** / **UpdateApiKey** / **EnableApiKey** / **DisableApiKey** / **DeleteApiKey**: Manage API keys. - **ListRoleMappings** / **CreateRoleMapping** / **UpdateRoleMapping** / **DeleteRoleMapping**: Manage LDAP group-to-role mappings. - **ListScopeRules** / **AddScopeRule** / **DeleteScopeRule**: Manage site scope rules on role mappings. -- Every role-mapping and scope-rule mutation (Create/Update/DeleteRoleMapping, Add/DeleteScopeRule) calls `RoleMappingCache.Invalidate()` as a **post-commit** action, so the change takes effect for the next role mapping on the node that served the command instead of after the `RoleMappingCacheSeconds` TTL (default 30 s); the other central node still converges within that TTL. Post-commit, because invalidating inside the transaction would let a concurrent mapping re-cache the pre-change table. +- Every role-mapping and scope-rule mutation (Create/Update/DeleteRoleMapping, Add/DeleteScopeRule) calls `RoleMappingCache.Invalidate()` as a **post-commit** action, so the change takes effect for the next role mapping on the node that served the command instead of after the `RoleMappingCacheSeconds` TTL (default 30 s), and publishes `RoleMappingsInvalidated` on the central cache-invalidation topic so the other central node drops its snapshot too (audit D1, ScadaBridge#100 item 3); the TTL is only the fallback if that best-effort broadcast is lost. The external-definition edit handlers do the same with `ExternalDefinitionsInvalidated` (#65, #83). Post-commit, because invalidating inside the transaction would let a concurrent mapping re-cache the pre-change table. ### Audit Log diff --git a/docs/requirements/Component-Security.md b/docs/requirements/Component-Security.md index 4ebdd359..e3fedc71 100644 --- a/docs/requirements/Component-Security.md +++ b/docs/requirements/Component-Security.md @@ -53,7 +53,7 @@ Set in a local or docker-dev environment via the environment variable `ScadaBrid - **Idle timeout**: Configurable, default **30 minutes**. Without eligible activity within the idle window, the session is rejected and the user must re-login. Passive `/auth/ping`, activity prevalidation, and Blazor transport requests preserve the activity anchor and suppress sliding renewal; the activity endpoint renews only after authentication and antiforgery validation. Tracked via a `LastActivity` last-activity timestamp claim. The cookie's `ExpireTimeSpan` is set to the idle timeout and `SlidingExpiration` renews it on activity, so the cookie window and the explicit `OnValidatePrincipal` idle check use the **same** value and cannot contradict each other. - **Role-mapping refresh (LDAP-free)**: Configurable, default **15 minutes** (`SecurityOptions.RoleRefreshThresholdMinutes`). At login the session stores the user's raw LDAP groups (one `zb:group` claim each) plus a `zb:lastrolerefresh` anchor. Once the anchor is older than the threshold, `OnValidatePrincipal` re-runs the **DB-backed** `RoleMapper` on the stored groups — **with no LDAP call** — rebuilds the role/scope claims via the shared claim-builder, advances the anchor, and re-issues the cookie. Central role-mapping (DB) changes — including a **revoked** mapping that drops the user's roles, and changed site-scope rules — take effect within this window. Ordinary HTTP validation uses this threshold. A live circuit refreshes mappings in fresh scopes on authentication-state reads — display reads (`GetAuthenticationStateAsync`) at most once per `CookieAuthenticationStateProvider.NonAuthoritativeRefreshInterval` (10 s) per circuit — and mutation entrypoints (`GetAuthoritativeStateAsync`) require a successful current mapping before every dispatch. -- **Role-mapping cache**: `RoleMapper` reads the LDAP-group→role table and the scope rules of every Deployer mapping through a process-wide `RoleMappingCache` snapshot (`SecurityOptions.RoleMappingCacheSeconds`, default **30**; `0` disables). One refresh loads the whole table plus all Deployer scope rules, so a cache hit costs no database round-trip — previously every Basic-authenticated management request and every circuit auth-state read issued `GetAllMappingsAsync` plus one `GetScopeRulesForMappingAsync` per matched Deployer mapping. Without an explicit invalidation a mapping or scope-rule change is therefore visible on every node within the TTL. `RoleMappingCache.Invalidate()` discards the calling node's snapshot so its next mapping re-reads the database (a refresh started before an invalidation is never published over it); other nodes still rely on the TTL. `RoleMappingCacheSeconds = 0` bypasses the snapshot and reads only the matched mappings' scope rules per call. +- **Role-mapping cache**: `RoleMapper` reads the LDAP-group→role table and the scope rules of every Deployer mapping through a process-wide `RoleMappingCache` snapshot (`SecurityOptions.RoleMappingCacheSeconds`, default **30**; `0` disables). One refresh loads the whole table plus all Deployer scope rules, so a cache hit costs no database round-trip — previously every Basic-authenticated management request and every circuit auth-state read issued `GetAllMappingsAsync` plus one `GetScopeRulesForMappingAsync` per matched Deployer mapping. Without an explicit invalidation a mapping or scope-rule change is therefore visible on every node within the TTL. `RoleMappingCache.Invalidate()` discards the calling node's snapshot so its next mapping re-reads the database (a refresh started before an invalidation is never published over it); the Management Service's mapping/scope-rule handlers also publish `RoleMappingsInvalidated` on the central cache-invalidation topic, so the peer central node discards its snapshot too (audit D1, ScadaBridge#100 item 3) and the TTL is only the fallback. `RoleMappingCacheSeconds = 0` bypasses the snapshot and reads only the matched mappings' scope rules per call. #### Session Validation (`OnValidatePrincipal`) - The cookie principal is built at login by a **single shared claim-builder** (`SessionClaimBuilder`). The `OnValidatePrincipal` role-refresh path rebuilds the principal through the **same** builder, so the login and refresh claim shapes cannot drift. diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ICentralCacheInvalidationPublisher.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ICentralCacheInvalidationPublisher.cs new file mode 100644 index 00000000..6317e09d --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ICentralCacheInvalidationPublisher.cs @@ -0,0 +1,33 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; + +namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; + +/// +/// Tells the peer central node to drop a cache entry this node has just invalidated +/// (ScadaBridge#100 item 3). Central only; resolved optionally, so a host without it (a site +/// node, a plain test host) simply keeps invalidations local. +/// +/// +/// Callers invalidate their own node's cache first and then publish: the publishing node does +/// not apply its own broadcast. Publishing never throws; before the actor system is up it is a +/// no-op, and the peer's cache TTL bounds the staleness exactly as it did before. +/// +public interface ICentralCacheInvalidationPublisher +{ + /// Broadcasts to the other central node. + /// The typed invalidation. + void Publish(ICentralCacheInvalidation invalidation); +} + +/// +/// Applies a peer's to one of this node's caches. Each +/// cache-owning component registers one (as an enumerable service) and ignores the message types +/// it does not own; the central invalidation subscriber hands every received message to all of +/// them. +/// +public interface ICentralCacheInvalidationHandler +{ + /// Applies if this handler owns its type. + /// The received invalidation. + void Apply(ICentralCacheInvalidation invalidation); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Communication/CentralCacheInvalidations.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Communication/CentralCacheInvalidations.cs new file mode 100644 index 00000000..488c94c1 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Communication/CentralCacheInvalidations.cs @@ -0,0 +1,44 @@ +namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; + +/// +/// Marker for a typed cache-invalidation message exchanged between the two central nodes over +/// the single central DistributedPubSub topic central-cache-invalidation +/// (ScadaBridge#100 item 3, #83, #103). Each central node caches configuration-database state +/// in-process; a change made through one node drops that node's cache directly and publishes +/// one of these so the peer drops its copy too, instead of waiting out the cache TTL. +/// +/// +/// Delivery is best-effort and at-most-once: every cache that takes part still has a TTL or a +/// per-request revision check as its correctness fallback. Message contracts follow the +/// additive-only evolution rule — a node that does not know a message type ignores it. +/// +public interface ICentralCacheInvalidation; + +/// +/// The LDAP group → role mapping table or a Deployer site-scope rule changed: drop the +/// RoleMappingCache snapshot (audit D1). +/// +public sealed record RoleMappingsInvalidated : ICentralCacheInvalidation; + +/// +/// An external system, external-system method or database connection definition changed +/// (management edit or Transport bundle import): drop every ExternalDefinitionLookupCache +/// entry (audit D3, #83). +/// +public sealed record ExternalDefinitionsInvalidated : ICentralCacheInvalidation; + +/// +/// An inbound API method was created, updated, deleted or recompiled: drop the method's +/// compiled handler, its known-bad compile record and its cached row (#103). +/// +/// The API method's name. +public sealed record InboundApiMethodInvalidated(string MethodName) : ICentralCacheInvalidation; + +/// +/// What travels on the central-cache-invalidation topic: the invalidation plus the +/// publishing node's process-unique origin id, so a node can skip its own broadcast (it already +/// invalidated locally before publishing). +/// +/// Process-unique id of the publishing node's invalidation bus. +/// The typed invalidation. +public sealed record CentralCacheInvalidationEnvelope(string OriginId, ICentralCacheInvalidation Invalidation); diff --git a/src/ZB.MOM.WW.ScadaBridge.ExternalSystemGateway/ExternalDefinitionLookupCache.cs b/src/ZB.MOM.WW.ScadaBridge.ExternalSystemGateway/ExternalDefinitionLookupCache.cs index d3ae1746..1b6401af 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ExternalSystemGateway/ExternalDefinitionLookupCache.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ExternalSystemGateway/ExternalDefinitionLookupCache.cs @@ -2,6 +2,7 @@ using System.Collections.Concurrent; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Entities.ExternalSystems; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; namespace ZB.MOM.WW.ScadaBridge.ExternalSystemGateway; @@ -20,7 +21,7 @@ namespace ZB.MOM.WW.ScadaBridge.ExternalSystemGateway; /// deleted one is visible within one TTL. /// /// -public sealed class ExternalDefinitionLookupCache : IExternalDefinitionCacheInvalidator +public sealed class ExternalDefinitionLookupCache : IExternalDefinitionCacheInvalidator, ICentralCacheInvalidationHandler { private sealed record Entry(object? Value, DateTimeOffset ExpiresAtUtc); @@ -75,10 +76,20 @@ public sealed class ExternalDefinitionLookupCache : IExternalDefinitionCacheInva /// Drops every cached definition. /// /// Called post-commit by the ManagementActor edit handlers (Gitea #65) and by the Transport - /// bundle importer through (Gitea #83). + /// bundle importer through (Gitea #83). Both + /// also publish so the peer central node drops its + /// copy through (ScadaBridge#100 item 3). /// public void InvalidateAll() => _entries.Clear(); + /// Applies a peer central node's . + /// The received invalidation; other types are ignored. + public void Apply(ICentralCacheInvalidation invalidation) + { + if (invalidation is ExternalDefinitionsInvalidated) + InvalidateAll(); + } + private async Task GetAsync(string key, Func> fetch) where T : class { if (!Enabled) diff --git a/src/ZB.MOM.WW.ScadaBridge.ExternalSystemGateway/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.ExternalSystemGateway/ServiceCollectionExtensions.cs index 30bd274e..34dd647f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ExternalSystemGateway/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ExternalSystemGateway/ServiceCollectionExtensions.cs @@ -48,6 +48,10 @@ public static class ServiceCollectionExtensions // importer can drop it after a committed import without referencing this assembly. services.TryAddSingleton( sp => sp.GetRequiredService()); + // Audit D3 / #83 (ScadaBridge#100 item 3): an edit or import made through the other central + // node reaches this node's cache over the central cache-invalidation topic. + services.TryAddEnumerable(ServiceDescriptor.Singleton( + sp => sp.GetRequiredService())); services.AddScoped(); services.AddScoped(sp => sp.GetRequiredService()); services.AddScoped(); diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index a7ff5a13..5a5ce4b1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -463,16 +463,23 @@ akka {{ "CentralControlGrpcService readiness set (service bound: {Bound})", centralControlGrpc is not null); - // Peer half of site-PSK rotation invalidation (Gitea #96): subscribe to the topic, and give - // the invalidator the mediator so a rotation written on this node reaches the other one. - var pskRotation = _serviceProvider.GetService(); - if (pskRotation is not null) + // Peer half of central cache invalidation (ScadaBridge#100 item 3, generalising the #96 + // site-PSK topic): subscribe this node's cache handlers (site PSK, role mappings, external + // definitions, inbound API methods) to the shared topic, and give the bus the mediator so + // an invalidation made on this node reaches the other one. + var cacheInvalidationBus = _serviceProvider.GetService(); + if (cacheInvalidationBus is not null) { + var handlers = _serviceProvider + .GetServices() + .ToList(); _actorSystem!.ActorOf( - Props.Create(() => new SitePskInvalidationSubscriber(pskRotation)), - "site-psk-invalidation"); - pskRotation.SetMediator( + Props.Create(() => new CentralCacheInvalidationSubscriber(cacheInvalidationBus, handlers)), + "central-cache-invalidation"); + cacheInvalidationBus.SetMediator( Akka.Cluster.Tools.PublishSubscribe.DistributedPubSub.Get(_actorSystem!).Mediator); + _logger.LogInformation( + "Central cache invalidation subscribed ({HandlerCount} handlers)", handlers.Count); } // Wire up the CommunicationService with the actor reference diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/CentralCacheInvalidation.cs b/src/ZB.MOM.WW.ScadaBridge.Host/CentralCacheInvalidation.cs new file mode 100644 index 00000000..557a4267 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Host/CentralCacheInvalidation.cs @@ -0,0 +1,185 @@ +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; + +namespace ZB.MOM.WW.ScadaBridge.Host; + +/// +/// Central's cross-node cache-invalidation channel (ScadaBridge#100 item 3): one +/// DistributedPubSub topic, , carrying typed +/// messages inside a +/// . It generalises the per-cache topic #96 added +/// for site PSK rotation; the role-mapping cache (audit D1), the external-definition cache +/// (audit D3 / #83), the inbound API compile state (#103) and the site PSK cache all ride it. +/// +/// +/// +/// Origin filtering. A publisher has already invalidated its own caches, and a node +/// receives its own broadcast too. Re-applying is harmless for most caches but not free for all +/// of them (the inbound API would drop a handler it has just compiled, and recompile it on the +/// next request), so each envelope carries this bus's process-unique and +/// the subscriber skips its own. +/// +/// +/// Migration from site-psk-invalidation. A is also +/// published, bare, on , and the subscriber still listens there, +/// so a central pair that is briefly mixed-version during an upgrade keeps exchanging PSK +/// invalidations. Applying a PSK invalidation twice is harmless. Remove both once every central +/// node runs this build. +/// +/// +public sealed class CentralCacheInvalidationBus : ICentralCacheInvalidationPublisher +{ + /// The single central cache-invalidation topic. + public const string Topic = "central-cache-invalidation"; + + /// The #96 per-cache topic, kept for mixed-version central pairs (see remarks). + public const string LegacySitePskTopic = "site-psk-invalidation"; + + private readonly ILogger _logger; + private volatile IActorRef? _mediator; + + /// Creates the bus. + /// Logger. + public CentralCacheInvalidationBus(ILogger logger) + { + ArgumentNullException.ThrowIfNull(logger); + _logger = logger; + } + + /// This process's origin id, stamped on every envelope it publishes. + public string OriginId { get; } = Guid.NewGuid().ToString("N"); + + /// + /// Supplies the DistributedPubSub mediator once the actor system exists. Until then + /// publishing is a no-op (callers have already invalidated locally). + /// + /// The mediator to publish to. + public void SetMediator(IActorRef mediator) => _mediator = mediator; + + /// + public void Publish(ICentralCacheInvalidation invalidation) + { + ArgumentNullException.ThrowIfNull(invalidation); + var mediator = _mediator; + if (mediator is null) + { + _logger.LogDebug( + "No pub-sub mediator yet; {Invalidation} stays local to this node.", invalidation.GetType().Name); + return; + } + + try + { + mediator.Tell(new Publish(Topic, new CentralCacheInvalidationEnvelope(OriginId, invalidation))); + if (invalidation is SitePskInvalidated psk) + { + mediator.Tell(new Publish(LegacySitePskTopic, psk)); + } + } + catch (Exception ex) + { + // Never let a broadcast failure fail the write that triggered it: the peer's TTL + // (or its per-request revision check) still bounds the staleness. + _logger.LogWarning(ex, "Publishing {Invalidation} to the peer central node failed.", invalidation.GetType().Name); + } + } + + /// + /// True when was published by this process (and was therefore + /// already applied locally). + /// + /// The received envelope. + /// true for this node's own broadcast. + public bool IsOwn(CentralCacheInvalidationEnvelope envelope) => + string.Equals(envelope.OriginId, OriginId, StringComparison.Ordinal); +} + +/// +/// Central actor that subscribes to (and the +/// legacy ) and hands every +/// invalidation published by the peer central node to this node's +/// s. +/// +public sealed class CentralCacheInvalidationSubscriber : ReceiveActor +{ + private readonly IReadOnlyList _handlers; + + /// Creates the subscriber. + /// This node's bus, used to recognise this node's own broadcasts. + /// This node's cache handlers. + public CentralCacheInvalidationSubscriber( + CentralCacheInvalidationBus bus, IReadOnlyList handlers) + { + ArgumentNullException.ThrowIfNull(bus); + ArgumentNullException.ThrowIfNull(handlers); + _handlers = handlers; + + Receive(envelope => + { + if (!bus.IsOwn(envelope) && envelope.Invalidation is not null) + { + Apply(envelope.Invalidation); + } + }); + // Bare message on the #96 topic, from a peer that predates the shared topic. + Receive(Apply); + Receive(_ => { }); + } + + private void Apply(ICentralCacheInvalidation invalidation) + { + foreach (var handler in _handlers) + { + try + { + handler.Apply(invalidation); + } + catch (Exception ex) + { + // One failing cache must not stop the others; its own TTL still bounds it. + Context.GetLogger().Warning(ex, "Applying peer invalidation {0} in {1} failed.", + invalidation.GetType().Name, handler.GetType().Name); + } + } + } + + /// + protected override void PreStart() + { + try + { + var mediator = DistributedPubSub.Get(Context.System).Mediator; + mediator.Tell(new Subscribe(CentralCacheInvalidationBus.Topic, Self)); + mediator.Tell(new Subscribe(CentralCacheInvalidationBus.LegacySitePskTopic, Self)); + } + catch (Exception ex) + { + // Non-clustered host (TestKit): peer invalidation is simply unavailable. + Context.GetLogger().Debug("DistributedPubSub not available — peer cache invalidation disabled: {0}", ex.Message); + } + } +} + +/// DI wiring for . +public static class CentralCacheInvalidationServiceCollectionExtensions +{ + /// + /// Registers the central cache-invalidation bus as . + /// Central only; the actor system wires the mediator and the subscriber at startup. + /// + /// The service collection. + /// The same collection. + public static IServiceCollection AddCentralCacheInvalidation(this IServiceCollection services) + { + services.TryAddSingleton(); + services.TryAddSingleton( + sp => sp.GetRequiredService()); + return services; + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs index 80a63d24..298e4a7c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs @@ -413,7 +413,10 @@ try builder.Services.AddScadaBridgeSecrets(builder.Configuration, SecretsNodeRole.Central); // An in-place rotation of a site's SB-GRPC-PSK-{siteId} through /admin/secrets must drop // central's cached copy of that key on BOTH central nodes (Gitea #96). Wraps the library's - // ISecretCacheInvalidator; the peer half is wired in AkkaHostedService.RegisterCentralActors. + // ISecretCacheInvalidator; the peer half rides the central cache-invalidation topic + // (AddCentralCacheInvalidation, ScadaBridge#100 item 3 — also used by the role-mapping, + // external-definition and inbound-API caches), wired in AkkaHostedService.RegisterCentralActors. + builder.Services.AddCentralCacheInvalidation(); builder.Services.AddSitePskRotationInvalidation(); // Secrets UI authorization: adds the named policies secrets:manage + secrets:reveal // (role-based) consumed by the /admin/secrets page. AddSecretsAuthorization only ADDS diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/SitePskRotationInvalidator.cs b/src/ZB.MOM.WW.ScadaBridge.Host/SitePskRotationInvalidator.cs index 70a98b14..6076a4e4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/SitePskRotationInvalidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/SitePskRotationInvalidator.cs @@ -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; /// /// Why the peer needs telling. Both central nodes cache the key in their own /// (and their own resolver cache), and the write only ran on one of -/// them. The peer is told over the central DistributedPubSub mediator on -/// — 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 on the central cache-invalidation +/// topic (, ScadaBridge#100 item 3, which generalised +/// the #96 site-psk-invalidation topic); this class is also the peer's handler for it. /// /// /// Writes made outside this process (the secret CLI, a direct SQL write) never reach this @@ -36,23 +35,22 @@ namespace ZB.MOM.WW.ScadaBridge.Host; /// ControlPlaneCredentials.WithSiteCredentials). /// /// -public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator +public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator, ICentralCacheInvalidationHandler { - /// The DistributedPubSub topic the central nodes exchange PSK invalidations on. - public const string Topic = "site-psk-invalidation"; - private readonly ISecretCacheInvalidator _inner; private readonly ISitePskProvider _pskProvider; + private readonly ICentralCacheInvalidationPublisher? _publisher; private readonly ILogger _logger; - private volatile IActorRef? _mediator; /// Creates the invalidator. /// The library's own invalidator (the secret resolver's cache). /// Central's per-site PSK cache. + /// The central cache-invalidation publisher; null keeps evictions local. /// Logger. public SitePskRotationInvalidator( ISecretCacheInvalidator inner, ISitePskProvider pskProvider, + ICentralCacheInvalidationPublisher? publisher, ILogger logger) { ArgumentNullException.ThrowIfNull(inner); @@ -60,16 +58,10 @@ public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator ArgumentNullException.ThrowIfNull(logger); _inner = inner; _pskProvider = pskProvider; + _publisher = publisher; _logger = logger; } - /// - /// Supplies the DistributedPubSub mediator once the actor system exists. Until then - /// (and on a host without one) evictions stay local. - /// - /// The mediator to publish peer invalidations to. - public void SetMediator(IActorRef mediator) => _mediator = mediator; - /// 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)); + } + + /// + public void Apply(ICentralCacheInvalidation invalidation) + { + if (invalidation is SitePskInvalidated psk) + { + ApplyPeerInvalidation(psk.SiteId); + } } /// - /// 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. /// /// The site whose key was rotated. public void ApplyPeerInvalidation(string siteId) @@ -107,51 +108,22 @@ public sealed class SitePskRotationInvalidator : ISecretCacheInvalidator /// /// Peer message: the gRPC preshared key of was rotated on a central node. +/// Rides the central cache-invalidation topic; kept in this namespace because the #96 +/// site-psk-invalidation topic carries it bare by this type name (see +/// ). /// /// The site whose key changed. -public sealed record SitePskInvalidated(string SiteId); - -/// -/// Central actor that subscribes to and applies -/// each to this node's caches. -/// -public sealed class SitePskInvalidationSubscriber : ReceiveActor -{ - private readonly SitePskRotationInvalidator _invalidator; - - /// Creates the subscriber. - /// This node's invalidator. - public SitePskInvalidationSubscriber(SitePskRotationInvalidator invalidator) - { - _invalidator = invalidator; - Receive(m => _invalidator.ApplyPeerInvalidation(m.SiteId)); - Receive(_ => { }); - } - - /// - 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; /// DI wiring for . public static class SitePskRotationServiceCollectionExtensions { /// /// Replaces the secrets library's with - /// wrapping the library's resolver. Central only; - /// call after AddScadaBridgeSecrets and after is - /// registered. + /// wrapping the library's resolver, and registers it + /// as a central cache-invalidation handler. Central only; call after + /// AddScadaBridgeSecrets, after is registered and after + /// . /// /// The service collection. /// The same collection. @@ -160,9 +132,12 @@ public static class SitePskRotationServiceCollectionExtensions services.TryAddSingleton(sp => new SitePskRotationInvalidator( sp.GetRequiredService(), sp.GetRequiredService(), + sp.GetService(), sp.GetRequiredService>())); services.Replace(ServiceDescriptor.Singleton( sp => sp.GetRequiredService())); + services.AddSingleton( + sp => sp.GetRequiredService()); return services; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs index 4732adac..9ffe4ef0 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs @@ -21,6 +21,7 @@ using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Security; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; @@ -2386,15 +2387,22 @@ public class ManagementActor : ReceiveActor /// call on this node resolves the fresh definition rather than waiting out /// DefinitionCacheTtl (ScadaBridge#65). Queued post-commit: invalidating before the commit /// would let a concurrent lookup re-cache the pre-edit row. The cache is per-node, so the other - /// central node still converges within DefinitionCacheTtl. A no-op when the cache is not - /// registered (site roles, plain test hosts). + /// central node is told with an on the central + /// cache-invalidation topic (audit D3 / #83, ScadaBridge#100 item 3). A no-op when neither the + /// cache nor the publisher is registered (site roles, plain test hosts). /// private static Task InvalidateExternalDefinitionCacheAfterCommit(IServiceProvider sp) { var cache = sp.GetService(); - return cache is null + var peers = sp.GetService(); + return cache is null && peers is null ? Task.CompletedTask - : AfterConfigurationCommit(sp, () => { cache.InvalidateAll(); return Task.CompletedTask; }); + : AfterConfigurationCommit(sp, () => + { + cache?.InvalidateAll(); + peers?.Publish(new ExternalDefinitionsInvalidated()); + return Task.CompletedTask; + }); } private static async Task HandleListExternalSystems(IServiceProvider sp) @@ -2902,15 +2910,23 @@ public class ManagementActor : ReceiveActor /// Drops this node's snapshot once a mapping / scope-rule /// change has committed, so the next role mapping on this node sees it immediately rather /// than after the cache TTL (lane K review M1). Queued post-commit: invalidating before the - /// commit would let a concurrent mapping re-cache the pre-change table. Other central - /// nodes still converge within RoleMappingCacheSeconds. + /// commit would let a concurrent mapping re-cache the pre-change table. The peer central node + /// is told with a on the central cache-invalidation topic + /// (audit D1, ScadaBridge#100 item 3), so a revocation takes effect there at once too; + /// RoleMappingCacheSeconds is only the fallback if that broadcast is lost. /// private static Task InvalidateRoleMappingCacheAfterCommit(IServiceProvider sp) { var cache = sp.GetService(); - return cache is null + var peers = sp.GetService(); + return cache is null && peers is null ? Task.CompletedTask - : AfterConfigurationCommit(sp, () => { cache.Invalidate(); return Task.CompletedTask; }); + : AfterConfigurationCommit(sp, () => + { + cache?.Invalidate(); + peers?.Publish(new RoleMappingsInvalidated()); + return Task.CompletedTask; + }); } private static async Task HandleCreateRoleMapping(IServiceProvider sp, CreateRoleMappingCommand cmd, string user) diff --git a/src/ZB.MOM.WW.ScadaBridge.Security/RoleMappingCache.cs b/src/ZB.MOM.WW.ScadaBridge.Security/RoleMappingCache.cs index 2ee4f12e..d32ccbed 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Security/RoleMappingCache.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Security/RoleMappingCache.cs @@ -1,5 +1,7 @@ using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; namespace ZB.MOM.WW.ScadaBridge.Security; @@ -14,7 +16,10 @@ namespace ZB.MOM.WW.ScadaBridge.Security; /// an acceptable price: without an explicit , a mapping or scope-rule /// change is visible on every node within that window. discards this /// node's snapshot so the next mapping re-reads the database; calling it after a mutation makes -/// the change immediate on the node that performed it (other nodes still rely on the TTL). +/// the change immediate on the node that performed it. The peer central node is told over the +/// central cache-invalidation topic (, audit D1, +/// ScadaBridge#100 item 3), so a revoked mapping stops working there at once too; the TTL is +/// only the fallback if that best-effort broadcast is lost. /// A TTL of 0 disables the cache: is false and /// reads the repository directly on every call. /// @@ -30,7 +35,7 @@ namespace ZB.MOM.WW.ScadaBridge.Security; /// read of the pre-change table can never overwrite the invalidation. /// /// -public sealed class RoleMappingCache +public sealed class RoleMappingCache : ICentralCacheInvalidationHandler { private readonly TimeProvider _timeProvider; private readonly IOptionsMonitor? _options; @@ -65,6 +70,14 @@ public sealed class RoleMappingCache /// public bool Enabled => Ttl > TimeSpan.Zero; + /// Applies a peer central node's . + /// The received invalidation; other types are ignored. + public void Apply(ICentralCacheInvalidation invalidation) + { + if (invalidation is RoleMappingsInvalidated) + Invalidate(); + } + /// /// Discards the cached snapshot so the next mapping re-reads the database. Call after any /// LDAP-group-mapping or site-scope-rule create/update/delete. diff --git a/src/ZB.MOM.WW.ScadaBridge.Security/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.Security/ServiceCollectionExtensions.cs index 946bc6ae..5adc45b3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Security/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Security/ServiceCollectionExtensions.cs @@ -65,6 +65,10 @@ public static class ServiceCollectionExtensions // (review 2026-10-21 05/P1). Singleton so every scoped RoleMapper shares it; the // repository it reads through is still the caller's scoped one. services.AddSingleton(); + // Audit D1 (ScadaBridge#100 item 3): a mapping change made through the other central node + // reaches this node's snapshot over the central cache-invalidation topic. + services.TryAddEnumerable(ServiceDescriptor.Singleton( + sp => sp.GetRequiredService())); // The cookie OnValidatePrincipal core. Scoped to match the // IGroupRoleMapper it depends on (which depends on the Scoped diff --git a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleImporter.cs b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleImporter.cs index de72516e..dd16f212 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleImporter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleImporter.cs @@ -17,6 +17,7 @@ using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Scripting; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Scripting; using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport; @@ -98,6 +99,9 @@ public sealed class BundleImporter : IBundleImporter // Optional (Gitea #83): drops this node's external-definition lookup cache after a // committed import. Null on hosts without the External System Gateway. private readonly IExternalDefinitionCacheInvalidator? _definitionCacheInvalidator; + // Optional (ScadaBridge#100 item 3 / #83): tells the peer central node to drop its caches of + // what the import wrote. Null on hosts without the central cache-invalidation bus. + private readonly ICentralCacheInvalidationPublisher? _cacheInvalidationPublisher; private readonly ILogger? _logger; /// @@ -125,6 +129,7 @@ public sealed class BundleImporter : IBundleImporter /// Optional logger. /// Optional: template-graph version watermark, bumped once per apply attempt so the flatten-session caches and the process-wide staleness memo cannot serve pre-import (or rolled-back) readings. Null on hosts that don't register it. /// Optional (Gitea #83): drops this node's external-system / database-connection definition cache after a committed import, so imported definitions take effect immediately rather than after the cache TTL. Null on hosts without the External System Gateway. + /// Optional (ScadaBridge#100 item 3): tells the peer central node to drop its caches of what a committed import wrote. Null on hosts without the central cache-invalidation bus. public BundleImporter( BundleSerializer bundleSerializer, ManifestValidator manifestValidator, @@ -147,7 +152,8 @@ public sealed class BundleImporter : IBundleImporter IScriptArtifactChangeBus? scriptArtifactChangeBus = null, ILogger? logger = null, ITemplateGraphWatermark? watermark = null, - IExternalDefinitionCacheInvalidator? definitionCacheInvalidator = null) + IExternalDefinitionCacheInvalidator? definitionCacheInvalidator = null, + ICentralCacheInvalidationPublisher? cacheInvalidationPublisher = null) { _bundleSerializer = bundleSerializer ?? throw new ArgumentNullException(nameof(bundleSerializer)); _manifestValidator = manifestValidator ?? throw new ArgumentNullException(nameof(manifestValidator)); @@ -170,6 +176,7 @@ public sealed class BundleImporter : IBundleImporter _scriptArtifactChangeBus = scriptArtifactChangeBus; _watermark = watermark; _definitionCacheInvalidator = definitionCacheInvalidator; + _cacheInvalidationPublisher = cacheInvalidationPublisher; _logger = logger; } @@ -2498,18 +2505,20 @@ public sealed class BundleImporter : IBundleImporter /// served the pre-import definition for up to DefinitionCacheTtl. Invalidates on every /// committed import rather than only when an external entity was resolved: dependencies can /// pull definitions in implicitly, imports are rare, and a cleared cache only costs one - /// re-query per name. Never throws (the transaction has already committed). + /// re-query per name. The peer central node is told too (ScadaBridge#100 item 3), since its + /// copy of the cache is just as stale. Never throws (the transaction has already committed). /// private void InvalidateExternalDefinitionCache() { - if (_definitionCacheInvalidator is null) + if (_definitionCacheInvalidator is null && _cacheInvalidationPublisher is null) { return; } try { - _definitionCacheInvalidator.InvalidateAll(); + _definitionCacheInvalidator?.InvalidateAll(); + _cacheInvalidationPublisher?.Publish(new ExternalDefinitionsInvalidated()); } catch (Exception ex) { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralCacheInvalidationBusTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralCacheInvalidationBusTests.cs new file mode 100644 index 00000000..7b615e54 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralCacheInvalidationBusTests.cs @@ -0,0 +1,80 @@ +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; + +/// +/// 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. +/// +public class CentralCacheInvalidationBusTests : TestKit +{ + private sealed class RecordingHandler : ICentralCacheInvalidationHandler + { + public List 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.Instance); + + /// Publishing wraps the invalidation with this node's origin id on the shared topic. + [Fact] + public void Publish_SendsAnEnvelopeOnTheSharedTopic() + { + var mediator = CreateTestProbe(); + _bus.SetMediator(mediator.Ref); + + _bus.Publish(new RoleMappingsInvalidated()); + + var publish = mediator.ExpectMsg(); + Assert.Equal("central-cache-invalidation", publish.Topic); + var envelope = Assert.IsType(publish.Message); + Assert.Equal(_bus.OriginId, envelope.OriginId); + Assert.IsType(envelope.Invalidation); + mediator.ExpectNoMsg(TimeSpan.FromMilliseconds(100)); // only PSK also uses the legacy topic + } + + /// Before the mediator exists publishing is a silent no-op. + [Fact] + public void Publish_WithoutAMediator_DoesNotThrow() + { + _bus.Publish(new ExternalDefinitionsInvalidated()); + } + + /// The peer's invalidation reaches every handler; this node's own broadcast is skipped. + [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); + } + }); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CompositionRootTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CompositionRootTests.cs index 7de164c5..9c8d7e32 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CompositionRootTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CompositionRootTests.cs @@ -419,6 +419,26 @@ public class CentralCompositionRootTests : IDisposable _factory.Services.GetService()); } + /// + /// ScadaBridge#100 item 3: central publishes peer invalidations through the shared bus, and + /// every cache that takes part (site PSK, role mappings, external definitions) is registered + /// as a handler for the subscriber. + /// + [Fact] + public void Central_CacheInvalidation_BusAndHandlersAreRegistered() + { + Assert.IsType( + _factory.Services.GetService()); + var handlers = _factory.Services + .GetServices() + .Select(h => h.GetType()) + .ToList(); + Assert.Contains(typeof(SitePskRotationInvalidator), handlers); + Assert.Contains(typeof(ZB.MOM.WW.ScadaBridge.Security.RoleMappingCache), handlers); + Assert.Contains(typeof(ZB.MOM.WW.ScadaBridge.ExternalSystemGateway.ExternalDefinitionLookupCache), handlers); + Assert.Equal(handlers.Count, handlers.Distinct().Count()); + } + /// Verifies that the shared audit seam resolves to the central direct-write bridge. [Fact] public void Central_SharedAuditSeam_IsTheCentralDirectWriteBridge() diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SitePskRotationInvalidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SitePskRotationInvalidatorTests.cs index c03a9517..88d36a64 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SitePskRotationInvalidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SitePskRotationInvalidatorTests.cs @@ -3,6 +3,8 @@ using Akka.Cluster.Tools.PublishSubscribe; using Akka.TestKit.Xunit2; using Microsoft.Extensions.Logging.Abstractions; using NSubstitute; +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.Abstractions; @@ -10,35 +12,43 @@ namespace ZB.MOM.WW.ScadaBridge.Host.Tests; /// /// Gitea #96: an in-place rotation of SB-GRPC-PSK-{siteId} written through the secrets -/// write path must drop central's cached key on this node and on the peer central node. +/// write path must drop central's cached key on this node and on the peer central node. Since +/// ScadaBridge#100 item 3 the peer is told over the shared central cache-invalidation topic. /// public class SitePskRotationInvalidatorTests : TestKit { private readonly ISecretCacheInvalidator _inner = Substitute.For(); private readonly ISitePskProvider _psk = Substitute.For(); + private readonly CentralCacheInvalidationBus _bus = new(NullLogger.Instance); - private SitePskRotationInvalidator Create() => - new(_inner, _psk, NullLogger.Instance); + private SitePskRotationInvalidator Create(ICentralCacheInvalidationPublisher? publisher) => + new(_inner, _psk, publisher, NullLogger.Instance); /// /// Verifies that writing a site's PSK secret evicts the resolver cache, drops the site's - /// cached key and publishes the invalidation to the peer. + /// cached key and publishes the invalidation on the shared topic and, for a mixed-version + /// peer, bare on the legacy #96 topic. /// [Fact] public void WritingAPskSecret_DropsTheCachedKey_AndTellsThePeer() { var mediator = CreateTestProbe(); - var invalidator = Create(); - invalidator.SetMediator(mediator.Ref); + _bus.SetMediator(mediator.Ref); + var invalidator = Create(_bus); // The secrets library lowercases names, so the site id arrives lowercased too. invalidator.Invalidate(new SecretName("SB-GRPC-PSK-site-a")); _inner.Received(1).Invalidate(new SecretName("SB-GRPC-PSK-site-a")); _psk.Received(1).Invalidate("site-a"); - var publish = mediator.ExpectMsg(); - Assert.Equal(SitePskRotationInvalidator.Topic, publish.Topic); - Assert.Equal(new SitePskInvalidated("site-a"), publish.Message); + var shared = mediator.ExpectMsg(); + Assert.Equal(CentralCacheInvalidationBus.Topic, shared.Topic); + var envelope = Assert.IsType(shared.Message); + Assert.Equal(new SitePskInvalidated("site-a"), envelope.Invalidation); + Assert.Equal(_bus.OriginId, envelope.OriginId); + var legacy = mediator.ExpectMsg(); + Assert.Equal("site-psk-invalidation", legacy.Topic); + Assert.Equal(new SitePskInvalidated("site-a"), legacy.Message); } /// Negative control: any other secret only evicts the resolver cache. @@ -46,8 +56,8 @@ public class SitePskRotationInvalidatorTests : TestKit public void WritingAnyOtherSecret_OnlyEvictsTheResolverCache() { var mediator = CreateTestProbe(); - var invalidator = Create(); - invalidator.SetMediator(mediator.Ref); + _bus.SetMediator(mediator.Ref); + var invalidator = Create(_bus); invalidator.Invalidate(new SecretName("SMTP-PASSWORD")); @@ -56,13 +66,15 @@ public class SitePskRotationInvalidatorTests : TestKit mediator.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } - /// Verifies that before the actor system exists an invalidation still applies locally. + /// Verifies that without a publisher (or before the actor system exists) an invalidation still applies locally. [Fact] - public void WithoutAMediator_TheLocalInvalidationStillApplies() + public void WithoutAPublisher_TheLocalInvalidationStillApplies() { - Create().Invalidate(new SecretName("SB-GRPC-PSK-site-a")); + Create(null).Invalidate(new SecretName("SB-GRPC-PSK-site-a")); + Create(_bus).Invalidate(new SecretName("SB-GRPC-PSK-site-b")); // bus has no mediator yet _psk.Received(1).Invalidate("site-a"); + _psk.Received(1).Invalidate("site-b"); } /// @@ -73,11 +85,12 @@ public class SitePskRotationInvalidatorTests : TestKit public void Subscriber_AppliesAPeerInvalidation_WithoutRepublishing() { var mediator = CreateTestProbe(); - var invalidator = Create(); - invalidator.SetMediator(mediator.Ref); - var subscriber = Sys.ActorOf(Props.Create(() => new SitePskInvalidationSubscriber(invalidator))); + _bus.SetMediator(mediator.Ref); + var invalidator = Create(_bus); + var subscriber = Sys.ActorOf(Props.Create(() => new CentralCacheInvalidationSubscriber( + _bus, new ICentralCacheInvalidationHandler[] { invalidator }))); - subscriber.Tell(new SitePskInvalidated("site-b")); + subscriber.Tell(new CentralCacheInvalidationEnvelope("peer-node", new SitePskInvalidated("site-b"))); AwaitAssert(() => { @@ -86,4 +99,20 @@ public class SitePskRotationInvalidatorTests : TestKit }); mediator.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } + + /// + /// Migration: a bare from a peer still on the #96 topic is + /// applied too. + /// + [Fact] + public void Subscriber_AppliesALegacyBarePskInvalidation() + { + var invalidator = Create(_bus); + var subscriber = Sys.ActorOf(Props.Create(() => new CentralCacheInvalidationSubscriber( + _bus, new ICentralCacheInvalidationHandler[] { invalidator }))); + + subscriber.Tell(new SitePskInvalidated("site-c")); + + AwaitAssert(() => _psk.Received(1).Invalidate("site-c")); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/CentralCacheInvalidationTwoNodeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/CentralCacheInvalidationTwoNodeTests.cs new file mode 100644 index 00000000..874fbfdf --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/CentralCacheInvalidationTwoNodeTests.cs @@ -0,0 +1,87 @@ +using Akka.Actor; +using Akka.Cluster.Tools.PublishSubscribe; +using Microsoft.Extensions.Logging.Abstractions; +using Xunit; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; +using ZB.MOM.WW.ScadaBridge.Host; + +namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster; + +/// +/// ScadaBridge#100 item 3 / #83 / #103 across a REAL two-node central cluster built from the +/// production HOCON: an invalidation published on one node's bus is delivered, through +/// DistributedPubSub and the production serializer, to the other node's handlers — and +/// not back to the publishing node's own handlers. +/// +public class CentralCacheInvalidationTwoNodeTests +{ + private sealed class RecordingHandler : ICentralCacheInvalidationHandler + { + private readonly List _applied = new(); + + public IReadOnlyList Applied + { + get { lock (_applied) return _applied.ToList(); } + } + + public void Apply(ICentralCacheInvalidation invalidation) + { + lock (_applied) _applied.Add(invalidation); + } + } + + private static (CentralCacheInvalidationBus Bus, RecordingHandler Handler) Wire(ActorSystem node) + { + var bus = new CentralCacheInvalidationBus(NullLogger.Instance); + var handler = new RecordingHandler(); + node.ActorOf(Props.Create(() => new CentralCacheInvalidationSubscriber( + bus, new ICentralCacheInvalidationHandler[] { handler })), "central-cache-invalidation"); + bus.SetMediator(DistributedPubSub.Get(node).Mediator); + return (bus, handler); + } + + /// + /// Every typed invalidation crosses to the peer intact; the publisher never applies its own. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Invalidations_ReachThePeerNode_AndNotThePublisher() + { + await using var cluster = await TwoNodeClusterFixture.StartAsync(); + var a = Wire(cluster.NodeA); + var b = Wire(cluster.NodeB); + + // Subscriptions spread by gossip, so probe until the peer has registered its subscriber. + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(20); + while (b.Handler.Applied.Count == 0) + { + Assert.True(DateTime.UtcNow < deadline, "node B never received node A's invalidation"); + a.Bus.Publish(new RoleMappingsInvalidated()); + await Task.Delay(250); + } + + a.Bus.Publish(new ExternalDefinitionsInvalidated()); + a.Bus.Publish(new InboundApiMethodInvalidated("MesMoveIn")); + a.Bus.Publish(new SitePskInvalidated("site-a")); + + deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + while (!b.Handler.Applied.Any(i => i is SitePskInvalidated)) + { + Assert.True(DateTime.UtcNow < deadline, "node B did not receive the later invalidations"); + await Task.Delay(100); + } + + var applied = b.Handler.Applied; + Assert.Contains(applied, i => i is RoleMappingsInvalidated); + Assert.Contains(new ExternalDefinitionsInvalidated(), applied); + Assert.Contains(new InboundApiMethodInvalidated("MesMoveIn"), applied); + // The PSK invalidation arrives twice during the migration window (shared + legacy topic). + Assert.Contains(new SitePskInvalidated("site-a"), applied); + + // The publisher skips its own envelopes; only the bare legacy PSK message (no origin) + // comes back to it, which is the pre-existing #96 behaviour and idempotent. + await Task.Delay(500); + Assert.All(a.Handler.Applied, i => Assert.IsType(i)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ExternalDefinitionCacheInvalidationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ExternalDefinitionCacheInvalidationTests.cs index e72605bd..919347b8 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ExternalDefinitionCacheInvalidationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ExternalDefinitionCacheInvalidationTests.cs @@ -6,6 +6,7 @@ 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; @@ -23,6 +24,9 @@ public class ExternalDefinitionCacheInvalidationTests : TestKit private readonly IExternalSystemRepository _repo = Substitute.For(); // 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(); /// Initializes the harness with one external system, one method and one DB connection. @@ -37,6 +41,8 @@ public class ExternalDefinitionCacheInvalidationTests : TestKit _services.AddScoped(_ => _repo); _services.AddScoped(_ => Substitute.For()); _services.AddSingleton(_cache); + _publisher = new PeerCacheInvalidationPublisher(_peerCache); + _services.AddSingleton(_publisher); } private IActorRef CreateActor() => @@ -103,4 +109,50 @@ public class ExternalDefinitionCacheInvalidationTests : TestKit Assert.False(await IsStillCached()); // dropped by the edit, so the lookup re-fetched } } + + /// + /// 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 DefinitionCacheTtl. + /// + /// A task that represents the asynchronous operation. + [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 Fetch() + { + fetches++; + return Task.FromResult(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(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(i)); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/PeerCacheInvalidationPublisher.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/PeerCacheInvalidationPublisher.cs new file mode 100644 index 00000000..016f153f --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/PeerCacheInvalidationPublisher.cs @@ -0,0 +1,30 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; + +namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests; + +/// +/// Test stand-in for the central cache-invalidation bus (ScadaBridge#100 item 3): records every +/// published invalidation and hands it to the "peer" node's handlers, as the subscriber on the +/// other central node would. +/// +internal sealed class PeerCacheInvalidationPublisher : ICentralCacheInvalidationPublisher +{ + private readonly ICentralCacheInvalidationHandler[] _peerHandlers; + + /// Creates the publisher. + /// The peer node's handlers. + public PeerCacheInvalidationPublisher(params ICentralCacheInvalidationHandler[] peerHandlers) => + _peerHandlers = peerHandlers; + + /// Gets every invalidation published. + public List Published { get; } = new(); + + /// + public void Publish(ICentralCacheInvalidation invalidation) + { + lock (Published) Published.Add(invalidation); + foreach (var handler in _peerHandlers) + handler.Apply(invalidation); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RoleMappingCacheInvalidationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RoleMappingCacheInvalidationTests.cs index 42e758bd..7207bf4c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RoleMappingCacheInvalidationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RoleMappingCacheInvalidationTests.cs @@ -6,6 +6,7 @@ using NSubstitute; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Security; 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.Security; @@ -20,6 +21,9 @@ public class RoleMappingCacheInvalidationTests : TestKit { private readonly ISecurityRepository _repo = Substitute.For(); private readonly RoleMappingCache _cache = new(TimeProvider.System); + // The other central node's cache, reached only through the peer invalidation (audit D1). + private readonly RoleMappingCache _peerCache = new(TimeProvider.System); + private readonly PeerCacheInvalidationPublisher _publisher; private readonly ServiceCollection _services = new ServiceCollection().AddTestConfigurationTransaction(); private List _mappings = [new("Ops", "Deployer") { Id = 5 }]; @@ -33,6 +37,8 @@ public class RoleMappingCacheInvalidationTests : TestKit _services.AddScoped(_ => _repo); _services.AddScoped(_ => Substitute.For()); _services.AddSingleton(_cache); + _publisher = new PeerCacheInvalidationPublisher(_peerCache); + _services.AddSingleton(_publisher); } private IActorRef CreateActor() => @@ -84,4 +90,45 @@ public class RoleMappingCacheInvalidationTests : TestKit await _repo.Received(1).GetAllMappingsAsync(Arg.Any()); // re-read, not served stale } } + + /// + /// Audit D1 (ScadaBridge#100 item 3): a revocation made through this node reaches the peer + /// central node's snapshot at once, instead of after RoleMappingCacheSeconds. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DeleteRoleMapping_IsVisibleOnThePeerNode() + { + var peerMapper = new RoleMapper(_repo, _peerCache); + Assert.Contains("Deployer", (await peerMapper.MapGroupsToRolesAsync(["Ops"])).Roles); // primes the peer + + _mappings = []; + CreateActor().Tell(Admin(new DeleteRoleMappingCommand(5))); + ExpectMsg(TimeSpan.FromSeconds(5)); + + Assert.Empty((await peerMapper.MapGroupsToRolesAsync(["Ops"])).Roles); + } + + /// Every mapping / scope-rule mutation tells the peer, once, after commit. + [Fact] + public void EveryMappingMutation_TellsThePeer() + { + object[] commands = + [ + new CreateRoleMappingCommand("Eng", "Designer"), + new UpdateRoleMappingCommand(5, "Ops", "Viewer"), + new DeleteRoleMappingCommand(5), + new AddScopeRuleCommand(5, 2), + new DeleteScopeRuleCommand(9), + ]; + var actor = CreateActor(); + foreach (var command in commands) + { + actor.Tell(Admin(command)); + ExpectMsg(TimeSpan.FromSeconds(5)); + } + + Assert.Equal(commands.Length, _publisher.Published.Count); + Assert.All(_publisher.Published, i => Assert.IsType(i)); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Transport.IntegrationTests/Import/BundleImporterApplyTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Transport.IntegrationTests/Import/BundleImporterApplyTests.cs index 1880a8e5..642b0d0e 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Transport.IntegrationTests/Import/BundleImporterApplyTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Transport.IntegrationTests/Import/BundleImporterApplyTests.cs @@ -13,6 +13,7 @@ using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Scripting; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Scripting; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport; @@ -48,6 +49,7 @@ public sealed class BundleImporterApplyTests : IDisposable private readonly ServiceProvider _provider; private readonly RecordingScriptArtifactChangeBus _artifactBus = new(); private readonly CountingDefinitionCacheInvalidator _definitionCache = new(); + private readonly RecordingCacheInvalidationPublisher _peerPublisher = new(); /// Initializes a new instance of the BundleImporterApplyTests class. public BundleImporterApplyTests() @@ -103,6 +105,8 @@ public sealed class BundleImporterApplyTests : IDisposable services.AddSingleton(_artifactBus); // Gitea #83: observable post-commit drop of the external-definition lookup cache. services.AddSingleton(_definitionCache); + // ScadaBridge#100 item 3: observable post-commit broadcast to the peer central node. + services.AddSingleton(_peerPublisher); // The stale-instance probe is implemented in DeploymentManager // and the flatten/hash primitives in TemplateEngine — register both so // BundleImporter resolves a real IStaleInstanceProbe and ApplyAsync can @@ -1946,6 +1950,8 @@ public sealed class BundleImporterApplyTests : IDisposable } Assert.Equal(1, _definitionCache.Calls); + // ScadaBridge#100 item 3 / #83: the peer central node is told to drop its copy too. + Assert.Contains(_peerPublisher.Published, i => i is ExternalDefinitionsInvalidated); } /// Verifies that a failed apply publishes no notifications. @@ -1998,6 +2004,20 @@ public sealed class BundleImporterApplyTests : IDisposable Assert.Empty(_artifactBus.Received); // Gitea #83: nothing committed, so the definition cache is left alone. Assert.Equal(0, _definitionCache.Calls); + Assert.Empty(_peerPublisher.Published); + } + + /// Records peer cache invalidations (ScadaBridge#100 item 3). + private sealed class RecordingCacheInvalidationPublisher : ICentralCacheInvalidationPublisher + { + /// Gets the invalidations published. + public List Published { get; } = new(); + + /// + public void Publish(ICentralCacheInvalidation invalidation) + { + lock (Published) Published.Add(invalidation); + } } /// Counts post-commit external-definition cache invalidations (Gitea #83).