Implement slow-start budget for orphaned shard claims and enhance related logging

Author: benjaminpetit

src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShardManager.cs

@@ -66,12 +66,13 @@ public AzureStorageJobShardManager(
     {
     }
 
-    public override async Task<List<Orleans.DurableJobs.IJobShard>> AssignJobShardsAsync(DateTimeOffset maxShardStartTime, CancellationToken cancellationToken)
+    public override async Task<List<Orleans.DurableJobs.IJobShard>> AssignJobShardsAsync(DateTimeOffset maxShardStartTime, int maxNewClaims, CancellationToken cancellationToken)
     {
         await InitializeIfNeeded(cancellationToken);
         LogAssigningShards(_logger, SiloAddress, maxShardStartTime, _containerName);
 
         var result = new List<Orleans.DurableJobs.IJobShard>();
+        var newClaimCount = 0;
         await foreach (var blob in _client.GetBlobsAsync(traits: BlobTraits.Metadata, states: BlobStates.None, cancellationToken: cancellationToken, prefix: _blobPrefix))
         {
             // Get the owner and creator of the shard
@@ -125,6 +126,12 @@ public AzureStorageJobShardManager(
             // Determine if this is an adopted shard (taken from dead owner) vs orphaned (gracefully released)
             var isAdopted = owner is not null && ownerStatus == SiloStatus.Dead;
 
+            // Respect the slow-start budget: skip claiming if we've exhausted the budget
+            if (newClaimCount >= maxNewClaims)
+            {
+                continue;
+            }
+
             // Try to claim orphaned or adopted shard
             LogClaimingShard(_logger, blob.Name, SiloAddress, owner);
             var blobClient = _client.GetAppendBlobClient(blob.Name);
@@ -142,6 +149,7 @@ public AzureStorageJobShardManager(
             _jobShardCache[blob.Name] = orphanedShard;
             LogShardAssigned(_logger, blob.Name, SiloAddress);
             result.Add(orphanedShard);
+            newClaimCount++;
         }
 
         LogAssignmentCompleted(_logger, result.Count, SiloAddress);

src/Orleans.DurableJobs/Hosting/DurableJobsOptions.cs

@@ -88,6 +88,51 @@ public sealed class DurableJobsOptions
     /// </remarks>
     public int MaxAdoptedCount { get; set; } = 3;
 
+    /// <summary>
+    /// Gets or sets the maximum number of orphaned shards a silo may claim immediately
+    /// after startup. The cumulative budget grows linearly from this value to
+    /// <see cref="SlowStartMaxBudget"/> over <see cref="SlowStartRampUpDuration"/>,
+    /// after which the limit is removed entirely.
+    /// This prevents a freshly started silo from overwhelming itself by claiming all orphaned shards
+    /// at once during disaster-recovery scenarios.
+    /// Default: 2.
+    /// </summary>
+    /// <example>
+    /// <code>
+    /// options.SlowStartInitialBudget = 1;
+    /// </code>
+    /// </example>
+    public int SlowStartInitialBudget { get; set; } = 2;
+
+    /// <summary>
+    /// Gets or sets the total number of orphaned shards the silo is allowed to have claimed
+    /// by the end of the slow-start ramp-up period. The cumulative budget is linearly
+    /// interpolated between <see cref="SlowStartInitialBudget"/> at startup and this value
+    /// at <see cref="SlowStartRampUpDuration"/>.
+    /// Default: 20.
+    /// </summary>
+    /// <example>
+    /// <code>
+    /// options.SlowStartMaxBudget = 50;
+    /// </code>
+    /// </example>
+    public int SlowStartMaxBudget { get; set; } = 20;
+
+    /// <summary>
+    /// Gets or sets the duration of the slow-start ramp-up period after silo activation.
+    /// While the silo has been running for less than this duration, the number of orphaned shards
+    /// it may claim is limited by a linearly increasing budget. Once this period elapses the
+    /// silo claims all available orphaned shards without limit.
+    /// Set to <see cref="TimeSpan.Zero"/> to disable slow-start entirely.
+    /// Default: 5 minutes.
+    /// </summary>
+    /// <example>
+    /// <code>
+    /// options.SlowStartRampUpDuration = TimeSpan.FromMinutes(10);
+    /// </code>
+    /// </example>
+    public TimeSpan SlowStartRampUpDuration { get; set; } = TimeSpan.FromMinutes(5);
+
     private static DateTimeOffset? DefaultShouldRetry(IJobRunContext jobContext, Exception ex)
     {
         // Default retry logic: retry up to 5 times with exponential backoff
@@ -141,6 +186,18 @@ public void ValidateConfiguration()
         {
             throw new OrleansConfigurationException("DurableJobsOptions.MaxAdoptedCount must be greater than or equal to zero.");
         }
+        if (options.SlowStartInitialBudget < 0)
+        {
+            throw new OrleansConfigurationException("DurableJobsOptions.SlowStartInitialBudget must be non-negative.");
+        }
+        if (options.SlowStartMaxBudget < options.SlowStartInitialBudget)
+        {
+            throw new OrleansConfigurationException("DurableJobsOptions.SlowStartMaxBudget must be greater than or equal to SlowStartInitialBudget.");
+        }
+        if (options.SlowStartRampUpDuration < TimeSpan.Zero)
+        {
+            throw new OrleansConfigurationException("DurableJobsOptions.SlowStartRampUpDuration must be non-negative.");
+        }
         _logger.LogInformation("DurableJobsOptions validated: ShardDuration={ShardDuration}", options.ShardDuration);
     }
 }

src/Orleans.DurableJobs/JobShardManager.cs

@@ -31,9 +31,14 @@ protected JobShardManager(SiloAddress siloAddress)
     /// Assigns orphaned job shards to this silo.
     /// </summary>
     /// <param name="maxDueTime">Maximum due time for shards to consider.</param>
+    /// <param name="maxNewClaims">
+    /// The maximum number of orphaned shards to claim in this call.
+    /// Use <see cref="int.MaxValue"/> for unlimited.
+    /// Shards already owned by this silo are always returned regardless of this limit.
+    /// </param>
     /// <param name="cancellationToken">Cancellation token.</param>
     /// <returns>A list of job shards assigned to this silo.</returns>
-    public abstract Task<List<IJobShard>> AssignJobShardsAsync(DateTimeOffset maxDueTime, CancellationToken cancellationToken);
+    public abstract Task<List<IJobShard>> AssignJobShardsAsync(DateTimeOffset maxDueTime, int maxNewClaims, CancellationToken cancellationToken);
 
     /// <summary>
     /// Creates a new job shard owned by this silo.
@@ -112,7 +117,7 @@ internal static async Task ClearAllShardsAsync()
         }
     }
 
-    public override async Task<List<IJobShard>> AssignJobShardsAsync(DateTimeOffset maxDueTime, CancellationToken cancellationToken)
+    public override async Task<List<IJobShard>> AssignJobShardsAsync(DateTimeOffset maxDueTime, int maxNewClaims, CancellationToken cancellationToken)
     {
         var alreadyOwnedShards = new List<IJobShard>();
         var adoptedShards = new List<IJobShard>();
@@ -172,6 +177,12 @@ public override async Task<List<IJobShard>> AssignJobShardsAsync(DateTimeOffset
                             }
                         }
 
+                        // Respect the slow-start budget: skip claiming if we've exhausted the budget
+                        if (adoptedShards.Count >= maxNewClaims)
+                        {
+                            continue;
+                        }
+
                         ownership.OwnerSiloAddress = SiloAddress.ToString();
                         adoptedShards.Add(ownership.Shard);
                     }

