From 1859d15105e2ea45497969cfd325e3a28cf91fe0 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Mon, 5 Oct 2026 18:56:43 -0500 Subject: [PATCH 1/7] Reject renewal of expired or lost cache locks --- .agents/skills/foundatio/SKILL.md | 1 + docs/guide/locks.md | 64 +++-- .../Locks/LockTestBase.cs | 87 ++++++ src/Foundatio/Lock/CacheLockProvider.cs | 12 +- src/Foundatio/Lock/ILockProvider.cs | 34 ++- .../Locks/InMemoryLockTests.cs | 259 ++++++++++++++---- 6 files changed, 388 insertions(+), 69 deletions(-) diff --git a/.agents/skills/foundatio/SKILL.md b/.agents/skills/foundatio/SKILL.md index 5d72ee65a..7b059b6d2 100644 --- a/.agents/skills/foundatio/SKILL.md +++ b/.agents/skills/foundatio/SKILL.md @@ -310,6 +310,7 @@ public class OrderServiceTests : TestLoggerBase - **Failure semantics depend on job type**: `JobResult` only has `IsSuccess` -- there is no separate "failed but don't retry" status. For **queue-processed jobs** (`QueueJobBase.ProcessQueueEntryAsync`, or setting `context.Result` in a `WorkItemJob` handler), a non-success result triggers `AbandonAsync`, which re-queues the entry and eventually dead-letters it after `Retries` is exhausted -- reserve `FailedWithMessage`/`FromException` for transient errors you want retried, and log + return `JobResult.Success`/`SuccessWithMessage(...)` for permanent errors to avoid a pointless retry loop. For **standalone/manual jobs** (`JobBase`, a one-off `RunAsync()`/`RunInConsoleAsync()` run, or scheduled/cron jobs via `Foundatio.Extensions.Hosting`), there is no built-in retry or dead letter queue -- a failed result just produces an error-level log, a non-zero exit code from `RunInConsoleAsync`, or a failed entry in the job run history. Returning `FailedWithMessage`/`FromException` there is correct even for permanent errors, since nothing inside Foundatio will retry it. - **JobWithLockBase vs manual locking**: Use `JobWithLockBase` when the entire run must be single-instance (leader election). Use manual `ILockProvider.AcquireAsync` inside `JobBase` for finer-grained locking within a job. - **JobContext.RenewLockAsync**: Call in long-running jobs (both `JobBase` and `QueueJobBase`) to prevent lock expiration mid-processing. +- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; `WorkItemJob` progress renewal only logs them. Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. - **In-memory conditional updates**: Expired conditional replace/remove operations return false. Conditional removal reclaims expired entries and raises `ItemExpired`; live conditional removal and empty-list removal free memory immediately without that event. Rejected sized updates preserve the old entry. `Items` exposes values, honoring `CloneValues`, without affecting LRU. Default and built-in fixed-size large-list writes share unchanged storage; batch updates. Custom sizing, cloning, or raw dictionary access uses full dictionary copies. No-op removals avoid copying. - **In-memory list comparers**: List updates copy Dictionary, SortedDictionary, SortedList, ConcurrentDictionary, and OrderedDictionary (NET9+) with their comparer. Unsupported IDictionary types throw NotSupportedException instead of silently changing equality; the original entry remains unchanged. - **Register as singletons**: All infrastructure services (`ICacheClient`, `IMessageBus`, `IQueue`, `IFileStorage`, `ILockProvider`) maintain internal state and connections -- always register as singletons. diff --git a/docs/guide/locks.md b/docs/guide/locks.md index d8ffdcb1f..3fc996f88 100644 --- a/docs/guide/locks.md +++ b/docs/guide/locks.md @@ -190,10 +190,10 @@ try await DoWorkAsync(); } -catch (LockAcquisitionTimeoutException) +catch (LockAcquisitionTimeoutException ex) { // Lock could not be acquired before the timeout elapsed. - _logger.LogWarning("Timed out waiting for lock"); + _logger.LogWarning(ex, "Timed out waiting for lock: {Message}", ex.Message); } ``` @@ -250,36 +250,68 @@ if (lck is not null) } ``` +`CacheLockProvider.RenewAsync` only extends a lock you still hold. If its cache rejects renewal, including when the lock expired, was released, or was taken by another owner, it throws `LockException`. Ownership is no longer assured: stop the protected work, since another process may already be doing it. Custom cache providers must implement atomic conditional replacement and treat expired entries as missing. + +After a process pause or network outage, renewal can succeed if the lease is still live and its owner ID matches. Once it expires, the old handle cannot renew it, even if nobody else has acquired the resource. Acquire a new lock and revalidate the work before continuing; acquiring again does not restore uninterrupted ownership. This follows the [conditional renewal rule documented by Redis](https://redis.io/docs/latest/develop/clients/patterns/distributed-locks/#making-the-algorithm-more-reliable-extending-the-lock). + +A supplied renewal duration must be at least 5 milliseconds; a shorter duration throws `ArgumentOutOfRangeException` without changing the cache. Omitting the duration uses 20 minutes. Renewal is a no-op for `ThrottlingLockProvider` and `EmptyLock`; these do not provide renewable exclusive leases. + +::: warning Behavior change +Earlier versions of `CacheLockProvider.RenewAsync` returned successfully even when renewal failed, and could revive an expired lock. Rejecting renewal durations below 5 milliseconds is also new; the cache already enforced this minimum, but previously could remove the key before checking ownership. +::: + +A lock is not a fencing token. Losing it cannot cancel work you already sent to another system. Checking ownership before a write leaves a race between the check and the write; only validation enforced by the destination can fence a stale owner. Make operations idempotent where appropriate, but do not treat idempotency as exclusive ownership. + ### Automatic Renewal -For very long operations, set up automatic renewal: +For very long operations, observe renewal failures and cancel cooperative work before releasing the lock: ```csharp -await using var lck = await locker.TryAcquireAsync("my-resource"); +await using var lck = await locker.TryAcquireAsync( + "my-resource", timeUntilExpires: TimeSpan.FromMinutes(1)); if (lck is null) return; using var cts = new CancellationTokenSource(); - -// Start renewal task -var renewTask = Task.Run(async () => -{ - while (!cts.Token.IsCancellationRequested) - { - await Task.Delay(TimeSpan.FromSeconds(30), cts.Token); - await lck.RenewAsync(TimeSpan.FromMinutes(1)); - } -}); +var renewTask = RenewPeriodicallyAsync(); try { - await VeryLongRunningOperationAsync(); + await VeryLongRunningOperationAsync(cts.Token); } finally { - cts.Cancel(); + await cts.CancelAsync(); + await renewTask; +} + +async Task RenewPeriodicallyAsync() +{ + try + { + while (true) + { + await Task.Delay(TimeSpan.FromSeconds(30), cts.Token); + cts.Token.ThrowIfCancellationRequested(); + await lck.RenewAsync(TimeSpan.FromMinutes(1)); + } + } + catch (OperationCanceledException) when (cts.IsCancellationRequested) + { + // Normal shutdown after work completes or is cancelled. + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Lock renewal failed: {Message}", ex.Message); + await cts.CancelAsync(); + throw; + } } ``` +The local async function starts immediately and yields at the delay; it does not need `Task.Run`. The work must honor the cancellation token. The `finally` block observes renewal failure and waits for renewal to stop before `await using` releases the lock. An in-flight renewal has no cancellation-token parameter, so shutdown waits for that cache operation; configure provider timeouts. + +Choose an interval shorter than the lease duration, leaving time for cache latency and retries. A pause can still let the lease expire before renewal detects the loss, and cancellation cannot revoke an operation already dispatched to another system. `WorkItemJob`'s progress-triggered renewal currently logs renewal failures and continues; it does not implement this cancellation pattern. + ## Common Patterns ### Singleton Processing diff --git a/src/Foundatio.TestHarness/Locks/LockTestBase.cs b/src/Foundatio.TestHarness/Locks/LockTestBase.cs index d5b9b2a7a..c1af03f52 100644 --- a/src/Foundatio.TestHarness/Locks/LockTestBase.cs +++ b/src/Foundatio.TestHarness/Locks/LockTestBase.cs @@ -434,6 +434,93 @@ private static Task DoLockedWorkAsync(ILockProvider locker) return locker.TryUsingAsync("DoLockedWork", async () => await Task.Delay(500), TimeSpan.FromMinutes(1), TimeSpan.Zero); } + public virtual async Task RenewAsync_AfterRelease_ThrowsLockExceptionAndDoesNotRecreateLock() + { + // Arrange + var locker = GetLockProvider(); + if (locker is null) + return; + + string lockName = Guid.NewGuid().ToString("N")[..10]; + await using var lockInstance = await locker.TryAcquireAsync(lockName, timeUntilExpires: TimeSpan.FromSeconds(5)); + Assert.NotNull(lockInstance); + await lockInstance.ReleaseAsync(); + + // Act + await Assert.ThrowsAsync(() => lockInstance.RenewAsync()); + + // Assert + Assert.False(await locker.IsLockedAsync(lockName)); + Assert.Equal(0, lockInstance.RenewalCount); + } + + public virtual async Task RenewAsync_WithInvalidDuration_PreservesCurrentOwner(int milliseconds) + { + // Arrange + var locker = GetLockProvider(); + if (locker is null) + return; + + string lockName = Guid.NewGuid().ToString("N")[..10]; + await using var stale = await locker.AcquireAsync(lockName, timeUntilExpires: TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + await stale.ReleaseAsync(); + await using var current = await locker.AcquireAsync(lockName, timeUntilExpires: TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Act + var error = await Assert.ThrowsAsync(() => stale.RenewAsync(TimeSpan.FromMilliseconds(milliseconds))); + + // Assert + Assert.Equal("timeUntilExpires", error.ParamName); + Assert.Equal(0, stale.RenewalCount); + Assert.True(await locker.IsLockedAsync(lockName)); + await current.RenewAsync(); + Assert.Equal(1, current.RenewalCount); + } + + public virtual async Task RenewAsync_WithMissingLock_ThrowsLockException() + { + // Arrange + var locker = GetLockProvider(); + if (locker is null) + return; + + string lockName = Guid.NewGuid().ToString("N")[..10]; + string lockId = Guid.NewGuid().ToString("N"); + + // Act + await Assert.ThrowsAsync(() => locker.RenewAsync(lockName, lockId)); + + // Assert + Assert.False(await locker.IsLockedAsync(lockName)); + } + + public virtual async Task RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(bool scoped) + { + // Arrange + var locker = GetLockProvider(); + if (locker is null) + return; + if (scoped) + locker = new ScopedLockProvider(locker, "renewal"); + + string lockName = Guid.NewGuid().ToString("N")[..10]; + await using var stale = await locker.AcquireAsync(lockName, timeUntilExpires: TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + await locker.ReleaseAsync(stale.Resource); + await using var current = await locker.AcquireAsync(lockName, timeUntilExpires: TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Act + await Assert.ThrowsAsync(() => stale.RenewAsync()); + + // Assert + Assert.NotEqual(stale.LockId, current.LockId); + Assert.Equal(0, stale.RenewalCount); + Assert.True(await locker.IsLockedAsync(lockName)); + await stale.ReleaseAsync(); + await current.RenewAsync(); + Assert.True(await locker.IsLockedAsync(lockName)); + Assert.Equal(1, current.RenewalCount); + } + public virtual async Task WillThrottleCallsAsync() { Log.SetLogLevel(LogLevel.Information); diff --git a/src/Foundatio/Lock/CacheLockProvider.cs b/src/Foundatio/Lock/CacheLockProvider.cs index aae3d4d14..775b68f13 100644 --- a/src/Foundatio/Lock/CacheLockProvider.cs +++ b/src/Foundatio/Lock/CacheLockProvider.cs @@ -244,13 +244,21 @@ public async Task ReleaseAsync(string resource) _logger.LogDebug("Released lock: {Resource}", resource); } - public Task RenewAsync(string resource, string lockId, TimeSpan? timeUntilExpires = null) + /// + /// The cache rejected renewal. Ownership is no longer assured. + /// The supplied duration is less than 5 milliseconds. + public async Task RenewAsync(string resource, string lockId, TimeSpan? timeUntilExpires = null) { + if (timeUntilExpires < CacheClientExtensions.MinimumExpiration) + throw new ArgumentOutOfRangeException(nameof(timeUntilExpires), timeUntilExpires, "Lock renewal duration must be at least 5 milliseconds."); + if (!timeUntilExpires.HasValue) timeUntilExpires = TimeSpan.FromMinutes(20); _logger.LogDebug("Renewing lock {Resource} ({LockId}) for {Duration:g}", resource, lockId, timeUntilExpires); - return _resiliencePolicy.ExecuteAsync(async _ => await _cacheClient.ReplaceIfEqualAsync(resource, lockId, lockId, timeUntilExpires.Value)).AsTask(); + bool renewed = await _resiliencePolicy.ExecuteAsync(async _ => await _cacheClient.ReplaceIfEqualAsync(resource, lockId, lockId, timeUntilExpires.Value).AnyContext()).AnyContext(); + if (!renewed) + throw new LockException($"Unable to renew lock {resource} ({lockId}): the cache rejected renewal and ownership is no longer assured."); } private class ResetEventWithRefCount diff --git a/src/Foundatio/Lock/ILockProvider.cs b/src/Foundatio/Lock/ILockProvider.cs index e09baf64e..15624cf1b 100644 --- a/src/Foundatio/Lock/ILockProvider.cs +++ b/src/Foundatio/Lock/ILockProvider.cs @@ -92,6 +92,17 @@ public interface ILockProvider /// The resource identifier. /// The unique identifier of the lock to renew. /// The new expiration duration from now. + /// + /// Cache-backed leases reject lost ownership. Throttling and empty locks do not renew an exclusive lease. + /// Other providers define their own renewal semantics. + /// + /// + /// received a duration less than 5 milliseconds. + /// + /// + /// A lease provider such as could not renew ownership. + /// Stop the protected work when renewal fails. + /// Task RenewAsync(string resource, string lockId, TimeSpan? timeUntilExpires = null); } @@ -104,6 +115,16 @@ public interface ILock : IAsyncDisposable /// Extends the lock expiration to prevent automatic release during long-running operations. /// /// The new expiration duration from now. + /// + /// Cache-backed leases reject lost ownership. Throttling and empty locks do not renew an exclusive lease. + /// Other providers define their own renewal semantics. + /// + /// + /// received a duration less than 5 milliseconds. + /// + /// + /// The underlying lease provider could not renew ownership. Stop the protected work when renewal fails. + /// Task RenewAsync(TimeSpan? timeUntilExpires = null); /// @@ -308,7 +329,18 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc var locksToRenew = acquiredLocks.Where(al => al.LastRenewed < utcNow.Subtract(renewTime)).ToArray(); if (locksToRenew.Length > 0) { - await Task.WhenAll(locksToRenew.Select(al => al.Lock.RenewAsync(timeUntilExpires))).AnyContext(); + try + { + await Task.WhenAll(locksToRenew.Select(al => al.Lock.RenewAsync(timeUntilExpires))).AnyContext(); + } + catch (LockException ex) + { + // An earlier lock was lost while waiting for this one, so the set can no longer be held as a whole + logger.LogWarning(ex, "Lost an acquired lock while acquiring {Resource}, releasing acquired locks: {Message}", resource, ex.Message); + await Task.WhenAll(acquiredLocks.Select(al => al.Lock).Append(l).Select(al => al.ReleaseAsync())).AnyContext(); + return null; + } + locksToRenew.ForEach(al => al.LastRenewed = utcNow); logger.LogTrace("Renewed {LockCount} locks {Resource} RenewTime={RenewTime:g}", locksToRenew.Length, locksToRenew.Select(al => al.Lock.Resource), renewTime); diff --git a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs index 7420708c9..724102a8a 100644 --- a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs +++ b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs @@ -4,6 +4,7 @@ using Foundatio.Caching; using Foundatio.Lock; using Foundatio.Messaging; +using Foundatio.Utility; using Microsoft.Extensions.Time.Testing; using Xunit; @@ -31,45 +32,45 @@ protected override ILockProvider GetLockProvider() } [Fact] - public override Task CanAcquireAndReleaseLockAsync() + public override Task AcquireAsync_AfterPeriodExhausted_RecoversWithinNextPeriodAsync() { - return base.CanAcquireAndReleaseLockAsync(); + return base.AcquireAsync_AfterPeriodExhausted_RecoversWithinNextPeriodAsync(); } [Fact] - public override Task LockWillTimeoutAsync() + public override Task AcquireAsync_MultiResource_ThrowsWhenAnyLockUnavailableAsync() { - return base.LockWillTimeoutAsync(); + return base.AcquireAsync_MultiResource_ThrowsWhenAnyLockUnavailableAsync(); } [Fact] - public override Task LockWontTimeoutEarly() + public override Task AcquireAsync_ThrowsWhenCancellationTokenCancelledAsync() { - return base.LockWontTimeoutEarly(); + return base.AcquireAsync_ThrowsWhenCancellationTokenCancelledAsync(); } [Fact] - public override Task LockOneAtATimeAsync() + public override Task AcquireAsync_ThrowsWhenLockNotAvailableAsync() { - return base.LockOneAtATimeAsync(); + return base.AcquireAsync_ThrowsWhenLockNotAvailableAsync(); } [Fact] - public override Task CanAcquireMultipleResources() + public override Task AcquireAsync_WithReleaseOnDisposeFalse_DoesNotReleaseOnDispose() { - return base.CanAcquireMultipleResources(); + return base.AcquireAsync_WithReleaseOnDisposeFalse_DoesNotReleaseOnDispose(); } [Fact] - public override Task CanAcquireLocksInParallel() + public override Task CanAcquireAndReleaseLockAsync() { - return base.CanAcquireLocksInParallel(); + return base.CanAcquireAndReleaseLockAsync(); } [Fact] - public override Task CanAcquireScopedLocksInParallel() + public override Task CanAcquireLocksInParallel() { - return base.CanAcquireScopedLocksInParallel(); + return base.CanAcquireLocksInParallel(); } [Fact] @@ -78,6 +79,12 @@ public override Task CanAcquireMultipleLocksInParallel() return base.CanAcquireMultipleLocksInParallel(); } + [Fact] + public override Task CanAcquireMultipleResources() + { + return base.CanAcquireMultipleResources(); + } + [Fact] public override Task CanAcquireMultipleScopedResources() { @@ -85,81 +92,184 @@ public override Task CanAcquireMultipleScopedResources() } [Fact] - public override async Task WillThrottleCallsAsync() + public override Task CanAcquireScopedLocksInParallel() { - // Arrange - var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); - using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); - var period = TimeSpan.FromSeconds(2); - var locker = new ThrottlingLockProvider(cache, 25, period, timeProvider, null, Log); - for (int i = 0; i < 25; i++) - { - await using var hit = await locker.TryAcquireAsync("resource", cancellationToken: TestContext.Current.CancellationToken); - Assert.NotNull(hit); - } + return base.CanAcquireScopedLocksInParallel(); + } - // Act - await using var exhausted = await locker.TryAcquireAsync("resource", cancellationToken: new CancellationToken(true)); - timeProvider.Advance(period); - await using var nextPeriodHit = await locker.TryAcquireAsync("resource", cancellationToken: TestContext.Current.CancellationToken); + [Fact] + public override Task CanReleaseLockMultipleTimes() + { + return base.CanReleaseLockMultipleTimes(); + } - // Assert - Assert.Null(exhausted); - Assert.NotNull(nextPeriodHit); + [Fact] + public override Task Lock_AcquiredTimeUtc_ReturnsValidTimestamp() + { + return base.Lock_AcquiredTimeUtc_ReturnsValidTimestamp(); } [Fact] - public override Task AcquireAsync_AfterPeriodExhausted_RecoversWithinNextPeriodAsync() + public override Task Lock_LockIdAndResource_ReturnCorrectValues() { - return base.AcquireAsync_AfterPeriodExhausted_RecoversWithinNextPeriodAsync(); + return base.Lock_LockIdAndResource_ReturnCorrectValues(); } [Fact] - public override Task AcquireAsync_ThrowsWhenLockNotAvailableAsync() + public override Task LockOneAtATimeAsync() { - return base.AcquireAsync_ThrowsWhenLockNotAvailableAsync(); + return base.LockOneAtATimeAsync(); } [Fact] - public override Task AcquireAsync_ThrowsWhenCancellationTokenCancelledAsync() + public override Task LockWillTimeoutAsync() { - return base.AcquireAsync_ThrowsWhenCancellationTokenCancelledAsync(); + return base.LockWillTimeoutAsync(); } [Fact] - public override Task AcquireAsync_MultiResource_ThrowsWhenAnyLockUnavailableAsync() + public override Task LockWontTimeoutEarly() { - return base.AcquireAsync_MultiResource_ThrowsWhenAnyLockUnavailableAsync(); + return base.LockWontTimeoutEarly(); } [Fact] - public override Task CanReleaseLockMultipleTimes() + public override Task ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockId() { - return base.CanReleaseLockMultipleTimes(); + return base.ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockId(); } [Fact] - public override Task AcquireAsync_WithReleaseOnDisposeFalse_DoesNotReleaseOnDispose() + public override Task RenewAsync_AfterRelease_ThrowsLockExceptionAndDoesNotRecreateLock() { - return base.AcquireAsync_WithReleaseOnDisposeFalse_DoesNotReleaseOnDispose(); + return base.RenewAsync_AfterRelease_ThrowsLockExceptionAndDoesNotRecreateLock(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task RenewAsync_WhenLeaseExpired_ThrowsLockException(bool acquiredByAnotherOwner) + { + // Arrange + var timeProvider = new FakeTimeProvider(DateTimeOffset.UtcNow); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); + var locker = new CacheLockProvider(cache, null, timeProvider, null, Log); + await using var staleLock = await locker.AcquireAsync("resource", TimeSpan.FromMinutes(1), cancellationToken: TestContext.Current.CancellationToken); + timeProvider.Advance(TimeSpan.FromMinutes(2)); + await using var newLock = acquiredByAnotherOwner + ? await locker.AcquireAsync("resource", TimeSpan.FromMinutes(10), cancellationToken: TestContext.Current.CancellationToken) + : null; + + // Act + await Assert.ThrowsAsync(() => staleLock.RenewAsync()); + + // Assert + Assert.Equal(0, staleLock.RenewalCount); + Assert.Equal(acquiredByAnotherOwner, await locker.IsLockedAsync("resource")); + if (newLock is not null) + { + await staleLock.ReleaseAsync(); + Assert.True(await locker.IsLockedAsync("resource")); + await newLock.RenewAsync(); + } } [Fact] - public override Task Lock_AcquiredTimeUtc_ReturnsValidTimestamp() + public async Task RenewAsync_WithCurrentOwner_ExtendsLease() { - return base.Lock_AcquiredTimeUtc_ReturnsValidTimestamp(); + // Arrange + var timeProvider = new FakeTimeProvider(DateTimeOffset.UtcNow); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); + var locker = new CacheLockProvider(cache, null, timeProvider, null, Log); + await using var lockInstance = await locker.AcquireAsync("resource", TimeSpan.FromMinutes(1), cancellationToken: TestContext.Current.CancellationToken); + timeProvider.Advance(TimeSpan.FromSeconds(30)); + + // Act + await lockInstance.RenewAsync(TimeSpan.FromMinutes(3)); + timeProvider.Advance(TimeSpan.FromMinutes(1)); + + // Assert + Assert.True(await locker.IsLockedAsync("resource")); + Assert.Equal(1, lockInstance.RenewalCount); + } + + [Theory] + [InlineData(-1)] + [InlineData(0)] + [InlineData(4)] + public override Task RenewAsync_WithInvalidDuration_PreservesCurrentOwner(int milliseconds) + { + return base.RenewAsync_WithInvalidDuration_PreservesCurrentOwner(milliseconds); } [Fact] - public override Task Lock_LockIdAndResource_ReturnCorrectValues() + public async Task RenewAsync_WithMinimumDuration_RenewsCurrentOwner() { - return base.Lock_LockIdAndResource_ReturnCorrectValues(); + // Arrange + var timeProvider = new FakeTimeProvider(); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider)); + var locker = new CacheLockProvider(cache, null, timeProvider); + await using var lease = await locker.AcquireAsync("resource", TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Act + await lease.RenewAsync(TimeSpan.FromMilliseconds(5)); + + // Assert + Assert.True(await locker.IsLockedAsync("resource")); + Assert.Equal(1, lease.RenewalCount); + timeProvider.Advance(TimeSpan.FromMilliseconds(6)); + Assert.False(await locker.IsLockedAsync("resource")); } [Fact] - public override Task ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockId() + public override Task RenewAsync_WithMissingLock_ThrowsLockException() { - return base.ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockId(); + return base.RenewAsync_WithMissingLock_ThrowsLockException(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public override Task RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(bool scoped) + { + return base.RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(scoped); + } + + [Fact] + public async Task TryAcquireAsync_WithMultipleResources_WhenEarlierLockLost_ReleasesAcquiredLocksAndReturnsNull() + { + // Arrange + var timeProvider = new FakeTimeProvider(DateTimeOffset.UtcNow); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); + var inner = new CacheLockProvider(cache, null, timeProvider, null, Log); + ILock? otherOwner = null; + var locker = new BeforeAcquireLockProvider(inner, timeProvider, async resource => + { + // While "b" is being acquired, "a" expires and another owner takes it, so renewing "a" must fail + if (resource != "b") + return; + + timeProvider.Advance(TimeSpan.FromMinutes(2)); + otherOwner = await inner.AcquireAsync("a", TimeSpan.FromMinutes(10), cancellationToken: TestContext.Current.CancellationToken); + }); + + try + { + // Act + await using var lockInstance = await locker.TryAcquireAsync(["a", "b"], TimeSpan.FromMinutes(1), cancellationToken: TestContext.Current.CancellationToken); + + // Assert + Assert.Null(lockInstance); + Assert.False(await inner.IsLockedAsync("b")); + Assert.NotNull(otherOwner); + Assert.True(await inner.IsLockedAsync("a")); + await otherOwner.RenewAsync(); + } + finally + { + if (otherOwner is not null) + await otherOwner.DisposeAsync(); + } } [Fact] @@ -168,9 +278,58 @@ public override Task TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock( return base.TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock(); } + [Fact] + public override async Task WillThrottleCallsAsync() + { + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); + var period = TimeSpan.FromSeconds(2); + var locker = new ThrottlingLockProvider(cache, 25, period, timeProvider, null, Log); + for (int i = 0; i < 25; i++) + { + await using var hit = await locker.TryAcquireAsync("resource", cancellationToken: TestContext.Current.CancellationToken); + Assert.NotNull(hit); + } + + // Act + await using var exhausted = await locker.TryAcquireAsync("resource", cancellationToken: new CancellationToken(true)); + timeProvider.Advance(period); + await using var nextPeriodHit = await locker.TryAcquireAsync("resource", cancellationToken: TestContext.Current.CancellationToken); + + // Assert + Assert.Null(exhausted); + Assert.NotNull(nextPeriodHit); + } + public void Dispose() { _cache.Dispose(); _messageBus.Dispose(); } + + private sealed class BeforeAcquireLockProvider(ILockProvider inner, TimeProvider timeProvider, Func beforeAcquire) : ILockProvider, IHaveTimeProvider + { + public TimeProvider TimeProvider => timeProvider; + + public async Task AcquireAsync(string resource, TimeSpan? timeUntilExpires = null, bool releaseOnDispose = true, CancellationToken cancellationToken = default) + { + await beforeAcquire(resource); + return await inner.AcquireAsync(resource, timeUntilExpires, releaseOnDispose, cancellationToken); + } + + public async Task TryAcquireAsync(string resource, TimeSpan? timeUntilExpires = null, bool releaseOnDispose = true, CancellationToken cancellationToken = default) + { + await beforeAcquire(resource); + return await inner.TryAcquireAsync(resource, timeUntilExpires, releaseOnDispose, cancellationToken); + } + + public Task IsLockedAsync(string resource) => inner.IsLockedAsync(resource); + + public Task ReleaseAsync(string resource, string lockId) => inner.ReleaseAsync(resource, lockId); + + public Task ReleaseAsync(string resource) => inner.ReleaseAsync(resource); + + public Task RenewAsync(string resource, string lockId, TimeSpan? timeUntilExpires = null) => inner.RenewAsync(resource, lockId, timeUntilExpires); + } } From 266ecfeea5df1491b841cc489e67d447976e5d26 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Mon, 5 Oct 2026 23:21:22 -0500 Subject: [PATCH 2/7] Preserve all-or-nothing multi-lock acquisition and renewal intervals --- .agents/skills/foundatio/SKILL.md | 2 +- docs/guide/locks.md | 2 + .../Locks/LockTestBase.cs | 23 ++++ src/Foundatio/Lock/ILockProvider.cs | 103 +++++++++--------- .../Locks/InMemoryLockTests.cs | 71 ++++++++++++ 5 files changed, 147 insertions(+), 54 deletions(-) diff --git a/.agents/skills/foundatio/SKILL.md b/.agents/skills/foundatio/SKILL.md index 7b059b6d2..c61d243d3 100644 --- a/.agents/skills/foundatio/SKILL.md +++ b/.agents/skills/foundatio/SKILL.md @@ -310,7 +310,7 @@ public class OrderServiceTests : TestLoggerBase - **Failure semantics depend on job type**: `JobResult` only has `IsSuccess` -- there is no separate "failed but don't retry" status. For **queue-processed jobs** (`QueueJobBase.ProcessQueueEntryAsync`, or setting `context.Result` in a `WorkItemJob` handler), a non-success result triggers `AbandonAsync`, which re-queues the entry and eventually dead-letters it after `Retries` is exhausted -- reserve `FailedWithMessage`/`FromException` for transient errors you want retried, and log + return `JobResult.Success`/`SuccessWithMessage(...)` for permanent errors to avoid a pointless retry loop. For **standalone/manual jobs** (`JobBase`, a one-off `RunAsync()`/`RunInConsoleAsync()` run, or scheduled/cron jobs via `Foundatio.Extensions.Hosting`), there is no built-in retry or dead letter queue -- a failed result just produces an error-level log, a non-zero exit code from `RunInConsoleAsync`, or a failed entry in the job run history. Returning `FailedWithMessage`/`FromException` there is correct even for permanent errors, since nothing inside Foundatio will retry it. - **JobWithLockBase vs manual locking**: Use `JobWithLockBase` when the entire run must be single-instance (leader election). Use manual `ILockProvider.AcquireAsync` inside `JobBase` for finer-grained locking within a job. - **JobContext.RenewLockAsync**: Call in long-running jobs (both `JobBase` and `QueueJobBase`) to prevent lock expiration mid-processing. -- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; `WorkItemJob` progress renewal only logs them. Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. +- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; `WorkItemJob` progress renewal only logs them. Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. Multi-resource acquisition is all-or-nothing even for suffix-overlapping names; successful renewals reset their per-lock interval. Provider failures trigger partial-set cleanup while preserving the original error. - **In-memory conditional updates**: Expired conditional replace/remove operations return false. Conditional removal reclaims expired entries and raises `ItemExpired`; live conditional removal and empty-list removal free memory immediately without that event. Rejected sized updates preserve the old entry. `Items` exposes values, honoring `CloneValues`, without affecting LRU. Default and built-in fixed-size large-list writes share unchanged storage; batch updates. Custom sizing, cloning, or raw dictionary access uses full dictionary copies. No-op removals avoid copying. - **In-memory list comparers**: List updates copy Dictionary, SortedDictionary, SortedList, ConcurrentDictionary, and OrderedDictionary (NET9+) with their comparer. Unsupported IDictionary types throw NotSupportedException instead of silently changing equality; the original entry remains unchanged. - **Register as singletons**: All infrastructure services (`ICacheClient`, `IMessageBus`, `IQueue`, `IFileStorage`, `ILockProvider`) maintain internal state and connections -- always register as singletons. diff --git a/docs/guide/locks.md b/docs/guide/locks.md index 3fc996f88..20c2967f1 100644 --- a/docs/guide/locks.md +++ b/docs/guide/locks.md @@ -312,6 +312,8 @@ The local async function starts immediately and yields at the delay; it does not Choose an interval shorter than the lease duration, leaving time for cache latency and retries. A pause can still let the lease expire before renewal detects the loss, and cancellation cannot revoke an operation already dispatched to another system. `WorkItemJob`'s progress-triggered renewal currently logs renewal failures and continues; it does not implement this cancellation pattern. +When acquiring multiple resources, success means every requested lock was acquired. If any resource is unavailable, the provider releases the partial set and returns `null`, including when resource names share a suffix. Provider exceptions also trigger release attempts for every acquired lock before propagating the error. If cleanup fails too, the original acquisition error is preserved and the cleanup failure is logged; the affected leases may remain until expiration. While acquisition is in progress, each successful renewal resets that lock's renewal interval. + ## Common Patterns ### Singleton Processing diff --git a/src/Foundatio.TestHarness/Locks/LockTestBase.cs b/src/Foundatio.TestHarness/Locks/LockTestBase.cs index c1af03f52..a52bb24a8 100644 --- a/src/Foundatio.TestHarness/Locks/LockTestBase.cs +++ b/src/Foundatio.TestHarness/Locks/LockTestBase.cs @@ -739,6 +739,29 @@ public virtual async Task ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockI Assert.False(await locker.IsLockedAsync(lockName).AnyContext()); } + public virtual async Task TryAcquireAsync_WithMultipleResources_WhenResourceNamesShareSuffix_ReleasesAcquiredLocksAndReturnsNull(bool scoped) + { + // Arrange + var locker = GetLockProvider(); + if (locker is null) + return; + if (scoped) + locker = new ScopedLockProvider(locker, "suffix"); + + string heldResource = "b" + Guid.NewGuid().ToString("N"); + string availableResource = "a" + heldResource; + await using var owner = await locker.AcquireAsync(heldResource, TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Act + await using var lockInstance = await locker.TryAcquireAsync([availableResource, heldResource], TimeSpan.FromMinutes(1), cancellationToken: new CancellationToken(true)); + + // Assert + Assert.Null(lockInstance); + Assert.False(await locker.IsLockedAsync(availableResource)); + Assert.True(await locker.IsLockedAsync(heldResource)); + await owner.RenewAsync(); + } + public virtual async Task TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock() { var locker = GetLockProvider(); diff --git a/src/Foundatio/Lock/ILockProvider.cs b/src/Foundatio/Lock/ILockProvider.cs index 15624cf1b..967d71615 100644 --- a/src/Foundatio/Lock/ILockProvider.cs +++ b/src/Foundatio/Lock/ILockProvider.cs @@ -314,73 +314,70 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc var sw = Stopwatch.StartNew(); var acquiredLocks = new List<(ILock Lock, DateTimeOffset LastRenewed)>(resourceList.Length); - foreach (string resource in resourceList) + bool acquiredAll = false; + Exception? acquisitionError = null; + try { - var l = await provider.TryAcquireAsync(resource, timeUntilExpires, releaseOnDispose, cancellationToken).AnyContext(); - if (l is null) + foreach (string resource in resourceList) { - break; - } + var l = await provider.TryAcquireAsync(resource, timeUntilExpires, releaseOnDispose, cancellationToken).AnyContext(); + if (l is null) + return null; - // Renew any acquired locks, so they stay alive until we have all locks - if (acquiredLocks.Count > 0 && renewTime > TimeSpan.Zero) - { - var utcNow = provider.GetTimeProvider().GetUtcNow(); - var locksToRenew = acquiredLocks.Where(al => al.LastRenewed < utcNow.Subtract(renewTime)).ToArray(); - if (locksToRenew.Length > 0) + acquiredLocks.Add((l, provider.GetTimeProvider().GetUtcNow())); + + // Renew earlier locks while waiting for the remaining resources. + if (acquiredLocks.Count > 1 && renewTime > TimeSpan.Zero) { - try - { - await Task.WhenAll(locksToRenew.Select(al => al.Lock.RenewAsync(timeUntilExpires))).AnyContext(); - } - catch (LockException ex) + var utcNow = provider.GetTimeProvider().GetUtcNow(); + var renewBefore = utcNow.Subtract(renewTime); + var locksToRenew = Enumerable.Range(0, acquiredLocks.Count - 1).Where(index => acquiredLocks[index].LastRenewed < renewBefore).ToArray(); + if (locksToRenew.Length > 0) { - // An earlier lock was lost while waiting for this one, so the set can no longer be held as a whole - logger.LogWarning(ex, "Lost an acquired lock while acquiring {Resource}, releasing acquired locks: {Message}", resource, ex.Message); - await Task.WhenAll(acquiredLocks.Select(al => al.Lock).Append(l).Select(al => al.ReleaseAsync())).AnyContext(); - return null; + try + { + await Task.WhenAll(locksToRenew.Select(async index => await acquiredLocks[index].Lock.RenewAsync(timeUntilExpires).AnyContext())).AnyContext(); + } + catch (LockException ex) + { + logger.LogWarning(ex, "Lost an acquired lock while acquiring {Resource}, releasing acquired locks: {Message}", resource, ex.Message); + return null; + } + + foreach (int index in locksToRenew) + acquiredLocks[index] = (acquiredLocks[index].Lock, utcNow); + + logger.LogTrace("Renewed {LockCount} locks {Resource} RenewTime={RenewTime:g}", locksToRenew.Length, locksToRenew.Select(index => acquiredLocks[index].Lock.Resource), renewTime); } - - locksToRenew.ForEach(al => al.LastRenewed = utcNow); - - logger.LogTrace("Renewed {LockCount} locks {Resource} RenewTime={RenewTime:g}", locksToRenew.Length, locksToRenew.Select(al => al.Lock.Resource), renewTime); } } - acquiredLocks.Add((l, provider.GetTimeProvider().GetUtcNow())); + sw.Stop(); + var locks = acquiredLocks.Select(l => l.Lock).ToArray(); + var result = new DisposableLockCollection(locks, String.Join("+", locks.Select(l => l.LockId)), provider.GetTimeProvider().GetUtcNow().UtcDateTime, sw.Elapsed, logger); + logger.LogTrace("Acquired {LockCount} locks {Resource} after {Duration:g}", locks.Length, locks.Select(l => l.Resource), sw.Elapsed); + acquiredAll = true; + return result; } - - sw.Stop(); - - var locks = acquiredLocks.Select(l => l.Lock).ToArray(); - // if any lock is null, release any acquired and return null (all or nothing) - if (resourceList.Length > locks.Length) + catch (Exception ex) { - var unacquiredResources = new List(resourceList.Length); - string[] acquiredResources = locks.Select(l => l.Resource).ToArray(); - - foreach (string resource in resourceList) - { - // account for scoped lock providers with prefixes - if (!acquiredResources.Any(r => r.EndsWith(resource))) - unacquiredResources.Add(resource); - } - - if (unacquiredResources.Count > 0) + acquisitionError = ex; + throw; + } + finally + { + if (!acquiredAll) { - logger.LogTrace("Unable to acquire all {LockCount} locks {UnacquiredResources} releasing acquired locks: {Resource}", unacquiredResources.Count, unacquiredResources, acquiredResources); - - await Task.WhenAll(locks.Select(l => l.ReleaseAsync())).AnyContext(); - - logger.LogTrace("Released {LockCount} locks {Resource}", acquiredResources.Length, acquiredResources); - - return null; + try + { + await Task.WhenAll(acquiredLocks.Select(async acquired => await acquired.Lock.ReleaseAsync().AnyContext())).AnyContext(); + } + catch (Exception ex) when (acquisitionError is not null) + { + logger.LogError(ex, "Unable to release partially acquired locks after acquisition failed: {Message}", ex.Message); + } } } - - logger.LogTrace("Acquired {LockCount} locks {Resource} after {Duration:g}", resourceList.Length, resourceList, sw.Elapsed); - - return new DisposableLockCollection(locks, String.Join("+", locks.Select(l => l.LockId)), provider.GetTimeProvider().GetUtcNow().UtcDateTime, sw.Elapsed, logger); } public static async Task TryAcquireAsync(this ILockProvider provider, IEnumerable resources, TimeSpan? timeUntilExpires, TimeSpan? acquireTimeout) diff --git a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs index 724102a8a..92c4027cb 100644 --- a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs +++ b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs @@ -6,6 +6,7 @@ using Foundatio.Messaging; using Foundatio.Utility; using Microsoft.Extensions.Time.Testing; +using Moq; using Xunit; namespace Foundatio.Tests.Locks; @@ -235,6 +236,33 @@ public override Task RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreserve return base.RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(scoped); } + [Fact] + public async Task TryAcquireAsync_WithMultipleResources_AfterRenewal_WaitsUntilNextRenewalInterval() + { + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); + var inner = new CacheLockProvider(cache, null, timeProvider, null, Log); + var locker = new BeforeAcquireLockProvider(inner, timeProvider, resource => + { + if (resource == "b") + timeProvider.Advance(TimeSpan.FromSeconds(31)); + else if (resource == "c") + timeProvider.Advance(TimeSpan.FromSeconds(1)); + return Task.CompletedTask; + }); + + // Act + await using var lockInstance = await locker.TryAcquireAsync(["a", "b", "c"], TimeSpan.FromMinutes(1), cancellationToken: TestContext.Current.CancellationToken); + + // Assert + Assert.NotNull(lockInstance); + Assert.Equal(TimeSpan.FromSeconds(59), await cache.GetExpirationAsync("lock:a")); + Assert.True(await inner.IsLockedAsync("a")); + Assert.True(await inner.IsLockedAsync("b")); + Assert.True(await inner.IsLockedAsync("c")); + } + [Fact] public async Task TryAcquireAsync_WithMultipleResources_WhenEarlierLockLost_ReleasesAcquiredLocksAndReturnsNull() { @@ -272,6 +300,49 @@ public async Task TryAcquireAsync_WithMultipleResources_WhenEarlierLockLost_Rele } } + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task TryAcquireAsync_WithMultipleResources_WhenProviderThrows_ReleasesAcquiredLocks(bool duringRenewal, bool releaseThrows) + { + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + var first = new Mock(); + var second = new Mock(); + first.Setup(l => l.ReleaseAsync()).Returns(() => releaseThrows + ? throw new CacheException("Release unavailable") + : Task.CompletedTask); + second.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var error = new CacheException("Provider unavailable"); + first.Setup(l => l.RenewAsync(It.IsAny())).ThrowsAsync(error); + var locker = new Mock(); + locker.As().SetupGet(p => p.TimeProvider).Returns(timeProvider); + locker.Setup(p => p.TryAcquireAsync("a", It.IsAny(), It.IsAny(), It.IsAny())).ReturnsAsync(first.Object); + locker.Setup(p => p.TryAcquireAsync("b", It.IsAny(), It.IsAny(), It.IsAny())).Returns(() => + { + timeProvider.Advance(TimeSpan.FromSeconds(31)); + return duringRenewal ? Task.FromResult(second.Object) : Task.FromException(error); + }); + + // Act + var thrown = await Assert.ThrowsAsync(() => locker.Object.TryAcquireAsync(["a", "b"], TimeSpan.FromMinutes(1), cancellationToken: TestContext.Current.CancellationToken)); + + // Assert + Assert.Same(error, thrown); + first.Verify(l => l.ReleaseAsync(), Times.Once); + second.Verify(l => l.ReleaseAsync(), duringRenewal ? Times.Once() : Times.Never()); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public override Task TryAcquireAsync_WithMultipleResources_WhenResourceNamesShareSuffix_ReleasesAcquiredLocksAndReturnsNull(bool scoped) + { + return base.TryAcquireAsync_WithMultipleResources_WhenResourceNamesShareSuffix_ReleasesAcquiredLocksAndReturnsNull(scoped); + } + [Fact] public override Task TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock() { From 87bf11f908ba87a40e7796ba9242fd08fe241018 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Wed, 7 Oct 2026 16:05:48 -0500 Subject: [PATCH 3/7] Abandon work items after a lost lock, let LockException win over provider errors, document breaking changes WorkItemJob no longer completes an entry when the work item lock was lost, even if the handler returns early or swallows the cancellation. Lock-set operations rethrow the definite LockException and log the provider errors they drop. Raises the 250ms lease in the multi-resource renewal tests, restores the InMemoryLockTests member order, and adds a breaking-changes callout to the locks guide. --- .agents/skills/foundatio/SKILL.md | 2 +- docs/guide/jobs.md | 8 +- docs/guide/locks.md | 17 +- .../Locks/LockTestBase.cs | 31 ++- .../Jobs/WorkItemJob/WorkItemHandlers.cs | 4 +- src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs | 76 +++++- .../Lock/DisposableLockCollection.cs | 8 +- src/Foundatio/Lock/ILockProvider.cs | 96 ++++--- src/Foundatio/Lock/LockSetOperations.cs | 45 ++++ .../Foundatio.Tests/Jobs/WorkItemJobTests.cs | 116 +++++++++ .../Locks/InMemoryLockTests.cs | 244 ++++++++++++------ 11 files changed, 498 insertions(+), 149 deletions(-) create mode 100644 src/Foundatio/Lock/LockSetOperations.cs diff --git a/.agents/skills/foundatio/SKILL.md b/.agents/skills/foundatio/SKILL.md index 63511bd55..605463cb2 100644 --- a/.agents/skills/foundatio/SKILL.md +++ b/.agents/skills/foundatio/SKILL.md @@ -312,7 +312,7 @@ public class OrderServiceTests : TestLoggerBase - **Failure semantics depend on job type**: `JobResult` only has `IsSuccess` -- there is no separate "failed but don't retry" status. For **queue-processed jobs** (`QueueJobBase.ProcessQueueEntryAsync`, or setting `context.Result` in a `WorkItemJob` handler), a non-success result triggers `AbandonAsync`, which re-queues the entry and eventually dead-letters it after `Retries` is exhausted -- reserve `FailedWithMessage`/`FromException` for transient errors you want retried, and log + return `JobResult.Success`/`SuccessWithMessage(...)` for permanent errors to avoid a pointless retry loop. For **standalone/manual jobs** (`JobBase`, a one-off `RunAsync()`/`RunInConsoleAsync()` run, or scheduled/cron jobs via `Foundatio.Extensions.Hosting`), there is no built-in retry or dead letter queue -- a failed result just produces an error-level log, a non-zero exit code from `RunInConsoleAsync`, or a failed entry in the job run history. Returning `FailedWithMessage`/`FromException` there is correct even for permanent errors, since nothing inside Foundatio will retry it. - **JobWithLockBase vs manual locking**: Use `JobWithLockBase` when the entire run must be single-instance (leader election). Use manual `ILockProvider.AcquireAsync` inside `JobBase` for finer-grained locking within a job. - **JobContext.RenewLockAsync**: Call in long-running jobs (both `JobBase` and `QueueJobBase`) to prevent lock expiration mid-processing. -- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; `WorkItemJob` progress renewal only logs them. Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. Multi-resource acquisition is all-or-nothing even for suffix-overlapping names; successful renewals reset their per-lock interval. Provider failures trigger partial-set cleanup while preserving the original error. +- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; with `AutoRenewLockOnProgress`, `WorkItemJob` cancels `WorkItemContext.CancellationToken` when the work item lock is lost and abandons the entry instead of completing it, even if the handler returns normally (other renewal errors, including queue entry renewal, are logged). Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. Multi-resource acquisition is all-or-nothing even for suffix-overlapping names; successful renewals reset their per-lock interval. Lock-set operations attempt every member; `LockException` takes precedence over provider errors (which are logged), and partial-set cleanup failures are logged so callers get `null` or the original error. - **In-memory expired entries**: Writes treat an expired entry as missing before maintenance sweeps it (as Redis does): increments and `SetIfHigher`/`SetIfLower` start fresh, `ReplaceAsync`, `SetExpirationAsync`, `RemoveIfEqualAsync` and `ReplaceIfEqualAsync` miss. The write reclaims the expired entry and raises `ItemExpired` once. Successful live conditional removal and empty-list removal free memory without that event. Size-rejected updates preserve value, TTL, and tracked size. - **In-memory numeric TTLs**: `SetIfHigherAsync` and `SetIfLowerAsync` apply the requested expiration even when the numeric value is unchanged. A zero return value does not mean the TTL was preserved; `null` clears it. Size-rejected updates remain unchanged. - **In-memory lists and cloning**: List writes preserve supported dictionary types and comparers; copying an unsupported `IDictionary` throws `NotSupportedException` without changing the entry. Batch large updates because cloning, custom sizing, and raw dictionary access can require whole-list copies. `Items` honors `CloneValues` without affecting LRU. Collection snapshots do not isolate shared mutable elements when cloning is disabled. See the [caching guide](https://foundatio.readthedocs.io/guide/caching.html#in-memory-update-behavior) for details. diff --git a/docs/guide/jobs.md b/docs/guide/jobs.md index f5410fcb8..9deef430b 100644 --- a/docs/guide/jobs.md +++ b/docs/guide/jobs.md @@ -70,7 +70,7 @@ public class CleanupJob : JobBase |--------|-------------| | `CancellationToken` | Signals that the job should stop gracefully | | `Lock` | The distributed lock held by the job (`null` unless using `JobWithLockBase`) | -| `RenewLockAsync()` | Extends the lock lease — call this in long-running loops to prevent expiration. In `QueueEntryContext`, also renews the queue entry's visibility timeout so the message isn't redelivered to another consumer. | +| `RenewLockAsync()` | Extends the lock lease — call this in long-running loops to prevent expiration. In `QueueEntryContext`, also renews the queue entry's visibility timeout so the message isn't redelivered to another consumer. With a cache-backed lock, it throws `LockException` if the lock was lost; let it propagate (or stop the work) because another process may now own it. | ```csharp protected override async Task RunInternalAsync(JobContext context) @@ -441,10 +441,10 @@ public record DeleteEntityWorkItem | `Data` | The raw work item payload (use `GetData()` instead) | | `JobId` | Unique identifier for this job run | | `WorkItemLock` | Optional distributed lock for the work item | -| `CancellationToken` | Signals that processing should stop | +| `CancellationToken` | Signals that processing should stop, including when `AutoRenewLockOnProgress` detects that the work item lock was lost. After a lost lock the entry is abandoned and retried, even if the handler returns normally | | `Result` | Set to `JobResult.FailedWithMessage(...)` to indicate failure without throwing. As with `QueueJobBase`, a non-success `Result` abandons and retries the entry -- see [Retry vs Permanent Failure](#retry-vs-permanent-failure) | -| `ReportProgressAsync(progress, message)` | Publishes `WorkItemStatus` updates via `IMessageBus` | -| `RenewLockAsync()` | Extends the work item lock lease | +| `ReportProgressAsync(progress, message)` | Publishes `WorkItemStatus` updates via `IMessageBus`. With `AutoRenewLockOnProgress`, it also renews the queue entry and work item lock | +| `RenewLockAsync()` | Extends the work item lock lease. Throws `LockException` if a cache-backed lock was lost; stop the work | ### WorkItemHandlers diff --git a/docs/guide/locks.md b/docs/guide/locks.md index 20c2967f1..1d81e6d18 100644 --- a/docs/guide/locks.md +++ b/docs/guide/locks.md @@ -256,8 +256,12 @@ After a process pause or network outage, renewal can succeed if the lease is sti A supplied renewal duration must be at least 5 milliseconds; a shorter duration throws `ArgumentOutOfRangeException` without changing the cache. Omitting the duration uses 20 minutes. Renewal is a no-op for `ThrottlingLockProvider` and `EmptyLock`; these do not provide renewable exclusive leases. -::: warning Behavior change -Earlier versions of `CacheLockProvider.RenewAsync` returned successfully even when renewal failed, and could revive an expired lock. Rejecting renewal durations below 5 milliseconds is also new; the cache already enforced this minimum, but previously could remove the key before checking ownership. +::: danger Breaking changes +- `CacheLockProvider.RenewAsync` throws `LockException` when renewal fails. Earlier versions returned successfully, and could revive an expired lock. Renewal called from an `async void` method or timer callback that doesn't catch it will crash the process; observe renewal in an awaited task as shown below. +- `RenewAsync` throws `ArgumentOutOfRangeException` for durations under 5 milliseconds. The cache already enforced this minimum, but `RenewAsync(TimeSpan.Zero)` used to delete the lock before ownership was checked; it now throws and leaves the lock unchanged. A multi-resource acquisition with `timeUntilExpires` under 5 milliseconds can throw the same exception when it renews earlier locks. +- Multi-resource `AcquireAsync` throws `LockAcquisitionTimeoutException` when an earlier lease is lost while waiting for the remaining resources; `TryAcquireAsync` returns `null`. +- `ScopedLockProvider.RenewAsync` must be passed the held lock's `ILock.Resource`, which includes the scope, and `ILock.LockId`. Renewing with the unscoped resource name used to succeed silently without renewing anything; it now throws `LockException`. +- `WorkItemContext.CancellationToken` is a token linked to the job's token, cancelled also when `AutoRenewLockOnProgress` detects a lost work item lock. Compare it by cancellation state, not by identity with the job token. ::: A lock is not a fencing token. Losing it cannot cancel work you already sent to another system. Checking ownership before a write leaves a race between the check and the write; only validation enforced by the destination can fence a stale owner. Make operations idempotent where appropriate, but do not treat idempotency as exclusive ownership. @@ -291,7 +295,6 @@ async Task RenewPeriodicallyAsync() while (true) { await Task.Delay(TimeSpan.FromSeconds(30), cts.Token); - cts.Token.ThrowIfCancellationRequested(); await lck.RenewAsync(TimeSpan.FromMinutes(1)); } } @@ -310,9 +313,13 @@ async Task RenewPeriodicallyAsync() The local async function starts immediately and yields at the delay; it does not need `Task.Run`. The work must honor the cancellation token. The `finally` block observes renewal failure and waits for renewal to stop before `await using` releases the lock. An in-flight renewal has no cancellation-token parameter, so shutdown waits for that cache operation; configure provider timeouts. -Choose an interval shorter than the lease duration, leaving time for cache latency and retries. A pause can still let the lease expire before renewal detects the loss, and cancellation cannot revoke an operation already dispatched to another system. `WorkItemJob`'s progress-triggered renewal currently logs renewal failures and continues; it does not implement this cancellation pattern. +Choose an interval shorter than the lease duration, leaving time for cache latency and retries. A pause can still let the lease expire before renewal detects the loss, and cancellation cannot revoke an operation already dispatched to another system. -When acquiring multiple resources, success means every requested lock was acquired. If any resource is unavailable, the provider releases the partial set and returns `null`, including when resource names share a suffix. Provider exceptions also trigger release attempts for every acquired lock before propagating the error. If cleanup fails too, the original acquisition error is preserved and the cleanup failure is logged; the affected leases may remain until expiration. While acquisition is in progress, each successful renewal resets that lock's renewal interval. +`WorkItemJob` applies this pattern when a handler sets `AutoRenewLockOnProgress`. Each `ReportProgressAsync` call renews the queue entry and the work item lock. Renewal errors are logged and progress reporting continues. If the work item lock is lost (`LockException`), `WorkItemContext.CancellationToken` is cancelled, and the job abandons the queue entry for retry instead of completing it, whether the handler throws, returns early, or swallows the cancellation. A handler that ignores the token keeps running unprotected until it returns. A failed queue entry renewal is only logged: queues don't report a definite loss, and while the work item lock is held a redelivered entry can't be processed concurrently. + +When acquiring multiple resources, success means every requested lock was acquired. If any resource is unavailable, the provider releases the partial set and returns `null`, including when resource names share a suffix. While acquisition is in progress, earlier locks are renewed once their renewal interval elapses. A lost lease (`LockException`) makes acquisition return `null`, and it takes precedence over a provider error when both occur, because the loss is definite; the provider error is logged. Otherwise a provider error is propagated. Either way, release is attempted for every acquired lock. Cleanup failures are logged rather than thrown, so callers still receive `null` or the original error; any lease that could not be released remains until it expires. + +Renewing a multi-resource lock renews every member, even if one fails. If a member's lease was lost, `RenewAsync` throws `LockException`, even when another member also hit a provider error (that error is logged). `RenewalCount` increases only when every member renews. ## Common Patterns diff --git a/src/Foundatio.TestHarness/Locks/LockTestBase.cs b/src/Foundatio.TestHarness/Locks/LockTestBase.cs index a52bb24a8..db710ef50 100644 --- a/src/Foundatio.TestHarness/Locks/LockTestBase.cs +++ b/src/Foundatio.TestHarness/Locks/LockTestBase.cs @@ -204,7 +204,8 @@ public virtual async Task CanAcquireMultipleResources() return; var resources = new List { "test1", "test2", "test3", "test4", "test5" }; - await using var testLock = await locker.TryAcquireAsync(resources, timeUntilExpires: TimeSpan.FromMilliseconds(250)); + // Long enough that the RenewAsync below can't race the lease expiring, which now throws + await using var testLock = await locker.TryAcquireAsync(resources, timeUntilExpires: TimeSpan.FromSeconds(5)); if (testLock is not null) _logger.LogInformation("Acquired lock attempt #1"); else @@ -243,7 +244,8 @@ public virtual async Task CanAcquireMultipleScopedResources() locker = new ScopedLockProvider(locker, "myscope"); var resources = new List { "test1", "test2", "test3", "test4", "test5" }; - await using var testLock = await locker.TryAcquireAsync(resources, timeUntilExpires: TimeSpan.FromMilliseconds(250)); + // Long enough that the RenewAsync below can't race the lease expiring, which now throws + await using var testLock = await locker.TryAcquireAsync(resources, timeUntilExpires: TimeSpan.FromSeconds(5)); if (testLock is not null) _logger.LogInformation("Acquired lock attempt #1"); else @@ -494,6 +496,31 @@ public virtual async Task RenewAsync_WithMissingLock_ThrowsLockException() Assert.False(await locker.IsLockedAsync(lockName)); } + public virtual async Task RenewAsync_WithMultipleResources_WhenOneLockReplaced_ThrowsLockExceptionAndPreservesCurrentOwner() + { + // Arrange + var locker = GetLockProvider(); + if (locker is null) + return; + + string prefix = Guid.NewGuid().ToString("N")[..10]; + string replacedResource = prefix + "-a"; + string retainedResource = prefix + "-b"; + await using var lockInstance = await locker.AcquireAsync([replacedResource, retainedResource], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + await locker.ReleaseAsync(replacedResource); + await using var currentOwner = await locker.AcquireAsync(replacedResource, TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Act + await Assert.ThrowsAsync(() => lockInstance.RenewAsync()); + + // Assert + Assert.Equal(0, lockInstance.RenewalCount); + Assert.True(await locker.IsLockedAsync(replacedResource)); + Assert.True(await locker.IsLockedAsync(retainedResource)); + await currentOwner.RenewAsync(); + Assert.Equal(1, currentOwner.RenewalCount); + } + public virtual async Task RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(bool scoped) { // Arrange diff --git a/src/Foundatio/Jobs/WorkItemJob/WorkItemHandlers.cs b/src/Foundatio/Jobs/WorkItemJob/WorkItemHandlers.cs index 0b9082673..a99ddaba8 100644 --- a/src/Foundatio/Jobs/WorkItemJob/WorkItemHandlers.cs +++ b/src/Foundatio/Jobs/WorkItemJob/WorkItemHandlers.cs @@ -73,8 +73,10 @@ public interface IWorkItemHandler Task HandleItemAsync(WorkItemContext context); /// - /// When true, the lock on the queue entry is automatically renewed each time + /// When true, the queue entry and work item lock are automatically renewed each time /// the handler reports progress. This prevents long-running work items from timing out. + /// Renewal errors are logged; if the work item lock is lost, + /// is cancelled so cooperative handlers can stop. /// bool AutoRenewLockOnProgress { get; set; } diff --git a/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs b/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs index 5d37547e6..7db00a507 100644 --- a/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs +++ b/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs @@ -3,6 +3,7 @@ using System.Diagnostics; using System.Threading; using System.Threading.Tasks; +using Foundatio.Lock; using Foundatio.Messaging; using Foundatio.Queues; using Foundatio.Serializer; @@ -126,22 +127,11 @@ public async Task ProcessAsync(IQueueEntry queueEntry, return JobResult.CancelledWithMessage($"Unable to acquire work item lock. Abandoning {queueEntry.Value.Type} queue entry: {queueEntry.Id}"); } + using var workItemCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); var progressCallback = new Func(async (progress, message) => { if (handler.AutoRenewLockOnProgress) - { - try - { - await Task.WhenAll( - queueEntry.RenewLockAsync(), - lockValue.RenewAsync() - ).AnyContext(); - } - catch (Exception ex) - { - handler.Log.LogError(ex, "Error renewing work item locks: {Message}", ex.Message); - } - } + await RenewWorkItemLocksAsync(handler, queueEntry, lockValue, workItemCancellation).AnyContext(); await ReportProgressAsync(handler, queueEntry, progress, message).AnyContext(); handler.Log.LogInformation("{TypeName} Progress {Progress}%: {Message}", workItemDataType.Name, progress, message); @@ -150,9 +140,21 @@ await Task.WhenAll( try { handler.LogProcessingQueueEntry(queueEntry, workItemDataType, workItemData); - var workItemContext = new WorkItemContext(workItemData, JobId, lockValue, cancellationToken, progressCallback); + var workItemContext = new WorkItemContext(workItemData, JobId, lockValue, workItemCancellation.Token, progressCallback); await handler.HandleItemAsync(workItemContext).AnyContext(); + // A handler that stops cooperatively after a lost lease (returning early or swallowing the cancellation) + // hasn't finished the work, and another worker may now own it: never acknowledge it + if (workItemCancellation.IsCancellationRequested && !cancellationToken.IsCancellationRequested) + { + if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted) + await queueEntry.AbandonAsync().AnyContext(); + + var lostResult = JobResult.FailedWithMessage($"Abandoning {queueEntry.Value.Type} work item: {queueEntry.Id}: Lost work item lease in handler {workItemDataType.Name}"); + activity?.SetErrorStatus(message: lostResult.Message); + return lostResult; + } + if (!workItemContext.Result.IsSuccess) { if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted) @@ -194,6 +196,52 @@ await Task.WhenAll( } } + /// + /// Renews the queue entry and work item locks. Renewal errors are logged so progress reporting continues, + /// but a lost work item lock cancels because another + /// process may now own the work, and the job then abandons the entry instead of completing it. + /// + /// + /// A failed queue entry renewal doesn't cancel the work item: queues report only provider errors, not a definite + /// loss, and while the work item lock is held a redelivered entry can't be processed concurrently because its + /// worker fails to acquire the same lock and abandons it. + /// + private static async Task RenewWorkItemLocksAsync(IWorkItemHandler handler, IQueueEntry queueEntry, ILock workItemLock, CancellationTokenSource workItemCancellation) + { + var queueEntryRenewal = RenewQueueEntryLockAsync(); + var workItemLockRenewal = RenewWorkItemLockAsync(); + + try + { + await Task.WhenAll(queueEntryRenewal, workItemLockRenewal).AnyContext(); + } + catch (Exception ex) when (workItemLockRenewal.Exception?.InnerException is not LockException) + { + handler.Log.LogError(ex, "Error renewing work item locks: {Message}", ex.Message); + } + catch + { + if (queueEntryRenewal.Exception?.InnerException is { } queueError) + handler.Log.LogError(queueError, "Error renewing queue entry lock: {Message}", queueError.Message); + } + + if (workItemLockRenewal.Exception?.InnerException is not LockException) + return; + + handler.Log.LogWarning("Lost work item lock {Resource} for queue entry {Id}, cancelling work item", workItemLock.Resource, queueEntry.Id); + try + { + await workItemCancellation.CancelAsync().AnyContext(); + } + catch (ObjectDisposedException) + { + // A progress report that wasn't awaited can finish after the work item; nothing is left to cancel + } + + async Task RenewQueueEntryLockAsync() => await queueEntry.RenewLockAsync().AnyContext(); + async Task RenewWorkItemLockAsync() => await workItemLock.RenewAsync().AnyContext(); + } + protected virtual Activity? StartProcessWorkItemActivity(IQueueEntry entry, Type workItemDataType) { var activity = FoundatioDiagnostics.ActivitySource.StartActivity("ProcessQueueEntry", ActivityKind.Internal, entry.CorrelationId); diff --git a/src/Foundatio/Lock/DisposableLockCollection.cs b/src/Foundatio/Lock/DisposableLockCollection.cs index 198112f39..a409ebba9 100644 --- a/src/Foundatio/Lock/DisposableLockCollection.cs +++ b/src/Foundatio/Lock/DisposableLockCollection.cs @@ -42,10 +42,10 @@ public async Task RenewAsync(TimeSpan? lockExtension = null) { _logger.LogTrace("Renewing {LockCount} locks {Resource}", _locks.Count, Resource); - await Task.WhenAll(_locks.Select(l => l.RenewAsync(lockExtension))).AnyContext(); + await LockSetOperations.RenewAllAsync(_locks, lockExtension, _logger).AnyContext(); _renewalCount++; - _logger.LogDebug("Renewing {LockCount} locks {Resource}", _locks.Count, Resource); + _logger.LogDebug("Renewed {LockCount} locks {Resource}", _locks.Count, Resource); } public async Task ReleaseAsync() @@ -63,7 +63,7 @@ public async Task ReleaseAsync() _logger.LogDebug("Releasing {LockCount} locks {Resource} after {Duration:g}", _locks.Count, Resource, _duration.Elapsed); - await Task.WhenAll(_locks.Select(l => l.ReleaseAsync())).AnyContext(); + await LockSetOperations.ReleaseAllAsync(_locks, _logger).AnyContext(); } } @@ -73,7 +73,7 @@ public async ValueTask DisposeAsync() try { - await Task.WhenAll(_locks.Select(l => l.ReleaseAsync())).AnyContext(); + await LockSetOperations.ReleaseAllAsync(_locks, _logger).AnyContext(); } catch (Exception ex) { diff --git a/src/Foundatio/Lock/ILockProvider.cs b/src/Foundatio/Lock/ILockProvider.cs index 967d71615..8444eccf8 100644 --- a/src/Foundatio/Lock/ILockProvider.cs +++ b/src/Foundatio/Lock/ILockProvider.cs @@ -313,70 +313,84 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc logger.LogTrace("Acquiring {LockCount} locks {Resource} RenewTime={RenewTime:g}", resourceList.Length, resourceList, renewTime); var sw = Stopwatch.StartNew(); + var timeProvider = provider.GetTimeProvider(); var acquiredLocks = new List<(ILock Lock, DateTimeOffset LastRenewed)>(resourceList.Length); bool acquiredAll = false; - Exception? acquisitionError = null; try { foreach (string resource in resourceList) { var l = await provider.TryAcquireAsync(resource, timeUntilExpires, releaseOnDispose, cancellationToken).AnyContext(); if (l is null) + { + logger.LogTrace("Unable to acquire lock {Resource}, releasing {LockCount} acquired locks", resource, acquiredLocks.Count); return null; + } - acquiredLocks.Add((l, provider.GetTimeProvider().GetUtcNow())); + acquiredLocks.Add((l, timeProvider.GetUtcNow())); // Renew earlier locks while waiting for the remaining resources. - if (acquiredLocks.Count > 1 && renewTime > TimeSpan.Zero) - { - var utcNow = provider.GetTimeProvider().GetUtcNow(); - var renewBefore = utcNow.Subtract(renewTime); - var locksToRenew = Enumerable.Range(0, acquiredLocks.Count - 1).Where(index => acquiredLocks[index].LastRenewed < renewBefore).ToArray(); - if (locksToRenew.Length > 0) - { - try - { - await Task.WhenAll(locksToRenew.Select(async index => await acquiredLocks[index].Lock.RenewAsync(timeUntilExpires).AnyContext())).AnyContext(); - } - catch (LockException ex) - { - logger.LogWarning(ex, "Lost an acquired lock while acquiring {Resource}, releasing acquired locks: {Message}", resource, ex.Message); - return null; - } - - foreach (int index in locksToRenew) - acquiredLocks[index] = (acquiredLocks[index].Lock, utcNow); - - logger.LogTrace("Renewed {LockCount} locks {Resource} RenewTime={RenewTime:g}", locksToRenew.Length, locksToRenew.Select(index => acquiredLocks[index].Lock.Resource), renewTime); - } - } + if (acquiredLocks.Count > 1 && renewTime > TimeSpan.Zero && !await TryRenewEarlierLocksAsync(acquiredLocks, timeUntilExpires, renewTime, timeProvider, logger).AnyContext()) + return null; } sw.Stop(); var locks = acquiredLocks.Select(l => l.Lock).ToArray(); - var result = new DisposableLockCollection(locks, String.Join("+", locks.Select(l => l.LockId)), provider.GetTimeProvider().GetUtcNow().UtcDateTime, sw.Elapsed, logger); + var result = new DisposableLockCollection(locks, String.Join("+", locks.Select(l => l.LockId)), timeProvider.GetUtcNow().UtcDateTime, sw.Elapsed, logger); logger.LogTrace("Acquired {LockCount} locks {Resource} after {Duration:g}", locks.Length, locks.Select(l => l.Resource), sw.Elapsed); acquiredAll = true; return result; } - catch (Exception ex) + finally { - acquisitionError = ex; - throw; + if (!acquiredAll && acquiredLocks.Count > 0) + await ReleasePartialLocksAsync(acquiredLocks.Select(l => l.Lock).ToArray(), logger).AnyContext(); } - finally + } + + /// + /// Renews every lock except the most recently acquired one whose renewal interval has elapsed. + /// Returns false when a lease was lost; provider errors propagate. + /// + private static async Task TryRenewEarlierLocksAsync(List<(ILock Lock, DateTimeOffset LastRenewed)> acquiredLocks, TimeSpan? timeUntilExpires, TimeSpan renewTime, TimeProvider timeProvider, ILogger logger) + { + var utcNow = timeProvider.GetUtcNow(); + var renewBeforeUtc = utcNow.Subtract(renewTime); + int[] locksToRenew = Enumerable.Range(0, acquiredLocks.Count - 1).Where(index => acquiredLocks[index].LastRenewed < renewBeforeUtc).ToArray(); + if (locksToRenew.Length is 0) + return true; + + try { - if (!acquiredAll) - { - try - { - await Task.WhenAll(acquiredLocks.Select(async acquired => await acquired.Lock.ReleaseAsync().AnyContext())).AnyContext(); - } - catch (Exception ex) when (acquisitionError is not null) - { - logger.LogError(ex, "Unable to release partially acquired locks after acquisition failed: {Message}", ex.Message); - } - } + await LockSetOperations.RenewAllAsync(locksToRenew.Select(index => acquiredLocks[index].Lock), timeUntilExpires, logger).AnyContext(); + } + catch (LockException ex) + { + logger.LogWarning(ex, "Lost an acquired lock while acquiring {Resource}, releasing acquired locks: {Message}", acquiredLocks[^1].Lock.Resource, ex.Message); + return false; + } + + foreach (int index in locksToRenew) + acquiredLocks[index] = (acquiredLocks[index].Lock, utcNow); + + logger.LogTrace("Renewed {LockCount} locks {Resource} RenewTime={RenewTime:g}", locksToRenew.Length, locksToRenew.Select(index => acquiredLocks[index].Lock.Resource), renewTime); + return true; + } + + /// + /// Attempts to release every partially acquired lock. Failures are logged rather than thrown so that callers receive + /// null or the original acquisition error; leases that could not be released remain held until they expire. + /// + private static async Task ReleasePartialLocksAsync(ILock[] locks, ILogger logger) + { + try + { + await LockSetOperations.ReleaseAllAsync(locks, logger).AnyContext(); + logger.LogTrace("Released {LockCount} partially acquired locks {Resource}", locks.Length, locks.Select(l => l.Resource)); + } + catch (Exception ex) + { + logger.LogError(ex, "Unable to release {LockCount} partially acquired locks {Resource}; unreleased locks remain held until they expire: {Message}", locks.Length, locks.Select(l => l.Resource), ex.Message); } } diff --git a/src/Foundatio/Lock/LockSetOperations.cs b/src/Foundatio/Lock/LockSetOperations.cs new file mode 100644 index 000000000..3b19baf40 --- /dev/null +++ b/src/Foundatio/Lock/LockSetOperations.cs @@ -0,0 +1,45 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.ExceptionServices; +using System.Threading.Tasks; +using Foundatio.Utility; +using Microsoft.Extensions.Logging; + +namespace Foundatio.Lock; + +/// +/// Runs an operation against every lock in a set. Every operation is attempted, even when one throws synchronously. +/// When operations fail, the first is rethrown, because a lost lease is definite while a +/// provider error may be transient; otherwise the first error is rethrown. Errors that are not rethrown are logged. +/// +internal static class LockSetOperations +{ + public static Task RenewAllAsync(IEnumerable locks, TimeSpan? timeUntilExpires, ILogger logger) + { + return WhenAllAsync(locks.Select(async l => await l.RenewAsync(timeUntilExpires).AnyContext()), logger); + } + + public static Task ReleaseAllAsync(IEnumerable locks, ILogger logger) + { + return WhenAllAsync(locks.Select(async l => await l.ReleaseAsync().AnyContext()), logger); + } + + private static async Task WhenAllAsync(IEnumerable operations, ILogger logger) + { + var task = Task.WhenAll(operations); + try + { + await task.AnyContext(); + } + catch (Exception) when (task.Exception is { } aggregate) + { + var errors = aggregate.InnerExceptions; + var error = errors.FirstOrDefault(e => e is LockException) ?? errors[0]; + foreach (var dropped in errors.Where(e => !ReferenceEquals(e, error))) + logger.LogError(dropped, "Lock operation failed alongside {ErrorType}: {Message}", error.GetType().Name, dropped.Message); + + ExceptionDispatchInfo.Throw(error); + } + } +} diff --git a/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs b/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs index 9b1cd4226..c6d5a5955 100644 --- a/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs +++ b/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs @@ -7,7 +7,9 @@ using System.Threading.Tasks; using Exceptionless; using Foundatio.AsyncEx; +using Foundatio.Caching; using Foundatio.Jobs; +using Foundatio.Lock; using Foundatio.Messaging; using Foundatio.Queues; using Foundatio.Tests.Extensions; @@ -334,6 +336,120 @@ await messageBus.SubscribeAsync(status => await countdown.WaitAsync(TimeSpan.FromSeconds(2)); Assert.Equal(0, countdown.CurrentCount); } + + [Theory] + [InlineData(LostLeaseResponse.Throw)] + [InlineData(LostLeaseResponse.ReturnEarly)] + [InlineData(LostLeaseResponse.SwallowCancellation)] + public async Task ProcessAsync_WhenAutoRenewedWorkItemLockIsLost_CancelsHandlerAndAbandonsEntry(LostLeaseResponse response) + { + // Arrange + using var cache = new InMemoryCacheClient(o => o.LoggerFactory(Log)); + using var queue = new InMemoryQueue(o => o.RetryDelay(TimeSpan.Zero).Retries(0).LoggerFactory(Log)); + using var messageBus = new InMemoryMessageBus(o => o.LoggerFactory(Log)); + var locker = new CacheLockProvider(cache, messageBus, null, null, Log); + var handler = new LockedWorkItemHandler(locker, Log, response: response) { AutoRenewLockOnProgress = true }; + var handlerRegistry = new WorkItemHandlers(); + handlerRegistry.Register(handler); + var job = new WorkItemJob(queue, messageBus, handlerRegistry, Log); + await queue.EnqueueAsync(new MyWorkItem { SomeData = "Test" }, true); + + // Act + var result = await job.RunAsync(TestCancellationToken); + + // Assert + Assert.False(result.IsSuccess); + Assert.True(handler.ObservedCancellation); + Assert.False(handler.CompletedWork); + var stats = await queue.GetQueueStatsAsync(); + Assert.Equal(0, stats.Completed); + Assert.Equal(1, stats.Abandoned); + } + + [Fact] + public async Task ProcessAsync_WhenAutoRenewedWorkItemLockIsHeld_DoesNotCancelHandler() + { + // Arrange + using var cache = new InMemoryCacheClient(o => o.LoggerFactory(Log)); + using var queue = new InMemoryQueue(o => o.LoggerFactory(Log)); + using var messageBus = new InMemoryMessageBus(o => o.LoggerFactory(Log)); + var locker = new CacheLockProvider(cache, messageBus, null, null, Log); + var handler = new LockedWorkItemHandler(locker, Log, loseLock: false) { AutoRenewLockOnProgress = true }; + var handlerRegistry = new WorkItemHandlers(); + handlerRegistry.Register(handler); + var job = new WorkItemJob(queue, messageBus, handlerRegistry, Log); + await queue.EnqueueAsync(new MyWorkItem { SomeData = "Test" }, true); + + // Act + var result = await job.RunAsync(TestCancellationToken); + + // Assert + Assert.True(result.IsSuccess); + Assert.False(handler.ObservedCancellation); + Assert.True(handler.CompletedWork); + var stats = await queue.GetQueueStatsAsync(); + Assert.Equal(1, stats.Completed); + } +} + +public enum LostLeaseResponse +{ + Throw, + ReturnEarly, + SwallowCancellation +} + +public class LockedWorkItemHandler : WorkItemHandlerBase +{ + private readonly ILockProvider _locker; + private readonly bool _loseLock; + private readonly LostLeaseResponse _response; + + public LockedWorkItemHandler(ILockProvider locker, ILoggerFactory loggerFactory, bool loseLock = true, LostLeaseResponse response = LostLeaseResponse.Throw) : base(loggerFactory) + { + _locker = locker; + _loseLock = loseLock; + _response = response; + } + + public bool ObservedCancellation { get; private set; } + public bool CompletedWork { get; private set; } + + public override Task GetWorkItemLockAsync(object workItem, CancellationToken cancellationToken = default) + { + return _locker.TryAcquireAsync("work-item", TimeSpan.FromMinutes(1), cancellationToken: cancellationToken); + } + + public override async Task HandleItemAsync(WorkItemContext context) + { + Assert.NotNull(context.WorkItemLock); + if (_loseLock) + await _locker.ReleaseAsync(context.WorkItemLock.Resource); + + await context.ReportProgressAsync(50); + ObservedCancellation = context.CancellationToken.IsCancellationRequested; + switch (_response) + { + case LostLeaseResponse.ReturnEarly when ObservedCancellation: + return; + case LostLeaseResponse.SwallowCancellation: + try + { + context.CancellationToken.ThrowIfCancellationRequested(); + } + catch (OperationCanceledException) + { + return; + } + + break; + default: + context.CancellationToken.ThrowIfCancellationRequested(); + break; + } + + CompletedWork = true; + } } public class MyWorkItem diff --git a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs index a2fa6d364..2a4ac8ed1 100644 --- a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs +++ b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs @@ -33,45 +33,45 @@ protected override ILockProvider GetLockProvider() } [Fact] - public override Task AcquireAsync_AfterPeriodExhausted_RecoversWithinNextPeriodAsync() + public override Task CanAcquireAndReleaseLockAsync() { - return base.AcquireAsync_AfterPeriodExhausted_RecoversWithinNextPeriodAsync(); + return base.CanAcquireAndReleaseLockAsync(); } [Fact] - public override Task AcquireAsync_MultiResource_ThrowsWhenAnyLockUnavailableAsync() + public override Task LockWillTimeoutAsync() { - return base.AcquireAsync_MultiResource_ThrowsWhenAnyLockUnavailableAsync(); + return base.LockWillTimeoutAsync(); } [Fact] - public override Task AcquireAsync_ThrowsWhenCancellationTokenCancelledAsync() + public override Task LockWontTimeoutEarly() { - return base.AcquireAsync_ThrowsWhenCancellationTokenCancelledAsync(); + return base.LockWontTimeoutEarly(); } [Fact] - public override Task AcquireAsync_ThrowsWhenLockNotAvailableAsync() + public override Task LockOneAtATimeAsync() { - return base.AcquireAsync_ThrowsWhenLockNotAvailableAsync(); + return base.LockOneAtATimeAsync(); } [Fact] - public override Task AcquireAsync_WithReleaseOnDisposeFalse_DoesNotReleaseOnDispose() + public override Task CanAcquireMultipleResources() { - return base.AcquireAsync_WithReleaseOnDisposeFalse_DoesNotReleaseOnDispose(); + return base.CanAcquireMultipleResources(); } [Fact] - public override Task CanAcquireAndReleaseLockAsync() + public override Task CanAcquireLocksInParallel() { - return base.CanAcquireAndReleaseLockAsync(); + return base.CanAcquireLocksInParallel(); } [Fact] - public override Task CanAcquireLocksInParallel() + public override Task CanAcquireScopedLocksInParallel() { - return base.CanAcquireLocksInParallel(); + return base.CanAcquireScopedLocksInParallel(); } [Fact] @@ -81,21 +81,39 @@ public override Task CanAcquireMultipleLocksInParallel() } [Fact] - public override Task CanAcquireMultipleResources() + public override Task CanAcquireMultipleScopedResources() { - return base.CanAcquireMultipleResources(); + return base.CanAcquireMultipleScopedResources(); } [Fact] - public override Task CanAcquireMultipleScopedResources() + public override Task WillThrottleCallsAsync() { - return base.CanAcquireMultipleScopedResources(); + return base.WillThrottleCallsAsync(); } [Fact] - public override Task CanAcquireScopedLocksInParallel() + public override Task AcquireAsync_AfterPeriodExhausted_RecoversWithinNextPeriodAsync() { - return base.CanAcquireScopedLocksInParallel(); + return base.AcquireAsync_AfterPeriodExhausted_RecoversWithinNextPeriodAsync(); + } + + [Fact] + public override Task AcquireAsync_ThrowsWhenLockNotAvailableAsync() + { + return base.AcquireAsync_ThrowsWhenLockNotAvailableAsync(); + } + + [Fact] + public override Task AcquireAsync_ThrowsWhenCancellationTokenCancelledAsync() + { + return base.AcquireAsync_ThrowsWhenCancellationTokenCancelledAsync(); + } + + [Fact] + public override Task AcquireAsync_MultiResource_ThrowsWhenAnyLockUnavailableAsync() + { + return base.AcquireAsync_MultiResource_ThrowsWhenAnyLockUnavailableAsync(); } [Fact] @@ -104,6 +122,12 @@ public override Task CanReleaseLockMultipleTimes() return base.CanReleaseLockMultipleTimes(); } + [Fact] + public override Task AcquireAsync_WithReleaseOnDisposeFalse_DoesNotReleaseOnDispose() + { + return base.AcquireAsync_WithReleaseOnDisposeFalse_DoesNotReleaseOnDispose(); + } + [Fact] public override Task Lock_AcquiredTimeUtc_ReturnsValidTimestamp() { @@ -117,27 +141,58 @@ public override Task Lock_LockIdAndResource_ReturnCorrectValues() } [Fact] - public override Task LockOneAtATimeAsync() + public override Task ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockId() { - return base.LockOneAtATimeAsync(); + return base.ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockId(); } [Fact] - public override Task LockWillTimeoutAsync() + public async Task TryAcquireAsync_WithExhaustedPeriod_WaitsForNextPeriod() { - return base.LockWillTimeoutAsync(); - } + // Arrange + const int allowedLocks = 25; + var period = TimeSpan.FromSeconds(2); + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); + var locker = new ThrottlingLockProvider(cache, allowedLocks, period, timeProvider, null, Log); + for (int i = 0; i < allowedLocks; i++) + { + await using var hit = await locker.TryAcquireAsync("resource", cancellationToken: TestCancellationToken); + Assert.NotNull(hit); + } + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestCancellationToken); - [Fact] - public override Task LockWontTimeoutEarly() - { - return base.LockWontTimeoutEarly(); + // Act + var waiting = locker.TryAcquireAsync("resource", cancellationToken: cancellation.Token); + try + { + bool waitsForQuota = !waiting.IsCompleted; + timeProvider.Advance(period - TimeSpan.FromMilliseconds(1)); + bool waitsBeforeBoundary = !waiting.IsCompleted; + bool exhaustedBeforeBoundary = await locker.IsLockedAsync("resource"); + timeProvider.Advance(TimeSpan.FromMilliseconds(1)); + bool exhaustedAtBoundary = await locker.IsLockedAsync("resource"); + // The provider wakes one millisecond after the boundary to avoid early system-timer wakeups. + timeProvider.Advance(TimeSpan.FromMilliseconds(1)); + await using var nextPeriodHit = await waiting.WaitAsync(TimeSpan.FromSeconds(5), TestCancellationToken); + + // Assert + Assert.True(waitsForQuota); + Assert.True(waitsBeforeBoundary); + Assert.True(exhaustedBeforeBoundary); + Assert.False(exhaustedAtBoundary); + Assert.NotNull(nextPeriodHit); + } + finally + { + await cancellation.CancelAsync(); + } } [Fact] - public override Task ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockId() + public override Task TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock() { - return base.ReleaseAsync_WithForceRelease_ReleasesLockWithoutLockId(); + return base.TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock(); } [Fact] @@ -228,55 +283,41 @@ public override Task RenewAsync_WithMissingLock_ThrowsLockException() return base.RenewAsync_WithMissingLock_ThrowsLockException(); } - [Theory] - [InlineData(false)] - [InlineData(true)] - public override Task RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(bool scoped) - { - return base.RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(scoped); - } - [Fact] - public async Task TryAcquireAsync_WithExhaustedPeriod_WaitsForNextPeriod() + public async Task RenewAsync_WithMultipleResources_WhenFirstLockThrowsSynchronously_AttemptsEveryRenewal() { // Arrange - const int allowedLocks = 25; - var period = TimeSpan.FromSeconds(2); var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); - using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); - var locker = new ThrottlingLockProvider(cache, allowedLocks, period, timeProvider, null, Log); - for (int i = 0; i < allowedLocks; i++) - { - await using var hit = await locker.TryAcquireAsync("resource", cancellationToken: TestCancellationToken); - Assert.NotNull(hit); - } - using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestCancellationToken); + var error = new CacheException("Provider unavailable"); + var first = new Mock(); + first.Setup(l => l.RenewAsync(It.IsAny())).Throws(error); + var second = new Mock(); + second.Setup(l => l.RenewAsync(It.IsAny())).Returns(Task.CompletedTask); + var locker = CreateMockLockProvider(timeProvider, ("a", first.Object), ("b", second.Object)); + await using var lockInstance = await locker.Object.TryAcquireAsync(["a", "b"], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + Assert.NotNull(lockInstance); // Act - var waiting = locker.TryAcquireAsync("resource", cancellationToken: cancellation.Token); - try - { - bool waitsForQuota = !waiting.IsCompleted; - timeProvider.Advance(period - TimeSpan.FromMilliseconds(1)); - bool waitsBeforeBoundary = !waiting.IsCompleted; - bool exhaustedBeforeBoundary = await locker.IsLockedAsync("resource"); - timeProvider.Advance(TimeSpan.FromMilliseconds(1)); - bool exhaustedAtBoundary = await locker.IsLockedAsync("resource"); - // The provider wakes one millisecond after the boundary to avoid early system-timer wakeups. - timeProvider.Advance(TimeSpan.FromMilliseconds(1)); - await using var nextPeriodHit = await waiting.WaitAsync(TimeSpan.FromSeconds(5), TestCancellationToken); + var thrown = await Assert.ThrowsAsync(() => lockInstance.RenewAsync()); - // Assert - Assert.True(waitsForQuota); - Assert.True(waitsBeforeBoundary); - Assert.True(exhaustedBeforeBoundary); - Assert.False(exhaustedAtBoundary); - Assert.NotNull(nextPeriodHit); - } - finally - { - await cancellation.CancelAsync(); - } + // Assert + Assert.Same(error, thrown); + Assert.Equal(0, lockInstance.RenewalCount); + second.Verify(l => l.RenewAsync(It.IsAny()), Times.Once); + } + + [Fact] + public override Task RenewAsync_WithMultipleResources_WhenOneLockReplaced_ThrowsLockExceptionAndPreservesCurrentOwner() + { + return base.RenewAsync_WithMultipleResources_WhenOneLockReplaced_ThrowsLockExceptionAndPreservesCurrentOwner(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public override Task RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(bool scoped) + { + return base.RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(scoped); } [Fact] @@ -387,15 +428,54 @@ public override Task TryAcquireAsync_WithMultipleResources_WhenResourceNamesShar } [Fact] - public override Task TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock() + public async Task TryAcquireAsync_WithMultipleResources_WhenRenewalLosesLockAndProviderThrows_ReturnsNullAndReleasesAcquiredLocks() { - return base.TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock(); + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + var providerError = new CacheException("Provider unavailable"); + var lost = new Mock(); + lost.Setup(l => l.RenewAsync(It.IsAny())).ThrowsAsync(new LockException("Lock lost")); + lost.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var failing = new Mock(); + failing.Setup(l => l.RenewAsync(It.IsAny())).ThrowsAsync(providerError); + failing.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var last = new Mock(); + last.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var locker = CreateMockLockProvider(timeProvider, ("a", lost.Object), ("b", failing.Object), ("c", last.Object)); + locker.Setup(p => p.TryAcquireAsync("c", It.IsAny(), It.IsAny(), It.IsAny())).Returns(() => + { + timeProvider.Advance(TimeSpan.FromSeconds(31)); + return Task.FromResult(last.Object); + }); + + // Act + var lockInstance = await locker.Object.TryAcquireAsync(["a", "b", "c"], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Assert + Assert.Null(lockInstance); + lost.Verify(l => l.ReleaseAsync(), Times.Once); + failing.Verify(l => l.ReleaseAsync(), Times.Once); + last.Verify(l => l.ReleaseAsync(), Times.Once); } [Fact] - public override Task WillThrottleCallsAsync() + public async Task TryAcquireAsync_WithMultipleResources_WhenUnavailableAndReleaseThrows_ReturnsNullAndAttemptsEveryRelease() { - return base.WillThrottleCallsAsync(); + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + var first = new Mock(); + first.Setup(l => l.ReleaseAsync()).Throws(new CacheException("Release unavailable")); + var second = new Mock(); + second.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var locker = CreateMockLockProvider(timeProvider, ("a", first.Object), ("b", second.Object), ("c", null)); + + // Act + var lockInstance = await locker.Object.TryAcquireAsync(["a", "b", "c"], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Assert + Assert.Null(lockInstance); + first.Verify(l => l.ReleaseAsync(), Times.Once); + second.Verify(l => l.ReleaseAsync(), Times.Once); } public void Dispose() @@ -404,6 +484,16 @@ public void Dispose() _messageBus.Dispose(); } + private static Mock CreateMockLockProvider(TimeProvider timeProvider, params (string Resource, ILock? Lock)[] locks) + { + var locker = new Mock(); + locker.As().SetupGet(p => p.TimeProvider).Returns(timeProvider); + foreach (var (resource, lockInstance) in locks) + locker.Setup(p => p.TryAcquireAsync(resource, It.IsAny(), It.IsAny(), It.IsAny())).ReturnsAsync(lockInstance); + + return locker; + } + private sealed class BeforeAcquireLockProvider(ILockProvider inner, TimeProvider timeProvider, Func beforeAcquire) : ILockProvider, IHaveTimeProvider { public TimeProvider TimeProvider => timeProvider; From 562ba954e5b8c77ed664a891a3fde339bef48aa3 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Wed, 7 Oct 2026 16:27:44 -0500 Subject: [PATCH 4/7] Make lock-set renewal deterministic and cancel WorkItemJob work on lost lock - Attempt every renew/release in a lock set and prefer provider errors over LockException - Track acquired locks by position instead of name suffix and log cleanup failures on the null-return path - Cancel WorkItemContext.CancellationToken when automatic progress renewal loses the lock - Add regression tests and update lock/job docs --- .agents/skills/foundatio/SKILL.md | 2 +- docs/guide/jobs.md | 8 +- docs/guide/locks.md | 9 +- .../Locks/LockTestBase.cs | 25 +++++ .../Jobs/WorkItemJob/WorkItemHandlers.cs | 4 +- src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs | 47 ++++++--- .../Lock/DisposableLockCollection.cs | 8 +- src/Foundatio/Lock/ILockProvider.cs | 96 +++++++++++-------- src/Foundatio/Lock/LockSetOperations.cs | 40 ++++++++ .../Foundatio.Tests/Jobs/WorkItemJobTests.cs | 85 ++++++++++++++++ .../Locks/InMemoryLockTests.cs | 90 +++++++++++++++++ 11 files changed, 346 insertions(+), 68 deletions(-) create mode 100644 src/Foundatio/Lock/LockSetOperations.cs diff --git a/.agents/skills/foundatio/SKILL.md b/.agents/skills/foundatio/SKILL.md index 63511bd55..ba1ba351c 100644 --- a/.agents/skills/foundatio/SKILL.md +++ b/.agents/skills/foundatio/SKILL.md @@ -312,7 +312,7 @@ public class OrderServiceTests : TestLoggerBase - **Failure semantics depend on job type**: `JobResult` only has `IsSuccess` -- there is no separate "failed but don't retry" status. For **queue-processed jobs** (`QueueJobBase.ProcessQueueEntryAsync`, or setting `context.Result` in a `WorkItemJob` handler), a non-success result triggers `AbandonAsync`, which re-queues the entry and eventually dead-letters it after `Retries` is exhausted -- reserve `FailedWithMessage`/`FromException` for transient errors you want retried, and log + return `JobResult.Success`/`SuccessWithMessage(...)` for permanent errors to avoid a pointless retry loop. For **standalone/manual jobs** (`JobBase`, a one-off `RunAsync()`/`RunInConsoleAsync()` run, or scheduled/cron jobs via `Foundatio.Extensions.Hosting`), there is no built-in retry or dead letter queue -- a failed result just produces an error-level log, a non-zero exit code from `RunInConsoleAsync`, or a failed entry in the job run history. Returning `FailedWithMessage`/`FromException` there is correct even for permanent errors, since nothing inside Foundatio will retry it. - **JobWithLockBase vs manual locking**: Use `JobWithLockBase` when the entire run must be single-instance (leader election). Use manual `ILockProvider.AcquireAsync` inside `JobBase` for finer-grained locking within a job. - **JobContext.RenewLockAsync**: Call in long-running jobs (both `JobBase` and `QueueJobBase`) to prevent lock expiration mid-processing. -- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; `WorkItemJob` progress renewal only logs them. Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. Multi-resource acquisition is all-or-nothing even for suffix-overlapping names; successful renewals reset their per-lock interval. Provider failures trigger partial-set cleanup while preserving the original error. +- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; with `AutoRenewLockOnProgress`, `WorkItemJob` cancels `WorkItemContext.CancellationToken` when the work item lock is lost (other renewal errors are logged). Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. Multi-resource acquisition is all-or-nothing even for suffix-overlapping names; successful renewals reset their per-lock interval. Lock-set operations attempt every member; provider errors take precedence over `LockException`, and partial-set cleanup failures are logged so callers get `null` or the original error. - **In-memory expired entries**: Writes treat an expired entry as missing before maintenance sweeps it (as Redis does): increments and `SetIfHigher`/`SetIfLower` start fresh, `ReplaceAsync`, `SetExpirationAsync`, `RemoveIfEqualAsync` and `ReplaceIfEqualAsync` miss. The write reclaims the expired entry and raises `ItemExpired` once. Successful live conditional removal and empty-list removal free memory without that event. Size-rejected updates preserve value, TTL, and tracked size. - **In-memory numeric TTLs**: `SetIfHigherAsync` and `SetIfLowerAsync` apply the requested expiration even when the numeric value is unchanged. A zero return value does not mean the TTL was preserved; `null` clears it. Size-rejected updates remain unchanged. - **In-memory lists and cloning**: List writes preserve supported dictionary types and comparers; copying an unsupported `IDictionary` throws `NotSupportedException` without changing the entry. Batch large updates because cloning, custom sizing, and raw dictionary access can require whole-list copies. `Items` honors `CloneValues` without affecting LRU. Collection snapshots do not isolate shared mutable elements when cloning is disabled. See the [caching guide](https://foundatio.readthedocs.io/guide/caching.html#in-memory-update-behavior) for details. diff --git a/docs/guide/jobs.md b/docs/guide/jobs.md index f5410fcb8..db2198755 100644 --- a/docs/guide/jobs.md +++ b/docs/guide/jobs.md @@ -70,7 +70,7 @@ public class CleanupJob : JobBase |--------|-------------| | `CancellationToken` | Signals that the job should stop gracefully | | `Lock` | The distributed lock held by the job (`null` unless using `JobWithLockBase`) | -| `RenewLockAsync()` | Extends the lock lease — call this in long-running loops to prevent expiration. In `QueueEntryContext`, also renews the queue entry's visibility timeout so the message isn't redelivered to another consumer. | +| `RenewLockAsync()` | Extends the lock lease — call this in long-running loops to prevent expiration. In `QueueEntryContext`, also renews the queue entry's visibility timeout so the message isn't redelivered to another consumer. With a cache-backed lock, it throws `LockException` if the lock was lost; let it propagate (or stop the work) because another process may now own it. | ```csharp protected override async Task RunInternalAsync(JobContext context) @@ -441,10 +441,10 @@ public record DeleteEntityWorkItem | `Data` | The raw work item payload (use `GetData()` instead) | | `JobId` | Unique identifier for this job run | | `WorkItemLock` | Optional distributed lock for the work item | -| `CancellationToken` | Signals that processing should stop | +| `CancellationToken` | Signals that processing should stop, including when `AutoRenewLockOnProgress` detects that the work item lock was lost | | `Result` | Set to `JobResult.FailedWithMessage(...)` to indicate failure without throwing. As with `QueueJobBase`, a non-success `Result` abandons and retries the entry -- see [Retry vs Permanent Failure](#retry-vs-permanent-failure) | -| `ReportProgressAsync(progress, message)` | Publishes `WorkItemStatus` updates via `IMessageBus` | -| `RenewLockAsync()` | Extends the work item lock lease | +| `ReportProgressAsync(progress, message)` | Publishes `WorkItemStatus` updates via `IMessageBus`. With `AutoRenewLockOnProgress`, it also renews the queue entry and work item lock | +| `RenewLockAsync()` | Extends the work item lock lease. Throws `LockException` if a cache-backed lock was lost; stop the work | ### WorkItemHandlers diff --git a/docs/guide/locks.md b/docs/guide/locks.md index 20c2967f1..5f916c82c 100644 --- a/docs/guide/locks.md +++ b/docs/guide/locks.md @@ -291,7 +291,6 @@ async Task RenewPeriodicallyAsync() while (true) { await Task.Delay(TimeSpan.FromSeconds(30), cts.Token); - cts.Token.ThrowIfCancellationRequested(); await lck.RenewAsync(TimeSpan.FromMinutes(1)); } } @@ -310,9 +309,13 @@ async Task RenewPeriodicallyAsync() The local async function starts immediately and yields at the delay; it does not need `Task.Run`. The work must honor the cancellation token. The `finally` block observes renewal failure and waits for renewal to stop before `await using` releases the lock. An in-flight renewal has no cancellation-token parameter, so shutdown waits for that cache operation; configure provider timeouts. -Choose an interval shorter than the lease duration, leaving time for cache latency and retries. A pause can still let the lease expire before renewal detects the loss, and cancellation cannot revoke an operation already dispatched to another system. `WorkItemJob`'s progress-triggered renewal currently logs renewal failures and continues; it does not implement this cancellation pattern. +Choose an interval shorter than the lease duration, leaving time for cache latency and retries. A pause can still let the lease expire before renewal detects the loss, and cancellation cannot revoke an operation already dispatched to another system. -When acquiring multiple resources, success means every requested lock was acquired. If any resource is unavailable, the provider releases the partial set and returns `null`, including when resource names share a suffix. Provider exceptions also trigger release attempts for every acquired lock before propagating the error. If cleanup fails too, the original acquisition error is preserved and the cleanup failure is logged; the affected leases may remain until expiration. While acquisition is in progress, each successful renewal resets that lock's renewal interval. +`WorkItemJob` applies this pattern when a handler sets `AutoRenewLockOnProgress`. Each `ReportProgressAsync` call renews the queue entry and the work item lock. Renewal errors are logged and progress reporting continues. If the work item lock is lost (`LockException`), `WorkItemContext.CancellationToken` is cancelled. A handler that honors the token stops, and its queue entry is abandoned and retried. A handler that ignores the token keeps running unprotected. + +When acquiring multiple resources, success means every requested lock was acquired. If any resource is unavailable, the provider releases the partial set and returns `null`, including when resource names share a suffix. While acquisition is in progress, earlier locks are renewed once their renewal interval elapses. A lost lease (`LockException`) makes acquisition return `null`. A provider error is propagated, and it takes precedence over `LockException` when both occur. Either way, release is attempted for every acquired lock. Cleanup failures are logged rather than thrown, so callers still receive `null` or the original error; any lease that could not be released remains until it expires. + +Renewing a multi-resource lock renews every member, even if one fails. If a member's lease was lost, `RenewAsync` throws `LockException`. If a provider error also occurs, the provider error is thrown instead. `RenewalCount` increases only when every member renews. ## Common Patterns diff --git a/src/Foundatio.TestHarness/Locks/LockTestBase.cs b/src/Foundatio.TestHarness/Locks/LockTestBase.cs index a52bb24a8..578a192e6 100644 --- a/src/Foundatio.TestHarness/Locks/LockTestBase.cs +++ b/src/Foundatio.TestHarness/Locks/LockTestBase.cs @@ -494,6 +494,31 @@ public virtual async Task RenewAsync_WithMissingLock_ThrowsLockException() Assert.False(await locker.IsLockedAsync(lockName)); } + public virtual async Task RenewAsync_WithMultipleResources_WhenOneLockReplaced_ThrowsLockExceptionAndPreservesCurrentOwner() + { + // Arrange + var locker = GetLockProvider(); + if (locker is null) + return; + + string prefix = Guid.NewGuid().ToString("N")[..10]; + string replacedResource = prefix + "-a"; + string retainedResource = prefix + "-b"; + await using var lockInstance = await locker.AcquireAsync([replacedResource, retainedResource], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + await locker.ReleaseAsync(replacedResource); + await using var currentOwner = await locker.AcquireAsync(replacedResource, TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Act + await Assert.ThrowsAsync(() => lockInstance.RenewAsync()); + + // Assert + Assert.Equal(0, lockInstance.RenewalCount); + Assert.True(await locker.IsLockedAsync(replacedResource)); + Assert.True(await locker.IsLockedAsync(retainedResource)); + await currentOwner.RenewAsync(); + Assert.Equal(1, currentOwner.RenewalCount); + } + public virtual async Task RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(bool scoped) { // Arrange diff --git a/src/Foundatio/Jobs/WorkItemJob/WorkItemHandlers.cs b/src/Foundatio/Jobs/WorkItemJob/WorkItemHandlers.cs index 0b9082673..a99ddaba8 100644 --- a/src/Foundatio/Jobs/WorkItemJob/WorkItemHandlers.cs +++ b/src/Foundatio/Jobs/WorkItemJob/WorkItemHandlers.cs @@ -73,8 +73,10 @@ public interface IWorkItemHandler Task HandleItemAsync(WorkItemContext context); /// - /// When true, the lock on the queue entry is automatically renewed each time + /// When true, the queue entry and work item lock are automatically renewed each time /// the handler reports progress. This prevents long-running work items from timing out. + /// Renewal errors are logged; if the work item lock is lost, + /// is cancelled so cooperative handlers can stop. /// bool AutoRenewLockOnProgress { get; set; } diff --git a/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs b/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs index 5d37547e6..127481117 100644 --- a/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs +++ b/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs @@ -3,6 +3,7 @@ using System.Diagnostics; using System.Threading; using System.Threading.Tasks; +using Foundatio.Lock; using Foundatio.Messaging; using Foundatio.Queues; using Foundatio.Serializer; @@ -126,22 +127,11 @@ public async Task ProcessAsync(IQueueEntry queueEntry, return JobResult.CancelledWithMessage($"Unable to acquire work item lock. Abandoning {queueEntry.Value.Type} queue entry: {queueEntry.Id}"); } + using var workItemCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); var progressCallback = new Func(async (progress, message) => { if (handler.AutoRenewLockOnProgress) - { - try - { - await Task.WhenAll( - queueEntry.RenewLockAsync(), - lockValue.RenewAsync() - ).AnyContext(); - } - catch (Exception ex) - { - handler.Log.LogError(ex, "Error renewing work item locks: {Message}", ex.Message); - } - } + await RenewWorkItemLocksAsync(handler, queueEntry, lockValue, workItemCancellation).AnyContext(); await ReportProgressAsync(handler, queueEntry, progress, message).AnyContext(); handler.Log.LogInformation("{TypeName} Progress {Progress}%: {Message}", workItemDataType.Name, progress, message); @@ -150,7 +140,7 @@ await Task.WhenAll( try { handler.LogProcessingQueueEntry(queueEntry, workItemDataType, workItemData); - var workItemContext = new WorkItemContext(workItemData, JobId, lockValue, cancellationToken, progressCallback); + var workItemContext = new WorkItemContext(workItemData, JobId, lockValue, workItemCancellation.Token, progressCallback); await handler.HandleItemAsync(workItemContext).AnyContext(); if (!workItemContext.Result.IsSuccess) @@ -194,6 +184,35 @@ await Task.WhenAll( } } + /// + /// Renews the queue entry and work item locks. Renewal errors are logged so progress reporting continues, + /// but a lost work item lock cancels because another + /// process may now own the work. + /// + private static async Task RenewWorkItemLocksAsync(IWorkItemHandler handler, IQueueEntry queueEntry, ILock workItemLock, CancellationTokenSource workItemCancellation) + { + var queueEntryRenewal = RenewQueueEntryLockAsync(); + var workItemLockRenewal = RenewWorkItemLockAsync(); + + try + { + await Task.WhenAll(queueEntryRenewal, workItemLockRenewal).AnyContext(); + } + catch (Exception ex) + { + handler.Log.LogError(ex, "Error renewing work item locks: {Message}", ex.Message); + } + + if (workItemLockRenewal.Exception?.InnerException is LockException) + { + handler.Log.LogWarning("Lost work item lock {Resource} for queue entry {Id}, cancelling work item", workItemLock.Resource, queueEntry.Id); + await workItemCancellation.CancelAsync().AnyContext(); + } + + async Task RenewQueueEntryLockAsync() => await queueEntry.RenewLockAsync().AnyContext(); + async Task RenewWorkItemLockAsync() => await workItemLock.RenewAsync().AnyContext(); + } + protected virtual Activity? StartProcessWorkItemActivity(IQueueEntry entry, Type workItemDataType) { var activity = FoundatioDiagnostics.ActivitySource.StartActivity("ProcessQueueEntry", ActivityKind.Internal, entry.CorrelationId); diff --git a/src/Foundatio/Lock/DisposableLockCollection.cs b/src/Foundatio/Lock/DisposableLockCollection.cs index 198112f39..ae68f6a07 100644 --- a/src/Foundatio/Lock/DisposableLockCollection.cs +++ b/src/Foundatio/Lock/DisposableLockCollection.cs @@ -42,10 +42,10 @@ public async Task RenewAsync(TimeSpan? lockExtension = null) { _logger.LogTrace("Renewing {LockCount} locks {Resource}", _locks.Count, Resource); - await Task.WhenAll(_locks.Select(l => l.RenewAsync(lockExtension))).AnyContext(); + await LockSetOperations.RenewAllAsync(_locks, lockExtension).AnyContext(); _renewalCount++; - _logger.LogDebug("Renewing {LockCount} locks {Resource}", _locks.Count, Resource); + _logger.LogDebug("Renewed {LockCount} locks {Resource}", _locks.Count, Resource); } public async Task ReleaseAsync() @@ -63,7 +63,7 @@ public async Task ReleaseAsync() _logger.LogDebug("Releasing {LockCount} locks {Resource} after {Duration:g}", _locks.Count, Resource, _duration.Elapsed); - await Task.WhenAll(_locks.Select(l => l.ReleaseAsync())).AnyContext(); + await LockSetOperations.ReleaseAllAsync(_locks).AnyContext(); } } @@ -73,7 +73,7 @@ public async ValueTask DisposeAsync() try { - await Task.WhenAll(_locks.Select(l => l.ReleaseAsync())).AnyContext(); + await LockSetOperations.ReleaseAllAsync(_locks).AnyContext(); } catch (Exception ex) { diff --git a/src/Foundatio/Lock/ILockProvider.cs b/src/Foundatio/Lock/ILockProvider.cs index 967d71615..3ce0a4c9b 100644 --- a/src/Foundatio/Lock/ILockProvider.cs +++ b/src/Foundatio/Lock/ILockProvider.cs @@ -313,70 +313,84 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc logger.LogTrace("Acquiring {LockCount} locks {Resource} RenewTime={RenewTime:g}", resourceList.Length, resourceList, renewTime); var sw = Stopwatch.StartNew(); + var timeProvider = provider.GetTimeProvider(); var acquiredLocks = new List<(ILock Lock, DateTimeOffset LastRenewed)>(resourceList.Length); bool acquiredAll = false; - Exception? acquisitionError = null; try { foreach (string resource in resourceList) { var l = await provider.TryAcquireAsync(resource, timeUntilExpires, releaseOnDispose, cancellationToken).AnyContext(); if (l is null) + { + logger.LogTrace("Unable to acquire lock {Resource}, releasing {LockCount} acquired locks", resource, acquiredLocks.Count); return null; + } - acquiredLocks.Add((l, provider.GetTimeProvider().GetUtcNow())); + acquiredLocks.Add((l, timeProvider.GetUtcNow())); // Renew earlier locks while waiting for the remaining resources. - if (acquiredLocks.Count > 1 && renewTime > TimeSpan.Zero) - { - var utcNow = provider.GetTimeProvider().GetUtcNow(); - var renewBefore = utcNow.Subtract(renewTime); - var locksToRenew = Enumerable.Range(0, acquiredLocks.Count - 1).Where(index => acquiredLocks[index].LastRenewed < renewBefore).ToArray(); - if (locksToRenew.Length > 0) - { - try - { - await Task.WhenAll(locksToRenew.Select(async index => await acquiredLocks[index].Lock.RenewAsync(timeUntilExpires).AnyContext())).AnyContext(); - } - catch (LockException ex) - { - logger.LogWarning(ex, "Lost an acquired lock while acquiring {Resource}, releasing acquired locks: {Message}", resource, ex.Message); - return null; - } - - foreach (int index in locksToRenew) - acquiredLocks[index] = (acquiredLocks[index].Lock, utcNow); - - logger.LogTrace("Renewed {LockCount} locks {Resource} RenewTime={RenewTime:g}", locksToRenew.Length, locksToRenew.Select(index => acquiredLocks[index].Lock.Resource), renewTime); - } - } + if (acquiredLocks.Count > 1 && renewTime > TimeSpan.Zero && !await TryRenewEarlierLocksAsync(acquiredLocks, timeUntilExpires, renewTime, timeProvider, logger).AnyContext()) + return null; } sw.Stop(); var locks = acquiredLocks.Select(l => l.Lock).ToArray(); - var result = new DisposableLockCollection(locks, String.Join("+", locks.Select(l => l.LockId)), provider.GetTimeProvider().GetUtcNow().UtcDateTime, sw.Elapsed, logger); + var result = new DisposableLockCollection(locks, String.Join("+", locks.Select(l => l.LockId)), timeProvider.GetUtcNow().UtcDateTime, sw.Elapsed, logger); logger.LogTrace("Acquired {LockCount} locks {Resource} after {Duration:g}", locks.Length, locks.Select(l => l.Resource), sw.Elapsed); acquiredAll = true; return result; } - catch (Exception ex) + finally { - acquisitionError = ex; - throw; + if (!acquiredAll && acquiredLocks.Count > 0) + await ReleasePartialLocksAsync(acquiredLocks.Select(l => l.Lock).ToArray(), logger).AnyContext(); } - finally + } + + /// + /// Renews every lock except the most recently acquired one whose renewal interval has elapsed. + /// Returns false when a lease was lost; provider errors propagate. + /// + private static async Task TryRenewEarlierLocksAsync(List<(ILock Lock, DateTimeOffset LastRenewed)> acquiredLocks, TimeSpan? timeUntilExpires, TimeSpan renewTime, TimeProvider timeProvider, ILogger logger) + { + var utcNow = timeProvider.GetUtcNow(); + var renewBefore = utcNow.Subtract(renewTime); + int[] locksToRenew = Enumerable.Range(0, acquiredLocks.Count - 1).Where(index => acquiredLocks[index].LastRenewed < renewBefore).ToArray(); + if (locksToRenew.Length is 0) + return true; + + try { - if (!acquiredAll) - { - try - { - await Task.WhenAll(acquiredLocks.Select(async acquired => await acquired.Lock.ReleaseAsync().AnyContext())).AnyContext(); - } - catch (Exception ex) when (acquisitionError is not null) - { - logger.LogError(ex, "Unable to release partially acquired locks after acquisition failed: {Message}", ex.Message); - } - } + await LockSetOperations.RenewAllAsync(locksToRenew.Select(index => acquiredLocks[index].Lock), timeUntilExpires).AnyContext(); + } + catch (LockException ex) + { + logger.LogWarning(ex, "Lost an acquired lock while acquiring {Resource}, releasing acquired locks: {Message}", acquiredLocks[^1].Lock.Resource, ex.Message); + return false; + } + + foreach (int index in locksToRenew) + acquiredLocks[index] = (acquiredLocks[index].Lock, utcNow); + + logger.LogTrace("Renewed {LockCount} locks {Resource} RenewTime={RenewTime:g}", locksToRenew.Length, locksToRenew.Select(index => acquiredLocks[index].Lock.Resource), renewTime); + return true; + } + + /// + /// Attempts to release every partially acquired lock. Failures are logged rather than thrown so that callers receive + /// null or the original acquisition error; leases that could not be released remain held until they expire. + /// + private static async Task ReleasePartialLocksAsync(ILock[] locks, ILogger logger) + { + try + { + await LockSetOperations.ReleaseAllAsync(locks).AnyContext(); + logger.LogTrace("Released {LockCount} partially acquired locks {Resource}", locks.Length, locks.Select(l => l.Resource)); + } + catch (Exception ex) + { + logger.LogError(ex, "Unable to release {LockCount} partially acquired locks {Resource}; unreleased locks remain held until they expire: {Message}", locks.Length, locks.Select(l => l.Resource), ex.Message); } } diff --git a/src/Foundatio/Lock/LockSetOperations.cs b/src/Foundatio/Lock/LockSetOperations.cs new file mode 100644 index 000000000..d6f9dbfa7 --- /dev/null +++ b/src/Foundatio/Lock/LockSetOperations.cs @@ -0,0 +1,40 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.ExceptionServices; +using System.Threading.Tasks; +using Foundatio.Utility; + +namespace Foundatio.Lock; + +/// +/// Runs an operation against every lock in a set. Every operation is attempted, even when one throws synchronously. +/// When operations fail, provider errors take precedence over so infrastructure failures +/// are not masked by a lost lease; otherwise the first is rethrown. +/// +internal static class LockSetOperations +{ + public static Task RenewAllAsync(IEnumerable locks, TimeSpan? timeUntilExpires) + { + return WhenAllAsync(locks.Select(async l => await l.RenewAsync(timeUntilExpires).AnyContext())); + } + + public static Task ReleaseAllAsync(IEnumerable locks) + { + return WhenAllAsync(locks.Select(async l => await l.ReleaseAsync().AnyContext())); + } + + private static async Task WhenAllAsync(IEnumerable operations) + { + var task = Task.WhenAll(operations); + try + { + await task.AnyContext(); + } + catch (Exception) when (task.Exception is { } aggregate) + { + var errors = aggregate.InnerExceptions; + ExceptionDispatchInfo.Throw(errors.FirstOrDefault(e => e is not LockException) ?? errors[0]); + } + } +} diff --git a/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs b/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs index 9b1cd4226..cbb260d0e 100644 --- a/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs +++ b/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs @@ -7,7 +7,9 @@ using System.Threading.Tasks; using Exceptionless; using Foundatio.AsyncEx; +using Foundatio.Caching; using Foundatio.Jobs; +using Foundatio.Lock; using Foundatio.Messaging; using Foundatio.Queues; using Foundatio.Tests.Extensions; @@ -334,6 +336,89 @@ await messageBus.SubscribeAsync(status => await countdown.WaitAsync(TimeSpan.FromSeconds(2)); Assert.Equal(0, countdown.CurrentCount); } + + [Fact] + public async Task ProcessAsync_WhenAutoRenewedWorkItemLockIsLost_CancelsHandlerAndAbandonsEntry() + { + // Arrange + using var cache = new InMemoryCacheClient(o => o.LoggerFactory(Log)); + using var queue = new InMemoryQueue(o => o.RetryDelay(TimeSpan.Zero).Retries(0).LoggerFactory(Log)); + using var messageBus = new InMemoryMessageBus(o => o.LoggerFactory(Log)); + var locker = new CacheLockProvider(cache, messageBus, null, null, Log); + var handler = new LockedWorkItemHandler(locker, Log) { AutoRenewLockOnProgress = true }; + var handlerRegistry = new WorkItemHandlers(); + handlerRegistry.Register(handler); + var job = new WorkItemJob(queue, messageBus, handlerRegistry, Log); + await queue.EnqueueAsync(new MyWorkItem { SomeData = "Test" }, true); + + // Act + var result = await job.RunAsync(TestCancellationToken); + + // Assert + Assert.False(result.IsSuccess); + Assert.True(handler.ObservedCancellation); + Assert.False(handler.CompletedWork); + var stats = await queue.GetQueueStatsAsync(); + Assert.Equal(0, stats.Completed); + Assert.Equal(1, stats.Abandoned); + } + + [Fact] + public async Task ProcessAsync_WhenAutoRenewedWorkItemLockIsHeld_DoesNotCancelHandler() + { + // Arrange + using var cache = new InMemoryCacheClient(o => o.LoggerFactory(Log)); + using var queue = new InMemoryQueue(o => o.LoggerFactory(Log)); + using var messageBus = new InMemoryMessageBus(o => o.LoggerFactory(Log)); + var locker = new CacheLockProvider(cache, messageBus, null, null, Log); + var handler = new LockedWorkItemHandler(locker, Log, loseLock: false) { AutoRenewLockOnProgress = true }; + var handlerRegistry = new WorkItemHandlers(); + handlerRegistry.Register(handler); + var job = new WorkItemJob(queue, messageBus, handlerRegistry, Log); + await queue.EnqueueAsync(new MyWorkItem { SomeData = "Test" }, true); + + // Act + var result = await job.RunAsync(TestCancellationToken); + + // Assert + Assert.True(result.IsSuccess); + Assert.False(handler.ObservedCancellation); + Assert.True(handler.CompletedWork); + var stats = await queue.GetQueueStatsAsync(); + Assert.Equal(1, stats.Completed); + } +} + +public class LockedWorkItemHandler : WorkItemHandlerBase +{ + private readonly ILockProvider _locker; + private readonly bool _loseLock; + + public LockedWorkItemHandler(ILockProvider locker, ILoggerFactory loggerFactory, bool loseLock = true) : base(loggerFactory) + { + _locker = locker; + _loseLock = loseLock; + } + + public bool ObservedCancellation { get; private set; } + public bool CompletedWork { get; private set; } + + public override Task GetWorkItemLockAsync(object workItem, CancellationToken cancellationToken = default) + { + return _locker.TryAcquireAsync("work-item", TimeSpan.FromMinutes(1), cancellationToken: cancellationToken); + } + + public override async Task HandleItemAsync(WorkItemContext context) + { + Assert.NotNull(context.WorkItemLock); + if (_loseLock) + await _locker.ReleaseAsync(context.WorkItemLock.Resource); + + await context.ReportProgressAsync(50); + ObservedCancellation = context.CancellationToken.IsCancellationRequested; + context.CancellationToken.ThrowIfCancellationRequested(); + CompletedWork = true; + } } public class MyWorkItem diff --git a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs index a2fa6d364..068c27e96 100644 --- a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs +++ b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs @@ -228,6 +228,35 @@ public override Task RenewAsync_WithMissingLock_ThrowsLockException() return base.RenewAsync_WithMissingLock_ThrowsLockException(); } + [Fact] + public async Task RenewAsync_WithMultipleResources_WhenFirstLockThrowsSynchronously_AttemptsEveryRenewal() + { + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + var error = new CacheException("Provider unavailable"); + var first = new Mock(); + first.Setup(l => l.RenewAsync(It.IsAny())).Throws(error); + var second = new Mock(); + second.Setup(l => l.RenewAsync(It.IsAny())).Returns(Task.CompletedTask); + var locker = CreateMockLockProvider(timeProvider, ("a", first.Object), ("b", second.Object)); + await using var lockInstance = await locker.Object.TryAcquireAsync(["a", "b"], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + Assert.NotNull(lockInstance); + + // Act + var thrown = await Assert.ThrowsAsync(() => lockInstance.RenewAsync()); + + // Assert + Assert.Same(error, thrown); + Assert.Equal(0, lockInstance.RenewalCount); + second.Verify(l => l.RenewAsync(It.IsAny()), Times.Once); + } + + [Fact] + public override Task RenewAsync_WithMultipleResources_WhenOneLockReplaced_ThrowsLockExceptionAndPreservesCurrentOwner() + { + return base.RenewAsync_WithMultipleResources_WhenOneLockReplaced_ThrowsLockExceptionAndPreservesCurrentOwner(); + } + [Theory] [InlineData(false)] [InlineData(true)] @@ -386,6 +415,57 @@ public override Task TryAcquireAsync_WithMultipleResources_WhenResourceNamesShar return base.TryAcquireAsync_WithMultipleResources_WhenResourceNamesShareSuffix_ReleasesAcquiredLocksAndReturnsNull(scoped); } + [Fact] + public async Task TryAcquireAsync_WithMultipleResources_WhenRenewalLosesLockAndProviderThrows_ThrowsProviderError() + { + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + var providerError = new CacheException("Provider unavailable"); + var lost = new Mock(); + lost.Setup(l => l.RenewAsync(It.IsAny())).ThrowsAsync(new LockException("Lock lost")); + lost.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var failing = new Mock(); + failing.Setup(l => l.RenewAsync(It.IsAny())).ThrowsAsync(providerError); + failing.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var last = new Mock(); + last.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var locker = CreateMockLockProvider(timeProvider, ("a", lost.Object), ("b", failing.Object), ("c", last.Object)); + locker.Setup(p => p.TryAcquireAsync("c", It.IsAny(), It.IsAny(), It.IsAny())).Returns(() => + { + timeProvider.Advance(TimeSpan.FromSeconds(31)); + return Task.FromResult(last.Object); + }); + + // Act + var thrown = await Assert.ThrowsAsync(() => locker.Object.TryAcquireAsync(["a", "b", "c"], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken)); + + // Assert + Assert.Same(providerError, thrown); + lost.Verify(l => l.ReleaseAsync(), Times.Once); + failing.Verify(l => l.ReleaseAsync(), Times.Once); + last.Verify(l => l.ReleaseAsync(), Times.Once); + } + + [Fact] + public async Task TryAcquireAsync_WithMultipleResources_WhenUnavailableAndReleaseThrows_ReturnsNullAndAttemptsEveryRelease() + { + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + var first = new Mock(); + first.Setup(l => l.ReleaseAsync()).Throws(new CacheException("Release unavailable")); + var second = new Mock(); + second.Setup(l => l.ReleaseAsync()).Returns(Task.CompletedTask); + var locker = CreateMockLockProvider(timeProvider, ("a", first.Object), ("b", second.Object), ("c", null)); + + // Act + var lockInstance = await locker.Object.TryAcquireAsync(["a", "b", "c"], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + + // Assert + Assert.Null(lockInstance); + first.Verify(l => l.ReleaseAsync(), Times.Once); + second.Verify(l => l.ReleaseAsync(), Times.Once); + } + [Fact] public override Task TryUsingAsync_WithSuccessfulAction_ExecutesAndReleasesLock() { @@ -404,6 +484,16 @@ public void Dispose() _messageBus.Dispose(); } + private static Mock CreateMockLockProvider(TimeProvider timeProvider, params (string Resource, ILock? Lock)[] locks) + { + var locker = new Mock(); + locker.As().SetupGet(p => p.TimeProvider).Returns(timeProvider); + foreach (var (resource, lockInstance) in locks) + locker.Setup(p => p.TryAcquireAsync(resource, It.IsAny(), It.IsAny(), It.IsAny())).ReturnsAsync(lockInstance); + + return locker; + } + private sealed class BeforeAcquireLockProvider(ILockProvider inner, TimeProvider timeProvider, Func beforeAcquire) : ILockProvider, IHaveTimeProvider { public TimeProvider TimeProvider => timeProvider; From 4fa23e3d5343392558b9f9fc471397b2b3ea3f0e Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Wed, 7 Oct 2026 16:42:04 -0500 Subject: [PATCH 5/7] Suffix LastRenewedUtc and flow the test logger and clock through new lock tests --- src/Foundatio/Lock/ILockProvider.cs | 6 +++--- tests/Foundatio.Tests/Locks/InMemoryLockTests.cs | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/Foundatio/Lock/ILockProvider.cs b/src/Foundatio/Lock/ILockProvider.cs index 8444eccf8..c2b6ac9d8 100644 --- a/src/Foundatio/Lock/ILockProvider.cs +++ b/src/Foundatio/Lock/ILockProvider.cs @@ -314,7 +314,7 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc var sw = Stopwatch.StartNew(); var timeProvider = provider.GetTimeProvider(); - var acquiredLocks = new List<(ILock Lock, DateTimeOffset LastRenewed)>(resourceList.Length); + var acquiredLocks = new List<(ILock Lock, DateTimeOffset LastRenewedUtc)>(resourceList.Length); bool acquiredAll = false; try { @@ -352,11 +352,11 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc /// Renews every lock except the most recently acquired one whose renewal interval has elapsed. /// Returns false when a lease was lost; provider errors propagate. /// - private static async Task TryRenewEarlierLocksAsync(List<(ILock Lock, DateTimeOffset LastRenewed)> acquiredLocks, TimeSpan? timeUntilExpires, TimeSpan renewTime, TimeProvider timeProvider, ILogger logger) + private static async Task TryRenewEarlierLocksAsync(List<(ILock Lock, DateTimeOffset LastRenewedUtc)> acquiredLocks, TimeSpan? timeUntilExpires, TimeSpan renewTime, TimeProvider timeProvider, ILogger logger) { var utcNow = timeProvider.GetUtcNow(); var renewBeforeUtc = utcNow.Subtract(renewTime); - int[] locksToRenew = Enumerable.Range(0, acquiredLocks.Count - 1).Where(index => acquiredLocks[index].LastRenewed < renewBeforeUtc).ToArray(); + int[] locksToRenew = Enumerable.Range(0, acquiredLocks.Count - 1).Where(index => acquiredLocks[index].LastRenewedUtc < renewBeforeUtc).ToArray(); if (locksToRenew.Length is 0) return true; diff --git a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs index f2172f4a2..0b7563206 100644 --- a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs +++ b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs @@ -263,9 +263,9 @@ public override Task RenewAsync_WithInvalidDuration_PreservesCurrentOwner(int mi public async Task RenewAsync_WithMinimumDuration_RenewsCurrentOwner() { // Arrange - var timeProvider = new FakeTimeProvider(); - using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider)); - var locker = new CacheLockProvider(cache, null, timeProvider); + var timeProvider = new FakeTimeProvider(DateTimeOffset.UtcNow); + using var cache = new InMemoryCacheClient(o => o.LoggerFactory(Log).TimeProvider(timeProvider)); + var locker = new CacheLockProvider(cache, null, timeProvider, null, Log); await using var lease = await locker.AcquireAsync("resource", TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); // Act From 0bc0f584a82f025d4332e45aa188df72257fb716 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Wed, 7 Oct 2026 17:09:06 -0500 Subject: [PATCH 6/7] Keep a lost work item lease from being completed when the job is cancelled afterward --- docs/guide/locks.md | 2 +- src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs | 43 +++++++++++++------ .../Foundatio.Tests/Jobs/WorkItemJobTests.cs | 34 ++++++++++++++- 3 files changed, 65 insertions(+), 14 deletions(-) diff --git a/docs/guide/locks.md b/docs/guide/locks.md index 1d81e6d18..65f54eb70 100644 --- a/docs/guide/locks.md +++ b/docs/guide/locks.md @@ -315,7 +315,7 @@ The local async function starts immediately and yields at the delay; it does not Choose an interval shorter than the lease duration, leaving time for cache latency and retries. A pause can still let the lease expire before renewal detects the loss, and cancellation cannot revoke an operation already dispatched to another system. -`WorkItemJob` applies this pattern when a handler sets `AutoRenewLockOnProgress`. Each `ReportProgressAsync` call renews the queue entry and the work item lock. Renewal errors are logged and progress reporting continues. If the work item lock is lost (`LockException`), `WorkItemContext.CancellationToken` is cancelled, and the job abandons the queue entry for retry instead of completing it, whether the handler throws, returns early, or swallows the cancellation. A handler that ignores the token keeps running unprotected until it returns. A failed queue entry renewal is only logged: queues don't report a definite loss, and while the work item lock is held a redelivered entry can't be processed concurrently. +`WorkItemJob` applies this pattern when a handler sets `AutoRenewLockOnProgress`. Each `ReportProgressAsync` call renews the queue entry and the work item lock. Renewal errors are logged and progress reporting continues. If the work item lock is lost (`LockException`), `WorkItemContext.CancellationToken` is cancelled, and the job abandons the queue entry for retry instead of completing it, whether the handler throws, returns early, or swallows the cancellation, and even if the job itself is cancelled afterward. A handler that ignores the token keeps running unprotected until it returns. A failed queue entry renewal is only logged: queues don't report a definite loss, and while the work item lock is held a redelivered entry can't be processed concurrently. When acquiring multiple resources, success means every requested lock was acquired. If any resource is unavailable, the provider releases the partial set and returns `null`, including when resource names share a suffix. While acquisition is in progress, earlier locks are renewed once their renewal interval elapses. A lost lease (`LockException`) makes acquisition return `null`, and it takes precedence over a provider error when both occur, because the loss is definite; the provider error is logged. Otherwise a provider error is propagated. Either way, release is attempted for every acquired lock. Cleanup failures are logged rather than thrown, so callers still receive `null` or the original error; any lease that could not be released remains until it expires. diff --git a/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs b/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs index 7db00a507..dd4b1bb14 100644 --- a/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs +++ b/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs @@ -128,10 +128,11 @@ public async Task ProcessAsync(IQueueEntry queueEntry, } using var workItemCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var leaseLoss = new WorkItemLeaseLoss(workItemCancellation); var progressCallback = new Func(async (progress, message) => { if (handler.AutoRenewLockOnProgress) - await RenewWorkItemLocksAsync(handler, queueEntry, lockValue, workItemCancellation).AnyContext(); + await RenewWorkItemLocksAsync(handler, queueEntry, lockValue, leaseLoss).AnyContext(); await ReportProgressAsync(handler, queueEntry, progress, message).AnyContext(); handler.Log.LogInformation("{TypeName} Progress {Progress}%: {Message}", workItemDataType.Name, progress, message); @@ -144,8 +145,9 @@ public async Task ProcessAsync(IQueueEntry queueEntry, await handler.HandleItemAsync(workItemContext).AnyContext(); // A handler that stops cooperatively after a lost lease (returning early or swallowing the cancellation) - // hasn't finished the work, and another worker may now own it: never acknowledge it - if (workItemCancellation.IsCancellationRequested && !cancellationToken.IsCancellationRequested) + // hasn't finished the work, and another worker may now own it: never acknowledge it, even if the job + // was also cancelled meanwhile + if (leaseLoss.IsLost) { if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted) await queueEntry.AbandonAsync().AnyContext(); @@ -206,7 +208,7 @@ public async Task ProcessAsync(IQueueEntry queueEntry, /// loss, and while the work item lock is held a redelivered entry can't be processed concurrently because its /// worker fails to acquire the same lock and abandons it. /// - private static async Task RenewWorkItemLocksAsync(IWorkItemHandler handler, IQueueEntry queueEntry, ILock workItemLock, CancellationTokenSource workItemCancellation) + private static async Task RenewWorkItemLocksAsync(IWorkItemHandler handler, IQueueEntry queueEntry, ILock workItemLock, WorkItemLeaseLoss leaseLoss) { var queueEntryRenewal = RenewQueueEntryLockAsync(); var workItemLockRenewal = RenewWorkItemLockAsync(); @@ -229,19 +231,36 @@ private static async Task RenewWorkItemLocksAsync(IWorkItemHandler handler, IQue return; handler.Log.LogWarning("Lost work item lock {Resource} for queue entry {Id}, cancelling work item", workItemLock.Resource, queueEntry.Id); - try - { - await workItemCancellation.CancelAsync().AnyContext(); - } - catch (ObjectDisposedException) - { - // A progress report that wasn't awaited can finish after the work item; nothing is left to cancel - } + await leaseLoss.SignalAsync().AnyContext(); async Task RenewQueueEntryLockAsync() => await queueEntry.RenewLockAsync().AnyContext(); async Task RenewWorkItemLockAsync() => await workItemLock.RenewAsync().AnyContext(); } + /// + /// Records that the work item lease was lost and cancels the work item. The flag, not the token, decides whether + /// the entry may be completed, because the job's own token can be cancelled at any time afterward. + /// + private sealed class WorkItemLeaseLoss(CancellationTokenSource workItemCancellation) + { + private volatile bool _isLost; + + public bool IsLost => _isLost; + + public async Task SignalAsync() + { + _isLost = true; + try + { + await workItemCancellation.CancelAsync().AnyContext(); + } + catch (ObjectDisposedException) + { + // A progress report that wasn't awaited can finish after the work item; nothing is left to cancel + } + } + } + protected virtual Activity? StartProcessWorkItemActivity(IQueueEntry entry, Type workItemDataType) { var activity = FoundatioDiagnostics.ActivitySource.StartActivity("ProcessQueueEntry", ActivityKind.Internal, entry.CorrelationId); diff --git a/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs b/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs index c6d5a5955..f28769dc0 100644 --- a/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs +++ b/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs @@ -390,6 +390,33 @@ public async Task ProcessAsync_WhenAutoRenewedWorkItemLockIsHeld_DoesNotCancelHa var stats = await queue.GetQueueStatsAsync(); Assert.Equal(1, stats.Completed); } + + [Fact] + public async Task ProcessAsync_WhenJobIsCancelledAfterWorkItemLockIsLost_AbandonsEntry() + { + // Arrange + using var cache = new InMemoryCacheClient(o => o.LoggerFactory(Log)); + using var queue = new InMemoryQueue(o => o.RetryDelay(TimeSpan.Zero).Retries(0).LoggerFactory(Log)); + using var messageBus = new InMemoryMessageBus(o => o.LoggerFactory(Log)); + var locker = new CacheLockProvider(cache, messageBus, null, null, Log); + using var jobCancellation = CancellationTokenSource.CreateLinkedTokenSource(TestCancellationToken); + var handler = new LockedWorkItemHandler(locker, Log, response: LostLeaseResponse.ReturnEarly, cancelAfterLoss: jobCancellation) { AutoRenewLockOnProgress = true }; + var handlerRegistry = new WorkItemHandlers(); + handlerRegistry.Register(handler); + var job = new WorkItemJob(queue, messageBus, handlerRegistry, Log); + await queue.EnqueueAsync(new MyWorkItem { SomeData = "Test" }, true); + + // Act + var result = await job.RunAsync(jobCancellation.Token); + + // Assert + Assert.False(result.IsSuccess); + Assert.True(handler.ObservedCancellation); + Assert.False(handler.CompletedWork); + var stats = await queue.GetQueueStatsAsync(); + Assert.Equal(0, stats.Completed); + Assert.Equal(1, stats.Abandoned); + } } public enum LostLeaseResponse @@ -404,12 +431,14 @@ public class LockedWorkItemHandler : WorkItemHandlerBase private readonly ILockProvider _locker; private readonly bool _loseLock; private readonly LostLeaseResponse _response; + private readonly CancellationTokenSource? _cancelAfterLoss; - public LockedWorkItemHandler(ILockProvider locker, ILoggerFactory loggerFactory, bool loseLock = true, LostLeaseResponse response = LostLeaseResponse.Throw) : base(loggerFactory) + public LockedWorkItemHandler(ILockProvider locker, ILoggerFactory loggerFactory, bool loseLock = true, LostLeaseResponse response = LostLeaseResponse.Throw, CancellationTokenSource? cancelAfterLoss = null) : base(loggerFactory) { _locker = locker; _loseLock = loseLock; _response = response; + _cancelAfterLoss = cancelAfterLoss; } public bool ObservedCancellation { get; private set; } @@ -428,6 +457,9 @@ public override async Task HandleItemAsync(WorkItemContext context) await context.ReportProgressAsync(50); ObservedCancellation = context.CancellationToken.IsCancellationRequested; + if (_cancelAfterLoss is not null) + await _cancelAfterLoss.CancelAsync(); + switch (_response) { case LostLeaseResponse.ReturnEarly when ObservedCancellation: From a68d22e5229e8615fa3fe415207559fdeb75ea3d Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Wed, 7 Oct 2026 20:49:37 -0500 Subject: [PATCH 7/7] Reject multi-resource acquisition when slow renewal outlives a lease --- .agents/skills/foundatio/SKILL.md | 2 +- docs/guide/locks.md | 2 +- src/Foundatio/Lock/ILockProvider.cs | 13 +++- .../Locks/InMemoryLockTests.cs | 71 +++++++++++++++++++ 4 files changed, 84 insertions(+), 4 deletions(-) diff --git a/.agents/skills/foundatio/SKILL.md b/.agents/skills/foundatio/SKILL.md index 605463cb2..8b9128395 100644 --- a/.agents/skills/foundatio/SKILL.md +++ b/.agents/skills/foundatio/SKILL.md @@ -312,7 +312,7 @@ public class OrderServiceTests : TestLoggerBase - **Failure semantics depend on job type**: `JobResult` only has `IsSuccess` -- there is no separate "failed but don't retry" status. For **queue-processed jobs** (`QueueJobBase.ProcessQueueEntryAsync`, or setting `context.Result` in a `WorkItemJob` handler), a non-success result triggers `AbandonAsync`, which re-queues the entry and eventually dead-letters it after `Retries` is exhausted -- reserve `FailedWithMessage`/`FromException` for transient errors you want retried, and log + return `JobResult.Success`/`SuccessWithMessage(...)` for permanent errors to avoid a pointless retry loop. For **standalone/manual jobs** (`JobBase`, a one-off `RunAsync()`/`RunInConsoleAsync()` run, or scheduled/cron jobs via `Foundatio.Extensions.Hosting`), there is no built-in retry or dead letter queue -- a failed result just produces an error-level log, a non-zero exit code from `RunInConsoleAsync`, or a failed entry in the job run history. Returning `FailedWithMessage`/`FromException` there is correct even for permanent errors, since nothing inside Foundatio will retry it. - **JobWithLockBase vs manual locking**: Use `JobWithLockBase` when the entire run must be single-instance (leader election). Use manual `ILockProvider.AcquireAsync` inside `JobBase` for finer-grained locking within a job. - **JobContext.RenewLockAsync**: Call in long-running jobs (both `JobBase` and `QueueJobBase`) to prevent lock expiration mid-processing. -- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; with `AutoRenewLockOnProgress`, `WorkItemJob` cancels `WorkItemContext.CancellationToken` when the work item lock is lost and abandons the entry instead of completing it, even if the handler returns normally (other renewal errors, including queue entry renewal, are logged). Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. Multi-resource acquisition is all-or-nothing even for suffix-overlapping names; successful renewals reset their per-lock interval. Lock-set operations attempt every member; `LockException` takes precedence over provider errors (which are logged), and partial-set cleanup failures are logged so callers get `null` or the original error. +- **Cache-backed renewal failure means ownership is unconfirmed**: `CacheLockProvider.RenewAsync` throws `LockException` when conditional renewal fails. After a pause, only a live lease with the same owner can renew; an expired handle requires a new acquisition and revalidation of the work. Stop cooperative work and observe background renewal failures; with `AutoRenewLockOnProgress`, `WorkItemJob` cancels `WorkItemContext.CancellationToken` when the work item lock is lost and abandons the entry instead of completing it, even if the handler returns normally (other renewal errors, including queue entry renewal, are logged). Durations below 5ms throw before mutation. Throttling and empty locks have no-op renewal. Ownership checks are not destination-enforced fencing. Multi-resource acquisition is all-or-nothing even for suffix-overlapping names; successful renewals reset their per-lock interval. After slow renewals, every acquired lease must still fit its conservative lifetime; an elapsed lifetime returns null and releases the partial set (omitted durations use the schedule's one-minute assumption). Lock-set operations attempt every member; `LockException` takes precedence over provider errors (which are logged), and partial-set cleanup failures are logged so callers get `null` or the original error. - **In-memory expired entries**: Writes treat an expired entry as missing before maintenance sweeps it (as Redis does): increments and `SetIfHigher`/`SetIfLower` start fresh, `ReplaceAsync`, `SetExpirationAsync`, `RemoveIfEqualAsync` and `ReplaceIfEqualAsync` miss. The write reclaims the expired entry and raises `ItemExpired` once. Successful live conditional removal and empty-list removal free memory without that event. Size-rejected updates preserve value, TTL, and tracked size. - **In-memory numeric TTLs**: `SetIfHigherAsync` and `SetIfLowerAsync` apply the requested expiration even when the numeric value is unchanged. A zero return value does not mean the TTL was preserved; `null` clears it. Size-rejected updates remain unchanged. - **In-memory lists and cloning**: List writes preserve supported dictionary types and comparers; copying an unsupported `IDictionary` throws `NotSupportedException` without changing the entry. Batch large updates because cloning, custom sizing, and raw dictionary access can require whole-list copies. `Items` honors `CloneValues` without affecting LRU. Collection snapshots do not isolate shared mutable elements when cloning is disabled. See the [caching guide](https://foundatio.readthedocs.io/guide/caching.html#in-memory-update-behavior) for details. diff --git a/docs/guide/locks.md b/docs/guide/locks.md index 65f54eb70..32d98dd31 100644 --- a/docs/guide/locks.md +++ b/docs/guide/locks.md @@ -317,7 +317,7 @@ Choose an interval shorter than the lease duration, leaving time for cache laten `WorkItemJob` applies this pattern when a handler sets `AutoRenewLockOnProgress`. Each `ReportProgressAsync` call renews the queue entry and the work item lock. Renewal errors are logged and progress reporting continues. If the work item lock is lost (`LockException`), `WorkItemContext.CancellationToken` is cancelled, and the job abandons the queue entry for retry instead of completing it, whether the handler throws, returns early, or swallows the cancellation, and even if the job itself is cancelled afterward. A handler that ignores the token keeps running unprotected until it returns. A failed queue entry renewal is only logged: queues don't report a definite loss, and while the work item lock is held a redelivered entry can't be processed concurrently. -When acquiring multiple resources, success means every requested lock was acquired. If any resource is unavailable, the provider releases the partial set and returns `null`, including when resource names share a suffix. While acquisition is in progress, earlier locks are renewed once their renewal interval elapses. A lost lease (`LockException`) makes acquisition return `null`, and it takes precedence over a provider error when both occur, because the loss is definite; the provider error is logged. Otherwise a provider error is propagated. Either way, release is attempted for every acquired lock. Cleanup failures are logged rather than thrown, so callers still receive `null` or the original error; any lease that could not be released remains until it expires. +When acquiring multiple resources, success means every requested lock was acquired. If any resource is unavailable, the provider releases the partial set and returns `null`, including when resource names share a suffix. While acquisition is in progress, earlier locks are renewed once their renewal interval elapses. After waiting for renewal, acquisition checks every member's conservative lease lifetime, including the newest lock. If that lifetime elapsed, acquisition returns `null` and releases the partial set. With no explicit duration, this check uses the renewal schedule's conservative one-minute lifetime, which may be shorter than a provider's default. A lost lease (`LockException`) makes acquisition return `null`, and it takes precedence over a provider error when both occur, because the loss is definite; the provider error is logged. Otherwise a provider error is propagated. Either way, release is attempted for every acquired lock. Cleanup failures are logged rather than thrown, so callers still receive `null` or the original error; any lease that could not be released remains until it expires. Renewing a multi-resource lock renews every member, even if one fails. If a member's lease was lost, `RenewAsync` throws `LockException`, even when another member also hit a provider error (that error is logged). `RenewalCount` increases only when every member renews. diff --git a/src/Foundatio/Lock/ILockProvider.cs b/src/Foundatio/Lock/ILockProvider.cs index c2b6ac9d8..1b29ee464 100644 --- a/src/Foundatio/Lock/ILockProvider.cs +++ b/src/Foundatio/Lock/ILockProvider.cs @@ -349,8 +349,8 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc } /// - /// Renews every lock except the most recently acquired one whose renewal interval has elapsed. - /// Returns false when a lease was lost; provider errors propagate. + /// Renews earlier locks whose renewal interval has elapsed and checks every acquired lease after awaiting renewal. + /// Returns false when a lease was lost or its conservative lifetime elapsed; provider errors propagate. /// private static async Task TryRenewEarlierLocksAsync(List<(ILock Lock, DateTimeOffset LastRenewedUtc)> acquiredLocks, TimeSpan? timeUntilExpires, TimeSpan renewTime, TimeProvider timeProvider, ILogger logger) { @@ -373,6 +373,15 @@ private static async Task TryRenewEarlierLocksAsync(List<(ILock Lock, Date foreach (int index in locksToRenew) acquiredLocks[index] = (acquiredLocks[index].Lock, utcNow); + var checkedAtUtc = timeProvider.GetUtcNow(); + // With no explicit duration, use the same conservative one-minute lifetime as the renewal schedule. + var leaseDuration = timeUntilExpires.GetValueOrDefault(TimeSpan.FromMinutes(1)); + if (acquiredLocks.Any(l => checkedAtUtc - l.LastRenewedUtc >= leaseDuration)) + { + logger.LogWarning("An acquired lease may have expired while renewing locks for {Resource}, releasing acquired locks", acquiredLocks[^1].Lock.Resource); + return false; + } + logger.LogTrace("Renewed {LockCount} locks {Resource} RenewTime={RenewTime:g}", locksToRenew.Length, locksToRenew.Select(index => acquiredLocks[index].Lock.Resource), renewTime); return true; } diff --git a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs index 0b7563206..0b6da1367 100644 --- a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs +++ b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs @@ -348,6 +348,77 @@ public async Task TryAcquireAsync_WithMultipleResources_AfterRenewal_WaitsUntilN Assert.True(await inner.IsLockedAsync("c")); } + [Theory] + [InlineData(59, false)] + [InlineData(60, false)] + [InlineData(61, true)] + public async Task TryAcquireAsync_AfterSlowRenewal_ReturnsOnlyUnexpiredLeases(int renewalSeconds, bool acquiredByAnotherOwner) + { + // Arrange + var timeProvider = new FakeTimeProvider(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider).LoggerFactory(Log)); + var inner = new CacheLockProvider(cache, null, timeProvider, null, Log); + var renewalStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var finishRenewal = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var first = new Mock(); + ILock? firstLease = null; + first.SetupGet(l => l.Resource).Returns("a"); + first.Setup(l => l.ReleaseAsync()).Returns(() => firstLease!.ReleaseAsync()); + first.Setup(l => l.RenewAsync(It.IsAny())).Returns(async (TimeSpan? extension) => + { + await firstLease!.RenewAsync(extension); + renewalStarted.SetResult(); + await finishRenewal.Task; + }); + var locker = CreateMockLockProvider(timeProvider); + locker.Setup(p => p.TryAcquireAsync("a", It.IsAny(), It.IsAny(), It.IsAny())).Returns(async () => + { + firstLease = await inner.TryAcquireAsync("a", TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + return first.Object; + }); + locker.Setup(p => p.TryAcquireAsync("b", It.IsAny(), It.IsAny(), It.IsAny())).Returns(async () => + { + timeProvider.Advance(TimeSpan.FromSeconds(31)); + return await inner.TryAcquireAsync("b", TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + }); + + // Act + var acquisition = locker.Object.TryAcquireAsync(["a", "b"], TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); + try + { + await renewalStarted.Task.WaitAsync(TimeSpan.FromSeconds(5), TestCancellationToken); + timeProvider.Advance(TimeSpan.FromSeconds(renewalSeconds)); + await using var newOwner = acquiredByAnotherOwner + ? await inner.AcquireAsync("b", TimeSpan.FromMinutes(10), cancellationToken: TestCancellationToken) + : null; + finishRenewal.SetResult(); + await using var lockInstance = await acquisition.WaitAsync(TimeSpan.FromSeconds(5), TestCancellationToken); + + // Assert + if (renewalSeconds < 60) + { + Assert.NotNull(lockInstance); + Assert.True(await inner.IsLockedAsync("a")); + Assert.True(await inner.IsLockedAsync("b")); + first.Verify(l => l.ReleaseAsync(), Times.Never); + } + else + { + Assert.Null(lockInstance); + Assert.False(await inner.IsLockedAsync("a")); + Assert.Equal(acquiredByAnotherOwner, await inner.IsLockedAsync("b")); + first.Verify(l => l.ReleaseAsync(), Times.Once); + } + if (newOwner is not null) + await newOwner.RenewAsync(); + } + finally + { + finishRenewal.TrySetResult(); + await using var remaining = await acquisition.WaitAsync(TimeSpan.FromSeconds(5), TestCancellationToken); + } + } + [Fact] public async Task TryAcquireAsync_WithMultipleResources_WhenEarlierLockLost_ReleasesAcquiredLocksAndReturnsNull() {