grain and range lease holds for even stronger single activation gurantees

Author: ledjon-behluli

src/Orleans.Core.Abstractions/Exceptions/DirectoryLeaseHoldException.cs

@@ -0,0 +1,14 @@
+namespace Orleans.Runtime;
+
+/// <summary>
+/// Indicates that a request to the grain directory was rejected because the target grain or directory range 
+/// is currently under a safety lease hold after an ungraceful silo crash.
+/// </summary>
+/// <remarks>
+/// Initializes a new instance of the <see cref="DirectoryLeaseHoldException"/> class.
+/// </remarks>
+/// <param name="message">The message.</param>
+[Serializable]
+[GenerateSerializer]
+[Alias("DirectoryLeaseHoldException")]
+public class DirectoryLeaseHoldException(string message) : OrleansException(message);

src/Orleans.Runtime/Configuration/Options/GrainDirectoryOptions.cs

@@ -1,99 +1,119 @@
-
-using System;
 using Orleans.Runtime.GrainDirectory;
 
-namespace Orleans.Configuration
+namespace Orleans.Configuration;
+
+public class GrainDirectoryOptions
 {
-    public class GrainDirectoryOptions
+    /// <summary>
+    /// Configuration type that controls the type of the grain directory caching algorithm that silo use.
+    /// </summary>
+    public enum CachingStrategyType
     {
-        /// <summary>
-        /// Configuration type that controls the type of the grain directory caching algorithm that silo use.
-        /// </summary>
-        public enum CachingStrategyType
-        {
-            /// <summary>Don't cache.</summary>
-            None,
-            /// <summary>Standard fixed-size LRU.</summary>
-            LRU,
-            /// <summary>Adaptive caching with fixed maximum size and refresh. This option should be used in production.</summary>
-            [Obsolete("Adaptive caching is deprecated in favor of LRU and will be removed in a future version. This value is now an alias for LRU.")]
-            Adaptive,
-            /// <summary>Custom cache implementation, configured by registering an <see cref="IGrainDirectoryCache"/> implementation in the dependency injection container.</summary>
-            Custom
-        }
-
-        /// <summary>
-        /// Gets or sets the caching strategy to use.
-        /// The options are None, which means don't cache directory entries locally;
-        /// LRU, which indicates that a standard fixed-size least recently used strategy should be used; and
-        /// Adaptive, which indicates that an adaptive strategy with a fixed maximum size should be used.
-        /// The LRU strategy is used by default.
-        /// </summary>
-        public CachingStrategyType CachingStrategy { get; set; } = DEFAULT_CACHING_STRATEGY;
-
-        /// <summary>
-        /// The default value for <see cref="CachingStrategy"/>.
-        /// </summary>
-        public const CachingStrategyType DEFAULT_CACHING_STRATEGY = CachingStrategyType.LRU;
-
-        /// <summary>
-        /// Gets or sets the maximum number of grains to cache directory information for.
-        /// </summary>
-        public int CacheSize { get; set; } = DEFAULT_CACHE_SIZE;
-
-        /// <summary>
-        /// The default value for <see cref="CacheSize"/>.
-        /// </summary>
-        public const int DEFAULT_CACHE_SIZE = 1_000_000;
-
-        /// <summary>
-        /// Gets or sets the initial (minimum) time, in seconds, to keep a cache entry before revalidating.
-        /// </summary>
-        [Obsolete("InitialCacheTTL is deprecated and will be removed in a future version.")]
-        public TimeSpan InitialCacheTTL { get; set; } = DEFAULT_INITIAL_CACHE_TTL;
-
-        /// <summary>
-        /// The default value for <see cref="InitialCacheTTL"/>.
-        /// </summary>
-        [Obsolete("DEFAULT_INITIAL_CACHE_TTL is deprecated and will be removed in a future version.")]
-        public static readonly TimeSpan DEFAULT_INITIAL_CACHE_TTL = TimeSpan.FromSeconds(30);
-
-        /// <summary>
-        /// Gets or sets the maximum time, in seconds, to keep a cache entry before revalidating.
-        /// </summary>
-        [Obsolete("MaximumCacheTTL is deprecated and will be removed in a future version.")]
-        public TimeSpan MaximumCacheTTL { get; set; } = DEFAULT_MAXIMUM_CACHE_TTL;
-
-        /// <summary>
-        /// The default value for <see cref="MaximumCacheTTL"/>.
-        /// </summary>
-        [Obsolete("DEFAULT_MAXIMUM_CACHE_TTL is deprecated and will be removed in a future version.")]
-        public static readonly TimeSpan DEFAULT_MAXIMUM_CACHE_TTL = TimeSpan.FromSeconds(240);
-
-        /// <summary>
-        /// Gets or sets the factor by which cache entry TTLs should be extended when they are found to be stable.
-        /// </summary>
-        [Obsolete("CacheTTLExtensionFactor is deprecated and will be removed in a future version.")]
-        public double CacheTTLExtensionFactor { get; set; } = DEFAULT_TTL_EXTENSION_FACTOR;
-
-        /// <summary>
-        /// The default value for <see cref="CacheTTLExtensionFactor"/>.
-        /// </summary>
-        [Obsolete("DEFAULT_TTL_EXTENSION_FACTOR is deprecated and will be removed in a future version.")]
-        public const double DEFAULT_TTL_EXTENSION_FACTOR = 2.0;
-
-        /// <summary>
-        /// Gets or sets the time span between when we have added an entry for an activation to the grain directory and when we are allowed
-        /// to conditionally remove that entry. 
-        /// Conditional deregistration is used for lazy clean-up of activations whose prompt deregistration failed for some reason (e.g., message failure).
-        /// This should always be at least one minute, since we compare the times on the directory partition, so message delays and clcks skues have
-        /// to be allowed.
-        /// </summary>
-        public TimeSpan LazyDeregistrationDelay { get; set; } = DEFAULT_UNREGISTER_RACE_DELAY;
-
-        /// <summary>
-        /// The default value for <see cref="LazyDeregistrationDelay"/>.
-        /// </summary>
-        public static readonly TimeSpan DEFAULT_UNREGISTER_RACE_DELAY = TimeSpan.FromMinutes(1);
+        /// <summary>Don't cache.</summary>
+        None,
+        /// <summary>Standard fixed-size LRU.</summary>
+        LRU,
+        /// <summary>Adaptive caching with fixed maximum size and refresh. This option should be used in production.</summary>
+        [Obsolete("Adaptive caching is deprecated in favor of LRU and will be removed in a future version. This value is now an alias for LRU.")]
+        Adaptive,
+        /// <summary>Custom cache implementation, configured by registering an <see cref="IGrainDirectoryCache"/> implementation in the dependency injection container.</summary>
+        Custom
     }
+
+    /// <summary>
+    /// Gets or sets the caching strategy to use.
+    /// The options are None, which means don't cache directory entries locally;
+    /// LRU, which indicates that a standard fixed-size least recently used strategy should be used; and
+    /// Adaptive, which indicates that an adaptive strategy with a fixed maximum size should be used.
+    /// The LRU strategy is used by default.
+    /// </summary>
+    public CachingStrategyType CachingStrategy { get; set; } = DEFAULT_CACHING_STRATEGY;
+
+    /// <summary>
+    /// The default value for <see cref="CachingStrategy"/>.
+    /// </summary>
+    public const CachingStrategyType DEFAULT_CACHING_STRATEGY = CachingStrategyType.LRU;
+
+    /// <summary>
+    /// Gets or sets the maximum number of grains to cache directory information for.
+    /// </summary>
+    public int CacheSize { get; set; } = DEFAULT_CACHE_SIZE;
+
+    /// <summary>
+    /// The default value for <see cref="CacheSize"/>.
+    /// </summary>
+    public const int DEFAULT_CACHE_SIZE = 1_000_000;
+
+    /// <summary>
+    /// Gets or sets the initial (minimum) time, in seconds, to keep a cache entry before revalidating.
+    /// </summary>
+    [Obsolete("InitialCacheTTL is deprecated and will be removed in a future version.")]
+    public TimeSpan InitialCacheTTL { get; set; } = DEFAULT_INITIAL_CACHE_TTL;
+
+    /// <summary>
+    /// The default value for <see cref="InitialCacheTTL"/>.
+    /// </summary>
+    [Obsolete("DEFAULT_INITIAL_CACHE_TTL is deprecated and will be removed in a future version.")]
+    public static readonly TimeSpan DEFAULT_INITIAL_CACHE_TTL = TimeSpan.FromSeconds(30);
+
+    /// <summary>
+    /// Gets or sets the maximum time, in seconds, to keep a cache entry before revalidating.
+    /// </summary>
+    [Obsolete("MaximumCacheTTL is deprecated and will be removed in a future version.")]
+    public TimeSpan MaximumCacheTTL { get; set; } = DEFAULT_MAXIMUM_CACHE_TTL;
+
+    /// <summary>
+    /// The default value for <see cref="MaximumCacheTTL"/>.
+    /// </summary>
+    [Obsolete("DEFAULT_MAXIMUM_CACHE_TTL is deprecated and will be removed in a future version.")]
+    public static readonly TimeSpan DEFAULT_MAXIMUM_CACHE_TTL = TimeSpan.FromSeconds(240);
+
+    /// <summary>
+    /// Gets or sets the factor by which cache entry TTLs should be extended when they are found to be stable.
+    /// </summary>
+    [Obsolete("CacheTTLExtensionFactor is deprecated and will be removed in a future version.")]
+    public double CacheTTLExtensionFactor { get; set; } = DEFAULT_TTL_EXTENSION_FACTOR;
+
+    /// <summary>
+    /// The default value for <see cref="CacheTTLExtensionFactor"/>.
+    /// </summary>
+    [Obsolete("DEFAULT_TTL_EXTENSION_FACTOR is deprecated and will be removed in a future version.")]
+    public const double DEFAULT_TTL_EXTENSION_FACTOR = 2.0;
+
+    /// <summary>
+    /// Gets or sets the time span between when we have added an entry for an activation to the grain directory and when we are allowed
+    /// to conditionally remove that entry. 
+    /// Conditional deregistration is used for lazy clean-up of activations whose prompt deregistration failed for some reason (e.g., message failure).
+    /// This should always be at least one minute, since we compare the times on the directory partition, so message delays and clcks skues have
+    /// to be allowed.
+    /// </summary>
+    public TimeSpan LazyDeregistrationDelay { get; set; } = DEFAULT_UNREGISTER_RACE_DELAY;
+
+    /// <summary>
+    /// The default value for <see cref="LazyDeregistrationDelay"/>.
+    /// </summary>
+    public static readonly TimeSpan DEFAULT_UNREGISTER_RACE_DELAY = TimeSpan.FromMinutes(1);
+
+    /// <summary>
+    /// Gets or sets the duration for the safety lease hold applied after an ungraceful silo failure.
+    /// This duration applies in two scenarios:
+    /// <list type="bullet">
+    /// <item>When a specific silo crashes ungracefully, grain lease holds prevent individual re-registration of its grains for this duration.</item>
+    /// <item>When a directory partition can not acquire a snapshot from a previous owner, range lease holds prevent new registrations in that whole range for this duration.</item>
+    /// </list>
+    /// </summary>
+    /// <remarks>
+    /// Depending on the value of this, the duration is understood as:
+    /// <list type="bullet">
+    /// <item><c>SafetyLeaseHoldDuration > TimeSpan.Zero</c> - The lease duration is explicitly controlled by the user.</item>
+    /// <item><c>SafetyLeaseHoldDuration &lt;= TimeSpan.Zero</c> - The system computes a lease duration as:
+    /// <c>2 × <see cref="ClusterMembershipOptions.ProbeTimeout"/> × <see cref="ClusterMembershipOptions.NumMissedProbesLimit"/></c>.</item>
+    /// </list>
+    /// </remarks>
+    public TimeSpan SafetyLeaseHoldDuration { get; set; }
+
+    /// <summary>
+    /// The default value for <see cref="SafetyLeaseHoldDuration"/>
+    /// </summary>
+    public static readonly TimeSpan DEFAULT_SAFETY_LEASE_HOLD_DURATION = TimeSpan.Zero;
 }

