diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Entities/Deployment.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Entities/Deployment.cs index 19d83728..d3044244 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Entities/Deployment.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Entities/Deployment.cs @@ -26,7 +26,11 @@ public sealed class Deployment public DateTime CreatedAtUtc { get; init; } = DateTime.UtcNow; /// Gets or sets the serialized artifact blob containing the configuration. - public byte[] ArtifactBlob { get; init; } = Array.Empty(); + /// + /// Settable (not init-only) so the retention sweep can EMPTY it on an old terminal row. An empty blob + /// is "no artifact" to every reader (#485), so the audit row survives with its bytes retired. + /// + public byte[] ArtifactBlob { get; set; } = Array.Empty(); /// Gets or sets the row version for optimistic concurrency control. public byte[] RowVersion { get; set; } = Array.Empty(); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Retention/DeploymentRetentionActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Retention/DeploymentRetentionActor.cs new file mode 100644 index 00000000..55ffd7dc --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Retention/DeploymentRetentionActor.cs @@ -0,0 +1,177 @@ +using Akka.Actor; +using Akka.Event; +using Microsoft.EntityFrameworkCore; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Retention; + +/// +/// Admin-role cluster singleton that periodically empties the ArtifactBlob of old, terminal +/// deployments (R03:1.2). See for the policy and +/// for the rule, which is pure over a context so it is unit-testable. +/// +/// +/// +/// Empties, never deletes, and never nulls. The column is NOT NULL varbinary(max), and +/// every reader (DriverHostActor.LoadArtifactBlob, OpcUaPublishActor, +/// DeploymentArtifactService) already treats a zero-length blob as "no artifact" (#485: empty +/// bytes are not an empty configuration — the apply fails and the node keeps last-known-good). An +/// empty blob therefore needs no schema change and no reader change, where a NULL would need both. +/// +/// +/// Never touches an in-flight row (Dispatching / AwaitingApplyAcks): a node may still be +/// fetching it. The newest terminal rows keep +/// their blob whatever their age, so the "latest sealed" every boot path reads is always intact. +/// +/// +public sealed class DeploymentRetentionActor : ReceiveActor, IWithTimers +{ + private readonly IDbContextFactory _dbFactory; + private readonly DeploymentRetentionOptions _options; + private readonly ILoggingAdapter _log = Context.GetLogger(); + private const string SweepTimerKey = "retention-sweep"; + + /// Gets or sets the timer scheduler. + public ITimerScheduler Timers { get; set; } = null!; + + /// Creates the actor props. + /// The config DB factory. + /// The retention policy. + /// The props. + public static Props Props(IDbContextFactory dbFactory, DeploymentRetentionOptions options) => + Akka.Actor.Props.Create(() => new DeploymentRetentionActor(dbFactory, options)); + + /// Initializes a new instance of the class. + /// The config DB factory. + /// The retention policy. + public DeploymentRetentionActor(IDbContextFactory dbFactory, DeploymentRetentionOptions options) + { + ArgumentNullException.ThrowIfNull(dbFactory); + ArgumentNullException.ThrowIfNull(options); + _dbFactory = dbFactory; + _options = options; + + Receive(_ => RunSweep()); + Receive(msg => + { + if (msg.Pruned > 0) + _log.Info("Deployment retention emptied the artifact blob on {Count} old deployment(s)", msg.Pruned); + else + _log.Debug("Deployment retention sweep found nothing to prune"); + if (msg.ReplyTo is not null) msg.ReplyTo.Tell(msg); + }); + Receive(msg => + { + _log.Warning(msg.Error, "Deployment retention sweep failed; the next sweep retries"); + if (msg.ReplyTo is not null) msg.ReplyTo.Tell(msg); + }); + } + + /// + protected override void PreStart() + { + if (!_options.Enabled) + { + _log.Info("Deployment retention is disabled (Deployments:Retention:Enabled=false); artifact blobs are kept forever"); + return; + } + var interval = TimeSpan.FromMinutes(Math.Max(1, _options.IntervalMinutes)); + var initial = TimeSpan.FromSeconds(Math.Max(0, _options.InitialDelaySeconds)); + Timers.StartPeriodicTimer(SweepTimerKey, Sweep.Instance, initial, interval); + } + + private void RunSweep() + { + var replyTo = Sender.IsNobody() ? null : Sender; + var self = Self; + var now = DateTime.UtcNow; + Task.Run(async () => + { + await using var db = await _dbFactory.CreateDbContextAsync().ConfigureAwait(false); + return await SweepAsync(db, _options, now).ConfigureAwait(false); + }) + .PipeTo(self, + success: pruned => new SweepCompleted(pruned, replyTo), + failure: ex => new SweepFailed(ex, replyTo)); + } + + /// + /// The retention rule. Empties the blob of every terminal deployment that is (a) older than the + /// newest terminal rows, (b) not referenced by a + /// NodeDeploymentState row started or applied within + /// , and (c) still carrying bytes — at most + /// per call. + /// + /// The config DB. + /// The policy. + /// The sweep's clock. + /// Cancellation. + /// How many rows were emptied. + public static async Task SweepAsync( + OtOpcUaConfigDbContext db, DeploymentRetentionOptions options, DateTime utcNow, CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(db); + ArgumentNullException.ThrowIfNull(options); + + var keep = Math.Max(1, options.KeepSealed); + var referenceFloor = utcNow.AddDays(-Math.Max(0, options.ReferencedDays)); + var batch = Math.Max(1, options.MaxRowsPerSweep); + + // Projected: ids only. The candidate scan must not materialise blobs for rows it will keep. + var terminal = await db.Deployments + .AsNoTracking() + .Where(d => d.Status == DeploymentStatus.Sealed + || d.Status == DeploymentStatus.PartiallyFailed + || d.Status == DeploymentStatus.TimedOut) + .OrderByDescending(d => d.CreatedAtUtc) + .Select(d => new { d.DeploymentId, HasBlob = d.ArtifactBlob.Length > 0 }) + .ToListAsync(ct) + .ConfigureAwait(false); + + var aged = terminal.Skip(keep).Where(d => d.HasBlob).Select(d => d.DeploymentId).ToList(); + if (aged.Count == 0) return 0; + + var referenced = (await db.NodeDeploymentStates + .AsNoTracking() + .Where(s => aged.Contains(s.DeploymentId) + && (s.StartedAtUtc >= referenceFloor + || (s.AppliedAtUtc != null && s.AppliedAtUtc >= referenceFloor))) + .Select(s => s.DeploymentId) + .Distinct() + .ToListAsync(ct) + .ConfigureAwait(false)) + .ToHashSet(); + + var targets = aged.Where(id => !referenced.Contains(id)).Take(batch).ToList(); + if (targets.Count == 0) return 0; + + // Tracked load of ONLY the rows being emptied: the update rides the row's real RowVersion on + // both providers, and the blob read is the one-time cost of retiring that row. + var rows = await db.Deployments + .Where(d => targets.Contains(d.DeploymentId)) + .ToListAsync(ct) + .ConfigureAwait(false); + foreach (var row in rows) row.ArtifactBlob = []; + await db.SaveChangesAsync(ct).ConfigureAwait(false); + return rows.Count; + } + + /// Asks for a sweep now; the sender (if any) receives the / . + public sealed class Sweep + { + /// The singleton instance. + public static readonly Sweep Instance = new(); + private Sweep() { } + } + + /// A sweep finished. + /// How many rows were emptied. + /// Who asked, when the sweep was requested rather than timer-driven. + public sealed record SweepCompleted(int Pruned, IActorRef? ReplyTo); + + /// A sweep threw. + /// The failure. + /// Who asked, when the sweep was requested rather than timer-driven. + public sealed record SweepFailed(Exception Error, IActorRef? ReplyTo); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Retention/DeploymentRetentionOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Retention/DeploymentRetentionOptions.cs new file mode 100644 index 00000000..6654ec1e --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Retention/DeploymentRetentionOptions.cs @@ -0,0 +1,54 @@ +using Microsoft.Extensions.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Retention; + +/// +/// Deployments:Retention — how long a sealed deployment keeps its ArtifactBlob (R03:1.2). +/// Every deploy stored the whole gzip'd fleet snapshot forever; nothing pruned it, so a year of routine +/// edits was gigabytes of blobs. The retention sweep EMPTIES the blob on old rows and never deletes a +/// row: the audit record (who deployed what revision, when, and how every node acked) survives. +/// +public sealed class DeploymentRetentionOptions +{ + /// The configuration section name. + public const string SectionName = "Deployments:Retention"; + + /// Master switch; keeps every blob forever (the previous behaviour). + public bool Enabled { get; set; } = true; + + /// + /// How many of the most recent terminal (Sealed / PartiallyFailed / TimedOut) deployments keep their + /// blob regardless of age. A node booting from central SQL reads the LATEST sealed row's blob, so this + /// must stay well above 1; 50 also keeps a comfortable re-deploy history. + /// + public int KeepSealed { get; set; } = 50; + + /// + /// A deployment any NodeDeploymentState row started or applied within this many days keeps its + /// blob even when older than the window — a node that just applied it may + /// still replay it on restart. + /// + public int ReferencedDays { get; set; } = 30; + + /// How often the sweep runs. + public int IntervalMinutes { get; set; } = 60; + + /// Delay before the first sweep after the singleton starts, so a boot is not a blob scan. + public int InitialDelaySeconds { get; set; } = 60; + + /// + /// Rows emptied per sweep. The sweep loads each candidate row (blob included) to update it under its + /// RowVersion, so this bounds one sweep's memory and transaction size; a backlog drains over sweeps. + /// + public int MaxRowsPerSweep { get; set; } = 200; + + /// Binds the section from configuration, with defaults for anything absent. + /// The app configuration; null yields the defaults. + /// The bound options. + public static DeploymentRetentionOptions Bind(IConfiguration? configuration) + { + var options = new DeploymentRetentionOptions(); + configuration?.GetSection(SectionName).Bind(options); + return options; + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs index cb8d605a..f838a004 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs @@ -15,6 +15,7 @@ using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using ZB.MOM.WW.OtOpcUa.ControlPlane.Fleet; using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Retention; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.ControlPlane; @@ -30,6 +31,7 @@ public static class ServiceCollectionExtensions public const string AuditWriterSingletonName = "audit-writer"; public const string RedundancyStateSingletonName = "redundancy-state"; public const string ClusterNodeAddressReconcilerSingletonName = "cluster-node-address-reconciler"; + public const string DeploymentRetentionSingletonName = "deployment-retention"; /// /// Registers the admin-role cluster singletons on the AkkaConfigurationBuilder — three with @@ -181,6 +183,21 @@ public static class ServiceCollectionExtensions singletonOptions, createProxyToo: false); + // Deployment artifact retention (R03:1.2): empties ArtifactBlob on old terminal deployments so the + // Deployment table stops growing by one fleet snapshot per edit. Admin singleton for the same + // reason as the reconciler (one sweeper per fleet, and only central holds the ConfigDb); + // proxy-less because nothing addresses it in production — a test sends Sweep directly. + builder.WithSingleton( + DeploymentRetentionSingletonName, + (system, registry, resolver) => + { + var dbFactory = resolver.GetService>(); + var options = DeploymentRetentionOptions.Bind(resolver.GetService()); + return DeploymentRetentionActor.Props(dbFactory, options); + }, + singletonOptions, + createProxyToo: false); + return builder; } @@ -256,4 +273,5 @@ public sealed class AdminOperationsActorKey { } public sealed class AuditWriterActorKey { } public sealed class RedundancyStateActorKey { } public sealed class ClusterNodeAddressReconcilerKey { } +public sealed class DeploymentRetentionKey { } public sealed class CentralCommunicationActorKey { } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/DeploymentRetentionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/DeploymentRetentionTests.cs new file mode 100644 index 00000000..ecd397c4 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/DeploymentRetentionTests.cs @@ -0,0 +1,174 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Configuration; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Retention; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests; + +/// +/// R03:1.2 — every deploy stored the whole fleet snapshot forever. The retention sweep empties the +/// blob on old terminal rows (never deletes a row, never touches an in-flight one, never the newest N, +/// never one a node applied recently). +/// +public sealed class DeploymentRetentionTests : ControlPlaneActorTestBase +{ + private static readonly DateTime SweepNow = new(2026, 10, 22, 12, 0, 0, DateTimeKind.Utc); + + [Fact] + public async Task Sweep_empties_blobs_beyond_the_keep_window_and_leaves_the_newest_intact() + { + var f = NewInMemoryDbFactory(); + var ids = Seed(f, sealedCount: 12); // ids[0] newest … ids[11] oldest + var options = new DeploymentRetentionOptions { KeepSealed = 5, ReferencedDays = 30, MaxRowsPerSweep = 200 }; + + int pruned; + await using (var db = f.CreateDbContext()) + pruned = await DeploymentRetentionActor.SweepAsync(db, options, SweepNow); + + pruned.ShouldBe(7); + await using var verify = f.CreateDbContext(); + foreach (var id in ids.Take(5)) + (await verify.Deployments.SingleAsync(d => d.DeploymentId == id)).ArtifactBlob.Length.ShouldBeGreaterThan(0, "newest 5 keep their blob"); + foreach (var id in ids.Skip(5)) + (await verify.Deployments.SingleAsync(d => d.DeploymentId == id)).ArtifactBlob.ShouldBeEmpty("older rows are emptied"); + // Rows survive: the audit record is intact. + (await verify.Deployments.CountAsync()).ShouldBe(12); + + // Idempotent: a second sweep finds nothing. + await using var again = f.CreateDbContext(); + (await DeploymentRetentionActor.SweepAsync(again, options, SweepNow)).ShouldBe(0); + } + + [Fact] + public async Task Sweep_keeps_a_row_a_node_applied_recently_and_never_touches_an_in_flight_row() + { + var f = NewInMemoryDbFactory(); + var ids = Seed(f, sealedCount: 4); + var inFlight = Guid.NewGuid(); + await using (var db = f.CreateDbContext()) + { + db.ClusterNodes.Add(new ClusterNode + { + NodeId = "site-a-1:4053", ClusterId = "SITE-A", Host = "site-a-1", + ApplicationUri = "urn:x", Enabled = true, CreatedBy = "test", + }); + // ids[3] is the oldest; a node applied it yesterday, so it stays. + db.NodeDeploymentStates.Add(new NodeDeploymentState + { + NodeId = "site-a-1:4053", DeploymentId = ids[3], Status = NodeDeploymentStatus.Applied, + StartedAtUtc = SweepNow.AddDays(-1), AppliedAtUtc = SweepNow.AddDays(-1), + }); + // ids[2]: applied long ago — not protected. + db.NodeDeploymentStates.Add(new NodeDeploymentState + { + NodeId = "site-a-1:4053", DeploymentId = ids[2], Status = NodeDeploymentStatus.Applied, + StartedAtUtc = SweepNow.AddDays(-90), AppliedAtUtc = SweepNow.AddDays(-90), + }); + // An ancient in-flight row with a blob — never touched. + db.Deployments.Add(new Deployment + { + DeploymentId = inFlight, RevisionHash = new string('f', 64), Status = DeploymentStatus.AwaitingApplyAcks, + CreatedBy = "test", CreatedAtUtc = SweepNow.AddDays(-400), ArtifactBlob = new byte[] { 1, 2, 3 }, + }); + await db.SaveChangesAsync(); + } + var options = new DeploymentRetentionOptions { KeepSealed = 1, ReferencedDays = 30 }; + + await using (var db = f.CreateDbContext()) + (await DeploymentRetentionActor.SweepAsync(db, options, SweepNow)).ShouldBe(2); // ids[1], ids[2] + + await using var verify = f.CreateDbContext(); + (await verify.Deployments.SingleAsync(d => d.DeploymentId == ids[0])).ArtifactBlob.ShouldNotBeEmpty("newest"); + (await verify.Deployments.SingleAsync(d => d.DeploymentId == ids[1])).ArtifactBlob.ShouldBeEmpty(); + (await verify.Deployments.SingleAsync(d => d.DeploymentId == ids[2])).ArtifactBlob.ShouldBeEmpty("applied 90 days ago"); + (await verify.Deployments.SingleAsync(d => d.DeploymentId == ids[3])).ArtifactBlob.ShouldNotBeEmpty("applied yesterday"); + (await verify.Deployments.SingleAsync(d => d.DeploymentId == inFlight)).ArtifactBlob.ShouldNotBeEmpty("in flight"); + } + + [Fact] + public async Task Sweep_is_bounded_per_call_and_drains_a_backlog_over_calls() + { + var f = NewInMemoryDbFactory(); + Seed(f, sealedCount: 10); + var options = new DeploymentRetentionOptions { KeepSealed = 2, MaxRowsPerSweep = 3 }; + + await using (var db = f.CreateDbContext()) + (await DeploymentRetentionActor.SweepAsync(db, options, SweepNow)).ShouldBe(3); + await using (var db = f.CreateDbContext()) + (await DeploymentRetentionActor.SweepAsync(db, options, SweepNow)).ShouldBe(3); + await using (var db = f.CreateDbContext()) + (await DeploymentRetentionActor.SweepAsync(db, options, SweepNow)).ShouldBe(2); + await using (var db = f.CreateDbContext()) + (await DeploymentRetentionActor.SweepAsync(db, options, SweepNow)).ShouldBe(0); + } + + /// The actor runs the same rule off its thread and reports back. + [Fact] + public void Actor_sweeps_on_request_and_reports_the_count() + { + var f = NewInMemoryDbFactory(); + Seed(f, sealedCount: 6); + var actor = Sys.ActorOf(DeploymentRetentionActor.Props(f, new DeploymentRetentionOptions + { + KeepSealed = 4, InitialDelaySeconds = 3600, IntervalMinutes = 60, + })); + + actor.Tell(DeploymentRetentionActor.Sweep.Instance); + + ExpectMsg(PresenceBudget).Pruned.ShouldBe(2); + using var verify = f.CreateDbContext(); + verify.Deployments.Count(d => d.ArtifactBlob.Length == 0).ShouldBe(2); + } + + [Fact] + public void Options_bind_from_configuration_with_defaults() + { + var defaults = DeploymentRetentionOptions.Bind(null); + defaults.Enabled.ShouldBeTrue(); + defaults.KeepSealed.ShouldBe(50); + defaults.ReferencedDays.ShouldBe(30); + defaults.IntervalMinutes.ShouldBe(60); + + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["Deployments:Retention:KeepSealed"] = "7", + ["Deployments:Retention:Enabled"] = "false", + }) + .Build(); + var bound = DeploymentRetentionOptions.Bind(config); + bound.KeepSealed.ShouldBe(7); + bound.Enabled.ShouldBeFalse(); + bound.ReferencedDays.ShouldBe(30); + } + + /// Seeds sealed deployments, newest first; each one hour older than the previous. + private static List Seed(IDbContextFactory f, int sealedCount) + { + var ids = new List(); + using var db = f.CreateDbContext(); + for (var i = 0; i < sealedCount; i++) + { + var id = Guid.NewGuid(); + ids.Add(id); + db.Deployments.Add(new Deployment + { + DeploymentId = id, + RevisionHash = new string('a', 64), + Status = i % 3 == 2 ? DeploymentStatus.TimedOut : DeploymentStatus.Sealed, + CreatedBy = "test", + CreatedAtUtc = SweepNow.AddHours(-i), + SealedAtUtc = SweepNow.AddHours(-i), + ArtifactBlob = new byte[] { 0x1f, 0x8b, (byte)i }, + }); + } + db.SaveChanges(); + return ids; + } +}