From 80a2af23ddbf80f480547cd1a4573d54c4383b10 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Mon, 12 Oct 2026 22:41:21 -0500 Subject: [PATCH 1/2] Cancel lost work-item leases without waiting for queue renewal --- .agents/skills/foundatio/SKILL.md | 2 +- docs/guide/jobs.md | 2 + docs/guide/locks.md | 2 +- src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs | 50 +++++----- .../Foundatio.Tests/Jobs/WorkItemJobTests.cs | 95 +++++++++++++++++++ 5 files changed, 126 insertions(+), 25 deletions(-) diff --git a/.agents/skills/foundatio/SKILL.md b/.agents/skills/foundatio/SKILL.md index 8b9128395..349f32c28 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. 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. +- **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). Cancellation on a lost work item lock does not wait for queue renewal. The default handler uses `EmptyLock`; preventing concurrent redelivery requires an exclusive lock from `GetWorkItemLockAsync`. 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/jobs.md b/docs/guide/jobs.md index 9deef430b..bb8f65d1c 100644 --- a/docs/guide/jobs.md +++ b/docs/guide/jobs.md @@ -446,6 +446,8 @@ public record DeleteEntityWorkItem | `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 | +With `AutoRenewLockOnProgress`, a lost work item lock cancels the context token without waiting for queue renewal to finish. Queue renewal errors are logged and do not themselves cancel the handler. The default `WorkItemHandlerBase` uses `EmptyLock`; preventing concurrent redelivery requires overriding `GetWorkItemLockAsync` with an exclusive lock shared by all processors of that work item. Progress reporting still awaits both renewals, so configure provider timeouts. + ### WorkItemHandlers `WorkItemHandlers` is a registry mapping work item data types to their handlers. You can register handlers in several ways: diff --git a/docs/guide/locks.md b/docs/guide/locks.md index 32d98dd31..39cc63ce3 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, 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. +`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 without waiting for the queue renewal to finish, 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 because it does not prove ownership was lost. Protection against concurrent redelivery requires the handler to override `GetWorkItemLockAsync` and acquire an exclusive lock shared by all processors of that work item. The default `WorkItemHandlerBase` returns `EmptyLock`, which has no ownership check: queue renewal failure can therefore allow concurrent processing. Progress reporting still awaits both renewal operations, so configure provider timeouts. 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. diff --git a/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs b/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs index dd4b1bb14..3b6863b57 100644 --- a/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs +++ b/src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs @@ -204,37 +204,41 @@ public async Task ProcessAsync(IQueueEntry queueEntry, /// 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. + /// Queue renewal errors do not prove ownership was lost. Protection against concurrent redelivery requires + /// an exclusive work item lock; the default handler uses an empty lock and provides no such protection. /// private static async Task RenewWorkItemLocksAsync(IWorkItemHandler handler, IQueueEntry queueEntry, ILock workItemLock, WorkItemLeaseLoss leaseLoss) { - var queueEntryRenewal = RenewQueueEntryLockAsync(); - var workItemLockRenewal = RenewWorkItemLockAsync(); + await Task.WhenAll(RenewQueueEntryLockAsync(), RenewWorkItemLockAsync()).AnyContext(); - try - { - await Task.WhenAll(queueEntryRenewal, workItemLockRenewal).AnyContext(); - } - catch (Exception ex) when (workItemLockRenewal.Exception?.InnerException is not LockException) + async Task RenewQueueEntryLockAsync() { - handler.Log.LogError(ex, "Error renewing work item locks: {Message}", ex.Message); + try + { + await queueEntry.RenewLockAsync().AnyContext(); + } + catch (Exception ex) + { + handler.Log.LogError(ex, "Error renewing queue entry lock: {Message}", ex.Message); + } } - catch + + async Task RenewWorkItemLockAsync() { - if (queueEntryRenewal.Exception?.InnerException is { } queueError) - handler.Log.LogError(queueError, "Error renewing queue entry lock: {Message}", queueError.Message); + try + { + await workItemLock.RenewAsync().AnyContext(); + } + catch (LockException ex) + { + handler.Log.LogWarning(ex, "Lost work item lock {Resource} for queue entry {Id}, cancelling work item: {Message}", workItemLock.Resource, queueEntry.Id, ex.Message); + await leaseLoss.SignalAsync().AnyContext(); + } + catch (Exception ex) + { + handler.Log.LogError(ex, "Error renewing work item lock: {Message}", ex.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); - await leaseLoss.SignalAsync().AnyContext(); - - async Task RenewQueueEntryLockAsync() => await queueEntry.RenewLockAsync().AnyContext(); - async Task RenewWorkItemLockAsync() => await workItemLock.RenewAsync().AnyContext(); } /// diff --git a/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs b/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs index f28769dc0..aad6eba9f 100644 --- a/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs +++ b/tests/Foundatio.Tests/Jobs/WorkItemJobTests.cs @@ -15,6 +15,7 @@ using Foundatio.Tests.Extensions; using Foundatio.Xunit; using Microsoft.Extensions.Logging; +using Moq; using Xunit; namespace Foundatio.Tests.Jobs; @@ -337,6 +338,70 @@ await messageBus.SubscribeAsync(status => Assert.Equal(0, countdown.CurrentCount); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ProcessAsync_WhenAutoRenewedWorkItemLockIsLost_CancelsBeforeQueueRenewalCompletes(bool queueRenewalFails) + { + // Arrange + using var cache = new InMemoryCacheClient(o => o.LoggerFactory(Log)); + using var messageBus = new InMemoryMessageBus(o => o.LoggerFactory(Log)); + var queueMock = new Mock>(new InMemoryQueueOptions + { + LoggerFactory = Log, + RetryDelay = TimeSpan.Zero, + Retries = 0 + }) + { CallBase = true }; + using var queue = queueMock.Object; + await queue.EnqueueAsync(new MyWorkItem { SomeData = "Test" }); + var entry = await queue.DequeueAsync(TestCancellationToken); + Assert.NotNull(entry); + + var queueRenewal = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var cancellationObserved = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + queueMock.Setup(q => q.RenewLockAsync(entry)).Returns(queueRenewal.Task); + var locker = new CacheLockProvider(cache, messageBus, null, null, Log); + await using var workItemLock = await locker.AcquireAsync("work-item", cancellationToken: TestCancellationToken); + await workItemLock.ReleaseAsync(); + var handler = new Mock(); + handler.SetupGet(h => h.Log).Returns(Log.CreateLogger()); + handler.SetupGet(h => h.AutoRenewLockOnProgress).Returns(true); + handler.Setup(h => h.GetWorkItemLockAsync(It.IsAny(), It.IsAny())).ReturnsAsync(workItemLock); + handler.Setup(h => h.HandleItemAsync(It.IsAny())).Returns(async (WorkItemContext context) => + { + using var registration = context.CancellationToken.Register(() => cancellationObserved.TrySetResult()); + await context.ReportProgressAsync(50); + }); + var handlers = new WorkItemHandlers(); + handlers.Register(handler.Object); + var job = new WorkItemJob(queue, messageBus, handlers, Log); + + // Act + var processing = job.ProcessAsync(entry, TestCancellationToken); + try + { + await cancellationObserved.Task.WaitAsync(TimeSpan.FromSeconds(5), TestCancellationToken); + + // Assert + Assert.False(queueRenewal.Task.IsCompleted); + Assert.False(processing.IsCompleted); + } + finally + { + if (queueRenewalFails) + queueRenewal.TrySetException(new InvalidOperationException("Queue renewal failed")); + else + queueRenewal.TrySetResult(); + + await processing.WaitAsync(TimeSpan.FromSeconds(5), TestCancellationToken); + } + + Assert.False((await processing).IsSuccess); + Assert.True(entry.IsAbandoned); + Assert.False(entry.IsCompleted); + } + [Theory] [InlineData(LostLeaseResponse.Throw)] [InlineData(LostLeaseResponse.ReturnEarly)] @@ -417,6 +482,36 @@ public async Task ProcessAsync_WhenJobIsCancelledAfterWorkItemLockIsLost_Abandon Assert.Equal(0, stats.Completed); Assert.Equal(1, stats.Abandoned); } + + [Fact] + public async Task ProcessAsync_WhenQueueRenewalFails_DoesNotCancelHandler() + { + // Arrange + using var cache = new InMemoryCacheClient(o => o.LoggerFactory(Log)); + using var messageBus = new InMemoryMessageBus(o => o.LoggerFactory(Log)); + var queueMock = new Mock>(new InMemoryQueueOptions { LoggerFactory = Log }) { CallBase = true }; + using var queue = queueMock.Object; + await queue.EnqueueAsync(new MyWorkItem { SomeData = "Test" }); + var entry = await queue.DequeueAsync(TestCancellationToken); + Assert.NotNull(entry); + queueMock.Setup(q => q.RenewLockAsync(entry)).ThrowsAsync(new InvalidOperationException("Queue renewal failed")); + var locker = new CacheLockProvider(cache, messageBus, null, null, Log); + var handler = new LockedWorkItemHandler(locker, Log, loseLock: false) { AutoRenewLockOnProgress = true }; + var handlers = new WorkItemHandlers(); + handlers.Register(handler); + var job = new WorkItemJob(queue, messageBus, handlers, Log); + + // Act + var result = await job.ProcessAsync(entry, TestCancellationToken); + + // Assert + Assert.True(result.IsSuccess); + Assert.False(handler.ObservedCancellation); + Assert.True(handler.CompletedWork); + Assert.True(entry.IsCompleted); + Assert.False(entry.IsAbandoned); + } + } public enum LostLeaseResponse From a7718f9148177ae504dd2f860d3f3ddc61e6bfe9 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Wed, 14 Oct 2026 22:22:09 -0500 Subject: [PATCH 2/2] Account for acquisition latency and always await renewal shutdown --- .agents/skills/foundatio/SKILL.md | 2 +- docs/guide/locks.md | 14 ++-- src/Foundatio/Lock/ILockProvider.cs | 13 ++-- .../Locks/InMemoryLockTests.cs | 69 ++++++++++++++----- 4 files changed, 70 insertions(+), 28 deletions(-) diff --git a/.agents/skills/foundatio/SKILL.md b/.agents/skills/foundatio/SKILL.md index 349f32c28..24dcf60e4 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). Cancellation on a lost work item lock does not wait for queue renewal. The default handler uses `EmptyLock`; preventing concurrent redelivery requires an exclusive lock from `GetWorkItemLockAsync`. 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. +- **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). Cancellation on a lost work item lock does not wait for queue renewal. The default handler uses `EmptyLock`; preventing concurrent redelivery requires an exclusive lock from `GetWorkItemLockAsync`. 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. Acquisition latency counts toward renewal scheduling, including the newest lease and one-resource collections. 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 39cc63ce3..40de6a036 100644 --- a/docs/guide/locks.md +++ b/docs/guide/locks.md @@ -284,8 +284,14 @@ try } finally { - await cts.CancelAsync(); - await renewTask; + try + { + await cts.CancelAsync(); + } + finally + { + await renewTask; + } } async Task RenewPeriodicallyAsync() @@ -311,13 +317,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. +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 nested `finally` observes renewal failure and waits for renewal to stop before `await using` releases the lock, even if a cancellation callback throws. 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` 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 without waiting for the queue renewal to finish, 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 because it does not prove ownership was lost. Protection against concurrent redelivery requires the handler to override `GetWorkItemLockAsync` and acquire an exclusive lock shared by all processors of that work item. The default `WorkItemHandlerBase` returns `EmptyLock`, which has no ownership check: queue renewal failure can therefore allow concurrent processing. Progress reporting still awaits both renewal operations, so configure provider timeouts. -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. +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, locks are renewed once their renewal interval elapses. The interval starts before each provider acquisition call, so a slow response also triggers renewal of the newest lease, even for a one-resource collection. An expired lease is rejected instead of being returned to protected work. 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 1b29ee464..5e9fd9a7a 100644 --- a/src/Foundatio/Lock/ILockProvider.cs +++ b/src/Foundatio/Lock/ILockProvider.cs @@ -320,6 +320,7 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc { foreach (string resource in resourceList) { + var acquireStartedUtc = timeProvider.GetUtcNow(); var l = await provider.TryAcquireAsync(resource, timeUntilExpires, releaseOnDispose, cancellationToken).AnyContext(); if (l is null) { @@ -327,10 +328,10 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc return null; } - acquiredLocks.Add((l, timeProvider.GetUtcNow())); + acquiredLocks.Add((l, acquireStartedUtc)); - // Renew earlier locks while waiting for the remaining resources. - if (acquiredLocks.Count > 1 && renewTime > TimeSpan.Zero && !await TryRenewEarlierLocksAsync(acquiredLocks, timeUntilExpires, renewTime, timeProvider, logger).AnyContext()) + // Include acquisition latency: even the newest lease may need renewal before it can be returned. + if (renewTime > TimeSpan.Zero && !await TryRenewAcquiredLocksAsync(acquiredLocks, timeUntilExpires, renewTime, timeProvider, logger).AnyContext()) return null; } @@ -349,14 +350,14 @@ public static Task TryUsingAsync(this ILockProvider locker, string resourc } /// - /// Renews earlier locks whose renewal interval has elapsed and checks every acquired lease after awaiting renewal. + /// Renews 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) + private static async Task TryRenewAcquiredLocksAsync(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].LastRenewedUtc < renewBeforeUtc).ToArray(); + int[] locksToRenew = Enumerable.Range(0, acquiredLocks.Count).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 0b6da1367..d6d01b8ef 100644 --- a/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs +++ b/tests/Foundatio.Tests/Locks/InMemoryLockTests.cs @@ -1,4 +1,5 @@ using System; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Foundatio.Caching; @@ -321,31 +322,38 @@ public override Task RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreserve return base.RenewAsync_WithReplacedOwner_ThrowsLockExceptionAndPreservesCurrentOwner(scoped); } - [Fact] - public async Task TryAcquireAsync_WithMultipleResources_AfterRenewal_WaitsUntilNextRenewalInterval() + [Theory] + [InlineData(1, 59)] + [InlineData(1, 61)] + [InlineData(2, 59)] + [InlineData(2, 61)] + public async Task TryAcquireAsync_AfterSlowAcquisition_DoesNotReturnExpiredLease(int resourceCount, int acquisitionSeconds) { // 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)); + using var cache = new InMemoryCacheClient(o => o.TimeProvider(timeProvider)); 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; - }); + var locker = CreateMockLockProvider(timeProvider); + var resources = Enumerable.Range(0, resourceCount).Select(i => i.ToString()).ToArray(); + locker.Setup(p => p.TryAcquireAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(async (string resource, TimeSpan? duration, bool release, CancellationToken token) => + { + var result = await inner.TryAcquireAsync(resource, duration, release, token); + if (resource == resources[^1]) + timeProvider.Advance(TimeSpan.FromSeconds(acquisitionSeconds)); + return result; + }); // Act - await using var lockInstance = await locker.TryAcquireAsync(["a", "b", "c"], TimeSpan.FromMinutes(1), cancellationToken: TestContext.Current.CancellationToken); + await using var acquired = await locker.Object.TryAcquireAsync(resources, TimeSpan.FromMinutes(1), cancellationToken: TestCancellationToken); // 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")); + if (acquisitionSeconds < 60) + Assert.NotNull(acquired); + else + Assert.Null(acquired); + foreach (string resource in resources) + Assert.Equal(acquisitionSeconds < 60, await inner.IsLockedAsync(resource)); } [Theory] @@ -419,6 +427,33 @@ public async Task TryAcquireAsync_AfterSlowRenewal_ReturnsOnlyUnexpiredLeases(in } } + [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() {