fix(worker): Usage-Limit-Auto-Continue verbrennt nicht mehr alle Kandidaten außer dem ersten
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).
This commit is contained in:
@@ -235,6 +235,16 @@ public sealed class TaskRepository
|
|||||||
return affected > 0;
|
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<int> FlipAllRunningToFailedAsync(string reason, CancellationToken ct = default)
|
internal async Task<int> FlipAllRunningToFailedAsync(string reason, CancellationToken ct = default)
|
||||||
{
|
{
|
||||||
var resultText = "[stale] " + reason;
|
var resultText = "[stale] " + reason;
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ namespace ClaudeDo.Worker.Queue;
|
|||||||
/// timer, never a crash, never a change from today's behaviour. Called once per
|
/// timer, never a crash, never a change from today's behaviour. Called once per
|
||||||
/// <see cref="QueueService"/> tick (both from its wake and its 30s backstop):
|
/// <see cref="QueueService"/> tick (both from its wake and its 30s backstop):
|
||||||
/// <see cref="GetScheduledWakeAtAsync"/> tells the caller when to schedule an exact wake so a
|
/// <see cref="GetScheduledWakeAtAsync"/> tells the caller when to schedule an exact wake so a
|
||||||
/// gate-blocked queue doesn't wait out the backstop, and <see cref="RunAsync"/> fires exactly one
|
/// gate-blocked queue doesn't wait out the backstop, and <see cref="RunAsync"/> fires at most one
|
||||||
/// <c>ContinueTask</c> per eligible task once the window has actually reset.
|
/// <c>ContinueTask</c> per eligible task once the window has actually reset.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public sealed class UsageLimitAutoContinueCoordinator
|
public sealed class UsageLimitAutoContinueCoordinator
|
||||||
@@ -47,9 +47,10 @@ public sealed class UsageLimitAutoContinueCoordinator
|
|||||||
return _usageState.Snapshot?.FiveHour?.ResetsAt;
|
return _usageState.Snapshot?.FiveHour?.ResetsAt;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>Fires exactly one <c>ContinueTask</c> per task failed on the usage limit, once the
|
/// <summary>Fires at most one <c>ContinueTask</c> 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
|
/// 5h window has actually reset — and at most one per tick, because <see cref="OverrideSlotService"/>
|
||||||
/// is known, or the window hasn't reset yet.</summary>
|
/// 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.</summary>
|
||||||
public async Task RunAsync(CancellationToken ct)
|
public async Task RunAsync(CancellationToken ct)
|
||||||
{
|
{
|
||||||
var settings = await TryReadSettingsAsync(ct);
|
var settings = await TryReadSettingsAsync(ct);
|
||||||
@@ -64,6 +65,11 @@ public sealed class UsageLimitAutoContinueCoordinator
|
|||||||
|
|
||||||
foreach (var task in candidates)
|
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;
|
bool claimed;
|
||||||
using (var context = _dbFactory.CreateDbContext())
|
using (var context = _dbFactory.CreateDbContext())
|
||||||
claimed = await new TaskRepository(context).TryClaimUsageLimitAutoContinueAsync(task.Id, DateTime.UtcNow, ct);
|
claimed = await new TaskRepository(context).TryClaimUsageLimitAutoContinueAsync(task.Id, DateTime.UtcNow, ct);
|
||||||
@@ -78,11 +84,29 @@ public sealed class UsageLimitAutoContinueCoordinator
|
|||||||
}
|
}
|
||||||
catch (Exception ex)
|
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);
|
_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<AppSettingsEntity?> TryReadSettingsAsync(CancellationToken ct)
|
private async Task<AppSettingsEntity?> TryReadSettingsAsync(CancellationToken ct)
|
||||||
{
|
{
|
||||||
try
|
try
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ using ClaudeDo.Worker.Queue;
|
|||||||
using ClaudeDo.Worker.Runner;
|
using ClaudeDo.Worker.Runner;
|
||||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||||
using ClaudeDo.Worker.Usage;
|
using ClaudeDo.Worker.Usage;
|
||||||
|
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;
|
||||||
|
|
||||||
@@ -29,12 +30,13 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable
|
|||||||
|
|
||||||
public void Dispose() { _db.Dispose(); try { Directory.Delete(_tempDir, true); } catch { } }
|
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<string, string, IReadOnlyList<string>, Func<string, Task>, CancellationToken, Task<RunResult>>? handler = null)
|
||||||
{
|
{
|
||||||
var dbFactory = _db.CreateFactory();
|
var dbFactory = _db.CreateFactory();
|
||||||
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 wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, _cfg, NullLogger<WorktreeManager>.Instance);
|
||||||
var fake = new FakeClaudeProcess();
|
var fake = new FakeClaudeProcess(handler);
|
||||||
var runner = new TaskRunner(fake, dbFactory, new HubBroadcaster(new CapturingHubContext()), wt,
|
var runner = new TaskRunner(fake, dbFactory, new HubBroadcaster(new CapturingHubContext()), wt,
|
||||||
new ClaudeArgsBuilder(), _cfg, NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(),
|
new ClaudeArgsBuilder(), _cfg, NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(),
|
||||||
new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
|
new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
|
||||||
@@ -44,9 +46,12 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable
|
|||||||
var coordinator = new UsageLimitAutoContinueCoordinator(
|
var coordinator = new UsageLimitAutoContinueCoordinator(
|
||||||
dbFactory, usageState, overrideSlot,
|
dbFactory, usageState, overrideSlot,
|
||||||
new HubBroadcaster(new CapturingHubContext()), NullLogger<UsageLimitAutoContinueCoordinator>.Instance);
|
new HubBroadcaster(new CapturingHubContext()), NullLogger<UsageLimitAutoContinueCoordinator>.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<string> SeedUsageLimitTaskAsync(bool withSessionId = true)
|
private async Task<string> SeedUsageLimitTaskAsync(bool withSessionId = true)
|
||||||
{
|
{
|
||||||
var listId = Guid.NewGuid().ToString();
|
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.Lists.Add(new ListEntity { Id = listId, Name = "L", CreatedAt = DateTime.UtcNow });
|
||||||
ctx.Tasks.Add(new TaskEntity
|
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,
|
FailureReason = "usage_limit", CreatedAt = DateTime.UtcNow, FinishedAt = DateTime.UtcNow,
|
||||||
});
|
});
|
||||||
if (withSessionId)
|
if (withSessionId)
|
||||||
@@ -95,7 +100,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public async Task Toggle_Off_Never_Schedules_Or_Continues()
|
public async Task Toggle_Off_Never_Schedules_Or_Continues()
|
||||||
{
|
{
|
||||||
var (coordinator, usageState, fake) = BuildCoordinator();
|
var (coordinator, usageState, fake, _) = BuildCoordinator();
|
||||||
var taskId = await SeedUsageLimitTaskAsync();
|
var taskId = await SeedUsageLimitTaskAsync();
|
||||||
usageState.ReportSuccess(new UsageSnapshot(
|
usageState.ReportSuccess(new UsageSnapshot(
|
||||||
new UsageBucket(85, DateTimeOffset.UtcNow.AddSeconds(-1)), null, Array.Empty<UsageLimitRow>(), DateTime.UtcNow));
|
new UsageBucket(85, DateTimeOffset.UtcNow.AddSeconds(-1)), null, Array.Empty<UsageLimitRow>(), DateTime.UtcNow));
|
||||||
@@ -114,7 +119,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public async Task Toggle_On_No_Snapshot_Yet_Schedules_Nothing_And_Does_Not_Crash()
|
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 SeedUsageLimitTaskAsync();
|
||||||
await EnableToggleAsync();
|
await EnableToggleAsync();
|
||||||
|
|
||||||
@@ -128,7 +133,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public async Task Toggle_On_Reset_In_Future_Schedules_Wake_But_Does_Not_Continue_Yet()
|
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 SeedUsageLimitTaskAsync();
|
||||||
await EnableToggleAsync();
|
await EnableToggleAsync();
|
||||||
var resetsAt = DateTimeOffset.UtcNow.AddMinutes(5);
|
var resetsAt = DateTimeOffset.UtcNow.AddMinutes(5);
|
||||||
@@ -146,7 +151,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public async Task Toggle_On_Reset_Already_Passed_Continues_Exactly_Once()
|
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();
|
var taskId = await SeedUsageLimitTaskAsync();
|
||||||
await EnableToggleAsync();
|
await EnableToggleAsync();
|
||||||
usageState.ReportSuccess(new UsageSnapshot(
|
usageState.ReportSuccess(new UsageSnapshot(
|
||||||
@@ -164,10 +169,63 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable
|
|||||||
Assert.Equal(1, fake.CallCount);
|
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<UsageLimitRow>(), 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<int> CountClaimedAsync()
|
||||||
|
{
|
||||||
|
using var ctx = _db.CreateContext();
|
||||||
|
return await ctx.Tasks.CountAsync(t => t.UsageLimitAutoContinuedAt != null);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static async Task PollAsync(Func<bool> 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]
|
[Fact]
|
||||||
public async Task Missing_Reset_Time_Never_Continues_And_Never_Throws()
|
public async Task Missing_Reset_Time_Never_Continues_And_Never_Throws()
|
||||||
{
|
{
|
||||||
var (coordinator, usageState, fake) = BuildCoordinator();
|
var (coordinator, usageState, fake, _) = BuildCoordinator();
|
||||||
await SeedUsageLimitTaskAsync();
|
await SeedUsageLimitTaskAsync();
|
||||||
await EnableToggleAsync();
|
await EnableToggleAsync();
|
||||||
// A snapshot exists, but this bucket has no reset time — fail open.
|
// A snapshot exists, but this bucket has no reset time — fail open.
|
||||||
@@ -185,7 +243,7 @@ public sealed class UsageLimitAutoContinueCoordinatorTests : IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public async Task Non_UsageLimit_Failure_Is_Never_Touched()
|
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 listId = Guid.NewGuid().ToString();
|
||||||
var taskId = Guid.NewGuid().ToString();
|
var taskId = Guid.NewGuid().ToString();
|
||||||
using (var ctx = _db.CreateContext())
|
using (var ctx = _db.CreateContext())
|
||||||
|
|||||||
Reference in New Issue
Block a user