chore(claude-do): merge fix(worker): Stuck-Running-Fenster in Continue- und Stop-Pfa
ClaudeDo-Task: 6db1a9e2-aa1d-4911-a763-76b136f97946
This commit is contained in:
@@ -348,9 +348,32 @@ public sealed class QueueService : BackgroundService
|
|||||||
}
|
}
|
||||||
catch (OperationCanceledException)
|
catch (OperationCanceledException)
|
||||||
{
|
{
|
||||||
// Cancellation is driven by the cancel path, which already wrote the terminal status.
|
// Most cancellation origins (TaskStateService.CancelAsync, the runner's own
|
||||||
// Marking the task Failed here would be a regression (it would stomp Cancelled).
|
// 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);
|
_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)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -17,6 +17,11 @@ public sealed class TaskRunTokenRegistry
|
|||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
public void Unregister(string token) => _tokenToTaskId.TryRemove(token, out _);
|
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()
|
public static string GenerateToken()
|
||||||
{
|
{
|
||||||
var bytes = RandomNumberGenerator.GetBytes(32);
|
var bytes = RandomNumberGenerator.GetBytes(32);
|
||||||
|
|||||||
@@ -124,7 +124,7 @@ public sealed class TaskRunner
|
|||||||
var runDir = prep.RunDir!;
|
var runDir = prep.RunDir!;
|
||||||
|
|
||||||
var resolvedConfig = await ResolveConfigAsync(task, list, listConfig, null, ct);
|
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);
|
await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct);
|
||||||
|
|
||||||
@@ -237,8 +237,14 @@ public sealed class TaskRunner
|
|||||||
runDir = Path.Combine(_cfg.SandboxRoot, taskId);
|
runDir = Path.Combine(_cfg.SandboxRoot, taskId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
string? mcpToken = null;
|
||||||
|
string? mcpConfigPath = null;
|
||||||
|
try
|
||||||
|
{
|
||||||
var now = DateTime.UtcNow;
|
var now = DateTime.UtcNow;
|
||||||
// See RunAsync: queue dispatches arrive pre-claimed by the picker.
|
// 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)
|
if (!alreadyClaimed)
|
||||||
{
|
{
|
||||||
var startResult = await _state.StartRunningAsync(taskId, now, ct);
|
var startResult = await _state.StartRunningAsync(taskId, now, ct);
|
||||||
@@ -258,11 +264,8 @@ public sealed class TaskRunner
|
|||||||
|
|
||||||
await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct);
|
await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct);
|
||||||
|
|
||||||
var (mcpToken, mcpConfigPath, mcpConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct);
|
(mcpToken, mcpConfigPath, resolvedConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct, token => mcpToken = token);
|
||||||
resolvedConfig = mcpConfig;
|
|
||||||
|
|
||||||
try
|
|
||||||
{
|
|
||||||
var nextRunNumber = lastRun.RunNumber + 1;
|
var nextRunNumber = lastRun.RunNumber + 1;
|
||||||
var result = await RunOnceAsync(taskId, task.Number, task.Title, slot, runDir, resolvedConfig, nextRunNumber, false, followUpPrompt, ct);
|
var result = await RunOnceAsync(taskId, task.Number, task.Title, slot, runDir, resolvedConfig, nextRunNumber, false, followUpPrompt, ct);
|
||||||
|
|
||||||
@@ -289,22 +292,30 @@ public sealed class TaskRunner
|
|||||||
await MarkFailed(taskId, task.Number, task.Title, slot, $"Unhandled error: {ex.Message}");
|
await MarkFailed(taskId, task.Number, task.Title, slot, $"Unhandled error: {ex.Message}");
|
||||||
}
|
}
|
||||||
finally
|
finally
|
||||||
|
{
|
||||||
|
if (mcpToken is not null)
|
||||||
{
|
{
|
||||||
_tokens.Unregister(mcpToken);
|
_tokens.Unregister(mcpToken);
|
||||||
|
if (mcpConfigPath is not null)
|
||||||
try { File.Delete(mcpConfigPath); } catch { /* best effort */ }
|
try { File.Delete(mcpConfigPath); } catch { /* best effort */ }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Every run gets a per-run MCP identity so the agent can ask the user a mid-run question
|
// Every run gets a per-run MCP identity so the agent can ask the user a mid-run question
|
||||||
// via AskUser. Improvement-eligible (standalone top-level) runs additionally get
|
// via AskUser. Improvement-eligible (standalone top-level) runs additionally get
|
||||||
// SuggestImprovement for filing out-of-scope follow-ups. Used by both a fresh run and a
|
// 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
|
// --resume continuation, which needs the same wiring or the resumed session loses every
|
||||||
// mcp__claudedo_run__* tool.
|
// 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(
|
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();
|
var mcpToken = TaskRunTokenRegistry.GenerateToken();
|
||||||
_tokens.Register(mcpToken, task.Id);
|
_tokens.Register(mcpToken, task.Id);
|
||||||
|
onTokenRegistered?.Invoke(mcpToken);
|
||||||
Directory.CreateDirectory(_cfg.LogRoot);
|
Directory.CreateDirectory(_cfg.LogRoot);
|
||||||
var mcpConfigPath = Path.Combine(_cfg.LogRoot, $"{task.Id}_mcp.json");
|
var mcpConfigPath = Path.Combine(_cfg.LogRoot, $"{task.Id}_mcp.json");
|
||||||
await File.WriteAllTextAsync(mcpConfigPath, BuildRunMcpConfigJson(mcpToken), ct);
|
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 record struct Call(string WorkingDir, IReadOnlyList<string> SkillNames, bool IsWorktree);
|
||||||
public readonly ConcurrentQueue<Call> Calls = new();
|
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)
|
public Task SeedAsync(string workingDir, IReadOnlyList<string> skillNames, bool isWorktree, CancellationToken ct)
|
||||||
{
|
{
|
||||||
Interlocked.Increment(ref CallCount);
|
Interlocked.Increment(ref CallCount);
|
||||||
Calls.Enqueue(new Call(workingDir, skillNames, isWorktree));
|
Calls.Enqueue(new Call(workingDir, skillNames, isWorktree));
|
||||||
|
if (ThrowOnSeed is not null) throw ThrowOnSeed;
|
||||||
return Task.CompletedTask;
|
return Task.CompletedTask;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ using ClaudeDo.Data.Repositories;
|
|||||||
using ClaudeDo.Worker.Config;
|
using ClaudeDo.Worker.Config;
|
||||||
using ClaudeDo.Worker.Hub;
|
using ClaudeDo.Worker.Hub;
|
||||||
using ClaudeDo.Worker.Runner;
|
using ClaudeDo.Worker.Runner;
|
||||||
|
using ClaudeDo.Worker.Skills;
|
||||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||||
using Microsoft.Extensions.Logging.Abstractions;
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
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 { } }
|
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 dbFactory = _db.CreateFactory();
|
||||||
var broadcaster = new HubBroadcaster(new CapturingHubContext());
|
var broadcaster = new HubBroadcaster(new CapturingHubContext());
|
||||||
var state = TaskStateServiceBuilder.Build(dbFactory).State;
|
var state = TaskStateServiceBuilder.Build(dbFactory).State;
|
||||||
var wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, _cfg, NullLogger<WorktreeManager>.Instance);
|
var effectiveCfg = cfg ?? _cfg;
|
||||||
return new TaskRunner(claude, dbFactory, broadcaster, wt, new ClaudeArgsBuilder(), _cfg,
|
var wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, effectiveCfg, NullLogger<WorktreeManager>.Instance);
|
||||||
NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
|
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]
|
[Fact]
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ using ClaudeDo.Worker.Runner;
|
|||||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||||
using ClaudeDo.Worker.Usage;
|
using ClaudeDo.Worker.Usage;
|
||||||
using Microsoft.Data.Sqlite;
|
using Microsoft.Data.Sqlite;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
using Microsoft.Extensions.Logging.Abstractions;
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||||
|
|
||||||
@@ -180,7 +181,7 @@ public sealed class QueueServiceSlotFailureTests : IDisposable
|
|||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[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 listId = await SeedListAsync();
|
||||||
var taskId = Guid.NewGuid().ToString();
|
var taskId = Guid.NewGuid().ToString();
|
||||||
@@ -203,21 +204,105 @@ public sealed class QueueServiceSlotFailureTests : IDisposable
|
|||||||
await service.StartAsync(outerCts.Token);
|
await service.StartAsync(outerCts.Token);
|
||||||
waker.Wake();
|
waker.Wake();
|
||||||
|
|
||||||
// Wait for the slot to be claimed and then released again (RunInSlotAsync's
|
var reloaded = await PollUntilLeftQueuedAsync(taskId);
|
||||||
// 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
|
|
||||||
|
|
||||||
TaskEntity? reloaded;
|
// This reproduces a plain Stop (QueueService.CancelTask cancels the slot's CTS
|
||||||
using (var verify = _db.CreateContext())
|
// directly, without going through TaskStateService) landing during the pre-dispatch DB
|
||||||
reloaded = await new TaskRepository(verify).GetByIdAsync(taskId);
|
// 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.
|
||||||
// The picker's atomic claim already flipped it to Running; the cancelled slot run must
|
Assert.Equal(TaskStatus.Cancelled, reloaded!.Status);
|
||||||
// leave it there rather than flipping it to Failed.
|
Assert.Contains(hub.Proxy.Calls,
|
||||||
Assert.Equal(TaskStatus.Running, reloaded!.Status);
|
|
||||||
Assert.DoesNotContain(hub.Proxy.Calls,
|
|
||||||
c => c.Method == "TaskUpdated" && (string)c.Args[0]! == taskId);
|
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