fix(worker): claim Running before creating run resources
TaskRunner.RunAsync created the worktree (PrepareRunDirectoryAsync) before claiming Running via StartRunningAsync. RunNow dispatches by task id with no atomic claim of their own, so a Queued task racing the queue picker's atomic SQL claim could hit WorktreeManager's branch-collision self-heal, which force-removes and recreates the winner's live worktree mid-run. Move the claim before any resource creation and bail out immediately when it's rejected. OverrideSlotService.RunNow also fast-rejects a task already Running in the DB (defense in depth). RunCancellationRegistry now refuses (and logs) a double registration instead of silently overwriting the first CTS, so a losing dispatch's cleanup can no longer unregister the winner's cancellation token.
This commit is contained in:
@@ -0,0 +1,122 @@
|
||||
using ClaudeDo.Data;
|
||||
using ClaudeDo.Data.Git;
|
||||
using ClaudeDo.Data.Models;
|
||||
using ClaudeDo.Data.Repositories;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Hub;
|
||||
using ClaudeDo.Worker.Queue;
|
||||
using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
|
||||
namespace ClaudeDo.Worker.Tests.Runner;
|
||||
|
||||
/// Regression test for the RunNow/queue-picker double-dispatch race: a Queued task
|
||||
/// dispatched both via the override slot (RunNow, unclaimed) and the queue picker (atomic
|
||||
/// SQL Queued->Running claim) must not create its worktree twice. Before the fix,
|
||||
/// TaskRunner.RunAsync created the worktree (PrepareRunDirectoryAsync) BEFORE claiming
|
||||
/// Running, so the losing dispatch reached WorktreeManager's "branch already exists"
|
||||
/// self-heal and force-removed the winner's live worktree out from under its still-running
|
||||
/// process. The fix claims Running first and bails out before touching git when the claim
|
||||
/// is rejected.
|
||||
public sealed class ConcurrentDispatchRaceTests : IDisposable
|
||||
{
|
||||
private readonly DbFixture _db = new();
|
||||
private readonly GitRepoFixture? _repo;
|
||||
private readonly string _tempDir;
|
||||
private readonly WorkerConfig _cfg;
|
||||
|
||||
private static bool GitAvailable => GitRepoFixture.IsGitAvailable();
|
||||
|
||||
public ConcurrentDispatchRaceTests()
|
||||
{
|
||||
_tempDir = Path.Combine(Path.GetTempPath(), $"cd_race_{Guid.NewGuid():N}");
|
||||
Directory.CreateDirectory(_tempDir);
|
||||
_cfg = new WorkerConfig { SandboxRoot = _tempDir, LogRoot = _tempDir };
|
||||
if (GitAvailable) _repo = new GitRepoFixture();
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
_db.Dispose();
|
||||
_repo?.Dispose();
|
||||
try { Directory.Delete(_tempDir, true); } catch { /* best effort */ }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RunNow_RacingQueuePicker_LoserBailsBeforeTouchingWorktree()
|
||||
{
|
||||
if (!GitAvailable) { Assert.True(true, "git not available -- skipping"); return; }
|
||||
|
||||
var dbFactory = _db.CreateFactory();
|
||||
string listId = Guid.NewGuid().ToString(), taskId = Guid.NewGuid().ToString();
|
||||
using (var ctx = _db.CreateContext())
|
||||
{
|
||||
ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", WorkingDir = _repo!.RepoDir, CreatedAt = DateTime.UtcNow });
|
||||
ctx.Tasks.Add(new TaskEntity
|
||||
{
|
||||
Id = taskId, ListId = listId, Title = "Racing task", Status = TaskStatus.Queued,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
});
|
||||
await ctx.SaveChangesAsync();
|
||||
}
|
||||
|
||||
var processStarted = new TaskCompletionSource();
|
||||
var releaseProcess = new TaskCompletionSource();
|
||||
var fake = new FakeClaudeProcess(async (_, _, _, _, _) =>
|
||||
{
|
||||
processStarted.TrySetResult();
|
||||
await releaseProcess.Task;
|
||||
return new RunResult { ExitCode = 0, ResultMarkdown = "ok" };
|
||||
});
|
||||
|
||||
var state = TaskStateServiceBuilder.Build(dbFactory).State;
|
||||
var wt = new WorktreeManager(new GitService(), dbFactory, _cfg, NullLogger<WorktreeManager>.Instance);
|
||||
var runner = new TaskRunner(fake, dbFactory, new HubBroadcaster(new CapturingHubContext()), wt,
|
||||
new ClaudeArgsBuilder(), _cfg, NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(),
|
||||
new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
|
||||
|
||||
// Winner: mirrors QueuePicker's atomic Queued->Running claim, then a queue-slot
|
||||
// dispatch (alreadyClaimed: true, matching QueueService.RunInSlotAsync).
|
||||
var picker = new QueuePicker(dbFactory);
|
||||
var claimed = await picker.ClaimNextAsync(DateTime.UtcNow, CancellationToken.None);
|
||||
Assert.NotNull(claimed);
|
||||
|
||||
var winnerTask = runner.RunAsync(claimed!, "queue", CancellationToken.None, alreadyClaimed: true);
|
||||
await processStarted.Task; // worktree exists, "Claude" is now running in it
|
||||
|
||||
WorktreeEntity? wtRowBefore;
|
||||
using (var ctx = _db.CreateContext())
|
||||
wtRowBefore = await new WorktreeRepository(ctx).GetByTaskIdAsync(taskId);
|
||||
Assert.NotNull(wtRowBefore);
|
||||
|
||||
var markerPath = Path.Combine(wtRowBefore!.Path, "in-progress.marker");
|
||||
File.WriteAllText(markerPath, "winner's work");
|
||||
|
||||
// Loser: a RunNow-style dispatch (unclaimed) of the SAME task arriving after the
|
||||
// picker already claimed it. Before the fix this would recreate the worktree via
|
||||
// WorktreeManager's self-heal, wiping the winner's live directory (and this marker
|
||||
// file) out from under the still-running process.
|
||||
TaskEntity taskForOverride;
|
||||
using (var ctx = _db.CreateContext())
|
||||
taskForOverride = (await new TaskRepository(ctx).GetByIdAsync(taskId))!;
|
||||
|
||||
await runner.RunAsync(taskForOverride, "override", CancellationToken.None);
|
||||
|
||||
Assert.True(File.Exists(markerPath), "loser dispatch touched the winner's live worktree");
|
||||
Assert.Equal(1, fake.CallCount);
|
||||
|
||||
releaseProcess.TrySetResult();
|
||||
await winnerTask;
|
||||
|
||||
using var verify = _db.CreateContext();
|
||||
Assert.Single(await new TaskRunRepository(verify).GetByTaskIdAsync(taskId));
|
||||
|
||||
var wtRowAfter = await new WorktreeRepository(verify).GetByTaskIdAsync(taskId);
|
||||
Assert.NotNull(wtRowAfter);
|
||||
Assert.Equal(WorktreeState.Active, wtRowAfter!.State);
|
||||
Assert.Equal(wtRowBefore.Path, wtRowAfter.Path);
|
||||
Assert.True(Directory.Exists(wtRowAfter.Path));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user