feat!: make queue and reconciliation leader-loss-aware

docs/docs/operator/building-blocks/controllers.mdx

@@ -194,8 +194,9 @@ public class V1DemoEntityController(
         V1DemoEntity entity,
         CancellationToken cancellationToken)
     {
-        // Schedule a follow-up reconciliation in 30 seconds without returning a result
-        queue(entity, ReconciliationType.Modified, ReconciliationTriggerSource.Operator,
+        // Schedule a follow-up reconciliation in 30 seconds without returning a result.
+        // The delegate returns Task<bool>; await it (false means the enqueue was dropped).
+        await queue(entity, ReconciliationType.Modified, ReconciliationTriggerSource.Operator,
             TimeSpan.FromSeconds(30), retryCount: 0, cancellationToken);
 
         return ReconciliationResult<V1DemoEntity>.Success(entity);

docs/docs/operator/caching.mdx

@@ -82,6 +82,18 @@ provided that configures the **active** cache (whichever strategy is selected).
 caches individually, register named FusionCache instances directly on the DI container using the
 FusionCache builder API.
 
+:::warning[Tagging must stay enabled]
+The resource watcher tags every deduplication entry with its entity type so that a leadership-aware
+watcher can, on leadership loss, drop only that type's entries via FusionCache's `RemoveByTag` (without
+wiping the dedup tokens of other entity types that share the same named cache). FusionCache tagging is
+enabled by default — if you supply a custom cache configuration, do **not** disable it
+(`FusionCacheOptions.DisableTagging`), otherwise the tagged writes throw at runtime.
+
+If you enable `OperatorSettings.ValidateRegistrations`, this is checked at startup: a cache with tagging
+disabled aborts host startup with a clear `InvalidRegistrationException` instead of failing later at
+runtime.
+:::
+
 **Example: L2 cache for the default `ByGeneration` strategy**
 
 ```csharp

examples/OtelOperator/Program.cs

@@ -7,8 +7,6 @@
 using KubeOps.Operator.Web.Builder;
 
 using OpenTelemetry;
-using OpenTelemetry.Metrics;
-using OpenTelemetry.Trace;
 
 const string operatorName = "otel-operator";
 

src/KubeOps.Abstractions/Builder/OperatorSettings.cs

@@ -128,4 +128,18 @@ public sealed record OperatorSettings
     /// scrape the metrics, register an OpenTelemetry exporter for the meter named <see cref="Name"/>.
     /// </remarks>
     public required bool EnableMetrics { get; init; }
+
+    /// <summary>
+    /// Indicates whether the operator validates, on host startup, that its dependency injection
+    /// registrations are complete and consistent with the configured
+    /// <see cref="LeaderElectionType"/> and <see cref="QueueStrategy"/>. Disabled by default.
+    /// </summary>
+    /// <remarks>
+    /// When enabled, the operator verifies — for every managed entity — that the components implied by
+    /// the configuration are registered. If anything is missing, host startup aborts with an
+    /// <c>InvalidRegistrationException</c> listing the gaps. This catches registration mistakes
+    /// (for example a forgotten watcher or queue consumer in a manually wired setup) that would
+    /// otherwise let the operator start without processing any resources.
+    /// </remarks>
+    public bool ValidateRegistrations { get; init; }
 }

src/KubeOps.Abstractions/Builder/OperatorSettingsBuilder.cs

@@ -113,6 +113,13 @@ public sealed partial class OperatorSettingsBuilder
     /// </summary>
     public bool EnableMetrics { get; set; } = true;
 
+    /// <summary>
+    /// Indicates whether the operator validates, on host startup, that its dependency injection
+    /// registrations are complete and consistent with the configuration. Disabled by default. See
+    /// <see cref="OperatorSettings.ValidateRegistrations"/> for details.
+    /// </summary>
+    public bool ValidateRegistrations { get; set; }
+
     /// <summary>
     /// Produces an immutable <see cref="OperatorSettings"/> record from the current configuration.
     /// </summary>
@@ -132,6 +139,7 @@ public sealed partial class OperatorSettingsBuilder
         ReconcileStrategy = ReconcileStrategy,
         ParallelReconciliation = ParallelReconciliation.Build(),
         EnableMetrics = EnableMetrics,
+        ValidateRegistrations = ValidateRegistrations,
     };
 
     [GeneratedRegex(@"(\W|_)", RegexOptions.CultureInvariant)]

src/KubeOps.Abstractions/Builder/OperatorSettingsBuilderExtensions.cs

@@ -161,6 +161,21 @@ public static OperatorSettingsBuilder WithMetrics(
         return builder;
     }
 
+    /// <summary>
+    /// Sets whether the operator validates, on host startup, that its dependency injection registrations
+    /// are complete and consistent with the configuration. Disabled by default. See
+    /// <see cref="OperatorSettings.ValidateRegistrations"/> for details.
+    /// </summary>
+    /// <param name="builder">The builder to configure.</param>
+    /// <param name="value"><c>true</c> to enable validation (default); <c>false</c> to disable.</param>
+    /// <returns>The same <paramref name="builder"/> instance for chaining.</returns>
+    public static OperatorSettingsBuilder WithRegistrationValidation(
+        this OperatorSettingsBuilder builder, bool value = true)
+    {
+        builder.ValidateRegistrations = value;
+        return builder;
+    }
+
     /// <summary>Configures parallel reconciliation settings inline via a delegate.</summary>
     /// <param name="builder">The builder to configure.</param>
     /// <param name="configure">An action that configures the <see cref="ParallelReconciliationSettingsBuilder"/>.</param>

src/KubeOps.Abstractions/Reconciliation/Queue/EntityQueue.cs

@@ -36,6 +36,12 @@ namespace KubeOps.Abstractions.Reconciliation.Queue;
 /// <param name="cancellationToken">
 /// A token to monitor for cancellation requests while waiting for the queue duration to elapse.
 /// </param>
+/// <returns>
+/// A task whose result is <see langword="true"/> if the entity was scheduled, or <see langword="false"/> if it
+/// was dropped (for example because <paramref name="cancellationToken"/> was already cancelled, or the queue's
+/// intake is suspended on a leadership transition). Await it to observe failures of a custom queue and to react
+/// to a dropped enqueue.
+/// </returns>
 /// <remarks>
 /// <para>
 /// This delegate is injected into controllers and other components via dependency injection to enable
@@ -62,6 +68,6 @@ namespace KubeOps.Abstractions.Reconciliation.Queue;
 /// <seealso cref="ReconciliationType"/>
 /// <seealso cref="ReconciliationTriggerSource"/>
 /// <seealso cref="IEntityQueueFactory"/>
-public delegate void EntityQueue<in TEntity>(
+public delegate Task<bool> EntityQueue<in TEntity>(
     TEntity entity, ReconciliationType type, ReconciliationTriggerSource reconciliationTriggerSource, TimeSpan queueIn, int retryCount, CancellationToken cancellationToken)
     where TEntity : IKubernetesObject<V1ObjectMeta>;

src/KubeOps.Operator/Builder/CacheExtensions.cs

@@ -27,9 +27,7 @@ internal static class CacheExtensions
     /// <returns>The modified service collection with resource watcher caching configured.</returns>
     internal static IServiceCollection WithResourceWatcherEntityCaching(this IServiceCollection services, OperatorSettings settings)
     {
-        var cacheName = settings.ReconcileStrategy == ReconcileStrategy.ByResourceVersion
-            ? CacheConstants.CacheNames.ResourceWatcherByResourceVersion
-            : CacheConstants.CacheNames.ResourceWatcher;
+        var cacheName = CacheConstants.ResourceWatcherCacheNameFor(settings.ReconcileStrategy);
 
         var builder = services.AddFusionCache(cacheName);
 

src/KubeOps.Operator/Builder/OperatorBuilder.cs

@@ -33,6 +33,8 @@ namespace KubeOps.Operator.Builder;
 
 internal sealed class OperatorBuilder : IOperatorBuilder
 {
+    private OperatorRegistrationRegistry? _registrationRegistry;
+
     public OperatorBuilder(IServiceCollection services, OperatorSettings settings)
     {
         Settings = settings;
@@ -51,6 +53,8 @@ public IOperatorBuilder AddController<TImplementation, TEntity>()
         Services.TryAddScoped<IEntityController<TEntity>, TImplementation>();
         Services.TryAddSingleton<IReconciler<TEntity>, Reconciler<TEntity>>();
 
+        RegisterRegistrationValidation(typeof(TEntity));
+
         // Queue
         Services.TryAddTransient<IEntityQueueFactory, EntityQueueFactory>();
         Services.TryAddTransient<EntityQueue<TEntity>>(services =>
@@ -59,7 +63,16 @@ public IOperatorBuilder AddController<TImplementation, TEntity>()
         if (Settings.QueueStrategy == QueueStrategy.InMemory)
         {
             Services.TryAddSingleton<ITimedEntityQueue<TEntity>, TimedEntityQueue<TEntity>>();
-            Services.AddHostedService<EntityQueueBackgroundService<TEntity>>();
+
+            switch (Settings.LeaderElectionType)
+            {
+                case LeaderElectionType.None:
+                    Services.AddHostedService<EntityQueueBackgroundService<TEntity>>();
+                    break;
+                case LeaderElectionType.Single:
+                    Services.AddHostedService<LeaderAwareEntityQueueBackgroundService<TEntity>>();
+                    break;
+            }
         }
 
         // Leader Election
@@ -108,6 +121,8 @@ public IOperatorBuilder AddFinalizer<TImplementation, TEntity>(string identifier
             services.GetRequiredService<IEventFinalizerAttacherFactory>()
                 .Create<TImplementation, TEntity>(identifier));
 
+        RegisterRegistrationValidation(typeof(TEntity));
+
         return this;
     }
 
@@ -165,4 +180,21 @@ private void AddOperatorBase()
             Services.AddLeaderElection();
         }
     }
+
+    private void RegisterRegistrationValidation(Type entityType)
+    {
+        if (!Settings.ValidateRegistrations)
+        {
+            return;
+        }
+
+        if (_registrationRegistry is null)
+        {
+            _registrationRegistry = new(Services);
+            Services.AddSingleton(_registrationRegistry);
+            Services.AddHostedService<OperatorRegistrationValidator>();
+        }
+
+        _registrationRegistry.Add(entityType);
+    }
 }

src/KubeOps.Operator/Builder/OperatorRegistrationRegistry.cs

@@ -0,0 +1,34 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information.
+
+using Microsoft.Extensions.DependencyInjection;
+
+namespace KubeOps.Operator.Builder;
+
+/// <summary>
+/// Bridges build-time information to the <see cref="OperatorRegistrationValidator"/> running at host
+/// startup: it tracks the entity types managed by the operator and exposes the service collection so
+/// the validator can inspect the registrations.
+/// </summary>
+/// <param name="services">The service collection the operator is registered into.</param>
+internal sealed class OperatorRegistrationRegistry(IServiceCollection services)
+{
+    private readonly HashSet<Type> _managedEntities = [];
+
+    /// <summary>
+    /// Gets the entity types that are managed by the operator (i.e. that have a controller registered).
+    /// </summary>
+    public IReadOnlyCollection<Type> ManagedEntities => _managedEntities;
+
+    /// <summary>
+    /// Gets the service collection the operator is registered into.
+    /// </summary>
+    public IServiceCollection Services => services;
+
+    /// <summary>
+    /// Registers an entity type as managed by the operator.
+    /// </summary>
+    /// <param name="entityType">The entity type to register.</param>
+    public void Add(Type entityType) => _managedEntities.Add(entityType);
+}