src/Orleans.DurableJobs/LocalDurableJobManager.Log.cs

@@ -131,4 +131,22 @@ internal partial class LocalDurableJobManager
         Message = "Creating new shard for key {ShardKey}"
     )]
     private static partial void LogCreatingNewShard(ILogger logger, DateTimeOffset shardKey);
+
+    [LoggerMessage(
+        Level = LogLevel.Information,
+        Message = "Claimed {NewClaims} new orphaned shard(s) this cycle (total claimed: {TotalClaimed})"
+    )]
+    private static partial void LogOrphanedShardsClaimed(ILogger logger, int newClaims, int totalClaimed);
+
+    [LoggerMessage(
+        Level = LogLevel.Debug,
+        Message = "Shard claim budget: {Budget} (totalBudget={TotalBudget}, alreadyClaimed={AlreadyClaimed}, elapsed={Elapsed}, rampUpDuration={RampUpDuration})"
+    )]
+    private static partial void LogShardClaimBudget(ILogger logger, int budget, int totalBudget, int alreadyClaimed, TimeSpan elapsed, TimeSpan rampUpDuration);
+
+    [LoggerMessage(
+        Level = LogLevel.Warning,
+        Message = "Silo is overloaded, pausing all new shard claims"
+    )]
+    private static partial void LogOverloadPausingShardClaims(ILogger logger);
 }

src/Orleans.DurableJobs/LocalDurableJobManager.cs

@@ -11,6 +11,7 @@
 using Orleans.Internal;
 using Orleans.Runtime;
 using Orleans.Runtime.Internal;
+using Orleans.Runtime.Messaging;
 
 namespace Orleans.DurableJobs;
 
