From 153609ef8a06f801d314903fabb9185682ddadb7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 9 Nov 2026 15:50:03 -0500 Subject: [PATCH] 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 --- docs/components/Security.md | 2 +- ...026-11-09-central-cache-staleness-audit.md | 19 +- docs/requirements/Component-Communication.md | 5 +- .../Component-ExternalSystemGateway.md | 2 +- docs/requirements/Component-Host.md | 3 +- .../Component-ManagementService.md | 2 +- docs/requirements/Component-Security.md | 2 +- .../ICentralCacheInvalidationPublisher.cs | 33 ++++ .../CentralCacheInvalidations.cs | 44 +++++ .../ExternalDefinitionLookupCache.cs | 15 +- .../ServiceCollectionExtensions.cs | 4 + .../Actors/AkkaHostedService.cs | 21 +- .../CentralCacheInvalidation.cs | 185 ++++++++++++++++++ src/ZB.MOM.WW.ScadaBridge.Host/Program.cs | 5 +- .../SitePskRotationInvalidator.cs | 91 ++++----- .../ManagementActor.cs | 32 ++- .../RoleMappingCache.cs | 17 +- .../ServiceCollectionExtensions.cs | 4 + .../Import/BundleImporter.cs | 17 +- .../CentralCacheInvalidationBusTests.cs | 80 ++++++++ .../CompositionRootTests.cs | 20 ++ .../SitePskRotationInvalidatorTests.cs | 65 ++++-- .../CentralCacheInvalidationTwoNodeTests.cs | 87 ++++++++ ...xternalDefinitionCacheInvalidationTests.cs | 52 +++++ .../PeerCacheInvalidationPublisher.cs | 30 +++ .../RoleMappingCacheInvalidationTests.cs | 47 +++++ .../Import/BundleImporterApplyTests.cs | 20 ++ 27 files changed, 794 insertions(+), 110 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ICentralCacheInvalidationPublisher.cs create mode 100644 src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Communication/CentralCacheInvalidations.cs create mode 100644 src/ZB.MOM.WW.ScadaBridge.Host/CentralCacheInvalidation.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralCacheInvalidationBusTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/CentralCacheInvalidationTwoNodeTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/PeerCacheInvalidationPublisher.cs 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).