fix(recycle): prune expired Admin/_Recycle requests with their reports (#6813)

src/MeshWeaver.Documentation/Data/Architecture/StaleStateUntilRecycle.md

@@ -220,6 +220,30 @@ bound NodeType is T"* (optionally narrowed to some addresses, or to the activati
 - **Only System's requests are acted on.** Each process re-validates what it reads and refuses a
   request node not created by System — a platform admin, who holds `Admin`, cannot bypass the checks
   by writing `Admin/_Recycle` directly.
+- **The audit is kept for a declared window, then pruned (#6813).** Every NodeType definition recycle
+  issues a request, so without a retention the rows under `Admin/_Recycle` grow with every recycle,
+  module landing and deploy. `ActivationRecycleRetention` is the policy, read from configuration:
+
+  | Key | Default | Meaning |
+  |---|---|---|
+  | `ActivationRecycle:Retention:Enabled` | `true` | `false` turns the prune off (no query is issued) |
+  | `ActivationRecycle:Retention:MaxAge` | `30.00:00:00` | a request untouched for this long is removed with its reports; a configured value below one day — zero and negative included — is clamped to one day; only an absent or unparseable value keeps the default |
+  | `ActivationRecycle:Retention:MaxDeletionsPerRun` | `50` | the most requests one prune removes, so a backlog drains over several requests |
+
+  Thirty days is long enough that the audit of a recent deploy or incident is still there when it
+  is read, whether for a post-incident review or to check which silos recycled after last week's
+  release, and it reaches back a full monthly release cycle. It is short enough that the rows stay
+  bounded by one month of recycle traffic rather than the portal's whole history. **The writer
+  prunes, and no timer does.** Right after `ActivationRecycle.Request` writes a new request it
+  removes the expired ones, as System, through `DeleteNode`. The delete cascades, so each
+  `{request}/{process}` report leaves with its request. This is the same owner-prunes-on-write shape
+  the compile activities use (`CompileActivityRetention`, in
+  [NodeType Compilation](../NodeTypeCompilation)). The rows grow only through that write, so pruning
+  on it bounds them. The request just written is never pruned. The decision is fail-closed: a row
+  that is undated, not a request, or not directly under `Admin/_Recycle` is kept. A failed listing
+  or delete costs that prune only. It is logged, and the next request retries; the caller's ticket
+  never depends on it — nor waits for it: the prune is a subscription of its own, started once the
+  request is committed, so the ticket is answered as soon as the write lands.
 
 - **A recycle of a NodeType DEFINITION is one of these requests (#6520).** A routed `DisposeRequest`
   on a definition hub (the direct `recycle` verb on a NodeType, a package install, a control-lane
@@ -242,7 +266,10 @@ not the admin, holds the rights. The direct `recycle` verb above is unchanged an
 
 Pinned by `ActivationRecycleTest` (monolith: two live activations of T and one of U, exactly the two
 disposed, re-activation on a fresh hub, redelivery, refusals) and `TwoSiloActivationRecycleTest` (both silos' activations,
-one report per silo whose counts match what each hosted).
+one report per silo whose counts match what each hosted). The retention is pinned by
+`ActivationRecycleRetentionTest`: an expired request goes with its report when the next one is
+written, a request inside the window survives with its report, and the decision is oldest first,
+capped and fail-closed.
 
 ### Three rules for posting one, each paid for
 

src/MeshWeaver.Graph/Configuration/ActivationRecycle.cs

@@ -7,6 +7,7 @@
 using MeshWeaver.Mesh.Security;
 using MeshWeaver.Mesh.Services;
 using MeshWeaver.Messaging;
+using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 
@@ -129,6 +130,10 @@ public static TBuilder AddActivationRecycle<TBuilder>(this TBuilder builder) whe
                 HubConfiguration = config => config
                     .AddMeshDataSource(source => source.WithContentType<ActivationRecycleReport>()),
             });
+        // Mesh-scoped singleton, read once from the host's configuration: how long a request and its
+        // reports are kept (Doc/Architecture/StaleStateUntilRecycle → "Recycling a type").
+        builder.ConfigureServices(services => services
+            .AddSingleton(sp => ActivationRecycleRetention.FromConfiguration(sp.GetService<IConfiguration>())));
         builder.ConfigureHub(c => c
             .WithType<ActivationRecycleRequest>(nameof(ActivationRecycleRequest))
             .WithType<ActivationRecycleReport>(nameof(ActivationRecycleReport))
@@ -228,14 +233,94 @@ IObservable<bool> TypeExists(string type) =>
                     Content = request with { RequestedAt = request.RequestedAt == default ? DateTimeOffset.UtcNow : request.RequestedAt },
                 };
                 // The id is the dedupe key: an existing request is the answer, never a second write.
+                // A NEW request is the moment the rows grow, so it is also when the expired ones go
+                // (ActivationRecycleRetention): the writer prunes, no timer does. The prune is its
+                // OWN subscription, started once the request is committed: the ticket is answered
+                // the moment the write lands and never waits for the listing or the deletes.
                 return Exists().SelectMany(there => there
                     ? Observable.Return(new ActivationRecycleTicket(path, null))
                     : access.RunAsSystem(() => mesh.CreateNode(node).Take(1))
+                        .Do(_ => StartPrune(hub, path))
                         .Select(_ => new ActivationRecycleTicket(path, null)));
             })
             .Take(1);
     }
 
