diff --git a/docs/requirements/Component-DeploymentManager.md b/docs/requirements/Component-DeploymentManager.md index 2f90d2ed..406dadca 100644 --- a/docs/requirements/Component-DeploymentManager.md +++ b/docs/requirements/Component-DeploymentManager.md @@ -137,6 +137,57 @@ When deploying artifacts (shared scripts, external system definitions, etc.) to - Deployment is performed at the **individual instance level**. - The UI may provide convenience operations (e.g., "deploy all out-of-date instances at Site A"), but these decompose into individual instance deployments. +### Bulk site deployment (`DeploySiteAsync`, WP2.5) + +`DeploySiteAsync(siteId, user)` deploys every instance at one site in a single +operation, surfaced as the `MgmtDeploySite` management command and the CLI +`deploy site --site-id`. It is the "deploy all at Site A" convenience above, made +first-class — it still decomposes into individual instance deployments, and each +instance keeps its own deployment id, revision hash, operation lock, and +optimistically-concurrent status record. + +It runs the ordinary deployment pipeline in three phases, of which only the middle +one is parallel: + +1. **Prepare (serial).** Validate transition, take the operation lock, flatten + + validate, run query-before-redeploy reconciliation, stage the + `PendingDeployment`, insert the `InProgress` record. Every step here touches the + scoped, non-thread-safe `DbContext`, so the phase is strictly serial. All + instances share ONE `FlattenSession`, so a template chain common to N instances + is walked once and the session-global queries (shared scripts, schema library, + the site's data connections) run once for the batch. +2. **Send (bounded parallel).** The `RefreshDeploymentCommand` round-trips run + concurrently up to `SiteDeploymentMaxParallelism` (default 4), each under a + `SiteDeploymentTimeoutPerInstance` deadline (default 120 s). This phase touches + no repository — that is exactly why it is the only phase allowed to run in + parallel. Shape mirrors `ArtifactDeploymentService.DeployCoreAsync`. +3. **Finalize (serial).** Commit terminal statuses, apply post-success side + effects, write audit rows, release each operation lock. + +**Not all-or-nothing.** An instance that fails for any reason (wrong state, failed +validation, lock already held by another operation, site round-trip timed out) is +reported as a failed row while the rest proceed, and is individually retryable via +the ordinary single-instance deploy. This matches the artifact-deployment policy: +successful targets are never rolled back because another target failed. + +`DeployInstanceAsync` is composed from the same three phase helpers with a batch of +one, so the two entry points cannot drift on deployment identity, idempotency, lock +coverage, or optimistic concurrency. + +### Terminal deployment-record retention + +Deployment records are insert-only — one row per deploy attempt — so without a +window the table grows for the life of the system and every unfiltered deployment +query degrades with age rather than with page size. +`TerminalDeploymentRecordRetention` (default **365 days**, deliberately generous +because deployment history is operator forensics) bounds it, swept opportunistically +and rate-limited on the deployment-list read path. + +Only **terminal** rows (`Success` / `Failed`) are eligible. An `InProgress` row is +never purged regardless of age: it is precisely the row the query-before-redeploy +reconciliation reads to decide whether a prior deploy actually landed at the site, +and expiring it by age would silently disable that idempotency guard. + ## Diff View Before deploying, the Deployment Manager can request a diff from the Template Engine showing: diff --git a/docs/requirements/Component-ManagementService.md b/docs/requirements/Component-ManagementService.md index 20c7219a..ac5fa61b 100644 --- a/docs/requirements/Component-ManagementService.md +++ b/docs/requirements/Component-ManagementService.md @@ -87,7 +87,7 @@ Both endpoints honour any site-scope rules attached to the caller's audit role b ### Templates -- **ListTemplates** / **GetTemplate**: Query template definitions. `ListTemplates` supports additive offset paging via `Skip` / `Take` (arch-review P2): `Take = null` (the default) preserves the historical unpaged behaviour (unlimited), a set `Take` is clamped to `1..1000` and `Skip` floors at `0`. +- **ListTemplates** / **GetTemplate**: Query template definitions. `ListTemplates` supports additive offset paging via `Skip` / `Take` (arch-review P2): `Take = null` (the default) preserves the historical unpaged behaviour (unlimited), a set `Take` is clamped to `1..1000` and `Skip` floors at `0`. **Paging and projection are DB-side (WP2.5):** the handler reads `TemplateSummary` rows (child collections reduced to counts, no script bodies) via `GetTemplateSummariesAsync`, instead of materialising every template's full five-Include child graph and then discarding all but one page in memory. A caller that needs a template's members already fetches it with `GetTemplate`. - **CreateTemplate** / **UpdateTemplate** / **DeleteTemplate**: Manage templates. - **ValidateTemplate**: Run on-demand pre-deployment validation (flattening, naming collisions, script compilation). - **GetTemplateDiff**: Compare deployed vs. template-derived configuration for an instance. @@ -115,9 +115,10 @@ The whole folder surface is reachable from the CLI as `template folder list|crea ### Instances -- **ListInstances** / **GetInstance**: Query instances, with filtering by site and area. `ListInstances` supports additive offset paging via `Skip` / `Take` (arch-review P2), applied **after** the in-memory site-scope filter so a site-scoped user never sees an out-of-scope instance surface into their page window; same `Take = null` unlimited default and `1..1000` clamp as `ListTemplates`. (`QueryDeployments` and `ExportBundle` still load the full table internally — a logged scale follow-up, arch-review P2 deferred.) +- **ListInstances** / **GetInstance**: Query instances, with filtering by site and area. `ListInstances` supports additive offset paging via `Skip` / `Take` (arch-review P2), applied **after** the in-memory site-scope filter so a site-scoped user never sees an out-of-scope instance surface into their page window; same `Take = null` unlimited default and `1..1000` clamp as `ListTemplates`. (`ExportBundle` still loads the full table internally — a logged scale follow-up, arch-review P2 deferred. `QueryDeployments` was closed by WP2.5, below.) - **CreateInstance**: Create a new instance from a template. -- **UpdateInstanceOverrides**: Set attribute overrides on an instance. +- **UpdateInstanceOverrides** (`SetInstanceOverrides`): Set attribute overrides on an instance. Every requested attribute is validated against the instance's template up front (unknown or locked ⇒ the whole batch is rejected before any write), then the batch is applied as **one bulk read of the existing override rows plus one commit** (WP2.5). It previously ran K independent read-modify-commit cycles — each re-reading the instance, the template's attributes and the existing overrides, then committing and writing its own audit row — so the all-or-nothing promise was only as good as the pre-validation. It is now a real single-transaction apply, with one audit row summarising the batch. +- **MgmtDeploySite** (`SiteId`, WP2.5): Bulk-deploy every deployable instance at one site. Site scope is enforced on the SITE (the command's target); every instance the batch touches belongs to that site by construction. Same `Deployer` authority as `MgmtDeployInstance` — it is N of those, not a new class of privilege — and registered as a long-running command so the caller does not time out mid-batch. Semantics, phasing and failure policy: see `Component-DeploymentManager.md` → Bulk site deployment. - **SetInstanceAlarmOverride** / **DeleteInstanceAlarmOverride** / **ListInstanceAlarmOverrides**: Manage per-instance computed-alarm overrides. - **SetInstanceNativeAlarmSourceOverride** / **DeleteInstanceNativeAlarmSourceOverride**: Retarget or clear a per-instance native alarm source binding, keyed by `SourceCanonicalName` — `ConnectionNameOverride` / `SourceReferenceOverride` / `ConditionFilterOverride` each apply only when non-null. Gated to the **Deployment** role. - **ListInstanceNativeAlarmSourceOverrides** (`InstanceId`): List an instance's native alarm source overrides (read-only). @@ -144,7 +145,9 @@ The whole folder surface is reachable from the CLI as `template folder list|crea - **DeployInstance**: Deploy configuration to a specific instance (includes pre-deployment validation). - **DeployArtifacts**: Deploy system-wide artifacts (shared scripts, external system definitions, DB connections, data connections) to all sites or a specific site. The command's `SiteId` is honored (arch-review C2): a value routes to the single-site deploy path (`ArtifactDeploymentService.DeployToSiteAsync`), while `null` deploys fleet-wide (`DeployToAllSitesAsync`). Site scope is enforced — a site-scoped (non-Administrator) Deployer may only target a site within its `PermittedSiteIds`, and may **not** deploy fleet-wide (`SiteId is null` is rejected with `SiteScopeViolationException` → `ManagementUnauthorized`, since `EnforceSiteScope(null)` is a deliberate no-op); fleet-wide deployment requires a system-wide Deployer or Administrator. +- **DeploySite** (`MgmtDeploySite`, WP2.5): Bulk-deploy every deployable instance at one site — see the Instances section above and `Component-DeploymentManager.md`. - **GetDeploymentStatus**: Query deployment status. +- **QueryDeployments** (`InstanceId?`, `Status?`, `Page`, `PageSize`): List deployment records. **All four arguments are honoured DB-side (WP2.5)** and the result is a `DeploymentRecordSummary` projection. The handler previously ignored `Status`, `Page` and `PageSize` entirely (the CLI had been sending them all along) and loaded the whole insert-only `DeploymentRecords` table; for a site-scoped user it additionally loaded every instance and intersected the two sets in memory. Site scope is now resolved to the set of in-scope instance ids and pushed into the query as an id filter — an **empty** scope stays a real filter (a user permitted no in-scope instances sees nothing), never a no-op. This read path also drives the opportunistic, rate-limited terminal-record retention sweep described in `Component-DeploymentManager.md`. ### Secured Writes (MxGateway, two-person) @@ -259,7 +262,7 @@ The ManagementActor receives the following services and repositories via DI (inj | Section | Options Class | Contents | |---------|--------------|----------| | `ScadaBridge:ManagementService` | `ManagementServiceOptions` | `CommandTimeout` (`TimeSpan`, default 30 s) — Ask timeout the HTTP endpoint applies when forwarding to the `ManagementActor`. A non-positive configured value falls back to the 30 s default. | -| | | `LongRunningCommandTimeout` (`TimeSpan`, default 5 min) — Ask timeout applied to long-running commands (`ImportBundle`, `PreviewBundle`, `ExportBundle`, `MgmtDeployArtifacts`, `MgmtDeployInstance`); all other commands use `CommandTimeout`. A non-positive configured value falls back to the 5 min default. | +| | | `LongRunningCommandTimeout` (`TimeSpan`, default 5 min) — Ask timeout applied to long-running commands (`ImportBundle`, `PreviewBundle`, `ExportBundle`, `MgmtDeployArtifacts`, `MgmtDeployInstance`, `MgmtDeploySite`); all other commands use `CommandTimeout`. A non-positive configured value falls back to the 5 min default. | | | | `SecuredWritePendingTtl` (`TimeSpan`, default 24 h) — age after which a `Pending` secured write is transitioned to `Expired` and can no longer be approved/executed; enforced at approve/reject and swept opportunistically on list. A non-positive value disables expiry (arch-review S2). | ## Dependencies diff --git a/docs/requirements/Component-ScriptAnalysis.md b/docs/requirements/Component-ScriptAnalysis.md index 86319200..44fea050 100644 --- a/docs/requirements/Component-ScriptAnalysis.md +++ b/docs/requirements/Component-ScriptAnalysis.md @@ -158,6 +158,19 @@ Mirrors `TriggerExpressionGlobals` in the same way. Used by `ValidationService.C `CheckExpressionSyntax` memoises its verdict in the Template Engine's process-wide `ScriptCompileVerdictCache`, whose key is the pair **(globals surface, SHA-256 of the code)** — not the code alone. The surface discriminator is load-bearing: a trigger expression vetted against `TriggerCompileSurface` is **not** interchangeable with a `ScriptCompileSurface` script-body verdict (different globals resolve different identifiers), so a code-only key could return a stale "clean" for code never compiled against the caller's surface. Keying by surface makes that cross-surface verdict reuse structurally impossible. +**Eviction is segmented, never wholesale (WP2.5).** The cache is bounded at two +generations of 2048 entries. New verdicts land in the *hot* generation; a hit in +*cold* is promoted back into hot; on overflow hot becomes the new cold and only the +old cold — the half nothing has touched for a full generation — is dropped. + +The bound must never be enforced by clearing the cache outright. That reads as +harmless ("a verdict is cheap to recompute") but is not: a recompute is a fresh +Roslyn compile, and every script compile loads an assembly through a +**non-collectible `InteractiveAssemblyLoader`**. Bounding that leak is the entire +reason this cache exists, so dropping every hot entry at the 4096th distinct script +would re-open it for the whole working set at once. Segmenting retains anything in +active use across an eviction while keeping the same ceiling. + #### Parity guard A reflection-based parity test in `SiteRuntime.Tests` compares the public member names on `ScriptCompileSurface` against `ScriptGlobals` (and `TriggerCompileSurface` against `TriggerExpressionGlobals`). Any drift between the stub and the real globals causes this test to fail, ensuring the stubs cannot silently fall out of sync. diff --git a/docs/requirements/Component-TemplateEngine.md b/docs/requirements/Component-TemplateEngine.md index 9b7abf88..875a4c1a 100644 --- a/docs/requirements/Component-TemplateEngine.md +++ b/docs/requirements/Component-TemplateEngine.md @@ -137,6 +137,64 @@ When an instance is deployed, the Template Engine resolves the full configuratio 5. Resolve data connection bindings — replace connection name references with concrete connection details from the site. 6. Output a flat structure: list of attributes with resolved values and data source addresses, list of alarms with resolved trigger definitions, list of scripts with resolved code and triggers. +### Flatten-session caching and the graph watermark (WP2.5) + +The flatten above is driven per instance, so an unmemoised implementation re-walks +the same template chain — one query per link, plus the compositions of every +template it reaches and of every composed chain those reach — once per instance, +and re-issues the three session-global queries (shared scripts, the shared-schema +library, the target site's data connections) each time too. + +A **`FlattenSession`** memoises all of that for the lifetime of ONE +flatten/validate operation. Callers that flatten a batch (`DeploySiteAsync`) pass +one shared session; a caller that passes none gets a private single-use session, +which still collapses the repeated composed-chain loads inside a single instance's +flatten. + +Cache validity is decided by **`ITemplateGraphWatermark`**, a process-wide set of +monotonic counters bumped by the configuration-database unit of work — the only +place every template-graph writer funnels through (`TemplateService`, the +`ManagementActor` native-alarm-source handlers, and the Transport bundle importer +all commit via the same `SaveChangesAsync`, and the change tracker is inspected +pre-commit to attribute each change to its owning template or instance): + +- a memoised **template** is keyed on `(id, template version)`; +- a memoised **chain** is valid only while BOTH the graph's `StructureVersion` + (bumped on add/remove, re-parent, or any composition change — i.e. anything that + can alter chain MEMBERSHIP) and every member's own version are unchanged. Both + halves are load-bearing: structure alone misses an ordinary member edit, member + versions alone miss a re-parent. + +Sessions are short-lived by construction, so the design's "template state is +captured at the time of flatten" guarantee is unchanged — the session narrows the +capture window, it never widens it. + +The watermark is **in-memory and process-local**, deliberately not persisted and +not replicated between central nodes. A restart, a failover, or any mutation simply +misses and falls back to the authoritative full flatten, so the watermark can only +ever cause EXTRA work, never stale work. + +The same watermark backs a **staleness fast path** in `StaleInstanceProbe`: a +previously computed revision hash is reused when the instance's version, the +structure version, and the version of every template that flatten walked are all +unchanged. This matters most where the probe is called per-instance across a whole +bundle import or a fleet-wide staleness sweep. + +### Design-time analysis reads + +The design-time checks — acyclicity (`CycleDetector`), naming collisions +(`CollisionDetector`) and canonical-name resolution (`TemplateResolver`) — read the +whole template graph but write through none of it and read no script bodies. They +use `GetAllTemplatesForAnalysisAsync`: `AsNoTracking`, with `TemplateScript.Code` +projected away. + +The tracked, body-bearing `GetAllTemplatesAsync` remains the read for the two paths +that genuinely need it — the inheritance reconciler (which compares and copies +script bodies and writes through the loaded entities) and the flattener (which must +observe rows an in-flight bundle import has staged on the shared change tracker). +`ReconcileDescendantsAsync` additionally accepts an already-loaded tracked graph so +a caller holding one does not force a second load. + ### Native Alarm Source Resolution The `FlatteningService` resolves native alarm sources alongside alarms, emitting a `ResolvedNativeAlarmSource` (CanonicalName, ConnectionName, SourceReference, ConditionFilter *(optional)*, and `Source` ∈ `Template` | `Inherited` | `Composed` | `Override`) for each. The resolved set is attached to `FlattenedConfiguration.NativeAlarmSources`. diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/DeployCommands.cs b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/DeployCommands.cs index f014f784..08b43a3d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/DeployCommands.cs +++ b/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/DeployCommands.cs @@ -19,6 +19,7 @@ public static class DeployCommands var command = new Command("deploy") { Description = "Deployment operations" }; command.Add(BuildInstance(urlOption, formatOption, usernameOption, passwordOption)); + command.Add(BuildSite(urlOption, formatOption, usernameOption, passwordOption)); command.Add(BuildArtifacts(urlOption, formatOption, usernameOption, passwordOption)); command.Add(BuildStatus(urlOption, formatOption, usernameOption, passwordOption)); @@ -39,6 +40,43 @@ public static class DeployCommands return cmd; } + /// + /// Builds deploy site — bulk-deploy every instance at one site. + /// + /// + /// Sits under deploy alongside deploy instance and + /// deploy artifacts, naming the SCOPE of the deploy as the verb, which + /// is the convention the group already follows. Note it is deliberately NOT + /// fleet-wide when --site-id is omitted (the pattern deploy + /// artifacts uses): a bulk instance deploy is far more consequential than + /// an artifact push, so the target site is required rather than defaulted. + /// + /// + private static Command BuildSite(Option urlOption, Option formatOption, Option usernameOption, Option passwordOption) + { + var siteIdOption = new Option("--site-id") { Description = "Target site ID", Required = true }; + var cmd = new Command("site") { Description = "Deploy every deployable instance at a site" }; + cmd.Add(siteIdOption); + cmd.SetAction(async (ParseResult result) => + { + var siteId = result.GetValue(siteIdOption); + return await CommandHelpers.ExecuteCommandAsync( + result, urlOption, formatOption, usernameOption, passwordOption, + new MgmtDeploySiteCommand(siteId), + // A bulk deploy is N instance round-trips; the default 30 s client + // timeout would abandon the request while the server is still + // applying. Matches the server-side long-running Ask window. + timeout: BulkDeployTimeout); + }); + return cmd; + } + + /// + /// Client-side timeout for the bulk site deploy, matching the management + /// service's long-running command Ask window. + /// + internal static readonly TimeSpan BulkDeployTimeout = TimeSpan.FromMinutes(5); + private static Command BuildArtifacts(Option urlOption, Option formatOption, Option usernameOption, Option passwordOption) { var siteIdOption = new Option("--site-id") { Description = "Target site ID (all sites if omitted)" }; diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/README.md b/src/ZB.MOM.WW.ScadaBridge.CLI/README.md index 28c7b912..e737dd63 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CLI/README.md +++ b/src/ZB.MOM.WW.ScadaBridge.CLI/README.md @@ -952,6 +952,37 @@ scadabridge --url deploy instance --id |--------|----------|-------------| | `--id` | yes | Instance ID | +#### `deploy site` + +Deploy **every deployable instance** at one site in a single batched operation. + +Each instance keeps the full single-instance semantics — its own deployment ID and +revision hash, its own per-instance operation lock, its own optimistically-concurrent +status record, and the same query-before-redeploy idempotency check. The batch is +**not** all-or-nothing: an instance that cannot be deployed (wrong state, failed +validation, lock already held, site round-trip timed out) is reported as a failed row +and the rest proceed. Each failed instance is individually retryable with +`instance deploy --id`. + +Site round-trips run with bounded concurrency +(`ScadaBridge:DeploymentManager:SiteDeploymentMaxParallelism`, default 4), each under +`SiteDeploymentTimeoutPerInstance` (default 120 s), so one wedged instance cannot stall +the batch. The client waits up to 5 minutes. + +`--site-id` is **required** — unlike `deploy artifacts`, omitting it does not mean +fleet-wide. + +```sh +scadabridge --url deploy site --site-id +``` + +| Option | Required | Description | +|--------|----------|-------------| +| `--site-id` | yes | Target site ID. A site-scoped (non-Administrator) Deployer must supply an in-scope site. | + +Output is a per-instance result matrix (`instanceId`, `uniqueName`, `deploymentId`, +`success`, `errorMessage`) plus `successCount` / `failureCount`. + #### `deploy artifacts` Deploy compiled artifacts to one or all sites (same as `site deploy-artifacts`). diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IDeploymentManagerRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IDeploymentManagerRepository.cs index 01b55d60..b7c59ced 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IDeploymentManagerRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IDeploymentManagerRepository.cs @@ -22,6 +22,48 @@ public interface IDeploymentManagerRepository /// A read-only list of all deployment records. Task> GetAllDeploymentRecordsAsync(CancellationToken cancellationToken = default); /// + /// Database-side paged and filtered deployment query returning row summaries. + /// Backs the QueryDeployments management command, which previously + /// loaded the ENTIRE deployment-record table (insert-only, one row per deploy + /// attempt for the life of the system) and ignored the page arguments the CLI + /// was already sending. + /// + /// Restrict to one instance's history, or for all. + /// Restrict to one deployment status, or for all. + /// + /// When non-null, restrict to these instance ids. Used to apply a site-scoped + /// user's permitted-site filter IN THE DATABASE rather than by loading every + /// record and every instance and filtering in memory. + /// + /// Rows to skip; negative values are floored at 0. + /// Page size; non-positive or null returns every remaining row. + /// A cancellation token that can be used to cancel the operation. + /// One page of deployment summaries, newest first. + Task> QueryDeploymentSummariesAsync( + int? instanceId, + DeploymentStatus? status, + IReadOnlyCollection? instanceIdScope, + int skip, + int? take, + CancellationToken cancellationToken = default); + /// + /// Deletes TERMINAL deployment records ( / + /// ) that completed before + /// , in one bounded batch. + /// + /// + /// Non-terminal rows are never touched: an InProgress record is what the + /// query-before-redeploy reconciliation path reads to decide whether a prior + /// deploy actually landed at the site, so purging one by age would silently + /// disable that idempotency guard. + /// + /// + /// Records completing before this instant are eligible. + /// Maximum rows deleted in this call. + /// A cancellation token that can be used to cancel the operation. + /// The number of rows deleted; fewer than means the sweep is complete. + Task PurgeTerminalDeploymentRecordsAsync(DateTimeOffset cutoffUtc, int batchSize, CancellationToken cancellationToken = default); + /// /// Gets all deployment records for a specific instance. /// /// The instance ID. diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ITemplateEngineRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ITemplateEngineRepository.cs index 3cc1d1d2..dc2d294f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ITemplateEngineRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ITemplateEngineRepository.cs @@ -35,6 +35,41 @@ public interface ITemplateEngineRepository /// A task that resolves to a read-only list of all templates. Task> GetAllTemplatesAsync(CancellationToken cancellationToken = default); /// + /// Read-only, no-tracking variant of for the + /// design-time ANALYSIS walks — acyclicity (CycleDetector), naming + /// collisions (CollisionDetector) and canonical-name resolution + /// (TemplateResolver). Script bodies are NOT loaded + /// ( comes back empty) because no analysis + /// consumer reads them; everything those walks do read — ids, parent and + /// composition edges, member names and lock flags — is present. + /// + /// + /// Do NOT use this for the inheritance reconciler or the flattener: the + /// reconciler compares and copies script bodies and writes through tracked + /// entities, and the flattener must observe rows an in-flight bundle import + /// has staged on the shared change tracker. Both keep using + /// / . + /// + /// + /// Cancellation token. + /// A task that resolves to a read-only list of all templates without script bodies. + Task> GetAllTemplatesForAnalysisAsync(CancellationToken cancellationToken = default); + /// + /// Database-side paged listing of templates as row summaries (child + /// collections reduced to counts, no script bodies). Backs the + /// ListTemplates management command, which previously materialised + /// every template's full child graph and then paged the list in memory. + /// + /// Number of rows to skip; negative values are floored at 0. + /// + /// Page size. or non-positive returns every remaining + /// row (the historical unpaged behaviour); otherwise clamped to + /// . + /// + /// Cancellation token. + /// A task that resolves to one page of template summaries, ordered by id. + Task> GetTemplateSummariesAsync(int skip, int? take, CancellationToken cancellationToken = default); + /// /// Returns every template that contains a composition referencing /// . Each result is eager-loaded with /// its Attributes / Scripts / Compositions so the caller can build a diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ITemplateGraphWatermark.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ITemplateGraphWatermark.cs new file mode 100644 index 00000000..2d2952d5 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ITemplateGraphWatermark.cs @@ -0,0 +1,82 @@ +namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; + +/// +/// Process-wide monotonic version watermark over the template/instance +/// configuration graph. +/// +/// +/// Every mutation that lands through the configuration database bumps the +/// version of the affected template (or instance), plus a global counter and — +/// for edge-shaped changes (template added/removed, parent or composition edge +/// changed) — a separate . Consumers use the +/// watermark two ways: +/// +/// +/// +/// as the discriminator in a flatten-session cache key +/// (templateId + version), so a memoised template chain can never be +/// served after that template changed; and +/// +/// +/// as a staleness fast path — when neither the instance nor any template in +/// its chain has moved since a previously computed revision hash, the hash +/// is still current and a full re-flatten can be skipped. +/// +/// +/// +/// +/// Versions are in-memory and process-local. They are deliberately NOT +/// persisted: a restart (or the standby central node) simply starts from zero, +/// which invalidates every cached entry and falls back to the authoritative +/// full flatten. The watermark is therefore only ever able to cause EXTRA work, +/// never stale work — a correctness property that matters because the value is +/// not replicated between the two central nodes. +/// +/// +public interface ITemplateGraphWatermark +{ + /// + /// Monotonic counter bumped on every template- or instance-graph mutation. + /// Useful as a coarse "did anything change at all" gate. + /// + long Global { get; } + + /// + /// Monotonic counter bumped only when the SHAPE of the graph changes — a + /// template row added or removed, a parent-template edge changed, or a + /// composition row added/removed/repointed. A cached template-chain + /// MEMBERSHIP is valid only while this value is unchanged; the per-template + /// versions then cover the contents of each member. + /// + long StructureVersion { get; } + + /// Current version of a single template. Zero when never mutated in this process. + /// Template id to read. + /// The template's monotonic version. + long GetTemplateVersion(int templateId); + + /// Current version of a single instance (its override/binding rows). Zero when never mutated in this process. + /// Instance id to read. + /// The instance's monotonic version. + long GetInstanceVersion(int instanceId); + + /// Records a mutation of the given template. + /// Template whose version is bumped. + /// + /// When true the change altered the graph shape (add/remove/parent or + /// composition edge) and is bumped too. + /// + void BumpTemplate(int templateId, bool structural = false); + + /// Records a mutation of the given instance. + /// Instance whose version is bumped. + void BumpInstance(int instanceId); + + /// + /// Records a mutation whose affected template/instance could not be + /// attributed to a specific id. Bumps and + /// , invalidating every chain-membership and + /// staleness fast path — the safe, conservative fallback. + /// + void BumpAll(); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/DeploymentCommands.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/DeploymentCommands.cs index 3896a455..e193e292 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/DeploymentCommands.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/DeploymentCommands.cs @@ -1,5 +1,26 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; public record MgmtDeployArtifactsCommand(int? SiteId = null); + +/// +/// Deploys every deployable instance at one site in a single operation. +/// +/// +/// Distinct from , which ships the +/// SYSTEM-WIDE artifact sets (shared scripts, external systems, DB/data +/// connections) to a site. This command deploys the site's INSTANCE +/// configurations — the same per-instance pipeline +/// MgmtDeployInstanceCommand runs, batched. +/// +/// +/// +/// Not all-or-nothing: each instance keeps its own deployment id, revision hash, +/// operation lock and status record, and a failure on one is reported as a failed +/// row while the rest proceed. Individually retryable via the single-instance +/// deploy. Additive contract; new fields go on the end with a default. +/// +/// +/// Database id of the target site. Required — this command is never fleet-wide. +public record MgmtDeploySiteCommand(int SiteId); public record QueryDeploymentsCommand(int? InstanceId = null, string? Status = null, int Page = 1, int PageSize = 50); public record GetDeploymentDiffCommand(int InstanceId); diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Deployment/DeploymentRecordSummary.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Deployment/DeploymentRecordSummary.cs new file mode 100644 index 00000000..804ba256 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Deployment/DeploymentRecordSummary.cs @@ -0,0 +1,54 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Deployment; + +/// +/// Row-shaped projection of a DeploymentRecord for list surfaces (CLI +/// deploy list / instance history, the Central UI deployment-status +/// page). +/// +/// +/// Exists so a deployment listing can be paged and filtered in the DATABASE. +/// Deployment records are insert-only — one row per deploy attempt, retained for +/// the configured retention window — so the previous "load the whole table, then +/// filter in memory" read path degraded with the age of the system rather than +/// with the size of the requested page. +/// +/// +/// +/// The RowVersion optimistic-concurrency token is deliberately absent: this +/// is a READ projection, and a summary must never be mistaken for something that +/// can be written back. Mutating paths keep loading the tracked entity. +/// +/// +/// +/// Message-contract evolution rule: additive-only. New fields go on the end with a +/// default. +/// +/// +/// Deployment record row id. +/// The logical deployment id (GUID, "N" format). +/// Instance the deployment targeted. +/// Terminal or in-flight deployment status. +/// Revision hash of the deployed configuration. +/// User who initiated the deployment. +/// When the deployment was initiated. +/// When the deployment reached a terminal status, if it has. +/// Failure detail when the deployment did not succeed. +public record DeploymentRecordSummary( + int Id, + string DeploymentId, + int InstanceId, + DeploymentStatus Status, + string? RevisionHash, + string DeployedBy, + DateTimeOffset DeployedAt, + DateTimeOffset? CompletedAt, + string? ErrorMessage) +{ + /// + /// Hard ceiling applied to a requested page size, mirroring the in-memory + /// Page clamp the management actor already applies elsewhere. + /// + public const int MaxPageSize = 1000; +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Templates/TemplateGraphWatermark.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Templates/TemplateGraphWatermark.cs new file mode 100644 index 00000000..0aced5b1 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Templates/TemplateGraphWatermark.cs @@ -0,0 +1,72 @@ +using System.Collections.Concurrent; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; + +namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Templates; + +/// +/// In-memory . Registered as a singleton so +/// the writer (the configuration-database repository's unit-of-work commit) and +/// the readers (the flatten-session cache, the staleness fast path) observe the +/// same counters within a process. +/// +/// +/// Lives in Commons rather than a component project because both the +/// ConfigurationDatabase writer and the DeploymentManager readers need it and +/// neither references the other. It is a pure in-memory primitive with no +/// infrastructure dependency, so it does not violate the "implementations live +/// with their component" convention the way a repository would. +/// +/// +/// +/// All counters are -updated and the per-id maps are +/// , so concurrent management +/// commands (which each run on their own DI scope) can bump safely. Reads are +/// deliberately unsynchronised relative to each other: a reader that observes a +/// half-applied batch of bumps sees a CHANGED value for at least one member, +/// which invalidates rather than falsely validates a cache entry. +/// +/// +public sealed class TemplateGraphWatermark : ITemplateGraphWatermark +{ + private readonly ConcurrentDictionary _templateVersions = new(); + private readonly ConcurrentDictionary _instanceVersions = new(); + private long _global; + private long _structure; + + /// + public long Global => Interlocked.Read(ref _global); + + /// + public long StructureVersion => Interlocked.Read(ref _structure); + + /// + public long GetTemplateVersion(int templateId) => + _templateVersions.TryGetValue(templateId, out var v) ? v : 0L; + + /// + public long GetInstanceVersion(int instanceId) => + _instanceVersions.TryGetValue(instanceId, out var v) ? v : 0L; + + /// + public void BumpTemplate(int templateId, bool structural = false) + { + _templateVersions.AddOrUpdate(templateId, 1L, static (_, current) => current + 1); + Interlocked.Increment(ref _global); + if (structural) + Interlocked.Increment(ref _structure); + } + + /// + public void BumpInstance(int instanceId) + { + _instanceVersions.AddOrUpdate(instanceId, 1L, static (_, current) => current + 1); + Interlocked.Increment(ref _global); + } + + /// + public void BumpAll() + { + Interlocked.Increment(ref _global); + Interlocked.Increment(ref _structure); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Templates/TemplateSummary.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Templates/TemplateSummary.cs new file mode 100644 index 00000000..c9294eca --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Templates/TemplateSummary.cs @@ -0,0 +1,53 @@ +namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Templates; + +/// +/// Row-shaped projection of a template for list surfaces (CLI template list, +/// the Central UI template tree, the ListTemplates management command). +/// +/// +/// Exists so a template listing can be paged in the DATABASE rather than by +/// materialising every template's full child graph — attributes, alarms, script +/// bodies, compositions and native alarm sources — and then discarding all but +/// one page of it in memory. Child collections are reduced to counts, which is +/// what every list surface actually renders. +/// +/// +/// +/// Message-contract evolution rule: this record is additive-only. New fields go +/// on the end with a default so an older peer deserialising a newer payload is +/// unaffected. +/// +/// +/// Template id. +/// Template name. +/// Optional description. +/// Parent template id when this template inherits. +/// Containing folder id. +/// True when the template was auto-derived to back a composition slot. +/// Owning composition row for a derived template. +/// Number of directly declared attributes. +/// Number of directly declared alarms. +/// Number of directly declared scripts. +/// Number of composition slots. +/// Number of native alarm source bindings. +public record TemplateSummary( + int Id, + string Name, + string? Description, + int? ParentTemplateId, + int? FolderId, + bool IsDerived, + int? OwnerCompositionId, + int AttributeCount, + int AlarmCount, + int ScriptCount, + int CompositionCount, + int NativeAlarmSourceCount) +{ + /// + /// Hard ceiling applied to a requested page size, mirroring the existing + /// in-memory Page clamp in the management actor so DB-side paging + /// cannot be used to pull an unbounded result set. + /// + public const int MaxPageSize = 1000; +} diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs index 1b5e5413..0383adcf 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs @@ -277,6 +277,89 @@ public class DeploymentManagerRepository : IDeploymentManagerRepository return expired.Count; } + /// + public async Task> QueryDeploymentSummariesAsync( + int? instanceId, + DeploymentStatus? status, + IReadOnlyCollection? instanceIdScope, + int skip, + int? take, + CancellationToken cancellationToken = default) + { + var query = _dbContext.DeploymentRecords.AsNoTracking(); + + if (instanceId.HasValue) + query = query.Where(d => d.InstanceId == instanceId.Value); + + if (status.HasValue) + query = query.Where(d => d.Status == status.Value); + + if (instanceIdScope != null) + { + // Site scoping applied DB-side. An empty scope is a real, meaningful + // filter — a user permitted no in-scope instances sees nothing — so it + // must not be short-circuited into "no filter". + var scope = instanceIdScope as int[] ?? instanceIdScope.ToArray(); + query = query.Where(d => scope.Contains(d.InstanceId)); + } + + // ThenByDescending(Id) for the same reason GetCurrentDeploymentStatusAsync + // does it: DeployedAt ties on rapid redeploys, and an unstable sort key + // makes Skip/Take paging non-deterministic (rows repeat or vanish between + // pages). + var paged = query + .OrderByDescending(d => d.DeployedAt) + .ThenByDescending(d => d.Id) + .Skip(Math.Max(0, skip)); + + if (take is > 0) + paged = paged.Take(Math.Min(take.Value, DeploymentRecordSummary.MaxPageSize)); + + return await paged + .Select(d => new DeploymentRecordSummary( + d.Id, + d.DeploymentId, + d.InstanceId, + d.Status, + d.RevisionHash, + d.DeployedBy, + d.DeployedAt, + d.CompletedAt, + d.ErrorMessage)) + .ToListAsync(cancellationToken); + } + + /// + public async Task PurgeTerminalDeploymentRecordsAsync( + DateTimeOffset cutoffUtc, + int batchSize, + CancellationToken cancellationToken = default) + { + // Terminal statuses ONLY. An InProgress (or Pending) row is never purged + // regardless of age: it is precisely the row TryReconcileWithSiteAsync + // looks for when deciding whether a prior deploy actually landed, and + // deleting it would silently disable the query-before-redeploy idempotency + // guard for that instance. + // + // Bounded batch rather than one unbounded DELETE so the first purge on a + // long-lived system cannot take a table-scale lock on DeploymentRecords. + // The caller re-invokes until it returns fewer rows than the batch size. + var stale = await _dbContext.DeploymentRecords + .Where(d => d.CompletedAt != null + && d.CompletedAt < cutoffUtc + && (d.Status == DeploymentStatus.Success || d.Status == DeploymentStatus.Failed)) + .OrderBy(d => d.Id) + .Take(batchSize) + .ToListAsync(cancellationToken); + + if (stale.Count == 0) + return 0; + + _dbContext.DeploymentRecords.RemoveRange(stale); + await _dbContext.SaveChangesAsync(cancellationToken); + return stale.Count; + } + // --- Startup reconciliation --- /// diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/TemplateEngineRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/TemplateEngineRepository.cs index a0a1e632..83e7d017 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/TemplateEngineRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/TemplateEngineRepository.cs @@ -3,20 +3,32 @@ using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Scripts; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Templates; namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories; public class TemplateEngineRepository : ITemplateEngineRepository { private readonly ScadaBridgeDbContext _context; + private readonly ITemplateGraphWatermark _watermark; /// /// Initializes a new instance of the TemplateEngineRepository class. /// /// The database context used to access template and instance data. - public TemplateEngineRepository(ScadaBridgeDbContext context) + /// + /// Process-wide template/instance version watermark. This repository is the + /// single unit-of-work through which EVERY template-graph mutation commits — + /// TemplateService, the ManagementActor native-alarm-source + /// handlers, and the Transport bundle importer all funnel through the same + /// — so bumping here (rather than in each + /// caller) is the only placement that cannot be bypassed. + /// + public TemplateEngineRepository(ScadaBridgeDbContext context, ITemplateGraphWatermark watermark) { _context = context ?? throw new ArgumentNullException(nameof(context)); + _watermark = watermark ?? throw new ArgumentNullException(nameof(watermark)); } // Template @@ -78,6 +90,95 @@ public class TemplateEngineRepository : ITemplateEngineRepository .ToListAsync(cancellationToken); } + /// + /// + /// Two deliberate departures from : + /// + /// + /// AsNoTracking — the analysis walks (cycle detection, collision + /// detection, canonical-name resolution) never write through the loaded + /// graph, so paying for a change-tracker snapshot of every attribute, + /// alarm, script and composition in the database is pure waste. + /// + /// + /// TemplateScript.Code is projected away (left empty). Script + /// bodies are by far the largest column in the graph and NONE of the + /// analysis consumers read them — CycleDetector reads only ids and + /// edges, CollisionDetector and TemplateResolver read only + /// member names and lock flags. + /// + /// + /// Callers that DO need bodies or tracked entities — the inheritance + /// reconciler and the flattener — must keep using + /// / . + /// + public async Task> GetAllTemplatesForAnalysisAsync(CancellationToken cancellationToken = default) + { + return await _context.Templates + .AsNoTracking() + .Select(t => new Template(t.Name) + { + Id = t.Id, + Description = t.Description, + ParentTemplateId = t.ParentTemplateId, + FolderId = t.FolderId, + IsDerived = t.IsDerived, + OwnerCompositionId = t.OwnerCompositionId, + Attributes = t.Attributes.ToList(), + Alarms = t.Alarms.ToList(), + NativeAlarmSources = t.NativeAlarmSources.ToList(), + Compositions = t.Compositions.ToList(), + Scripts = t.Scripts + .Select(s => new TemplateScript(s.Name, string.Empty) + { + Id = s.Id, + TemplateId = s.TemplateId, + IsLocked = s.IsLocked, + TriggerType = s.TriggerType, + TriggerConfiguration = s.TriggerConfiguration, + ParameterDefinitions = s.ParameterDefinitions, + ReturnDefinition = s.ReturnDefinition, + MinTimeBetweenRuns = s.MinTimeBetweenRuns, + ExecutionTimeoutSeconds = s.ExecutionTimeoutSeconds, + IsInherited = s.IsInherited, + LockedInDerived = s.LockedInDerived + }) + .ToList() + }) + .ToListAsync(cancellationToken); + } + + /// + public async Task> GetTemplateSummariesAsync( + int skip, + int? take, + CancellationToken cancellationToken = default) + { + var query = _context.Templates + .AsNoTracking() + .OrderBy(t => t.Id) + .Skip(Math.Max(0, skip)); + + if (take is > 0) + query = query.Take(Math.Min(take.Value, TemplateSummary.MaxPageSize)); + + return await query + .Select(t => new TemplateSummary( + t.Id, + t.Name, + t.Description, + t.ParentTemplateId, + t.FolderId, + t.IsDerived, + t.OwnerCompositionId, + t.Attributes.Count, + t.Alarms.Count, + t.Scripts.Count, + t.Compositions.Count, + t.NativeAlarmSources.Count)) + .ToListAsync(cancellationToken); + } + /// public async Task> GetTemplatesComposingAsync(int composedTemplateId, CancellationToken cancellationToken = default) { @@ -659,8 +760,129 @@ public class TemplateEngineRepository : ITemplateEngineRepository } /// + /// + /// Inspects the change tracker BEFORE committing (afterwards every entry is + /// and the attribution is lost) and bumps + /// the for each affected template / + /// instance. The bump is applied only AFTER a successful commit, so a failed + /// save does not invalidate caches for a change that never landed. + /// public async Task SaveChangesAsync(CancellationToken cancellationToken = default) { - return await _context.SaveChangesAsync(cancellationToken); + var pending = CollectWatermarkBumps(); + var written = await _context.SaveChangesAsync(cancellationToken); + ApplyWatermarkBumps(pending); + return written; + } + + /// + /// Snapshot of the watermark bumps implied by the current change-tracker + /// contents. Captured pre-commit, applied post-commit. + /// + private readonly record struct WatermarkBumps( + HashSet Templates, + HashSet StructuralTemplates, + HashSet Instances, + bool Unattributed); + + /// + /// Walks the change tracker and maps every added/modified/deleted + /// template-graph or instance-graph entry back to the template (or instance) + /// whose flattened output it can affect. + /// + /// + /// A change is treated as STRUCTURAL — invalidating cached chain membership, + /// not just chain contents — when it adds or removes a + /// row, changes a , or touches a + /// row. Those are exactly the edges the + /// flattener walks to build a chain. + /// + /// + /// + /// An entry whose owning id cannot be resolved (a child row deleted before + /// its FK was materialised, or an entity type added later that this switch + /// does not know) sets Unattributed, which conservatively invalidates + /// every cached chain rather than silently letting a stale entry survive. + /// + /// + private WatermarkBumps CollectWatermarkBumps() + { + var templates = new HashSet(); + var structural = new HashSet(); + var instances = new HashSet(); + var unattributed = false; + + foreach (var entry in _context.ChangeTracker.Entries()) + { + if (entry.State is not (EntityState.Added or EntityState.Modified or EntityState.Deleted)) + continue; + + switch (entry.Entity) + { + case Template t: + templates.Add(t.Id); + // Add/Delete changes chain membership; a re-parent changes it too. + if (entry.State != EntityState.Modified + || entry.Property(nameof(Template.ParentTemplateId)).IsModified) + { + structural.Add(t.Id); + } + break; + case TemplateAttribute a: + templates.Add(a.TemplateId); + break; + case TemplateAlarm al: + templates.Add(al.TemplateId); + break; + case TemplateScript s: + templates.Add(s.TemplateId); + break; + case TemplateNativeAlarmSource ns: + templates.Add(ns.TemplateId); + break; + case TemplateComposition c: + templates.Add(c.TemplateId); + structural.Add(c.TemplateId); + break; + case Instance i: + instances.Add(i.Id); + break; + case InstanceAttributeOverride ao: + instances.Add(ao.InstanceId); + break; + case InstanceAlarmOverride alo: + instances.Add(alo.InstanceId); + break; + case InstanceNativeAlarmSourceOverride nso: + instances.Add(nso.InstanceId); + break; + case InstanceConnectionBinding cb: + instances.Add(cb.InstanceId); + break; + case SharedScript: + // Shared scripts are a session-global input to validation, not + // owned by any one template — invalidate everything. + unattributed = true; + break; + default: + // Areas, folders and anything else do not feed the flattener. + break; + } + } + + return new WatermarkBumps(templates, structural, instances, unattributed); + } + + /// Applies a previously collected snapshot. + private void ApplyWatermarkBumps(WatermarkBumps bumps) + { + if (bumps.Unattributed) + _watermark.BumpAll(); + + foreach (var templateId in bumps.Templates) + _watermark.BumpTemplate(templateId, bumps.StructuralTemplates.Contains(templateId)); + + foreach (var instanceId in bumps.Instances) + _watermark.BumpInstance(instanceId); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ServiceCollectionExtensions.cs index 1752e17e..e8718dac 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ServiceCollectionExtensions.cs @@ -1,10 +1,12 @@ using Microsoft.AspNetCore.DataProtection; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Templates; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Services; @@ -49,6 +51,13 @@ public static class ServiceCollectionExtensions return new ScadaBridgeDbContext(options, protectionProvider); }); + // Process-wide template/instance version watermark. Singleton because the + // writer (TemplateEngineRepository.SaveChangesAsync, scoped) and the readers + // (the flatten-session cache and the staleness fast path, also scoped) must + // observe the same counters across DI scopes. TryAdd so the DeploymentManager + // registration of the same interface is idempotent. + services.TryAddSingleton(); + services.AddScoped(); services.AddScoped(); services.AddScoped(); diff --git a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptions.cs b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptions.cs index 7fc4dabf..46f905c7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptions.cs @@ -17,4 +17,43 @@ public class DeploymentManagerOptions /// Timeout for acquiring an operation lock on an instance. public TimeSpan OperationLockTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Maximum number of instance deployments whose SITE ROUND-TRIP runs + /// concurrently during a bulk DeploySiteAsync. The flatten/validate and + /// persistence phases stay serial regardless (they share one non-thread-safe + /// DbContext); only the network wait is fanned out. + /// + /// + /// Bounded rather than unbounded so a large site cannot open one in-flight + /// gRPC command per instance against a two-node site pair. Four mirrors the + /// conservative end of the site's own apply concurrency; raise it only with + /// evidence from the target-scale load test. + /// + /// + public int SiteDeploymentMaxParallelism { get; set; } = 4; + + /// + /// Per-instance deadline applied to the site round-trip inside a bulk + /// DeploySiteAsync. One unreachable or wedged instance is recorded as a + /// failed row and the rest of the batch continues — the bulk operation never + /// inherits a single instance's hang. + /// + public TimeSpan SiteDeploymentTimeoutPerInstance { get; set; } = TimeSpan.FromSeconds(120); + + /// + /// Retention window for TERMINAL deployment records (Success, + /// Failed). Deployments are insert-only — one row per attempt, forever — + /// so without a window the table grows without bound and every unfiltered + /// deployment query gets slower with the age of the system. + /// + /// + /// Deliberately generous (one year) because deployment history is + /// operator-facing forensics, and deliberately terminal-only: an + /// InProgress or Pending row is never purged regardless of age, + /// because it is exactly the row the query-before-redeploy reconciliation + /// path needs to find. Set to to disable purging. + /// + /// + public TimeSpan TerminalDeploymentRecordRetention { get; set; } = TimeSpan.FromDays(365); } diff --git a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptionsValidator.cs index d041088e..a4431212 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptionsValidator.cs @@ -27,5 +27,20 @@ public sealed class DeploymentManagerOptionsValidator : OptionsValidatorBase TimeSpan.Zero, $"ScadaBridge:DeploymentManager:OperationLockTimeout must be a positive duration " + $"(was {options.OperationLockTimeout}); it bounds acquiring an instance operation lock."); + + builder.RequireThat(options.SiteDeploymentMaxParallelism > 0, + $"ScadaBridge:DeploymentManager:SiteDeploymentMaxParallelism must be at least 1 " + + $"(was {options.SiteDeploymentMaxParallelism}); it is the SemaphoreSlim bound on a bulk " + + $"site deployment's site round-trips, and a non-positive value would deadlock the fan-out."); + + builder.RequireThat(options.SiteDeploymentTimeoutPerInstance > TimeSpan.Zero, + $"ScadaBridge:DeploymentManager:SiteDeploymentTimeoutPerInstance must be a positive duration " + + $"(was {options.SiteDeploymentTimeoutPerInstance}); it bounds each instance's site round-trip " + + $"inside a bulk site deployment."); + + builder.RequireThat(options.TerminalDeploymentRecordRetention >= TimeSpan.Zero, + $"ScadaBridge:DeploymentManager:TerminalDeploymentRecordRetention must not be negative " + + $"(was {options.TerminalDeploymentRecordRetention}); use TimeSpan.Zero to disable purging of " + + $"terminal deployment records."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs index 321d6fe6..3fef99e7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs @@ -57,6 +57,24 @@ public class DeploymentService /// private const string TimeoutFailurePrefix = "Communication failure:"; + /// Rows deleted per terminal-record purge batch. + private const int TerminalPurgeBatchSize = 500; + + /// Batches deleted per opportunistic sweep, bounding one caller's latency cost. + private const int TerminalPurgeBatchesPerSweep = 4; + + /// Minimum interval between opportunistic terminal-record sweeps. + private static readonly TimeSpan TerminalPurgeMinInterval = TimeSpan.FromHours(6); + + /// + /// Ticks of the last opportunistic terminal-record sweep, process-wide. + /// is scoped, so this must be static to + /// rate-limit across requests; makes the + /// claim-the-sweep step a compare-and-swap so concurrent readers do not all + /// sweep at once. + /// + private static long _lastTerminalPurgeTicks; + /// /// Initializes a new instance of with all required dependencies. /// @@ -161,28 +179,113 @@ public class DeploymentService int instanceId, string user, CancellationToken cancellationToken = default) + { + // Single-instance deploy = the same three phases the bulk path runs, with a + // batch of one and a private flatten session. Composing both entry points + // out of the same phase helpers is what keeps deployment identity, + // idempotency, lock coverage and optimistic concurrency from drifting apart + // between them. + var prepared = await PrepareDeploymentAsync( + instanceId, user, session: null, cancellationToken); + + if (prepared.EarlyResult is { } early) + return early; + + using (prepared.LockHandle) + { + var outcome = await SendDeploymentAsync(prepared, cancellationToken); + return await FinalizeDeploymentAsync(prepared, outcome, user, cancellationToken); + } + } + + /// + /// Phase 1 of a deployment (serial, database-bound): validate the state + /// transition, take the per-instance operation lock, mint the deployment id, + /// flatten + validate, run query-before-redeploy reconciliation, stage the + /// PendingDeployment row and insert the InProgress + /// . + /// + /// + /// Everything here touches the scoped, non-thread-safe DbContext, so it + /// must run serially — a bulk site deploy loops this phase before fanning out + /// the phase-2 network waits. On success the returned + /// owns the operation lock; the CALLER is + /// responsible for disposing it once phase 3 has completed. + /// + /// + /// Instance being deployed. + /// User attributed on the deployment record and audit rows. + /// + /// Shared flatten session for a batch, or for a private + /// one. Sharing it across a batch is what collapses N identical template-chain + /// walks into one. + /// + /// Cancellation token. + /// + /// A prepared deployment ready to send, or one carrying + /// when the deploy resolved before + /// any site round-trip (validation failure, or a reconciled prior deployment). + /// + private async Task PrepareDeploymentAsync( + int instanceId, + string user, + FlattenSession? session, + CancellationToken cancellationToken) { // Load instance var instance = await _repository.GetInstanceByIdAsync(instanceId, cancellationToken); if (instance == null) - return Result.Failure($"Instance with ID {instanceId} not found."); + return PreparedDeployment.Resolved(Result.Failure($"Instance with ID {instanceId} not found.")); // Validate state transition var transitionError = StateTransitionValidator.ValidateTransition(instance.State, "deploy"); if (transitionError != null) - return Result.Failure(transitionError); + return PreparedDeployment.Resolved(Result.Failure(transitionError)); // Acquire per-instance operation lock - using var lockHandle = await _lockManager.AcquireAsync( + var lockHandle = await _lockManager.AcquireAsync( instance.UniqueName, _options.OperationLockTimeout, cancellationToken); + try + { + return await PrepareUnderLockAsync(instance, user, session, lockHandle, cancellationToken); + } + catch + { + // The lock is handed to the caller only on the success path; any fault + // during preparation must release it here or a failed prepare would + // wedge the instance until the process restarts. + lockHandle.Dispose(); + throw; + } + } + + /// + /// The body of that runs while the + /// per-instance operation lock is held. Split out so the lock can be released + /// on every fault path with one try/catch rather than a nested one per step. + /// + private async Task PrepareUnderLockAsync( + Instance instance, + string user, + FlattenSession? session, + IDisposable lockHandle, + CancellationToken cancellationToken) + { + var instanceId = instance.Id; + // Generate unique deployment ID var deploymentId = Guid.NewGuid().ToString("N"); // Flatten configuration (captures template state at this point in time) - var flattenResult = await _flatteningPipeline.FlattenAndValidateAsync(instanceId, cancellationToken); + var flattenResult = await _flatteningPipeline.FlattenAndValidateAsync( + instanceId, cancellationToken, validateScripts: true, session); if (flattenResult.IsFailure) - return Result.Failure($"Validation failed: {flattenResult.Error}"); + { + lockHandle.Dispose(); + return PreparedDeployment.Resolved( + Result.Failure($"Validation failed: {flattenResult.Error}")); + } var flattenedConfig = flattenResult.Value.Configuration; var revisionHash = flattenResult.Value.RevisionHash; @@ -200,8 +303,9 @@ public class DeploymentService validationResult.Errors.Count, string.Join("; ", validationResult.Errors.Select(e => e.Message))); - return Result.Failure( - $"Pre-deployment validation failed: {validationResult.SummarizeErrors()}"); + lockHandle.Dispose(); + return PreparedDeployment.Resolved(Result.Failure( + $"Pre-deployment validation failed: {validationResult.SummarizeErrors()}")); } // Serialize for transmission (also the payload stored in the deployed @@ -218,7 +322,10 @@ public class DeploymentService var reconciled = await TryReconcileWithSiteAsync( instance, revisionHash, configJson, user, cancellationToken); if (reconciled != null) - return Result.Success(reconciled); + { + lockHandle.Dispose(); + return PreparedDeployment.Resolved(Result.Success(reconciled)); + } // Notify-and-fetch: the site fetches the staged config from // CentralFetchBaseUrl, so a deploy is impossible without it. Fail fast @@ -227,8 +334,11 @@ public class DeploymentService // confusing downstream site-fetch failure (and no InProgress record is // stranded). if (string.IsNullOrEmpty(_commOptions.CentralFetchBaseUrl)) - return Result.Failure( - "CentralFetchBaseUrl is not configured — required for deployment (notify-and-fetch)."); + { + lockHandle.Dispose(); + return PreparedDeployment.Resolved(Result.Failure( + "CentralFetchBaseUrl is not configured — required for deployment (notify-and-fetch).")); + } // Create the deployment record directly in InProgress. // @@ -275,12 +385,8 @@ public class DeploymentService deploymentId, instance.UniqueName, revisionHash, user, stagedAt, _commOptions.CentralFetchBaseUrl, token); - _logger.LogInformation( - "Sending deployment {DeploymentId} for instance {Instance} to site {SiteId} (notify-and-fetch)", - deploymentId, instance.UniqueName, siteId); - // Cleanup of the staged PendingDeployment is TTL-based ONLY — the row - // is deliberately NOT deleted on success or in the catch. On a + // is deliberately NOT deleted on success or on failure. On a // central-side Ask timeout the site may have applied AND told the // standby node to fetch; deleting now would 404 that in-flight // standby fetch and break failover. Supersession bounds pending rows @@ -288,8 +394,89 @@ public class DeploymentService // leaving rows for TTL purge is safe. Expired rows are swept by the // central PendingDeploymentPurgeActor singleton on its // CommunicationOptions.PendingDeploymentPurgeInterval cadence. - var response = await _communicationService.RefreshDeploymentAsync(siteId, command, cancellationToken); + return new PreparedDeployment( + instance, deploymentId, record, revisionHash, configJson, + siteId, command, lockHandle, EarlyResult: null); + } + catch (Exception ex) + { + // Staging failed before anything was sent. Record the failure exactly + // as the post-send path does (never leave the record InProgress) and + // release the lock — there is no phase 2 or 3 to run. + await MarkDeploymentFailedAsync(record, instance, deploymentId, user, ex); + lockHandle.Dispose(); + return PreparedDeployment.Resolved(FailureResultFor(ex)); + } + } + /// + /// Phase 2 of a deployment (parallelisable, network-bound): the + /// RefreshDeploymentCommand round-trip to the site. Touches no + /// repository and mutates no shared state, which is what makes it safe to fan + /// out across a batch while phases 1 and 3 stay serial. + /// + /// The prepared deployment to send. + /// Cancellation token, already carrying any per-instance deadline. + /// The site's response, or the exception that prevented one. + private async Task SendDeploymentAsync( + PreparedDeployment prepared, + CancellationToken cancellationToken) + { + try + { + _logger.LogInformation( + "Sending deployment {DeploymentId} for instance {Instance} to site {SiteId} (notify-and-fetch)", + prepared.DeploymentId, prepared.Instance.UniqueName, prepared.SiteIdentifier); + + var response = await _communicationService.RefreshDeploymentAsync( + prepared.SiteIdentifier, prepared.Command, cancellationToken); + return new SendOutcome(response, null); + } + catch (Exception ex) + { + return new SendOutcome(null, ex); + } + } + + /// + /// Phase 3 of a deployment (serial, database-bound): commit the terminal + /// status, apply post-success side effects, and write the audit row. + /// + /// The prepared deployment whose send just completed. + /// The site response or the fault from phase 2. + /// User attributed on the audit rows. + /// Cancellation token. + /// The deployment result for the caller. + private async Task> FinalizeDeploymentAsync( + PreparedDeployment prepared, + SendOutcome outcome, + string user, + CancellationToken cancellationToken) + { + var instance = prepared.Instance; + var record = prepared.Record; + var deploymentId = prepared.DeploymentId; + var instanceId = instance.Id; + + if (outcome.Error is { } sendError) + { + // Any fault out of the site round-trip (timeout, cancellation, + // transport, serialization) must leave the deployment record as + // Failed -- the design requires an interrupted deployment to be + // treated as failed, never stuck in InProgress. + await MarkDeploymentFailedAsync(record, instance, deploymentId, user, sendError); + + _logger.LogError(sendError, + "Deployment {DeploymentId} for instance {Instance} failed", + deploymentId, instance.UniqueName); + + return FailureResultFor(sendError); + } + + var response = outcome.Response!; + + try + { // Update status based on site response. record.Status = response.Status; record.ErrorMessage = response.ErrorMessage; @@ -322,7 +509,7 @@ public class DeploymentService // logged loudly for operator reconciliation but must not flip // the already-committed Success record back to Failed. await ApplyPostSuccessSideEffectsAsync( - instance, deploymentId, revisionHash, configJson, + instance, deploymentId, prepared.RevisionHash, prepared.ConfigJson, forceEnabledState: true, cancellationToken); } @@ -345,61 +532,366 @@ public class DeploymentService } catch (Exception ex) { - // Any exception out of the try (timeout, - // cancellation, transport, serialization, DB) must leave the - // deployment record as Failed -- the design requires an interrupted - // deployment to be treated as failed, never stuck in InProgress. - // - // The failure-status write must NOT use the - // operation's cancellation token. If the operation was cancelled or - // timed out, that token is already cancelled and the cleanup writes - // would themselves throw before the Failed status is persisted. - // Use CancellationToken.None so the failure is durably recorded. - var isTimeout = ex is TimeoutException or OperationCanceledException or Akka.Actor.AskTimeoutException; - - record.Status = DeploymentStatus.Failed; - record.ErrorMessage = isTimeout - ? $"{TimeoutFailurePrefix} {ex.Message}" - : $"Deployment error: {ex.Message}"; - record.CompletedAt = DateTimeOffset.UtcNow; - - try - { - await _repository.UpdateDeploymentRecordAsync(record, CancellationToken.None); - // Note: if the staging SaveChangesAsync above was interrupted, an - // Added PendingDeployment may still be tracked and will be - // committed by this cleanup save. That row is orphaned (no - // RefreshDeploymentCommand was sent, so no site holds its token) - // and is removed by TTL purge / superseded by the next deploy -- - // harmless. - await _repository.SaveChangesAsync(CancellationToken.None); - NotifyStatusChange(record); - - await _auditService.LogAsync(user, "DeployFailed", "Instance", instanceId.ToString(), - instance.UniqueName, new { DeploymentId = deploymentId, Error = ex.Message }, - CancellationToken.None); - } - catch (Exception cleanupEx) - { - // The deployment already failed; a failed cleanup write must not - // mask the original error. Log loudly so an operator can reconcile. - _logger.LogError(cleanupEx, - "Failed to persist Failed status for deployment {DeploymentId} of instance {Instance} " + - "after deployment error: {Error}", - deploymentId, instance.UniqueName, ex.Message); - } + // A fault in the post-send persistence path is handled identically to + // a send fault: the record must never be left InProgress. + await MarkDeploymentFailedAsync(record, instance, deploymentId, user, ex); _logger.LogError(ex, "Deployment {DeploymentId} for instance {Instance} failed", deploymentId, instance.UniqueName); - return Result.Failure( - isTimeout - ? $"Deployment timed out: {ex.Message}" - : $"Deployment failed: {ex.Message}"); + return FailureResultFor(ex); } } + /// + /// Writes the terminal status (plus the + /// failure audit row) for a deployment that faulted, on any phase. + /// + /// + /// The failure-status write deliberately uses : + /// if the operation was cancelled or timed out, the operation's own token is + /// already cancelled and these cleanup writes would themselves throw before + /// the Failed status was persisted, leaving the record stuck InProgress. + /// + /// + /// + /// A fault DURING cleanup is logged loudly and swallowed — it must not mask + /// the original error the caller is about to report. + /// + /// + private async Task MarkDeploymentFailedAsync( + DeploymentRecord record, + Instance instance, + string deploymentId, + string user, + Exception cause) + { + record.Status = DeploymentStatus.Failed; + record.ErrorMessage = IsTimeoutFault(cause) + ? $"{TimeoutFailurePrefix} {cause.Message}" + : $"Deployment error: {cause.Message}"; + record.CompletedAt = DateTimeOffset.UtcNow; + + try + { + await _repository.UpdateDeploymentRecordAsync(record, CancellationToken.None); + // Note: if the staging SaveChangesAsync was interrupted, an Added + // PendingDeployment may still be tracked and will be committed by this + // cleanup save. That row is orphaned (no RefreshDeploymentCommand was + // sent, so no site holds its token) and is removed by TTL purge / + // superseded by the next deploy -- harmless. + await _repository.SaveChangesAsync(CancellationToken.None); + NotifyStatusChange(record); + + await _auditService.LogAsync(user, "DeployFailed", "Instance", instance.Id.ToString(), + instance.UniqueName, new { DeploymentId = deploymentId, Error = cause.Message }, + CancellationToken.None); + } + catch (Exception cleanupEx) + { + _logger.LogError(cleanupEx, + "Failed to persist Failed status for deployment {DeploymentId} of instance {Instance} " + + "after deployment error: {Error}", + deploymentId, instance.UniqueName, cause.Message); + } + } + + /// + /// True when a deployment fault is a timeout/cancellation rather than a hard + /// error. Drives both the marker (which the + /// query-before-redeploy trigger reads) and the caller-facing wording. + /// + private static bool IsTimeoutFault(Exception ex) => + ex is TimeoutException or OperationCanceledException or Akka.Actor.AskTimeoutException; + + /// Maps a deployment fault to the caller-facing failure result. + private static Result FailureResultFor(Exception ex) => + Result.Failure( + IsTimeoutFault(ex) + ? $"Deployment timed out: {ex.Message}" + : $"Deployment failed: {ex.Message}"); + + /// + /// A deployment that has cleared phase 1 and is ready for its site round-trip, + /// OR one that resolved during phase 1 (carrying ). + /// + /// + /// When is this record OWNS the + /// per-instance operation lock in , and the caller must + /// dispose it after phase 3. When is set the lock has + /// already been released and every other member is a placeholder. + /// + /// + private sealed record PreparedDeployment( + Instance Instance, + string DeploymentId, + DeploymentRecord Record, + string RevisionHash, + string ConfigJson, + string SiteIdentifier, + RefreshDeploymentCommand Command, + IDisposable? LockHandle, + Result? EarlyResult) + { + /// Builds a phase-1-resolved deployment carrying the given result and holding no lock. + /// The result to hand back to the caller. + /// A prepared deployment whose is set. + public static PreparedDeployment Resolved(Result result) => + new(null!, string.Empty, null!, string.Empty, string.Empty, string.Empty, null!, null, result); + } + + /// + /// Result of phase 2: either the site's response or the exception that + /// prevented one. Exactly one is non-null. + /// + private readonly record struct SendOutcome(DeploymentStatusResponse? Response, Exception? Error); + + /// + /// Deploy every deployable instance at a site in one operation. + /// + /// + /// Structured as the three deployment phases with the middle one fanned out: + /// + /// + /// + /// Prepare (serial). Every instance is flattened, validated, staged + /// and given an InProgress record on the caller's single scoped + /// DbContext. All instances share ONE , + /// so a template chain common to N instances is walked once, and the + /// session-global queries (shared scripts, schema library, the site's data + /// connections) run once for the whole batch instead of once per instance. + /// + /// + /// Send (bounded parallel). Site round-trips run concurrently up to + /// , + /// each under its own + /// + /// deadline, so one wedged instance cannot stall the batch. This phase + /// touches no repository — that is precisely why it is the only phase that + /// may run in parallel against a non-thread-safe DbContext. + /// + /// + /// Finalize (serial). Terminal statuses, post-success side effects + /// and audit rows are committed one instance at a time, then each + /// instance's operation lock is released. + /// + /// + /// + /// + /// Semantics per instance are byte-for-byte those of + /// — its own deployment id and revision hash, + /// its own operation lock held across the whole operation, its own + /// optimistically-concurrent status record, and the same query-before-redeploy + /// idempotency check. Instances that cannot be deployed (wrong state, failed + /// validation, lock already held) are reported as failed rows; the batch is + /// NOT all-or-nothing, mirroring the artifact deployment path. + /// + /// + /// The database ID of the site whose instances are deployed. + /// The username initiating the deployment, recorded on every record and audit row. + /// Cancellation token for the operation. + /// A per-instance result matrix, or a failure result when the site itself cannot be resolved. + public async Task> DeploySiteAsync( + int siteId, + string user, + CancellationToken cancellationToken = default) + { + var site = await _siteRepository.GetSiteByIdAsync(siteId, cancellationToken); + if (site == null) + return Result.Failure($"Site with ID {siteId} not found."); + + var instances = await _siteRepository.GetInstancesBySiteIdAsync(siteId, cancellationToken); + if (instances.Count == 0) + return Result.Success(new SiteDeploymentSummary(site.SiteIdentifier, [], 0, 0)); + + // ---- Phase 1: prepare, serially, on ONE shared flatten session. ---- + var session = _flatteningPipeline.CreateSession(); + var prepared = new List(instances.Count); + var results = new List(instances.Count); + + foreach (var instance in instances) + { + cancellationToken.ThrowIfCancellationRequested(); + + PreparedDeployment step; + try + { + step = await PrepareDeploymentAsync(instance.Id, user, session, cancellationToken); + } + catch (Exception ex) + { + // A prepare fault (most commonly a TimeoutException from the + // operation lock because another operation holds this instance) + // fails only this instance. Recording it and moving on is what + // makes a bulk deploy usable while individual instances are busy. + _logger.LogWarning(ex, + "Preparing instance {Instance} for bulk deployment of site {SiteId} failed", + instance.UniqueName, site.SiteIdentifier); + results.Add(new InstanceDeploymentResult( + instance.Id, instance.UniqueName, null, false, ex.Message)); + continue; + } + + if (step.EarlyResult is { } early) + { + results.Add(ToInstanceResult(instance.Id, instance.UniqueName, early)); + continue; + } + + prepared.Add(step); + } + + // ---- Phase 2: bounded-parallel site round-trips. ---- + var outcomes = await SendPreparedAsync(prepared, cancellationToken); + + // ---- Phase 3: finalize, serially, releasing each lock as we go. ---- + for (var i = 0; i < prepared.Count; i++) + { + var step = prepared[i]; + try + { + var result = await FinalizeDeploymentAsync(step, outcomes[i], user, cancellationToken); + results.Add(ToInstanceResult(step.Instance.Id, step.Instance.UniqueName, result, step.DeploymentId)); + } + finally + { + step.LockHandle?.Dispose(); + } + } + + var summary = new SiteDeploymentSummary( + site.SiteIdentifier, + results, + results.Count(r => r.Success), + results.Count(r => !r.Success)); + + _logger.LogInformation( + "Bulk deployment of site {SiteId} complete: {SuccessCount} succeeded, {FailureCount} failed " + + "({ChainLoads} template-chain load(s) across {InstanceCount} instance(s))", + site.SiteIdentifier, summary.SuccessCount, summary.FailureCount, + session.ChainLoads, instances.Count); + + return Result.Success(summary); + } + + /// + /// Runs phase 2 for a whole batch: every prepared deployment's site round-trip, + /// concurrent up to + /// and each bounded by + /// . + /// Outcomes are returned positionally so phase 3 can pair them back with their + /// prepared deployment. + /// + private async Task SendPreparedAsync( + IReadOnlyList prepared, + CancellationToken cancellationToken) + { + var outcomes = new SendOutcome[prepared.Count]; + if (prepared.Count == 0) + return outcomes; + + using var gate = new SemaphoreSlim(_options.SiteDeploymentMaxParallelism); + + var sends = prepared.Select(async (step, index) => + { + await gate.WaitAsync(cancellationToken); + try + { + using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + cts.CancelAfter(_options.SiteDeploymentTimeoutPerInstance); + outcomes[index] = await SendDeploymentAsync(step, cts.Token); + } + finally + { + gate.Release(); + } + }); + + await Task.WhenAll(sends); + return outcomes; + } + + /// Maps a per-instance deployment result into the bulk summary row shape. + private static InstanceDeploymentResult ToInstanceResult( + int instanceId, + string uniqueName, + Result result, + string? deploymentId = null) => + result.IsSuccess + ? new InstanceDeploymentResult( + instanceId, uniqueName, result.Value.DeploymentId, true, null) + : new InstanceDeploymentResult( + instanceId, uniqueName, deploymentId, false, result.Error); + + /// + /// Opportunistic retention sweep for TERMINAL deployment records, rate-limited + /// to once every per process and bounded + /// to batches per call. + /// + /// + /// Swept on read (the deployment-list path) rather than from a timer for the + /// same reason expired secured writes are: the component owns no scheduling + /// infrastructure of its own, and the read path is exactly the operation that + /// suffers when the table is allowed to grow without bound. Both central nodes + /// may sweep; the delete is idempotent, so a duplicate pass is a no-op. + /// + /// + /// + /// Best-effort: a failed sweep is logged and swallowed. Retention maintenance + /// must never fail the operator's actual query. + /// + /// + /// Cancellation token. + /// The number of records purged by this call (0 when the sweep was skipped). + public async Task TryPurgeTerminalDeploymentRecordsAsync(CancellationToken cancellationToken = default) + { + var retention = _options.TerminalDeploymentRecordRetention; + if (retention <= TimeSpan.Zero) + return 0; + + var now = DateTimeOffset.UtcNow; + var last = Interlocked.Read(ref _lastTerminalPurgeTicks); + if (last != 0 && now.UtcTicks - last < TerminalPurgeMinInterval.Ticks) + return 0; + + // Claim the sweep. A loser of this CAS returns immediately rather than + // running a redundant second pass. + if (Interlocked.CompareExchange(ref _lastTerminalPurgeTicks, now.UtcTicks, last) != last) + return 0; + + var cutoff = now - retention; + var purged = 0; + + try + { + for (var batch = 0; batch < TerminalPurgeBatchesPerSweep; batch++) + { + var deleted = await _repository.PurgeTerminalDeploymentRecordsAsync( + cutoff, TerminalPurgeBatchSize, cancellationToken); + purged += deleted; + if (deleted < TerminalPurgeBatchSize) + break; + } + + if (purged > 0) + { + _logger.LogInformation( + "Purged {Count} terminal deployment record(s) completed before {Cutoff} " + + "(retention {Retention}).", + purged, cutoff, retention); + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Terminal deployment-record retention sweep failed; retrying after {Interval}.", + TerminalPurgeMinInterval); + } + + return purged; + } + /// /// Disable an instance. Stops Instance Actor, retains config, S&F drains. /// @@ -1178,3 +1670,37 @@ public record DeploymentComparisonResult( bool IsStale, DateTimeOffset DeployedAt, ConfigurationDiff? Diff = null); + +/// +/// One instance's outcome inside a bulk site deployment. +/// +/// Database id of the instance. +/// The instance's unique name (also its operation-lock key). +/// +/// The deployment id minted for this instance, or when the +/// instance never got that far (state-transition rejection, lock contention). +/// +/// Whether the site confirmed the apply. +/// Failure detail; on success. +public record InstanceDeploymentResult( + int InstanceId, + string UniqueName, + string? DeploymentId, + bool Success, + string? ErrorMessage); + +/// +/// Result matrix of a bulk site deployment. Mirrors the shape of +/// ArtifactDeploymentSummary: successes are not rolled back when other +/// instances fail, and each failed instance is individually retryable through the +/// ordinary single-instance deploy. +/// +/// The site's string identifier (e.g. site-a). +/// Per-instance outcomes. +/// Number of instances the site confirmed. +/// Number of instances that failed for any reason. +public record SiteDeploymentSummary( + string SiteIdentifier, + IReadOnlyList InstanceResults, + int SuccessCount, + int FailureCount); diff --git a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/FlattenSession.cs b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/FlattenSession.cs new file mode 100644 index 00000000..6c0fe23c --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/FlattenSession.cs @@ -0,0 +1,257 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening; + +namespace ZB.MOM.WW.ScadaBridge.DeploymentManager; + +/// +/// Memoises the repository reads that performs, +/// for the lifetime of ONE flatten/validate session. +/// +/// +/// Without a session every instance flatten re-walks its whole template +/// inheritance chain (one query per link), re-loads the compositions of every +/// template in that chain and of every composed chain it reaches, and re-issues +/// the three session-global queries — shared scripts, the shared-schema library, +/// and the target site's data connections. For a bulk deploy of N instances off +/// the same template that is N× the same work; even a single instance re-loads a +/// composed chain once per composing template. +/// +/// +/// +/// Cache key. Templates are keyed on (id, ITemplateGraphWatermark +/// version) and chain MEMBERSHIP additionally on the watermark's +/// , so a template edited +/// mid-session is re-read rather than served from the memo. The session is also +/// short-lived by construction — it is created per deploy/validate operation and +/// discarded with it — so a cached graph can never outlive the operation that +/// captured it. That is the design's "template state is captured at the time of +/// flatten" guarantee, unchanged: the session narrows the capture window, it does +/// not widen it. +/// +/// +/// +/// Threading. A session is NOT thread-safe and must not be shared across +/// threads. It caches entities materialised by one scoped DbContext, which +/// is itself single-threaded, and it hands out tracked entities that the +/// in-flight bundle importer relies on observing. DeploySiteAsync honours +/// this by running the whole prepare (flatten) phase serially on one scope and +/// parallelising only the site round-trips, which touch no repository at all. +/// +/// +public sealed class FlattenSession +{ + private readonly ITemplateGraphWatermark _watermark; + + private readonly Dictionary _templates = new(); + private readonly Dictionary _chains = new(); + private readonly Dictionary Compositions)> _compositions = new(); + private readonly Dictionary> _dataConnections = new(); + + private IReadOnlyList? _sharedScripts; + private IReadOnlyDictionary? _schemaLibrary; + + /// Initializes a session bound to the process-wide graph watermark. + /// Watermark supplying the per-template versions used as cache keys. + public FlattenSession(ITemplateGraphWatermark watermark) + { + _watermark = watermark ?? throw new ArgumentNullException(nameof(watermark)); + } + + /// + /// Number of template-chain walks that actually hit the repository. Reset + /// never; used by tests to assert that N instances sharing a template load + /// that chain exactly once. + /// + public int ChainLoads { get; private set; } + + /// Number of single-template repository reads that missed the memo. + public int TemplateLoads { get; private set; } + + /// Number of per-template composition queries that missed the memo. + public int CompositionLoads { get; private set; } + + /// Number of times the session-global queries (shared scripts, schema library) ran. + public int GlobalLoads { get; private set; } + + /// Number of per-site data-connection queries that missed the memo. + public int DataConnectionLoads { get; private set; } + + /// + /// Every template id this session resolved — the union of all inheritance and + /// composed chains it walked. The staleness fast path records the watermark + /// version of exactly this set, so a change to ANY template that fed a + /// flatten invalidates the memoised revision hash. + /// + public IReadOnlyCollection VisitedTemplateIds => _templates.Keys; + + /// + /// Returns the inheritance chain for (the + /// template itself, then each ancestor), loading it through + /// on a miss. + /// + /// Root of the chain to build. + /// Repository read for one template with its children. + /// The chain, ordered leaf-first. + public async Task> GetChainAsync( + int templateId, + Func> loadTemplate) + { + var structure = _watermark.StructureVersion; + if (_chains.TryGetValue(templateId, out var cached) && IsChainCurrent(cached, structure)) + return cached.Chain; + + var chain = new List