Scanner: Laeufe deckeln, stille Kappung bei 500 beseitigen (#682)
Drei Fehler im Taktgeber des Taskboards, zwei gemeldet, einer beim Bau gefunden: 1. Unbegrenzte Parallelitaet. Jede faellige Aufgabe startete sofort. Bei Cron-Terminen auf runden Zeiten hiess das: alle Agenten gleichzeitig, und alle Laeufe passierten die Budgetpruefung gemeinsam, weil sie je Lauf vor dem Start laeuft. Neu: maxParallelTaskRuns je Instanz (Vorgabe 4). Der Rest wartet, sortiert nach Prioritaet und dann nach Faelligkeit. 2. Stille Kappung. Die Kandidaten kamen ueber ListAsync mit Limit 500; darueber fielen Termine lautlos aus dem Takt. Neu: ListRunnableAsync liefert alle Aufgaben auf todo, ungekappt. Muessen Aufgaben warten, steht das im Log, allerdings nur, wenn sich die Zahl aendert. 3. Der Takt wartete auf jeden Lauf, den er startete. Ein zehnminuetiger Lauf hielt damit jeden anderen Termin zehn Minuten an, und mit einem Deckel haetten die freien Plaetze leer gestanden. Neu: Die Schleife startet und wartet nicht; DisposeAsync wartet stattdessen auf die laufenden Laeufe. Beansprucht wird nur, was sofort starten kann. Alles andere bleibt auf todo, sonst liefe beim Warten der Lease ab und die Aufgabe wuerde ein zweites Mal beansprucht. Ein Cron-Termin geht dadurch nicht verloren, weil DueOccurrence bis zum letzten Lauf zuruecksucht. Ein von Hand gestarteter Lauf wartet auf keinen Platz, belegt aber einen. Bleibt offen: Bis zu maxParallelTaskRuns Laeufe koennen die Budgetpruefung weiterhin gemeinsam passieren. Die Luecke ist damit begrenzt, aber nicht geschlossen; dafuer braeuchte es eine Reservierung vor dem Lauf. Tests: ScannerCapacityTests (9 neu). 1.229 gruen. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Richard
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
3ba9ebc565
commit
8df5d004e4
@@ -0,0 +1,288 @@
|
||||
using ClawdDotNet.Core.Storage;
|
||||
using ClawdDotNet.Core.Tasks;
|
||||
using ClawdDotNet.Core.Tests.Infrastructure;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Shouldly;
|
||||
|
||||
namespace ClawdDotNet.Core.Tests.Tasks;
|
||||
|
||||
/// <summary>
|
||||
/// #682: The scanner caps parallel runs and no longer drops candidates beyond 500.
|
||||
///
|
||||
/// Before, every due task started at once — fifty agents on a round cron time meant fifty
|
||||
/// simultaneous model calls, all passing the instance budget check together. And the
|
||||
/// candidate list was clamped to 500 without a word, so schedules beyond that silently
|
||||
/// fell out of the tick.
|
||||
/// </summary>
|
||||
public sealed class ScannerCapacityTests : IDisposable
|
||||
{
|
||||
private readonly string _directory;
|
||||
private readonly SqliteStorage _storage;
|
||||
private readonly SqliteTaskRepository _repo;
|
||||
private readonly FakeTimeProvider _clock = new(new DateTimeOffset(2026, 7, 31, 12, 0, 0, TimeSpan.Zero));
|
||||
|
||||
public ScannerCapacityTests()
|
||||
{
|
||||
_directory = Path.Combine(Path.GetTempPath(), "clawd-tests", Guid.NewGuid().ToString("N"));
|
||||
_storage = new SqliteStorage(Path.Combine(_directory, "state.db"));
|
||||
_repo = new SqliteTaskRepository(_storage);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
|
||||
try { Directory.Delete(_directory, recursive: true); }
|
||||
catch { /* cleanup is secondary */ }
|
||||
}
|
||||
|
||||
private TaskScanner Scanner(ITaskDispatcher dispatcher, int maxParallelRuns)
|
||||
=> new(_repo, dispatcher, NullLoggerFactory.Instance, clock: _clock, maxParallelRuns: maxParallelRuns);
|
||||
|
||||
private Task<TaskItem> SeedDue(string id, int priority = 3, TaskWhen? when = null)
|
||||
=> _repo.UpsertAsync(new TaskItem
|
||||
{
|
||||
Id = id,
|
||||
Title = id,
|
||||
Assignee = "@new",
|
||||
Status = TaskItemStatus.Todo,
|
||||
Priority = priority,
|
||||
When = when
|
||||
}, default);
|
||||
|
||||
private static async Task WaitUntil(Func<bool> condition)
|
||||
{
|
||||
var deadline = DateTime.UtcNow.AddSeconds(10);
|
||||
while (!condition())
|
||||
{
|
||||
if (DateTime.UtcNow > deadline)
|
||||
throw new TimeoutException("condition not reached within 10 s");
|
||||
await Task.Delay(10);
|
||||
}
|
||||
}
|
||||
|
||||
// ═══════════════════ the cap ═══════════════════
|
||||
|
||||
[Fact]
|
||||
public async Task No_more_runs_than_the_cap_are_ever_in_flight()
|
||||
{
|
||||
for (var i = 0; i < 7; i++)
|
||||
await SeedDue($"t-{i}");
|
||||
|
||||
var dispatcher = new TrackingDispatcher(holdFor: TimeSpan.FromMilliseconds(40));
|
||||
var scanner = Scanner(dispatcher, maxParallelRuns: 2);
|
||||
|
||||
// Several ticks until everything has run — each tick fills only the free slots.
|
||||
for (var tick = 0; tick < 20 && dispatcher.Started.Count < 7; tick++)
|
||||
await scanner.ScanOnceAsync(default);
|
||||
|
||||
dispatcher.Started.Count.ShouldBe(7, "every due task runs eventually");
|
||||
dispatcher.MaxConcurrent.ShouldBeLessThanOrEqualTo(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Tasks_beyond_the_cap_stay_todo_and_are_not_claimed()
|
||||
{
|
||||
await SeedDue("t-a", priority: 5);
|
||||
await SeedDue("t-b");
|
||||
await SeedDue("t-c");
|
||||
|
||||
var dispatcher = new TrackingDispatcher(gated: true);
|
||||
var scanner = Scanner(dispatcher, maxParallelRuns: 1);
|
||||
|
||||
var scan = scanner.ScanOnceAsync(default);
|
||||
await WaitUntil(() => dispatcher.Started.Count == 1);
|
||||
await Task.Delay(100); // give a (wrong) second claim the chance to happen
|
||||
|
||||
dispatcher.Started.ShouldBe(["t-a"]);
|
||||
(await _repo.GetAsync("t-b", default))!.Status.ShouldBe(TaskItemStatus.Todo,
|
||||
"a waiting task must not be claimed — its lease would run out while it waits");
|
||||
(await _repo.GetAsync("t-c", default))!.Status.ShouldBe(TaskItemStatus.Todo);
|
||||
|
||||
dispatcher.ReleaseAll();
|
||||
(await scan).ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_task_that_had_to_wait_keeps_its_occurrence()
|
||||
{
|
||||
var at = new TaskWhen { Kind = TaskWhenKind.At, Value = "2026-07-31T11:00:00Z", TimeZone = "UTC" };
|
||||
await SeedDue("t-first", priority: 5);
|
||||
await SeedDue("t-later", when: at);
|
||||
|
||||
var dispatcher = new TrackingDispatcher(gated: true);
|
||||
var scanner = Scanner(dispatcher, maxParallelRuns: 1);
|
||||
|
||||
var first = scanner.ScanOnceAsync(default);
|
||||
await WaitUntil(() => dispatcher.Started.Count == 1);
|
||||
dispatcher.ReleaseAll();
|
||||
await first;
|
||||
|
||||
// Ten minutes later, the one-time occurrence at 11:00 is still due.
|
||||
_clock.Advance(TimeSpan.FromMinutes(10));
|
||||
var second = scanner.ScanOnceAsync(default);
|
||||
await WaitUntil(() => dispatcher.Started.Count == 2);
|
||||
dispatcher.ReleaseAll();
|
||||
await second;
|
||||
|
||||
dispatcher.Started.ShouldBe(["t-first", "t-later"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_run_started_by_hand_occupies_a_slot()
|
||||
{
|
||||
await SeedDue("t-manual");
|
||||
await SeedDue("t-scheduled");
|
||||
|
||||
var dispatcher = new TrackingDispatcher(gated: true);
|
||||
var scanner = Scanner(dispatcher, maxParallelRuns: 1);
|
||||
|
||||
var manual = scanner.RunTaskNowAsync("t-manual", default);
|
||||
await WaitUntil(() => dispatcher.Started.Count == 1);
|
||||
|
||||
(await scanner.ScanOnceAsync(default)).ShouldBe(0, "the only slot is taken by the manual run");
|
||||
|
||||
dispatcher.ReleaseAll();
|
||||
(await manual).ShouldBeTrue();
|
||||
|
||||
// The slot is free again: the waiting task starts on the next tick.
|
||||
var next = scanner.ScanOnceAsync(default);
|
||||
await WaitUntil(() => dispatcher.Started.Count == 2);
|
||||
dispatcher.ReleaseAll();
|
||||
(await next).ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_value_below_one_still_lets_work_through()
|
||||
{
|
||||
await SeedDue("t-1");
|
||||
var scanner = Scanner(new TrackingDispatcher(), maxParallelRuns: 0);
|
||||
|
||||
scanner.MaxParallelRuns.ShouldBe(1);
|
||||
(await scanner.ScanOnceAsync(default)).ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_long_run_does_not_hold_back_the_next_tick()
|
||||
{
|
||||
// Real clock and a short tick: this is about the loop, not a single scan.
|
||||
await SeedDue("t-slow");
|
||||
var dispatcher = new TrackingDispatcher(gated: true);
|
||||
await using var scanner = new TaskScanner(_repo, dispatcher, NullLoggerFactory.Instance,
|
||||
tick: TimeSpan.FromMilliseconds(50), maxParallelRuns: 2);
|
||||
|
||||
scanner.Start();
|
||||
await WaitUntil(() => dispatcher.Started.Count == 1);
|
||||
|
||||
// While t-slow still runs, new work arrives. Before #682 the loop awaited every
|
||||
// run it started, so t-new would have waited for t-slow to finish.
|
||||
await SeedDue("t-new");
|
||||
await WaitUntil(() => dispatcher.Started.Count == 2);
|
||||
|
||||
dispatcher.Started.ShouldBe(["t-slow", "t-new"]);
|
||||
scanner.OpenCount.ShouldBe(2);
|
||||
|
||||
scanner.StopAccepting();
|
||||
dispatcher.ReleaseAll();
|
||||
await WaitUntil(() => scanner.OpenCount == 0);
|
||||
}
|
||||
|
||||
// ═══════════════════ order ═══════════════════
|
||||
|
||||
[Fact]
|
||||
public async Task Higher_priority_goes_first()
|
||||
{
|
||||
await SeedDue("t-low", priority: 1);
|
||||
await SeedDue("t-high", priority: 5);
|
||||
await SeedDue("t-mid", priority: 3);
|
||||
|
||||
var dispatcher = new TrackingDispatcher();
|
||||
await Scanner(dispatcher, maxParallelRuns: 1).ScanOnceAsync(default);
|
||||
|
||||
dispatcher.Started.ShouldBe(["t-high", "t-mid", "t-low"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task At_equal_priority_the_longest_waiting_goes_first()
|
||||
{
|
||||
TaskWhen At(string utc) => new() { Kind = TaskWhenKind.At, Value = utc, TimeZone = "UTC" };
|
||||
await SeedDue("t-1130", when: At("2026-07-31T11:30:00Z"));
|
||||
await SeedDue("t-0900", when: At("2026-07-31T09:00:00Z"));
|
||||
await SeedDue("t-1000", when: At("2026-07-31T10:00:00Z"));
|
||||
|
||||
var dispatcher = new TrackingDispatcher();
|
||||
await Scanner(dispatcher, maxParallelRuns: 1).ScanOnceAsync(default);
|
||||
|
||||
dispatcher.Started.ShouldBe(["t-0900", "t-1000", "t-1130"]);
|
||||
}
|
||||
|
||||
// ═══════════════════ no silent clamp ═══════════════════
|
||||
|
||||
[Fact]
|
||||
public async Task More_than_500_due_tasks_are_all_considered()
|
||||
{
|
||||
for (var i = 0; i < 520; i++)
|
||||
await SeedDue($"t-{i:000}");
|
||||
|
||||
var dispatcher = new TrackingDispatcher();
|
||||
var dispatched = await Scanner(dispatcher, maxParallelRuns: 1000).ScanOnceAsync(default);
|
||||
|
||||
dispatched.ShouldBe(520, "before #682 the tick stopped at 500 without a word");
|
||||
}
|
||||
|
||||
// ═══════════════════ test double ═══════════════════
|
||||
|
||||
/// <summary>
|
||||
/// Records starts and the highest number of runs in flight at once. Either finishes
|
||||
/// immediately, after <c>holdFor</c>, or — <c>gated</c> — only on <see cref="ReleaseAll"/>.
|
||||
/// </summary>
|
||||
private sealed class TrackingDispatcher(TimeSpan? holdFor = null, bool gated = false) : ITaskDispatcher
|
||||
{
|
||||
private readonly Lock _lock = new();
|
||||
private readonly List<TaskCompletionSource> _gates = [];
|
||||
private int _current;
|
||||
|
||||
public List<string> Started { get; } = [];
|
||||
public int MaxConcurrent { get; private set; }
|
||||
|
||||
public async Task<TaskLauf> DispatchAsync(TaskItem task, CancellationToken ct)
|
||||
{
|
||||
TaskCompletionSource? gate = null;
|
||||
lock (_lock)
|
||||
{
|
||||
Started.Add(task.Id);
|
||||
_current++;
|
||||
MaxConcurrent = Math.Max(MaxConcurrent, _current);
|
||||
if (gated)
|
||||
{
|
||||
gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
_gates.Add(gate);
|
||||
}
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
if (gate is not null)
|
||||
await gate.Task;
|
||||
else if (holdFor is { } delay)
|
||||
await Task.Delay(delay, ct);
|
||||
|
||||
return TaskLauf.Erledigt;
|
||||
}
|
||||
finally
|
||||
{
|
||||
lock (_lock)
|
||||
_current--;
|
||||
}
|
||||
}
|
||||
|
||||
public void ReleaseAll()
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
foreach (var gate in _gates)
|
||||
gate.TrySetResult();
|
||||
_gates.Clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user