src/Orleans.Runtime/GrainDirectory/DistributedGrainDirectory.cs

@@ -1,13 +1,11 @@
-using System;
-using System.Collections.Generic;
 using System.Collections.Immutable;
-using System.Linq;
 using System.Runtime.CompilerServices;
-using System.Threading;
-using System.Threading.Tasks;
+using System.Timers;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 using Orleans.Concurrency;
+using Orleans.Configuration;
 using Orleans.GrainDirectory;
 using Orleans.Internal;
 using Orleans.Runtime.Internal;
@@ -63,6 +61,7 @@ internal sealed partial class DistributedGrainDirectory : SystemTarget, IGrainDi
     private readonly IServiceProvider _serviceProvider;
     private readonly ImmutableArray<GrainDirectoryPartition> _partitions;
     private readonly CancellationTokenSource _stoppedCts = new();
+    private readonly TimeSpan _leaseHoldDuration;
 
     internal CancellationToken OnStoppedToken => _stoppedCts.Token;
     internal ClusterMembershipSnapshot ClusterMembershipSnapshot => _membershipService.CurrentView.ClusterMembershipSnapshot;
@@ -75,25 +74,37 @@ internal sealed partial class DistributedGrainDirectory : SystemTarget, IGrainDi
     // precise by also tracking the sets of ranges which need to be recovered, but that complicates things somewhat since it would require tracking the ranges
     // for each recovery version.
     private long _recoveryMembershipVersion;
