fix(worker): Stuck-Running-Fenster in Continue- und Stop-Pfad schließen
TaskRunner.ContinueAsync: Claim, SeedAsync und SetupMcpConfigAsync liefen vor dem try/catch (anders als RunAsync) - warf einer davon nach dem Running-Claim, propagierte die Exception ungefangen bis zu OverrideSlotService.RunContinueInSlotAsync, das nur loggt. Task blieb Running. Fix: derselbe Aufbau wie RunAsync (Claim+Seed+MCP-Setup im try, MarkFailed im catch, mcpToken/mcpConfigPath vor dem try auf null). SetupMcpConfigAsync bekommt zusätzlich einen onTokenRegistered-Callback, damit die äußere mcpToken-Variable den Token sofort nach dem Register sieht - sonst hätte ein Fehler zwischen Register und Rückgabe (z.B. File.WriteAllTextAsync) den Token in der TaskRunTokenRegistry geleakt (betrifft denselben Aufruf in RunAsync mit, daher dort ebenfalls verdrahtet - RunAsync-Struktur selbst unverändert). QueueService.RunInSlotAsync: der Stop-Button (CancelTask) cancelt die Slot-CTS direkt ohne TaskStateService-Schreibzugriff. Traf das die Pre-Dispatch-DB-Reads, loggte der OCE-Catch nur und die vom Picker bereits auf Running geclaimte Task blieb dort für immer hängen. Fix: Status im Catch neu lesen und nur wenn er noch Running ist über TaskStateService.CancelAsync auf Cancelled setzen - ein Ursprung, der bereits selbst einen Terminalstatus geschrieben hat (z.B. CancelReview), wird nicht überschrieben. Kommentar korrigiert. TDD: neue Tests in ContinueAsyncExceptionTests (Seed-/MCP-Setup-Fehler nach Claim -> Failed, kein Token-Leak) und QueueServiceSlotFailureTests (Stop während Pre-Dispatch -> Cancelled statt Running; ein bereits terminal geschriebener Status wird nicht gestompt) vorher rot, jetzt grün. Worker.Tests: 1213/1213 grün, Worker baut in Release.
This commit is contained in:
@@ -348,9 +348,32 @@ public sealed class QueueService : BackgroundService
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Cancellation is driven by the cancel path, which already wrote the terminal status.
|
||||
// Marking the task Failed here would be a regression (it would stomp Cancelled).
|
||||
// Most cancellation origins (TaskStateService.CancelAsync, the runner's own
|
||||
// OCE handling) already write a terminal status before the CTS is actually
|
||||
// cancelled, so by the time we get here the task has already left Running.
|
||||
// A plain Stop (QueueService.CancelTask cancels the slot's CTS directly, with no
|
||||
// TaskStateService write) landing during the pre-dispatch reads above is the one
|
||||
// gap nothing else covers — the picker's claim already committed status='running'
|
||||
// and nothing since has moved it off that. Only close that specific gap: write
|
||||
// Cancelled when (and only when) the task is still Running, so an origin that
|
||||
// already wrote its own terminal status is never stomped.
|
||||
_logger.LogInformation("Slot runner cancelled for task {TaskId}", taskId);
|
||||
try
|
||||
{
|
||||
TaskStatus current;
|
||||
using (var context = _dbFactory.CreateDbContext())
|
||||
current = await context.Tasks.AsNoTracking()
|
||||
.Where(t => t.Id == taskId)
|
||||
.Select(t => t.Status)
|
||||
.FirstAsync(CancellationToken.None);
|
||||
|
||||
if (current == TaskStatus.Running)
|
||||
await _state.CancelAsync(taskId, DateTime.UtcNow, CancellationToken.None);
|
||||
}
|
||||
catch (Exception cancelEx)
|
||||
{
|
||||
_logger.LogError(cancelEx, "Could not finalize cancelled task {TaskId} after slot cancellation", taskId);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
|
||||
@@ -17,6 +17,11 @@ public sealed class TaskRunTokenRegistry
|
||||
return false;
|
||||
}
|
||||
public void Unregister(string token) => _tokenToTaskId.TryRemove(token, out _);
|
||||
|
||||
// Test-only leak check: a run must never leave its token registered after it ends,
|
||||
// no matter which step of setup failed.
|
||||
public bool HasAnyTokenFor(string taskId) => _tokenToTaskId.Values.Contains(taskId);
|
||||
|
||||
public static string GenerateToken()
|
||||
{
|
||||
var bytes = RandomNumberGenerator.GetBytes(32);
|
||||
|
||||
@@ -124,7 +124,7 @@ public sealed class TaskRunner
|
||||
var runDir = prep.RunDir!;
|
||||
|
||||
var resolvedConfig = await ResolveConfigAsync(task, list, listConfig, null, ct);
|
||||
(mcpToken, mcpConfigPath, resolvedConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct);
|
||||
(mcpToken, mcpConfigPath, resolvedConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct, token => mcpToken = token);
|
||||
|
||||
await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct);
|
||||
|
||||
@@ -237,32 +237,35 @@ public sealed class TaskRunner
|
||||
runDir = Path.Combine(_cfg.SandboxRoot, taskId);
|
||||
}
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
// See RunAsync: queue dispatches arrive pre-claimed by the picker.
|
||||
if (!alreadyClaimed)
|
||||
{
|
||||
var startResult = await _state.StartRunningAsync(taskId, now, ct);
|
||||
if (!startResult.Ok)
|
||||
{
|
||||
_logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", taskId, startResult.Reason);
|
||||
return;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// Queue-claimed dispatches skip StartRunningAsync, so broadcast TaskUpdated here
|
||||
// (see RunAsync for the full rationale).
|
||||
await _broadcaster.TaskUpdated(taskId);
|
||||
}
|
||||
await _broadcaster.TaskStarted(slot, taskId, now);
|
||||
|
||||
await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct);
|
||||
|
||||
var (mcpToken, mcpConfigPath, mcpConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct);
|
||||
resolvedConfig = mcpConfig;
|
||||
|
||||
string? mcpToken = null;
|
||||
string? mcpConfigPath = null;
|
||||
try
|
||||
{
|
||||
var now = DateTime.UtcNow;
|
||||
// See RunAsync: queue dispatches arrive pre-claimed by the picker. Claim, skill
|
||||
// seeding, and MCP setup all live inside this try (mirroring RunAsync) so a failure
|
||||
// in any of them still lands the task in Failed instead of stuck Running.
|
||||
if (!alreadyClaimed)
|
||||
{
|
||||
var startResult = await _state.StartRunningAsync(taskId, now, ct);
|
||||
if (!startResult.Ok)
|
||||
{
|
||||
_logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", taskId, startResult.Reason);
|
||||
return;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// Queue-claimed dispatches skip StartRunningAsync, so broadcast TaskUpdated here
|
||||
// (see RunAsync for the full rationale).
|
||||
await _broadcaster.TaskUpdated(taskId);
|
||||
}
|
||||
await _broadcaster.TaskStarted(slot, taskId, now);
|
||||
|
||||
await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct);
|
||||
|
||||
(mcpToken, mcpConfigPath, resolvedConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct, token => mcpToken = token);
|
||||
|
||||
var nextRunNumber = lastRun.RunNumber + 1;
|
||||
var result = await RunOnceAsync(taskId, task.Number, task.Title, slot, runDir, resolvedConfig, nextRunNumber, false, followUpPrompt, ct);
|
||||
|
||||
@@ -290,8 +293,12 @@ public sealed class TaskRunner
|
||||
}
|
||||
finally
|
||||
{
|
||||
_tokens.Unregister(mcpToken);
|
||||
try { File.Delete(mcpConfigPath); } catch { /* best effort */ }
|
||||
if (mcpToken is not null)
|
||||
{
|
||||
_tokens.Unregister(mcpToken);
|
||||
if (mcpConfigPath is not null)
|
||||
try { File.Delete(mcpConfigPath); } catch { /* best effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -300,11 +307,15 @@ public sealed class TaskRunner
|
||||
// SuggestImprovement for filing out-of-scope follow-ups. Used by both a fresh run and a
|
||||
// --resume continuation, which needs the same wiring or the resumed session loses every
|
||||
// mcp__claudedo_run__* tool.
|
||||
// onTokenRegistered fires the instant the token is registered, before any I/O that could
|
||||
// still fail — callers assign it straight into their outer (pre-try) mcpToken variable so a
|
||||
// later throw in this method still leaves the finally block able to unregister it.
|
||||
private async Task<(string McpToken, string McpConfigPath, ClaudeRunConfig Config)> SetupMcpConfigAsync(
|
||||
TaskEntity task, ClaudeRunConfig resolvedConfig, CancellationToken ct)
|
||||
TaskEntity task, ClaudeRunConfig resolvedConfig, CancellationToken ct, Action<string>? onTokenRegistered = null)
|
||||
{
|
||||
var mcpToken = TaskRunTokenRegistry.GenerateToken();
|
||||
_tokens.Register(mcpToken, task.Id);
|
||||
onTokenRegistered?.Invoke(mcpToken);
|
||||
Directory.CreateDirectory(_cfg.LogRoot);
|
||||
var mcpConfigPath = Path.Combine(_cfg.LogRoot, $"{task.Id}_mcp.json");
|
||||
await File.WriteAllTextAsync(mcpConfigPath, BuildRunMcpConfigJson(mcpToken), ct);
|
||||
|
||||
@@ -9,10 +9,15 @@ public sealed class FakeSessionSkillSeeder : ISessionSkillSeeder
|
||||
public readonly record struct Call(string WorkingDir, IReadOnlyList<string> SkillNames, bool IsWorktree);
|
||||
public readonly ConcurrentQueue<Call> Calls = new();
|
||||
|
||||
// Lets a test simulate a seeding failure (e.g. a broken skill install) to verify callers
|
||||
// don't leave a task stuck Running when SeedAsync throws.
|
||||
public Exception? ThrowOnSeed;
|
||||
|
||||
public Task SeedAsync(string workingDir, IReadOnlyList<string> skillNames, bool isWorktree, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref CallCount);
|
||||
Calls.Enqueue(new Call(workingDir, skillNames, isWorktree));
|
||||
if (ThrowOnSeed is not null) throw ThrowOnSeed;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ using ClaudeDo.Data.Repositories;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Hub;
|
||||
using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Skills;
|
||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
@@ -29,14 +30,101 @@ public sealed class ContinueAsyncExceptionTests : IDisposable
|
||||
|
||||
public void Dispose() { _db.Dispose(); try { Directory.Delete(_tempDir, true); } catch { } }
|
||||
|
||||
private TaskRunner BuildRunner(IClaudeProcess claude, ClaudeDoDbContext ctx)
|
||||
private TaskRunner BuildRunner(
|
||||
IClaudeProcess claude, ClaudeDoDbContext ctx,
|
||||
ISessionSkillSeeder? skillSeeder = null, TaskRunTokenRegistry? tokens = null, WorkerConfig? cfg = null)
|
||||
{
|
||||
var dbFactory = _db.CreateFactory();
|
||||
var broadcaster = new HubBroadcaster(new CapturingHubContext());
|
||||
var state = TaskStateServiceBuilder.Build(dbFactory).State;
|
||||
var wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, _cfg, NullLogger<WorktreeManager>.Instance);
|
||||
return new TaskRunner(claude, dbFactory, broadcaster, wt, new ClaudeArgsBuilder(), _cfg,
|
||||
NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
|
||||
var effectiveCfg = cfg ?? _cfg;
|
||||
var wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, effectiveCfg, NullLogger<WorktreeManager>.Instance);
|
||||
return new TaskRunner(claude, dbFactory, broadcaster, wt, new ClaudeArgsBuilder(), effectiveCfg,
|
||||
NullLogger<TaskRunner>.Instance, state, tokens ?? new TaskRunTokenRegistry(), new AttachmentStore(),
|
||||
skillSeeder ?? new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
|
||||
}
|
||||
|
||||
private async Task<string> SeedContinuableTaskAsync(string sessionId)
|
||||
{
|
||||
string listId, taskId;
|
||||
using var ctx = _db.CreateContext();
|
||||
listId = Guid.NewGuid().ToString();
|
||||
ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", WorkingDir = null, CreatedAt = DateTime.UtcNow });
|
||||
|
||||
taskId = Guid.NewGuid().ToString();
|
||||
ctx.Tasks.Add(new TaskEntity
|
||||
{
|
||||
Id = taskId,
|
||||
ListId = listId,
|
||||
Title = "Continue me",
|
||||
Status = TaskStatus.WaitingForReview,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
});
|
||||
await ctx.SaveChangesAsync();
|
||||
|
||||
await new TaskRunRepository(ctx).AddAsync(new TaskRunEntity
|
||||
{
|
||||
Id = Guid.NewGuid().ToString(),
|
||||
TaskId = taskId,
|
||||
RunNumber = 1,
|
||||
IsRetry = false,
|
||||
Prompt = "original prompt",
|
||||
SessionId = sessionId,
|
||||
StartedAt = DateTime.UtcNow.AddMinutes(-5),
|
||||
FinishedAt = DateTime.UtcNow.AddMinutes(-1),
|
||||
ExitCode = 0,
|
||||
ResultMarkdown = "first result",
|
||||
});
|
||||
return taskId;
|
||||
}
|
||||
|
||||
// Bug: SeedAsync used to run after the Running claim but before ContinueAsync's own
|
||||
// try/catch started, so a throw here propagated straight out of ContinueAsync to the
|
||||
// caller (OverrideSlotService.RunContinueInSlotAsync), which only logs — leaving the task
|
||||
// stuck Running forever.
|
||||
[Fact]
|
||||
public async Task ContinueAsync_SkillSeederThrowsAfterClaim_MarksTaskFailed_NotStuckRunning()
|
||||
{
|
||||
var taskId = await SeedContinuableTaskAsync("sess-seed-throws");
|
||||
var throwingSeeder = new FakeSessionSkillSeeder { ThrowOnSeed = new InvalidOperationException("skill seed exploded") };
|
||||
|
||||
using var ctx2 = _db.CreateContext();
|
||||
var runner = BuildRunner(new FakeClaudeProcess(), ctx2, skillSeeder: throwingSeeder);
|
||||
|
||||
await runner.ContinueAsync(taskId, "please continue", "slot-1", CancellationToken.None);
|
||||
|
||||
using var verify = _db.CreateContext();
|
||||
var task = await new TaskRepository(verify).GetByIdAsync(taskId);
|
||||
Assert.NotNull(task);
|
||||
Assert.Equal(TaskStatus.Failed, task.Status);
|
||||
}
|
||||
|
||||
// Bug: SetupMcpConfigAsync registers the per-run MCP token *before* it can fail (writing
|
||||
// the config file to disk). When it threw after Register but before the try/catch existed,
|
||||
// the token was never unregistered — a leak in TaskRunTokenRegistry on top of the stuck
|
||||
// Running task.
|
||||
[Fact]
|
||||
public async Task ContinueAsync_McpSetupThrowsAfterTokenRegistered_MarksTaskFailed_AndUnregistersToken()
|
||||
{
|
||||
var taskId = await SeedContinuableTaskAsync("sess-mcp-throws");
|
||||
|
||||
// SetupMcpConfigAsync calls Directory.CreateDirectory(cfg.LogRoot) right after
|
||||
// registering the token; pointing LogRoot at an existing file makes that call throw.
|
||||
var badLogRoot = Path.Combine(_tempDir, "logroot_is_a_file");
|
||||
await File.WriteAllTextAsync(badLogRoot, "not a directory");
|
||||
var cfg = new WorkerConfig { SandboxRoot = _tempDir, LogRoot = badLogRoot };
|
||||
var tokens = new TaskRunTokenRegistry();
|
||||
|
||||
using var ctx2 = _db.CreateContext();
|
||||
var runner = BuildRunner(new FakeClaudeProcess(), ctx2, tokens: tokens, cfg: cfg);
|
||||
|
||||
await runner.ContinueAsync(taskId, "please continue", "slot-1", CancellationToken.None);
|
||||
|
||||
using var verify = _db.CreateContext();
|
||||
var task = await new TaskRepository(verify).GetByIdAsync(taskId);
|
||||
Assert.NotNull(task);
|
||||
Assert.Equal(TaskStatus.Failed, task.Status);
|
||||
Assert.False(tokens.HasAnyTokenFor(taskId));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
||||
@@ -9,6 +9,7 @@ using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||
using ClaudeDo.Worker.Usage;
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
|
||||
@@ -180,7 +181,7 @@ public sealed class QueueServiceSlotFailureTests : IDisposable
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_cancelled_slot_run_does_not_mark_the_task_Failed()
|
||||
public async Task A_cancelled_slot_run_is_marked_Cancelled_not_left_Running()
|
||||
{
|
||||
var listId = await SeedListAsync();
|
||||
var taskId = Guid.NewGuid().ToString();
|
||||
@@ -203,21 +204,105 @@ public sealed class QueueServiceSlotFailureTests : IDisposable
|
||||
await service.StartAsync(outerCts.Token);
|
||||
waker.Wake();
|
||||
|
||||
// Wait for the slot to be claimed and then released again (RunInSlotAsync's
|
||||
// ContinueWith removes it once the catch block — ours or a stray one — finishes).
|
||||
var deadline = DateTime.UtcNow.AddSeconds(10);
|
||||
while (service.GetActive().Any(a => a.taskId == taskId) && DateTime.UtcNow < deadline)
|
||||
await Task.Delay(25);
|
||||
await Task.Delay(100); // let the fire-and-forget continuation fully settle
|
||||
var reloaded = await PollUntilLeftQueuedAsync(taskId);
|
||||
|
||||
TaskEntity? reloaded;
|
||||
using (var verify = _db.CreateContext())
|
||||
reloaded = await new TaskRepository(verify).GetByIdAsync(taskId);
|
||||
|
||||
// The picker's atomic claim already flipped it to Running; the cancelled slot run must
|
||||
// leave it there rather than flipping it to Failed.
|
||||
Assert.Equal(TaskStatus.Running, reloaded!.Status);
|
||||
Assert.DoesNotContain(hub.Proxy.Calls,
|
||||
// This reproduces a plain Stop (QueueService.CancelTask cancels the slot's CTS
|
||||
// directly, without going through TaskStateService) landing during the pre-dispatch DB
|
||||
// reads — before TaskRunner's own claim/try-catch ever starts. Nothing else writes a
|
||||
// terminal status for that window, so the OCE catch itself must close it.
|
||||
Assert.Equal(TaskStatus.Cancelled, reloaded!.Status);
|
||||
Assert.Contains(hub.Proxy.Calls,
|
||||
c => c.Method == "TaskUpdated" && (string)c.Args[0]! == taskId);
|
||||
}
|
||||
|
||||
// A_cancelled_slot_run_is_marked_Cancelled_not_left_Running above reproduces a plain Stop.
|
||||
// TaskStateService.CancelAsync (the hub's CancelReview action) is a different origin: it
|
||||
// writes Cancelled *before* it cancels the run's CTS via RunCancellationRegistry, so by the
|
||||
// time RunInSlotAsync's OCE catch runs, the task has already left Running. The catch must
|
||||
// recognize that and not blindly stomp whatever terminal status is already there.
|
||||
[Fact]
|
||||
public async Task A_cancelled_slot_run_does_not_stomp_a_status_already_written_terminal()
|
||||
{
|
||||
var listId = await SeedListAsync();
|
||||
var taskId = Guid.NewGuid().ToString();
|
||||
|
||||
using (var ctx = _db.CreateContext())
|
||||
{
|
||||
ctx.Tasks.Add(new TaskEntity
|
||||
{
|
||||
Id = taskId, ListId = listId, Title = "T", Status = TaskStatus.Queued,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
});
|
||||
await ctx.SaveChangesAsync();
|
||||
}
|
||||
|
||||
var outerCts = new CancellationTokenSource();
|
||||
var realPicker = new QueuePicker(_db.CreateFactory());
|
||||
var picker = new ClaimMarkFailedThenCancelPicker(realPicker, outerCts, _db);
|
||||
var (service, hub, waker) = CreateService(picker);
|
||||
|
||||
await service.StartAsync(outerCts.Token);
|
||||
waker.Wake();
|
||||
|
||||
var reloaded = await PollUntilLeftQueuedAsync(taskId);
|
||||
|
||||
// Some other terminal write (simulating CancelAsync/FailAsync having already run) must
|
||||
// survive — the OCE catch must not overwrite it with Cancelled just because it observed
|
||||
// a cancellation.
|
||||
Assert.Equal(TaskStatus.Failed, reloaded!.Status);
|
||||
}
|
||||
|
||||
// Polls until the task leaves its initial Queued status (the claim + dispatch + OCE-catch
|
||||
// chain settling), or gives up at the deadline; then waits a further grace period so the
|
||||
// fire-and-forget continuation (queue-slot cleanup, any terminal-status write) fully lands
|
||||
// before the caller reads the final state.
|
||||
private async Task<TaskEntity?> PollUntilLeftQueuedAsync(string taskId)
|
||||
{
|
||||
TaskEntity? reloaded = null;
|
||||
var deadline = DateTime.UtcNow.AddSeconds(10);
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
using var verify = _db.CreateContext();
|
||||
reloaded = await new TaskRepository(verify).GetByIdAsync(taskId);
|
||||
if (reloaded!.Status != TaskStatus.Queued) break;
|
||||
await Task.Delay(25);
|
||||
}
|
||||
await Task.Delay(150); // let the fire-and-forget continuation fully settle
|
||||
|
||||
using var final = _db.CreateContext();
|
||||
return await new TaskRepository(final).GetByIdAsync(taskId);
|
||||
}
|
||||
|
||||
// Simulates a terminal status already having been written (by TaskStateService, from some
|
||||
// other origin) between the picker's claim and RunInSlotAsync's OCE catch running.
|
||||
private sealed class ClaimMarkFailedThenCancelPicker : IQueuePicker
|
||||
{
|
||||
private readonly IQueuePicker _inner;
|
||||
private readonly CancellationTokenSource _cancelAfterClaim;
|
||||
private readonly DbFixture _db;
|
||||
|
||||
public ClaimMarkFailedThenCancelPicker(IQueuePicker inner, CancellationTokenSource cancelAfterClaim, DbFixture db)
|
||||
{
|
||||
_inner = inner;
|
||||
_cancelAfterClaim = cancelAfterClaim;
|
||||
_db = db;
|
||||
}
|
||||
|
||||
public async Task<TaskEntity?> ClaimNextAsync(DateTime now, CancellationToken ct)
|
||||
{
|
||||
var claimed = await _inner.ClaimNextAsync(now, ct);
|
||||
if (claimed is not null)
|
||||
{
|
||||
using (var ctx = _db.CreateContext())
|
||||
{
|
||||
await ctx.Tasks.Where(t => t.Id == claimed.Id)
|
||||
.ExecuteUpdateAsync(s => s
|
||||
.SetProperty(t => t.Status, TaskStatus.Failed)
|
||||
.SetProperty(t => t.FinishedAt, DateTime.UtcNow));
|
||||
}
|
||||
_cancelAfterClaim.Cancel();
|
||||
}
|
||||
return claimed;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user