feat(claude-do): feat(queue): Option pro Liste — Tasks mit überlappendem Date
## Befund (Batch-Lauf 2026-08-06, Liste "Bandel.Hub") maxParallelExecutions = 5, 23 Geschwister-Tasks auf demselben Plugin. Die Parallelitaet hat die Ausfuehrung verkuerzt, aber die Konfliktlast erhoeht: eine CSS-Datei wurde von 10 Tasks angefasst, drei Razor-Dateien von je 6. Der Engpass des Gesamtdurchlaufs war nicht die Ausfuehrung, sondern Review und Merge — die sind zwingend seriell. Netto wa ClaudeDo-Task: 89f989e0-4fcf-43f5-a802-11d63df7a6d1
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
using ClaudeDo.Data;
|
||||
using ClaudeDo.Data.Models;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
|
||||
namespace ClaudeDo.Worker.Queue;
|
||||
|
||||
@@ -13,14 +14,30 @@ public sealed class QueuePicker : IQueuePicker
|
||||
|
||||
public async Task<TaskEntity?> ClaimNextAsync(DateTime now, CancellationToken ct)
|
||||
{
|
||||
// Atomic queue claim: UPDATE + RETURNING in a single statement prevents TOCTOU races.
|
||||
// Raw SQL because EF cannot express UPDATE...RETURNING.
|
||||
// Eligible task must be Queued, unblocked, not manual, and due (or unscheduled).
|
||||
// EF SQLite stores DateTime as "yyyy-MM-dd HH:mm:ss.fffffff" — same format used here for comparison.
|
||||
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
||||
var nowStr = now.ToUniversalTime().ToString("yyyy-MM-dd HH:mm:ss.fffffff");
|
||||
var startedAtStr = DateTime.UtcNow.ToString("yyyy-MM-dd HH:mm:ss.fffffff");
|
||||
|
||||
var serializingListIds = await ctx.ListConfigs
|
||||
.Where(c => c.SerializeOnFileOverlap)
|
||||
.Select(c => c.ListId)
|
||||
.ToListAsync(ct);
|
||||
|
||||
// Fast path: no list has opted into file-scope serialization, so behavior and cost stay
|
||||
// identical to before that option existed -- single atomic UPDATE...RETURNING.
|
||||
if (serializingListIds.Count == 0)
|
||||
return await ClaimTopEligibleAsync(ctx, nowStr, startedAtStr, ct);
|
||||
|
||||
return await ClaimRespectingScopeAsync(ctx, now, startedAtStr, serializingListIds.ToHashSet(StringComparer.Ordinal), ct);
|
||||
}
|
||||
|
||||
private static async Task<TaskEntity?> ClaimTopEligibleAsync(
|
||||
ClaudeDoDbContext ctx, string nowStr, string startedAtStr, CancellationToken ct)
|
||||
{
|
||||
// Atomic queue claim: UPDATE + RETURNING in a single statement prevents TOCTOU races.
|
||||
// Raw SQL because EF cannot express UPDATE...RETURNING.
|
||||
// Eligible task must be Queued, unblocked, not manual, and due (or unscheduled).
|
||||
// EF SQLite stores DateTime as "yyyy-MM-dd HH:mm:ss.fffffff" — same format used here for comparison.
|
||||
var rows = await ctx.Tasks.FromSqlRaw("""
|
||||
UPDATE tasks SET status = 'running', started_at = {1}
|
||||
WHERE id = (
|
||||
@@ -37,4 +54,41 @@ public sealed class QueuePicker : IQueuePicker
|
||||
|
||||
return rows.FirstOrDefault();
|
||||
}
|
||||
|
||||
// At least one list wants overlapping-scope tasks serialized: walk eligible candidates in the
|
||||
// usual order and skip any whose declared scope overlaps a running or awaiting-merge sibling
|
||||
// in the same list. A candidate with no declared scope, or belonging to a list not in
|
||||
// serializingListIds, is claimed exactly as before -- there is no attempt to predict a scope
|
||||
// that was never declared and no finished sibling to infer it from.
|
||||
private async Task<TaskEntity?> ClaimRespectingScopeAsync(
|
||||
ClaudeDoDbContext ctx, DateTime now, string startedAtStr, HashSet<string> serializingListIds, CancellationToken ct)
|
||||
{
|
||||
var candidates = await ctx.Tasks
|
||||
.AsNoTracking()
|
||||
.Where(t => t.Status == TaskStatus.Queued
|
||||
&& t.BlockedByTaskId == null
|
||||
&& !t.IsManual
|
||||
&& (t.ScheduledFor == null || t.ScheduledFor <= now))
|
||||
.OrderBy(t => t.SortOrder).ThenBy(t => t.CreatedAt)
|
||||
.ToListAsync(ct);
|
||||
|
||||
foreach (var candidate in candidates)
|
||||
{
|
||||
if (serializingListIds.Contains(candidate.ListId)
|
||||
&& await ScopeOverlap.FindBlockingSiblingAsync(ctx, candidate, ct) is not null)
|
||||
continue;
|
||||
|
||||
var rows = await ctx.Tasks.FromSqlRaw("""
|
||||
UPDATE tasks SET status = 'running', started_at = {1}
|
||||
WHERE id = {0} AND status = 'queued'
|
||||
RETURNING *
|
||||
""", candidate.Id, startedAtStr).ToListAsync(ct);
|
||||
|
||||
var claimed = rows.FirstOrDefault();
|
||||
if (claimed is not null) return claimed;
|
||||
// Lost the race for this row to a concurrent picker -- try the next candidate.
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user