-    private Task _runTask = Task.CompletedTask;
-    private ActivationDirectory _localActivations;
+    private readonly ActivationDirectory _localActivations;
     private GrainDirectoryResolver? _grainDirectoryResolver;
+    private Task? _runTask;
+    private Task? _leaseCleanupTask;
 
     public DistributedGrainDirectory(
         DirectoryMembershipService membershipService,
         ILogger<DistributedGrainDirectory> logger,
         IServiceProvider serviceProvider,
         IInternalGrainFactory grainFactory,
+        IOptions<GrainDirectoryOptions> directoryOptions,
+        IOptions<ClusterMembershipOptions> membershipOptions,
+        TimeProvider timeProvider,
         SystemTargetShared shared) : base(Constants.GrainDirectoryType, shared)
     {
         _localActivations = shared.ActivationDirectory;
         _serviceProvider = serviceProvider;
         _membershipService = membershipService;
         _logger = logger;
+        _leaseHoldDuration = directoryOptions.Value.SafetyLeaseHoldDuration;
+
+        if (_leaseHoldDuration <= TimeSpan.Zero)
+        {
+            _leaseHoldDuration = 2 * membershipOptions.Value.ProbeTimeout * membershipOptions.Value.NumMissedProbesLimit;
+        }
+
         var partitions = ImmutableArray.CreateBuilder<GrainDirectoryPartition>(DirectoryMembershipSnapshot.PartitionsPerSilo);
+
         for (var i = 0; i < DirectoryMembershipSnapshot.PartitionsPerSilo; i++)
         {
-            partitions.Add(new GrainDirectoryPartition(i, this, grainFactory, shared));
+            partitions.Add(new GrainDirectoryPartition(i, this, _leaseHoldDuration, grainFactory, timeProvider, shared));
         }
 
         _partitions = partitions.ToImmutable();
@@ -312,19 +323,27 @@ void ILifecycleParticipant<ISiloLifecycle>.Participate(ISiloLifecycle observer)
         Task OnRuntimeInitializeStart(CancellationToken cancellationToken)
         {
             using var _ = new ExecutionContextSuppressor();
-            WorkItemGroup.QueueAction(() => _runTask = ProcessMembershipUpdates());
+
+            WorkItemGroup.QueueAction(() =>
+            {
+                _runTask = ProcessMembershipUpdates();
+                _leaseCleanupTask = RequestExpiredLeaseCleanups();
+            });
 
             return Task.CompletedTask;
         }
 
         async Task OnRuntimeInitializeStop(CancellationToken cancellationToken)
         {
             _stoppedCts.Cancel();
+
             if (_runTask is { } task)
             {
                 // Try to wait for hand-off to complete.
                 await this.RunOrQueueTask(async () => await task.WaitAsync(cancellationToken).SuppressThrowing());
             }
+
+            // No need to wait on the cleanup task since it does not have any external effects.
         }
 
         async Task OnShuttingDown(CancellationToken token)
@@ -359,9 +378,10 @@ private async Task ProcessMembershipUpdates()
                     {
                         if (change.Status == SiloStatus.Dead)
                         {
+                            var previousStatus = previousUpdate.GetSiloStatus(change.SiloAddress);
                             foreach (var partition in _partitions)
                             {
-                                tasks.Add(partition.OnSiloRemovedFromClusterAsync(change));
+                                tasks.Add(partition.OnSiloRemovedFromClusterAsync(change, previousStatus));
                             }
                         }
                     }
@@ -396,6 +416,37 @@ private async Task ProcessMembershipUpdates()
         await Task.WhenAll(tasks).SuppressThrowing();
     }
 
+    private async Task RequestExpiredLeaseCleanups()
+    {
+        // We request cleanups periodically to not let expired leases linger in the directory for too long.
+        // We do it here as opposed to in the partitions to avoid having 30 (by default, maybe more) timers.
+
+        var period = 1.1 * _leaseHoldDuration;
+
+        if (period < TimeSpan.FromMinutes(1))
+        {
+            // We create a lower-bound to avoid creting too much overhead in the partitions.
+            period = TimeSpan.FromMinutes(1);
+        }
+
+        using var timer = new PeriodicTimer(period);
+
+        try
+        {
+            while (await timer.WaitForNextTickAsync(_stoppedCts.Token))
+            {
+                foreach (var partition in _partitions)
+                {
+                    partition.CleanupExpiredLeases();
+                }
+            }
+        }
+        catch (OperationCanceledException) when (_stoppedCts.IsCancellationRequested)
+        {
+            // Ignore
+        }
+    }
+
     SiloAddress? ITestHooks.GetPrimaryForGrain(GrainId grainId)
     {
         _membershipService.CurrentView.TryGetOwner(grainId, out var owner, out _);