feat(controlplane): deployment artifact retention on the admin singleton

Every deploy stored the whole gzip'd fleet snapshot in Deployment.ArtifactBlob forever (R03:1.2).
DeploymentRetentionActor (admin singleton, proxy-less) sweeps every Deployments:Retention:
IntervalMinutes (60): terminal rows (Sealed / PartiallyFailed / TimedOut) older than the newest
KeepSealed (50) and not referenced by a NodeDeploymentState started/applied within ReferencedDays
(30) get their blob EMPTIED — never deleted, never NULLed: the column stays NOT NULL and every
reader already treats zero bytes as "no artifact" (#485), so no schema migration and no reader
change. In-flight rows are never touched; MaxRowsPerSweep (200) bounds one sweep.

Refs #733

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 10:27:59 -04:00
parent f72af76e99
commit 1bf5ef916c
5 changed files with 428 additions and 1 deletions
@@ -26,7 +26,11 @@ public sealed class Deployment
public DateTime CreatedAtUtc { get; init; } = DateTime.UtcNow;
/// <summary>Gets or sets the serialized artifact blob containing the configuration.</summary>
public byte[] ArtifactBlob { get; init; } = Array.Empty<byte>();
/// <remarks>
/// 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.
/// </remarks>
public byte[] ArtifactBlob { get; set; } = Array.Empty<byte>();
/// <summary>Gets or sets the row version for optimistic concurrency control.</summary>
public byte[] RowVersion { get; set; } = Array.Empty<byte>();
@@ -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;
/// <summary>
/// Admin-role cluster singleton that periodically empties the <c>ArtifactBlob</c> of old, terminal
/// deployments (R03:1.2). See <see cref="DeploymentRetentionOptions"/> for the policy and
/// <see cref="SweepAsync"/> for the rule, which is pure over a context so it is unit-testable.
/// </summary>
/// <remarks>
/// <para>
/// <b>Empties, never deletes, and never nulls.</b> The column is <c>NOT NULL varbinary(max)</c>, and
/// every reader (<c>DriverHostActor.LoadArtifactBlob</c>, <c>OpcUaPublishActor</c>,
/// <c>DeploymentArtifactService</c>) 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.
/// </para>
/// <para>
/// <b>Never touches an in-flight row</b> (Dispatching / AwaitingApplyAcks): a node may still be
/// fetching it. The newest <see cref="DeploymentRetentionOptions.KeepSealed"/> terminal rows keep
/// their blob whatever their age, so the "latest sealed" every boot path reads is always intact.
/// </para>
/// </remarks>
public sealed class DeploymentRetentionActor : ReceiveActor, IWithTimers
{
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
private readonly DeploymentRetentionOptions _options;
private readonly ILoggingAdapter _log = Context.GetLogger();
private const string SweepTimerKey = "retention-sweep";
/// <summary>Gets or sets the timer scheduler.</summary>
public ITimerScheduler Timers { get; set; } = null!;
/// <summary>Creates the actor props.</summary>
/// <param name="dbFactory">The config DB factory.</param>
/// <param name="options">The retention policy.</param>
/// <returns>The props.</returns>
public static Props Props(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, DeploymentRetentionOptions options) =>
Akka.Actor.Props.Create(() => new DeploymentRetentionActor(dbFactory, options));
/// <summary>Initializes a new instance of the <see cref="DeploymentRetentionActor"/> class.</summary>
/// <param name="dbFactory">The config DB factory.</param>
/// <param name="options">The retention policy.</param>
public DeploymentRetentionActor(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, DeploymentRetentionOptions options)
{
ArgumentNullException.ThrowIfNull(dbFactory);
ArgumentNullException.ThrowIfNull(options);
_dbFactory = dbFactory;
_options = options;
Receive<Sweep>(_ => RunSweep());
Receive<SweepCompleted>(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<SweepFailed>(msg =>
{
_log.Warning(msg.Error, "Deployment retention sweep failed; the next sweep retries");
if (msg.ReplyTo is not null) msg.ReplyTo.Tell(msg);
});
}
/// <inheritdoc />
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));
}
/// <summary>
/// The retention rule. Empties the blob of every terminal deployment that is (a) older than the
/// newest <see cref="DeploymentRetentionOptions.KeepSealed"/> terminal rows, (b) not referenced by a
/// <c>NodeDeploymentState</c> row started or applied within
/// <see cref="DeploymentRetentionOptions.ReferencedDays"/>, and (c) still carrying bytes — at most
/// <see cref="DeploymentRetentionOptions.MaxRowsPerSweep"/> per call.
/// </summary>
/// <param name="db">The config DB.</param>
/// <param name="options">The policy.</param>
/// <param name="utcNow">The sweep's clock.</param>
/// <param name="ct">Cancellation.</param>
/// <returns>How many rows were emptied.</returns>
public static async Task<int> 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;
}
/// <summary>Asks for a sweep now; the sender (if any) receives the <see cref="SweepCompleted"/> / <see cref="SweepFailed"/>.</summary>
public sealed class Sweep
{
/// <summary>The singleton instance.</summary>
public static readonly Sweep Instance = new();
private Sweep() { }
}
/// <summary>A sweep finished.</summary>
/// <param name="Pruned">How many rows were emptied.</param>
/// <param name="ReplyTo">Who asked, when the sweep was requested rather than timer-driven.</param>
public sealed record SweepCompleted(int Pruned, IActorRef? ReplyTo);
/// <summary>A sweep threw.</summary>
/// <param name="Error">The failure.</param>
/// <param name="ReplyTo">Who asked, when the sweep was requested rather than timer-driven.</param>
public sealed record SweepFailed(Exception Error, IActorRef? ReplyTo);
}
@@ -0,0 +1,54 @@
using Microsoft.Extensions.Configuration;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Retention;
/// <summary>
/// <c>Deployments:Retention</c> — how long a sealed deployment keeps its <c>ArtifactBlob</c> (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.
/// </summary>
public sealed class DeploymentRetentionOptions
{
/// <summary>The configuration section name.</summary>
public const string SectionName = "Deployments:Retention";
/// <summary>Master switch; <see langword="false"/> keeps every blob forever (the previous behaviour).</summary>
public bool Enabled { get; set; } = true;
/// <summary>
/// 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.
/// </summary>
public int KeepSealed { get; set; } = 50;
/// <summary>
/// A deployment any <c>NodeDeploymentState</c> row started or applied within this many days keeps its
/// blob even when older than the <see cref="KeepSealed"/> window — a node that just applied it may
/// still replay it on restart.
/// </summary>
public int ReferencedDays { get; set; } = 30;
/// <summary>How often the sweep runs.</summary>
public int IntervalMinutes { get; set; } = 60;
/// <summary>Delay before the first sweep after the singleton starts, so a boot is not a blob scan.</summary>
public int InitialDelaySeconds { get; set; } = 60;
/// <summary>
/// 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.
/// </summary>
public int MaxRowsPerSweep { get; set; } = 200;
/// <summary>Binds the section from configuration, with defaults for anything absent.</summary>
/// <param name="configuration">The app configuration; null yields the defaults.</param>
/// <returns>The bound options.</returns>
public static DeploymentRetentionOptions Bind(IConfiguration? configuration)
{
var options = new DeploymentRetentionOptions();
configuration?.GetSection(SectionName).Bind(options);
return options;
}
}
@@ -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";
/// <summary>
/// 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<DeploymentRetentionKey>(
DeploymentRetentionSingletonName,
(system, registry, resolver) =>
{
var dbFactory = resolver.GetService<IDbContextFactory<OtOpcUaConfigDbContext>>();
var options = DeploymentRetentionOptions.Bind(resolver.GetService<IConfiguration>());
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 { }
@@ -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;
/// <summary>
/// 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).
/// </summary>
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);
}
/// <summary>The actor runs the same rule off its thread and reports back.</summary>
[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<DeploymentRetentionActor.SweepCompleted>(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<string, string?>
{
["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);
}
/// <summary>Seeds sealed deployments, newest first; each one hour older than the previous.</summary>
private static List<Guid> Seed(IDbContextFactory<OtOpcUaConfigDbContext> f, int sealedCount)
{
var ids = new List<Guid>();
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;
}
}