Merge task branch for: fix(worker): planning-chain cascade stalls at an Idle middle link
This commit is contained in:
@@ -92,14 +92,7 @@ public sealed class PlanningChainCoordinator
|
|||||||
public async Task<string?> OnChildFinishedAsync(
|
public async Task<string?> OnChildFinishedAsync(
|
||||||
string childTaskId, TaskStatus finalStatus, CancellationToken ct = default)
|
string childTaskId, TaskStatus finalStatus, CancellationToken ct = default)
|
||||||
{
|
{
|
||||||
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
var nextId = await FindSuccessorAsync(childTaskId, ct);
|
||||||
// The successor is whichever sibling explicitly blocks on this child.
|
|
||||||
var nextId = await ctx.Tasks
|
|
||||||
.AsNoTracking()
|
|
||||||
.Where(t => t.BlockedByTaskId == childTaskId)
|
|
||||||
.OrderBy(t => t.SortOrder).ThenBy(t => t.CreatedAt)
|
|
||||||
.Select(t => t.Id)
|
|
||||||
.FirstOrDefaultAsync(ct);
|
|
||||||
if (nextId is null) return null;
|
if (nextId is null) return null;
|
||||||
|
|
||||||
if (finalStatus == TaskStatus.Done)
|
if (finalStatus == TaskStatus.Done)
|
||||||
@@ -108,10 +101,33 @@ public sealed class PlanningChainCoordinator
|
|||||||
return nextId;
|
return nextId;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Child failed or was cancelled: cancel the immediate successor so the chain
|
// Child failed or was cancelled: cancel the immediate successor so the chain is
|
||||||
// is not left wedged. CancelAsync triggers OnChildTerminalAsync → OnChildFinishedAsync
|
// not left wedged. If it's cancellable, CancelAsync's own OnChildTerminalAsync
|
||||||
// for that successor, cascading cancellation through the rest of the chain.
|
// callback recurses into this method for its successor, cascading cancellation
|
||||||
await _state().CancelAsync(nextId, DateTime.UtcNow, ct);
|
// through the rest of the chain. If it's not (e.g. it was parked back to Idle
|
||||||
return null;
|
// out of band), CancelAsync is a no-op and nothing will ever call back for it —
|
||||||
|
// keep walking the chain ourselves so the tail isn't left wedged forever.
|
||||||
|
var predecessorId = nextId;
|
||||||
|
while (true)
|
||||||
|
{
|
||||||
|
var result = await _state().CancelAsync(predecessorId, DateTime.UtcNow, ct);
|
||||||
|
if (result.Ok) return null;
|
||||||
|
|
||||||
|
var following = await FindSuccessorAsync(predecessorId, ct);
|
||||||
|
if (following is null) return null;
|
||||||
|
predecessorId = following;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The successor is whichever sibling explicitly blocks on this task.
|
||||||
|
private async Task<string?> FindSuccessorAsync(string taskId, CancellationToken ct)
|
||||||
|
{
|
||||||
|
await using var ctx = await _dbFactory.CreateDbContextAsync(ct);
|
||||||
|
return await ctx.Tasks
|
||||||
|
.AsNoTracking()
|
||||||
|
.Where(t => t.BlockedByTaskId == taskId)
|
||||||
|
.OrderBy(t => t.SortOrder).ThenBy(t => t.CreatedAt)
|
||||||
|
.Select(t => t.Id)
|
||||||
|
.FirstOrDefaultAsync(ct);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -178,6 +178,37 @@ public sealed class PlanningChainCoordinatorTests : IDisposable
|
|||||||
Assert.Equal(TaskStatus.Cancelled, kids[3].Status);
|
Assert.Equal(TaskStatus.Cancelled, kids[3].Status);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task OnChildFailed_MidChain_IdleSuccessor_DoesNotWedgeTail()
|
||||||
|
{
|
||||||
|
// Chain: c0 → c1 → c2 → c3. c1 is parked back to Idle (e.g. a manual reset)
|
||||||
|
// while still wired into the chain (BlockedByTaskId = c0). c0 then fails:
|
||||||
|
// CancelAsync(c1) is a no-op because Idle isn't a cancellable state, so the
|
||||||
|
// cascade must keep walking past c1 instead of leaving c2/c3 wedged forever.
|
||||||
|
await SeedPlanningFamilyAsync("P", 4);
|
||||||
|
await _sut.SetupChainAsync("P", enqueue: true, default);
|
||||||
|
|
||||||
|
await using (var ctx = _factory.CreateDbContext())
|
||||||
|
{
|
||||||
|
var c0 = await ctx.Tasks.FirstAsync(t => t.Id == "P-c0");
|
||||||
|
c0.Status = TaskStatus.Failed;
|
||||||
|
var c1 = await ctx.Tasks.FirstAsync(t => t.Id == "P-c1");
|
||||||
|
c1.Status = TaskStatus.Idle;
|
||||||
|
await ctx.SaveChangesAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
var advanced = await _sut.OnChildFinishedAsync("P-c0", TaskStatus.Failed, default);
|
||||||
|
|
||||||
|
Assert.Null(advanced);
|
||||||
|
var kids = await GetChildrenAsync("P");
|
||||||
|
Assert.Equal(TaskStatus.Failed, kids[0].Status);
|
||||||
|
// c1 was already parked Idle; the coordinator must not resurrect it.
|
||||||
|
Assert.Equal(TaskStatus.Idle, kids[1].Status);
|
||||||
|
// c2/c3 must not stay wedged as Queued+blocked forever.
|
||||||
|
Assert.Equal(TaskStatus.Cancelled, kids[2].Status);
|
||||||
|
Assert.Equal(TaskStatus.Cancelled, kids[3].Status);
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task OnChildDone_LastChild_ReturnsNull()
|
public async Task OnChildDone_LastChild_ReturnsNull()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user