feat(claude-code): single-flight login coordinator
This commit is contained in:
@@ -0,0 +1,49 @@
|
||||
namespace Volcanic.AgenticCore.Extensions.ClaudeCode;
|
||||
|
||||
/// <summary>
|
||||
/// Keeps Claude Code logins single-flight: every session shares one <c>claude</c> configuration, so concurrent expired
|
||||
/// requests wait on the same flow. The flow runs detached from any caller's token; a caller's cancellation only stops
|
||||
/// its own wait. The slot clears when the flow completes, so a later expiry starts a fresh one.
|
||||
/// </summary>
|
||||
public sealed class ClaudeLoginCoordinator
|
||||
{
|
||||
/// <summary>The coordinator shared by every extension instance in the process.</summary>
|
||||
public static ClaudeLoginCoordinator Shared { get; } = new();
|
||||
|
||||
private readonly Lock _lock = new();
|
||||
private Task<LoginOutcome>? _current;
|
||||
|
||||
/// <summary>Joins the in-flight login, or starts one. <paramref name="runFlow"/> only ever runs for the caller that starts it.</summary>
|
||||
public Task<LoginOutcome> EnsureLoggedInAsync(Func<CancellationToken, Task<LoginOutcome>> runFlow, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(runFlow);
|
||||
Task<LoginOutcome> flow;
|
||||
lock (_lock)
|
||||
{
|
||||
flow = _current ??= StartAsync(runFlow);
|
||||
}
|
||||
return flow.WaitAsync(cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<LoginOutcome> StartAsync(Func<CancellationToken, Task<LoginOutcome>> runFlow)
|
||||
{
|
||||
// Unconditional suspension before touching _current again: without it, a runFlow that completes (or throws)
|
||||
// synchronously would run this whole method - including the finally below - inside the very statement in
|
||||
// EnsureLoggedInAsync that assigns _current, so that assignment would run *after* the finally's clear and
|
||||
// overwrite it with the now-finished task. The slot would then stay stuck on a stale result forever, since
|
||||
// nothing else ever clears it. Task.Yield() guarantees the finally always runs on a later, separate turn.
|
||||
await Task.Yield();
|
||||
try
|
||||
{
|
||||
return await runFlow(CancellationToken.None).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return LoginOutcome.Failed(ex.Message);
|
||||
}
|
||||
finally
|
||||
{
|
||||
lock (_lock) _current = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
using Volcanic.AgenticCore.Extensions.ClaudeCode;
|
||||
|
||||
namespace Volcanic.AgenticCore.Extensions.Tests.ClaudeCode;
|
||||
|
||||
public class ClaudeLoginCoordinatorTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Concurrent_callers_share_one_flow()
|
||||
{
|
||||
var coordinator = new ClaudeLoginCoordinator();
|
||||
var gate = new TaskCompletionSource<LoginOutcome>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
var runs = 0;
|
||||
Func<CancellationToken, Task<LoginOutcome>> flow = _ => { Interlocked.Increment(ref runs); return gate.Task; };
|
||||
|
||||
var first = coordinator.EnsureLoggedInAsync(flow, CancellationToken.None);
|
||||
var second = coordinator.EnsureLoggedInAsync(flow, CancellationToken.None);
|
||||
gate.SetResult(LoginOutcome.Ok);
|
||||
|
||||
Assert.True((await first).Success);
|
||||
Assert.True((await second).Success);
|
||||
Assert.Equal(1, runs);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Failure_reaches_every_waiter_and_the_next_call_starts_a_new_flow()
|
||||
{
|
||||
var coordinator = new ClaudeLoginCoordinator();
|
||||
var runs = 0;
|
||||
var a = coordinator.EnsureLoggedInAsync(_ => { runs++; return Task.FromResult(LoginOutcome.Failed("no code was entered")); }, CancellationToken.None);
|
||||
Assert.Equal("no code was entered", (await a).Error);
|
||||
var b = await coordinator.EnsureLoggedInAsync(_ => { runs++; return Task.FromResult(LoginOutcome.Ok); }, CancellationToken.None);
|
||||
Assert.True(b.Success);
|
||||
Assert.Equal(2, runs);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_cancelled_waiter_does_not_cancel_the_flow()
|
||||
{
|
||||
var coordinator = new ClaudeLoginCoordinator();
|
||||
var gate = new TaskCompletionSource<LoginOutcome>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
CancellationToken flowToken = default;
|
||||
Func<CancellationToken, Task<LoginOutcome>> flow = ct => { flowToken = ct; return gate.Task; };
|
||||
using var cts = new CancellationTokenSource();
|
||||
|
||||
var cancelled = coordinator.EnsureLoggedInAsync(flow, cts.Token);
|
||||
var other = coordinator.EnsureLoggedInAsync(flow, CancellationToken.None);
|
||||
cts.Cancel();
|
||||
|
||||
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => cancelled);
|
||||
Assert.False(flowToken.IsCancellationRequested);
|
||||
gate.SetResult(LoginOutcome.Ok);
|
||||
Assert.True((await other).Success);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_throwing_flow_becomes_a_failed_outcome()
|
||||
{
|
||||
var coordinator = new ClaudeLoginCoordinator();
|
||||
var outcome = await coordinator.EnsureLoggedInAsync(_ => throw new InvalidOperationException("boom"), CancellationToken.None);
|
||||
Assert.False(outcome.Success);
|
||||
Assert.Contains("boom", outcome.Error);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Regression test for a real bug found while reviewing the design: without an unconditional suspension point
|
||||
/// (<c>Task.Yield()</c>) before the slot is ever cleared, a flow that completes synchronously (as
|
||||
/// <c>Task.FromResult</c> does here) runs its whole body - including the <c>finally</c> that clears
|
||||
/// <c>_current</c> - inside the very statement that assigns <c>_current</c> in the first place. The clear then
|
||||
/// happens before the assignment, so the assignment silently overwrites it with the already-finished task,
|
||||
/// wedging the coordinator: every later call would replay that stale result and <c>runFlow</c> would never run
|
||||
/// again. This drives many concurrent synchronous completions through the same coordinator to make sure the
|
||||
/// slot is freed every time.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_synchronously_completing_flow_still_frees_its_slot_for_the_next_call()
|
||||
{
|
||||
var coordinator = new ClaudeLoginCoordinator();
|
||||
var runs = 0;
|
||||
Func<CancellationToken, Task<LoginOutcome>> flow = _ => { Interlocked.Increment(ref runs); return Task.FromResult(LoginOutcome.Ok); };
|
||||
|
||||
for (var i = 0; i < 50; i++)
|
||||
{
|
||||
var outcome = await coordinator.EnsureLoggedInAsync(flow, CancellationToken.None);
|
||||
Assert.True(outcome.Success);
|
||||
}
|
||||
|
||||
Assert.Equal(50, runs);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user