+    /// <summary>
+    /// Starts the prune that follows a newly committed request as a subscription of its own, so the
+    /// request's ticket never waits for it. <see cref="Prune"/> already maps every fault to a logged
+    /// zero; the error handler here is the backstop for anything that escapes it.
+    /// </summary>
+    private static void StartPrune(IMessageHub hub, string keepPath)
+    {
+        var logger = hub.ServiceProvider.GetService<ILoggerFactory>()?.CreateLogger(typeof(ActivationRecycle));
+        Prune(hub, hub.ServiceProvider.GetRequiredService<ActivationRecycleRetention>(), DateTimeOffset.UtcNow, keepPath)
+            .Subscribe(
+                _ => { },
+                ex => logger?.LogWarning(ex,
+                    "[ActivationRecycle] retention: the prune after {Path} faulted; expired requests stay until the next request prunes them",
+                    keepPath));
+    }
+
+    /// <summary>
+    /// Removes the requests <paramref name="retention"/> calls expired at <paramref name="now"/>,
+    /// each with its per-process reports, never <paramref name="keepPath"/>. Cold; emits once, the
+    /// number of requests removed.
+    ///
+    /// <para>The listing is a QUERY, which is a valid use here: it lists children, and a stale answer
+    /// is harmless — a row the index has not seen yet is a NEWER row and is kept, a row already gone
+    /// is an idempotent delete. Each removal goes through <see cref="IMeshService.DeleteNode"/>,
+    /// which cascades, so the <c>{request}/{process}</c> reports leave with their request. It runs as
+    /// System, as the request's own create did: infrastructure acting on records it wrote.</para>
+    ///
+    /// <para>A failed removal is per ROW, and a failed listing costs this prune only: each is logged,
+    /// the rows stay, and the next request's prune tries again. The request that triggered the prune
+    /// is already written and acted on, so its caller's ticket never depends on the prune.</para>
+    /// </summary>
+    /// <param name="hub">Any surviving hub of the mesh.</param>
+    /// <param name="retention">The policy to prune to.</param>
+    /// <param name="now">The cutoff instant.</param>
+    /// <param name="keepPath">The request just written; always kept.</param>
+    public static IObservable<int> Prune(IMessageHub hub, ActivationRecycleRetention retention, DateTimeOffset now, string? keepPath = null)
+    {
+        if (!retention.Enabled)
+            return Observable.Return(0);
+        var mesh = hub.ServiceProvider.GetRequiredService<IMeshService>();
+        var access = hub.ServiceProvider.GetRequiredService<AccessService>();
+        var logger = hub.ServiceProvider.GetService<ILoggerFactory>()?.CreateLogger(typeof(ActivationRecycle));
+        return access.RunAsSystem(() => mesh.Query<MeshNode>(MeshQueryRequest.FromQuery(retention.WindowQuery))
+                .Where(change => change.ChangeType == QueryChangeType.Initial)
+                .Take(1)
+                .Timeout(ReadBudget))
+            .Select(change => retention.SelectForPruning(change.Items, now, keepPath))
+            .SelectMany(paths =>
+            {
+                if (paths.IsEmpty)
+                    return Observable.Return(0);
+                // Information on purpose: the platform is removing audit rows, and an operator must be
+                // able to see that. One line per prune, and only when something is removed.
+                logger?.LogInformation(
+                    "[ActivationRecycle] retention: removing {Count} recycle request(s) older than {MaxAge}, with their reports",
+                    paths.Count, retention.MaxAge);
+                return paths
+                    .Select(p => access.RunAsSystem(() => mesh.DeleteNode(p))
+                        .Take(1)
+                        .Select(removed => removed ? 1 : 0)
+                        .Catch((Exception ex) =>
+                        {
+                            logger?.LogWarning(ex, "[ActivationRecycle] retention: {Path} was not removed; the next request's prune retries it", p);
+                            return Observable.Return(0);
+                        }))
+                    .Concat()
+                    .Sum();
+            })
+            .Catch((Exception ex) =>
+            {
+                logger?.LogWarning(ex, "[ActivationRecycle] retention: the request window could not be read; nothing was pruned, the next request retries");
+                return Observable.Return(0);
+            });
+    }
+
     /// <summary>The reports written so far for a request, read as System. Cold; one snapshot.</summary>
     public static IObservable<ImmutableList<ActivationRecycleReport>> Reports(IMessageHub hub, string requestPath)
     {

src/MeshWeaver.Graph/Configuration/ActivationRecycleRetention.cs

@@ -0,0 +1,146 @@
+using System.Collections.Immutable;
+using MeshWeaver.Mesh;
+using Microsoft.Extensions.Configuration;
+
+namespace MeshWeaver.Graph.Configuration;
+
+/// <summary>
+/// How long an <see cref="ActivationRecycleRequest"/> at <c>Admin/_Recycle/{id}</c> — and with it
+/// every per-process <see cref="ActivationRecycleReport"/> under it — is kept, and how many one
+/// prune may remove. The whole policy: the selection <see cref="SelectForPruning"/> makes and the
+/// bounds it works within, with no mesh, no hub and no clock of its own.
+///
+/// <para><b>Why requests get a retention.</b> A request plus its reports is the audit of a recycle:
+/// who asked, when, why, which types, and what each process disposed. Nothing ever removed one, and
+/// once every NodeType definition recycle issues a request, the rows grow with every recycle, module
+/// landing and deploy (Systemorph/MeshWeaver#6813).</para>
+///
+/// <para><b>Who prunes.</b> The writer does, at the moment it writes a new request:
+/// <see cref="ActivationRecycle.Request"/> prunes right after its create, the same owner-prunes-on-write
+/// shape as <c>CompileActivityRetention</c>. No timer and no sweeper: the rows grow only through
+/// that write, so pruning on it bounds them, and a mesh that stops recycling stops growing. Each
+/// prune is capped at <see cref="MaxDeletionsPerRun"/>, so a backlog drains over successive
+/// requests.</para>
+///
+/// <para>Declared, not derived: every bound comes from configuration
+/// (<see cref="FromConfiguration"/>) with the shipped values as the fallback.</para>
+/// </summary>
+public sealed record ActivationRecycleRetention
+{
+    /// <summary>Config key turning the prune off entirely. Default <c>true</c>.</summary>
+    public const string EnabledConfigKey = "ActivationRecycle:Retention:Enabled";
+
+    /// <summary>Config key overriding <see cref="MaxAge"/> (a <see cref="TimeSpan"/> string, e.g. <c>30.00:00:00</c>).</summary>
+    public const string MaxAgeConfigKey = "ActivationRecycle:Retention:MaxAge";
+
+    /// <summary>Config key overriding <see cref="MaxDeletionsPerRun"/>.</summary>
+    public const string MaxDeletionsPerRunConfigKey = "ActivationRecycle:Retention:MaxDeletionsPerRun";
+
+    /// <summary>
+    /// The floor <see cref="FromConfiguration"/> clamps a CONFIGURED <see cref="MaxAge"/> to. A
+    /// window of zero in a chart is a typo, not a request to delete the audit of the recycle that
+    /// just ran. The clamp lives at the configuration edge only; a directly constructed policy is
+    /// applied exactly as written.
+    /// </summary>
+    public static readonly TimeSpan MinimumMaxAge = TimeSpan.FromDays(1);
+
+    /// <summary>The shipped policy: 30 days, 50 removals per prune, enabled.</summary>
+    public static readonly ActivationRecycleRetention Default = new();
+
+    /// <summary>Whether a prune removes anything at all. Off means no query is issued.</summary>
+    public bool Enabled { get; init; } = true;
+
+    /// <summary>
+    /// A request (and its reports) is removed once its node has not been written for this long.
+    ///
+    /// <para>Thirty days: long enough that the audit of a recent deploy or incident is still there
+    /// when someone reads it — a post-incident review, or the question "which silos actually
+    /// recycled after last week's release" — and a full monthly release cycle back. Short enough
+    /// that the rows stay bounded by one month of recycle traffic instead of the portal's whole
+    /// history.</para>
+    /// </summary>
+    public TimeSpan MaxAge { get; init; } = TimeSpan.FromDays(30);
+
+    /// <summary>
+    /// The cap on how many requests ONE prune removes (each with its reports). A backlog drains by
+    /// this many per new request, instead of one request turning into a mass delete.
+    /// </summary>
+    public int MaxDeletionsPerRun { get; init; } = 50;
+
+    /// <summary>
+    /// The query that reads the oldest window of requests. It bounds the work; the decision is
+    /// <see cref="SelectForPruning"/>'s, so a backend that ignored the sort would only drain slower,
+    /// never delete a row the policy has not called expired. One more than the cap, so the request
+    /// the caller just wrote never crowds out a row that may go.
+    /// </summary>
+    public string WindowQuery
+        => $"path:{ActivationRecycleRequest.Namespace} scope:children nodeType:{ActivationRecycleRequest.NodeType} "
+           + $"sort:LastModified-asc limit:{MaxDeletionsPerRun + 1}";
+
+    /// <summary>
+    /// Whether <paramref name="node"/> is a request this policy removes at <paramref name="now"/>.
+    /// The clock is <see cref="MeshNode.LastModified"/> — the quantity the window query orders by.
+    ///
+    /// <para>🚨 Fail closed: not a request, not directly under <c>Admin/_Recycle</c>, or undated ⇒
+    /// kept. This decides what is deleted, so every uncertainty resolves to keeping the row.</para>
+    /// </summary>
+    /// <param name="node">The candidate row.</param>
+    /// <param name="now">The current time, passed in so the rule is a pure function.</param>
+    public bool IsExpired(MeshNode? node, DateTimeOffset now)
+    {
+        if (!Enabled || node is null)
+            return false;
+        if (!string.Equals(node.NodeType, ActivationRecycleRequest.NodeType, StringComparison.OrdinalIgnoreCase))
+            return false;
+        if (!string.Equals(node.Namespace?.Trim('/'), ActivationRecycleRequest.Namespace, StringComparison.OrdinalIgnoreCase))
+            return false;
+        if (node.LastModified == default)
+            return false;
+        return node.LastModified <= now - MaxAge;
+    }
+
+    /// <summary>
+    /// The request paths among <paramref name="requests"/> this policy removes at
+    /// <paramref name="now"/>: expired ones, oldest first, never <paramref name="keepPath"/>, at most
+    /// <see cref="MaxDeletionsPerRun"/>.
+    /// </summary>
+    /// <param name="requests">Candidate rows, in any order.</param>
+    /// <param name="now">The current time.</param>
+    /// <param name="keepPath">The request the caller just wrote; never selected.</param>
+    public ImmutableList<string> SelectForPruning(IEnumerable<MeshNode> requests, DateTimeOffset now, string? keepPath = null)
+        => !Enabled
+            ? ImmutableList<string>.Empty
+            : requests
+                .Where(n => IsExpired(n, now))
+                .Where(n => !string.Equals(n.Path, keepPath, StringComparison.OrdinalIgnoreCase))
+                .OrderBy(n => n.LastModified)
+                .ThenBy(n => n.Id, StringComparer.Ordinal)
+                .Select(n => n.Path)
+                .Distinct(StringComparer.OrdinalIgnoreCase)
+                .Take(MaxDeletionsPerRun)
+                .ToImmutableList();
+
+    /// <summary>
+    /// Reads the policy from configuration. Every key falls back to its default when absent or
+    /// malformed; a configured window below <see cref="MinimumMaxAge"/> is clamped to it, so a typo
+    /// never widens what gets deleted.
+    /// </summary>
+    /// <param name="configuration">The host's configuration, or null on a host that has none.</param>
+    public static ActivationRecycleRetention FromConfiguration(IConfiguration? configuration)
+    {
+        var retention = Default;
+        if (configuration is null)
+            return retention;
+
+        if (bool.TryParse(configuration[EnabledConfigKey], out var enabled))
+            retention = retention with { Enabled = enabled };
+        // Every PARSED span is clamped, zero and negative included: only an absent or unparseable
+        // value falls back to the default.
+        if (TimeSpan.TryParse(configuration[MaxAgeConfigKey], System.Globalization.CultureInfo.InvariantCulture, out var maxAge))
+            retention = retention with { MaxAge = maxAge < MinimumMaxAge ? MinimumMaxAge : maxAge };
+        if (int.TryParse(configuration[MaxDeletionsPerRunConfigKey], out var max) && max >= 1)
+            retention = retention with { MaxDeletionsPerRun = max };
+
+        return retention;
+    }
+}