From 3dfae30fff966b1bd23858db58ab17e77e9438d1 Mon Sep 17 00:00:00 2001 From: mika kuns Date: Mon, 24 Aug 2026 09:05:04 +0200 Subject: [PATCH] =?UTF-8?q?fix(worker):=20Usage-Limit-Auto-Continue=20verb?= =?UTF-8?q?rennt=20nicht=20mehr=20alle=20Kandidaten=20au=C3=9Fer=20dem=20e?= =?UTF-8?q?rsten?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OverrideSlotService hält genau einen Slot, und ein Usage-Limit stoppt naturgemäß alle laufenden Tasks gleichzeitig — "mehr Kandidaten als Kapazität" ist also der Normalfall. Der Coordinator stempelte UsageLimitAutoContinuedAt aber VOR dem ContinueTask (der Marker ist die Dedupe-Guard), sodass Kandidat 2..n den Marker bekamen, deren ContinueTask mit "override slot busy" in den catch flog und sie danach dauerhaft aus GetUsageLimitAutoContinueCandidatesAsync ausgeschlossen waren — ein Continue, der nie lief. - Vorab-Check auf CurrentSlot: bei belegtem Slot bricht der Tick ab, statt die restlichen Kandidaten zu verbrennen. - Neuer TaskRepository.ReleaseUsageLimitAutoContinueClaimAsync gibt den Claim zurück, wenn ContinueTask wirft; danach return, der 30s-Backstop holt den Rest im nächsten Tick. - Regressionstest mit zwei Kandidaten, deterministisch über eine TaskCompletionSource im FakeClaudeProcess (StartInSlot setzt _slot synchron unter dem Lock, bevor die Arbeit startet). --- .../Repositories/TaskRepository.cs | 10 +++ .../UsageLimitAutoContinueCoordinator.cs | 32 +++++++- .../UsageLimitAutoContinueCoordinatorTests.cs | 78 ++++++++++++++++--- 3 files changed, 106 insertions(+), 14 deletions(-) diff --git a/src/ClaudeDo.Data/Repositories/TaskRepository.cs b/src/ClaudeDo.Data/Repositories/TaskRepository.cs index 6ad5d575..e70b3b83 100644 --- a/src/ClaudeDo.Data/Repositories/TaskRepository.cs +++ b/src/ClaudeDo.Data/Repositories/TaskRepository.cs @@ -235,6 +235,16 @@ public sealed class TaskRepository return affected > 0; } + /// Undoes a claim whose ContinueTask never actually started (the single override slot was + /// busy, the run was rejected). Without this the task keeps a marker for a run that never + /// happened and the candidate query excludes it forever. + public async Task ReleaseUsageLimitAutoContinueClaimAsync(string taskId, CancellationToken ct = default) + { + await _context.Tasks + .Where(t => t.Id == taskId) + .ExecuteUpdateAsync(s => s.SetProperty(t => t.UsageLimitAutoContinuedAt, (DateTime?)null), ct); + } + internal async Task FlipAllRunningToFailedAsync(string reason, CancellationToken ct = default) { var resultText = "[stale] " + reason; diff --git a/src/ClaudeDo.Worker/Queue/UsageLimitAutoContinueCoordinator.cs b/src/ClaudeDo.Worker/Queue/UsageLimitAutoContinueCoordinator.cs index 041d19a0..b11d5ee6 100644 --- a/src/ClaudeDo.Worker/Queue/UsageLimitAutoContinueCoordinator.cs +++ b/src/ClaudeDo.Worker/Queue/UsageLimitAutoContinueCoordinator.cs @@ -13,7 +13,7 @@ namespace ClaudeDo.Worker.Queue; /// timer, never a crash, never a change from today's behaviour. Called once per /// tick (both from its wake and its 30s backstop): /// tells the caller when to schedule an exact wake so a -/// gate-blocked queue doesn't wait out the backstop, and fires exactly one +/// gate-blocked queue doesn't wait out the backstop, and fires at most one /// ContinueTask per eligible task once the window has actually reset. /// public sealed class UsageLimitAutoContinueCoordinator @@ -47,9 +47,10 @@ public sealed class UsageLimitAutoContinueCoordinator return _usageState.Snapshot?.FiveHour?.ResetsAt; } - /// Fires exactly one ContinueTask per task failed on the usage limit, once the - /// 5h window has actually reset. No-ops (never throws) when the toggle is off, no reset time - /// is known, or the window hasn't reset yet. + /// Fires at most one ContinueTask per task failed on the usage limit, once the + /// 5h window has actually reset — and at most one per tick, because + /// has a single slot. Remaining candidates are picked up by later ticks. No-ops (never throws) + /// when the toggle is off, no reset time is known, or the window hasn't reset yet. public async Task RunAsync(CancellationToken ct) { var settings = await TryReadSettingsAsync(ct); @@ -64,6 +65,11 @@ public sealed class UsageLimitAutoContinueCoordinator foreach (var task in candidates) { + // OverrideSlotService holds exactly ONE slot, and a usage limit typically stops every + // running task at once — so most ticks have more candidates than capacity. Stop at the + // first occupied slot instead of burning the rest; the queue's backstop retries. + if (_override.CurrentSlot is not null) return; + bool claimed; using (var context = _dbFactory.CreateDbContext()) claimed = await new TaskRepository(context).TryClaimUsageLimitAutoContinueAsync(task.Id, DateTime.UtcNow, ct); @@ -78,11 +84,29 @@ public sealed class UsageLimitAutoContinueCoordinator } catch (Exception ex) { + // The claim is the dedupe guard, so it has to be taken *before* the run starts — + // which means a run that never started has to give it back, or the task is + // permanently excluded from the candidate query for a continue that never ran. + await ReleaseClaimAsync(task.Id, ct); _logger.LogError(ex, "Failed to auto-continue task {TaskId} after usage-limit reset", task.Id); + return; // slot busy or the run was rejected — leave the rest for the next tick. } } } + private async Task ReleaseClaimAsync(string taskId, CancellationToken ct) + { + try + { + using var context = _dbFactory.CreateDbContext(); + await new TaskRepository(context).ReleaseUsageLimitAutoContinueClaimAsync(taskId, ct); + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to release the usage-limit auto-continue claim for task {TaskId}", taskId); + } + } + private async Task TryReadSettingsAsync(CancellationToken ct) { try diff --git a/tests/ClaudeDo.Worker.Tests/Queue/UsageLimitAutoContinueCoordinatorTests.cs b/tests/ClaudeDo.Worker.Tests/Queue/UsageLimitAutoContinueCoordinatorTests.cs index cb596851..679b07fa 100644 --- a/tests/ClaudeDo.Worker.Tests/Queue/UsageLimitAutoContinueCoordinatorTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Queue/UsageLimitAutoContinueCoordinatorTests.cs @@ -7,6 +7,7 @@ using ClaudeDo.Worker.Queue; using ClaudeDo.Worker.Runner; using ClaudeDo.Worker.Tests.Infrastructure; using ClaudeDo.Worker.Usage; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging.Abstractions; using TaskStatus = ClaudeDo.Data.Models.TaskStatus; @@ -29,12 +30,13 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable public void Dispose() { _db.Dispose(); try { Directory.Delete(_tempDir, true); } catch { } } - private (UsageLimitAutoContinueCoordinator Coordinator, UsageState UsageState, FakeClaudeProcess Process) BuildCoordinator() + private (UsageLimitAutoContinueCoordinator Coordinator, UsageState UsageState, FakeClaudeProcess Process, OverrideSlotService Slot) BuildCoordinator( + Func, Func, CancellationToken, Task>? handler = null) { var dbFactory = _db.CreateFactory(); var state = TaskStateServiceBuilder.Build(dbFactory).State; var wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, _cfg, NullLogger.Instance); - var fake = new FakeClaudeProcess(); + var fake = new FakeClaudeProcess(handler); var runner = new TaskRunner(fake, dbFactory, new HubBroadcaster(new CapturingHubContext()), wt, new ClaudeArgsBuilder(), _cfg, NullLogger.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); @@ -44,9 +46,12 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable var coordinator = new UsageLimitAutoContinueCoordinator( dbFactory, usageState, overrideSlot, new HubBroadcaster(new CapturingHubContext()), NullLogger.Instance); - return (coordinator, usageState, fake); + return (coordinator, usageState, fake, overrideSlot); } + // Raw inserts bypass TaskNumberAllocator, so hand out distinct Numbers — tasks.number is unique. + private int _nextNumber = 1; + private async Task SeedUsageLimitTaskAsync(bool withSessionId = true) { var listId = Guid.NewGuid().ToString(); @@ -56,7 +61,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", CreatedAt = DateTime.UtcNow }); ctx.Tasks.Add(new TaskEntity { - Id = taskId, ListId = listId, Title = "T", Status = TaskStatus.Failed, + Id = taskId, ListId = listId, Title = "T", Status = TaskStatus.Failed, Number = _nextNumber++, FailureReason = "usage_limit", CreatedAt = DateTime.UtcNow, FinishedAt = DateTime.UtcNow, }); if (withSessionId) @@ -95,7 +100,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable [Fact] public async Task Toggle_Off_Never_Schedules_Or_Continues() { - var (coordinator, usageState, fake) = BuildCoordinator(); + var (coordinator, usageState, fake, _) = BuildCoordinator(); var taskId = await SeedUsageLimitTaskAsync(); usageState.ReportSuccess(new UsageSnapshot( new UsageBucket(85, DateTimeOffset.UtcNow.AddSeconds(-1)), null, Array.Empty(), DateTime.UtcNow)); @@ -114,7 +119,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable [Fact] public async Task Toggle_On_No_Snapshot_Yet_Schedules_Nothing_And_Does_Not_Crash() { - var (coordinator, _, fake) = BuildCoordinator(); + var (coordinator, _, fake, _) = BuildCoordinator(); await SeedUsageLimitTaskAsync(); await EnableToggleAsync(); @@ -128,7 +133,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable [Fact] public async Task Toggle_On_Reset_In_Future_Schedules_Wake_But_Does_Not_Continue_Yet() { - var (coordinator, usageState, fake) = BuildCoordinator(); + var (coordinator, usageState, fake, _) = BuildCoordinator(); await SeedUsageLimitTaskAsync(); await EnableToggleAsync(); var resetsAt = DateTimeOffset.UtcNow.AddMinutes(5); @@ -146,7 +151,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable [Fact] public async Task Toggle_On_Reset_Already_Passed_Continues_Exactly_Once() { - var (coordinator, usageState, fake) = BuildCoordinator(); + var (coordinator, usageState, fake, _) = BuildCoordinator(); var taskId = await SeedUsageLimitTaskAsync(); await EnableToggleAsync(); usageState.ReportSuccess(new UsageSnapshot( @@ -164,10 +169,63 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable Assert.Equal(1, fake.CallCount); } + /// A usage limit stops every running task at once, so "more candidates than override slots" + /// is the normal case, not an edge case. The claim has to be taken before the run starts + /// (it is the dedupe guard), so a candidate that never got a slot must not keep it — the + /// candidate query filters on that marker and would exclude the task forever. + [Fact] + public async Task Busy_Override_Slot_Leaves_The_Other_Candidate_For_A_Later_Tick() + { + var gate = new TaskCompletionSource(); + var (coordinator, usageState, fake, slot) = BuildCoordinator(async (_, _, _, _, _) => + { + await gate.Task; + return new RunResult { ExitCode = 0, ResultMarkdown = "ok" }; + }); + await SeedUsageLimitTaskAsync(); + await SeedUsageLimitTaskAsync(); + await EnableToggleAsync(); + usageState.ReportSuccess(new UsageSnapshot( + new UsageBucket(85, DateTimeOffset.UtcNow.AddSeconds(-1)), null, Array.Empty(), DateTime.UtcNow)); + + // Tick 1 — the first candidate takes the single slot and holds it on the gate. + await coordinator.RunAsync(default); + Assert.Equal(1, await CountClaimedAsync()); + + // Tick 2 while that slot is still busy — the second candidate must stay unclaimed. + await coordinator.RunAsync(default); + Assert.Equal(1, await CountClaimedAsync()); + + // Let the first run finish, freeing the slot. + await PollAsync(() => fake.CallCount >= 1); + gate.SetResult(); + await PollAsync(() => slot.CurrentSlot is null); + + // Tick 3 — the leftover candidate finally gets its turn. + await coordinator.RunAsync(default); + await PollAsync(() => fake.CallCount >= 2); + Assert.Equal(2, fake.CallCount); + Assert.Equal(2, await CountClaimedAsync()); + } + + private async Task CountClaimedAsync() + { + using var ctx = _db.CreateContext(); + return await ctx.Tasks.CountAsync(t => t.UsageLimitAutoContinuedAt != null); + } + + private static async Task PollAsync(Func condition) + { + var deadline = DateTime.UtcNow.AddSeconds(10); + while (DateTime.UtcNow < deadline && !condition()) + await Task.Delay(25); + Assert.True(condition(), "condition did not become true within 10s"); + } + [Fact] public async Task Missing_Reset_Time_Never_Continues_And_Never_Throws() { - var (coordinator, usageState, fake) = BuildCoordinator(); + var (coordinator, usageState, fake, _) = BuildCoordinator(); await SeedUsageLimitTaskAsync(); await EnableToggleAsync(); // A snapshot exists, but this bucket has no reset time — fail open. @@ -185,7 +243,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable [Fact] public async Task Non_UsageLimit_Failure_Is_Never_Touched() { - var (coordinator, usageState, fake) = BuildCoordinator(); + var (coordinator, usageState, fake, _) = BuildCoordinator(); var listId = Guid.NewGuid().ToString(); var taskId = Guid.NewGuid().ToString(); using (var ctx = _db.CreateContext())