@@ -20,6 +21,8 @@ internal partial class LocalDurableJobManager : SystemTarget, ILocalDurableJobMa
     private readonly JobShardManager _shardManager;
     private readonly ShardExecutor _shardExecutor;
     private readonly IAsyncEnumerable<ClusterMembershipSnapshot> _clusterMembershipUpdates;
+    private readonly IOverloadDetector _overloadDetector;
+    private readonly TimeProvider _timeProvider;
     private readonly ILogger<LocalDurableJobManager> _logger;
     private readonly DurableJobsOptions _options;
     private readonly CancellationTokenSource _cts = new();
@@ -33,12 +36,18 @@ internal partial class LocalDurableJobManager : SystemTarget, ILocalDurableJobMa
     private readonly SemaphoreSlim _shardCreationLock = new(1, 1);
     private readonly SemaphoreSlim _shardCheckSignal = new(0);
 
+    // Slow-start state
+    private long _startTimestamp;
+    private int _totalClaimedOrphanedShards;
+
     private static readonly IDictionary<string, string> EmptyMetadata = new Dictionary<string, string>();
 
     public LocalDurableJobManager(
         JobShardManager shardManager,
         ShardExecutor shardExecutor,
         IClusterMembershipService clusterMembership,
+        IOverloadDetector overloadDetector,
+        TimeProvider timeProvider,
         IOptions<DurableJobsOptions> options,
         SystemTargetShared shared,
         ILogger<LocalDurableJobManager> logger)
@@ -47,6 +56,8 @@ public LocalDurableJobManager(
         _shardManager = shardManager;
         _shardExecutor = shardExecutor;
         _clusterMembershipUpdates = clusterMembership.MembershipUpdates;
+        _overloadDetector = overloadDetector;
+        _timeProvider = timeProvider;
         _logger = logger;
         _options = options.Value;
     }
@@ -115,6 +126,8 @@ private Task Start(CancellationToken ct)
     {
         LogStarting(_logger);
 
+        _startTimestamp = _timeProvider.GetTimestamp();
+
         using (var _ = new ExecutionContextSuppressor())
         {
             _listenForClusterChangesTask = Task.Factory.StartNew(
@@ -247,14 +260,23 @@ private async Task PeriodicShardCheck()
                     }
                 }
 
+                // Compute the slow-start budget for this cycle
+                var budget = ComputeClaimBudget();
+
                 // Query ShardManager for assigned shards (source of truth)
-                var shards = await _shardManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), _cts.Token);
+                var shards = await _shardManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), budget, _cts.Token);
+
+                // Count newly claimed shards (those not already in our cache)
+                var newClaimsThisCycle = 0;
                 if (shards.Count > 0)
                 {
                     LogAssignedShards(_logger, shards.Count);
                     foreach (var shard in shards)
                     {
-                        _shardCache.TryAdd(shard.Id, shard);
+                        if (_shardCache.TryAdd(shard.Id, shard))
+                        {
+                            newClaimsThisCycle++;
+                        }
 
                         if (!_runningShards.ContainsKey(shard.Id))
                         {
@@ -266,6 +288,12 @@ private async Task PeriodicShardCheck()
                 {
                     LogNoShardsToAssign(_logger);
                 }
+
+                if (newClaimsThisCycle > 0)
+                {
+                    _totalClaimedOrphanedShards += newClaimsThisCycle;
+                    LogOrphanedShardsClaimed(_logger, newClaimsThisCycle, _totalClaimedOrphanedShards);
+                }
             }
             catch (OperationCanceledException)
             {
@@ -279,6 +307,45 @@ private async Task PeriodicShardCheck()
         }
     }
 
+    /// <summary>
+    /// Computes the maximum number of orphaned shards this silo may claim in the current check cycle.
+    /// Returns <see cref="int.MaxValue"/> when unlimited (ramp-up complete or disabled).
+    /// Returns <c>0</c> when the silo is overloaded, pausing all new claims.
+    /// </summary>
+    private int ComputeClaimBudget()
+    {
+        // If overloaded, claim nothing new regardless of ramp-up phase
+        if (_overloadDetector.IsOverloaded)
+        {
+            LogOverloadPausingShardClaims(_logger);
+            return 0;
+        }
+
+        // Slow-start disabled
+        if (_options.SlowStartRampUpDuration <= TimeSpan.Zero)
+        {
+            return int.MaxValue;
+        }
+
+        var elapsed = _timeProvider.GetElapsedTime(_startTimestamp);
+
+        // After ramp-up period, no limit
+        if (elapsed >= _options.SlowStartRampUpDuration)
+        {
+            return int.MaxValue;
+        }
+
+        // Linear interpolation: InitialBudget at t=0, MaxBudget at t=RampUpDuration
+        var progress = elapsed.TotalMilliseconds / _options.SlowStartRampUpDuration.TotalMilliseconds;
+        var totalBudget = _options.SlowStartInitialBudget
+                        + (int)(progress * (_options.SlowStartMaxBudget - _options.SlowStartInitialBudget));
+        var remaining = totalBudget - _totalClaimedOrphanedShards;
+        var budget = Math.Max(0, remaining);
+
+        LogShardClaimBudget(_logger, budget, totalBudget, _totalClaimedOrphanedShards, elapsed, _options.SlowStartRampUpDuration);
+        return budget;
+    }
+
     private void TryActivateShard(IJobShard shard)
     {
         // Only start if not already running

test/Extensions/Orleans.Azure.Tests/DurableJobs/AzureStorageJobShardBatchingTests.cs

@@ -119,7 +119,7 @@ public async Task AzureStorageJobShard_MultipleOperationsBatched()
         SetSiloStatus(newSiloAddress, SiloStatus.Active);
 
         var newManager = CreateManager(newSiloAddress);
-        var shards = await newManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), cancellationToken);
+        var shards = await newManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), maxNewClaims: int.MaxValue, cancellationToken);
         Assert.Single(shards);
 
         var consumedJobs = new List<string>();
@@ -168,7 +168,7 @@ public async Task AzureStorageJobShard_PartialBatchFlushesOnTimeout()
         SetSiloStatus(newSiloAddress, SiloStatus.Active);
 
         var newManager = CreateManager(newSiloAddress);
-        var shards = await newManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), cancellationToken);
+        var shards = await newManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), maxNewClaims: int.MaxValue, cancellationToken);
         Assert.Single(shards);
 
         var consumedJobs = new List<string>();
@@ -222,7 +222,7 @@ public async Task AzureStorageJobShard_MaxBatchSizeEnforced()
         SetSiloStatus(newSiloAddress, SiloStatus.Active);
 
         var newManager = CreateManager(newSiloAddress);
-        var shards = await newManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), cancellationToken);
+        var shards = await newManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), maxNewClaims: int.MaxValue, cancellationToken);
         Assert.Single(shards);
 
         var consumedJobs = new List<string>();
@@ -290,7 +290,7 @@ public async Task AzureStorageJobShard_MetadataOperationsBreakBatches()
         StorageOptions.Value.BatchFlushInterval = TimeSpan.FromMilliseconds(100);
 
         var newManager = CreateManager(newSiloAddress);
-        var shards = await newManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), cancellationToken);
+        var shards = await newManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), maxNewClaims: int.MaxValue, cancellationToken);
         Assert.Single(shards);
 
         var consumedJobs = new List<string>();

test/Extensions/Orleans.Azure.Tests/DurableJobs/AzureStorageJobShardManagerTests.cs

@@ -179,4 +179,37 @@ public async Task AzureStorageJobShardManager_UnregisterShard_WithJobsRemaining(
         using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
         await _runner.UnregisterShard_WithJobsRemaining(cts.Token);
     }
+
+    /// <summary>
+    /// Tests that maxNewClaims limits the number of orphaned shards claimed.
+    /// This test is delegated to the runner for reuse across providers.
+    /// </summary>
+    [SkippableFact, TestCategory("Azure"), TestCategory("Functional")]
+    public async Task AzureStorageJobShardManager_SlowStart_LimitsOrphanedShardClaims()
+    {
+        using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
+        await _runner.SlowStart_LimitsOrphanedShardClaims(cts.Token);
+    }
+
+    /// <summary>
+    /// Tests that maxNewClaims = 0 prevents claiming orphaned shards but returns owned shards.
+    /// This test is delegated to the runner for reuse across providers.
+    /// </summary>
+    [SkippableFact, TestCategory("Azure"), TestCategory("Functional")]
+    public async Task AzureStorageJobShardManager_SlowStart_ZeroBudgetClaimsNothing()
+    {
+        using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
+        await _runner.SlowStart_ZeroBudgetClaimsNothing(cts.Token);
+    }
+
+    /// <summary>
+    /// Tests that maxNewClaims = int.MaxValue (unlimited) claims all orphaned shards.
+    /// This test is delegated to the runner for reuse across providers.
+    /// </summary>
+    [SkippableFact, TestCategory("Azure"), TestCategory("Functional")]
+    public async Task AzureStorageJobShardManager_SlowStart_UnlimitedBudgetClaimsAll()
+    {
+        using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
+        await _runner.SlowStart_UnlimitedBudgetClaimsAll(cts.Token);
+    }
 }