Merge branch 'claudedo/f359858ac98a439593e459df9c5d0a5d'
This commit is contained in:
@@ -44,9 +44,18 @@ public sealed class WorkerConfig
|
||||
[JsonPropertyName("online_inbox")]
|
||||
public OnlineInboxConfig OnlineInbox { get; set; } = new();
|
||||
|
||||
/// <summary>Poll interval for the OAuth usage monitor. Clamped to a minimum of 15s on load.</summary>
|
||||
[JsonPropertyName("usage_poll_interval_seconds")]
|
||||
public int UsagePollIntervalSeconds { get; set; } = 60;
|
||||
/// <summary>
|
||||
/// Usage-monitor poll interval while at least one task is Running. Clamped to a minimum
|
||||
/// of 60s on load — the endpoint rate-limits (429) on tighter polling.
|
||||
/// </summary>
|
||||
[JsonPropertyName("usage_poll_interval_active_seconds")]
|
||||
public int UsagePollIntervalActiveSeconds { get; set; } = 300;
|
||||
|
||||
/// <summary>
|
||||
/// Usage-monitor poll interval while nothing is running. Clamped to a minimum of 60s on load.
|
||||
/// </summary>
|
||||
[JsonPropertyName("usage_poll_interval_idle_seconds")]
|
||||
public int UsagePollIntervalIdleSeconds { get; set; } = 900;
|
||||
|
||||
public static string DefaultConfigPath =>
|
||||
Path.Combine(Paths.AppDataRoot(), "worker.config.json");
|
||||
@@ -75,7 +84,8 @@ public sealed class WorkerConfig
|
||||
cfg.SandboxRoot = Paths.Expand(cfg.SandboxRoot);
|
||||
cfg.LogRoot = Paths.Expand(cfg.LogRoot);
|
||||
cfg.CentralWorktreeRoot = Paths.Expand(cfg.CentralWorktreeRoot);
|
||||
cfg.UsagePollIntervalSeconds = Math.Max(15, cfg.UsagePollIntervalSeconds);
|
||||
cfg.UsagePollIntervalActiveSeconds = Math.Max(60, cfg.UsagePollIntervalActiveSeconds);
|
||||
cfg.UsagePollIntervalIdleSeconds = Math.Max(60, cfg.UsagePollIntervalIdleSeconds);
|
||||
|
||||
return cfg;
|
||||
}
|
||||
|
||||
@@ -189,6 +189,7 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
private readonly Data.Git.GitService? _git;
|
||||
private readonly UsageSnapshotBuilder? _usageSnapshotBuilder;
|
||||
private readonly ITranscriptUsageReader? _usageReader;
|
||||
private readonly UsageMonitorService? _usageMonitor;
|
||||
|
||||
public WorkerHub(
|
||||
QueueService queue,
|
||||
@@ -220,7 +221,8 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
WorktreeManager? worktreeManager = null,
|
||||
Data.Git.GitService? git = null,
|
||||
UsageSnapshotBuilder? usageSnapshotBuilder = null,
|
||||
ITranscriptUsageReader? usageReader = null)
|
||||
ITranscriptUsageReader? usageReader = null,
|
||||
UsageMonitorService? usageMonitor = null)
|
||||
{
|
||||
_queue = queue;
|
||||
_waker = waker;
|
||||
@@ -252,6 +254,7 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
_git = git;
|
||||
_usageSnapshotBuilder = usageSnapshotBuilder;
|
||||
_usageReader = usageReader;
|
||||
_usageMonitor = usageMonitor;
|
||||
}
|
||||
|
||||
// Persistence boundary for the session_skills JSON-array columns (task/list/global).
|
||||
@@ -1049,6 +1052,18 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
return _usageSnapshotBuilder.BuildAsync(Context.ConnectionAborted);
|
||||
});
|
||||
|
||||
/// <summary>
|
||||
/// Manual "refresh now" for the usage monitor. Polls the endpoint out of band and returns the
|
||||
/// fresh snapshot; a refresh inside the monitor's cooldown reuses the last poll's result
|
||||
/// instead of risking a 429.
|
||||
/// </summary>
|
||||
public Task<UsageSnapshotDto> RefreshUsage() => HubGuard(() =>
|
||||
{
|
||||
if (_usageMonitor is null)
|
||||
throw new InvalidOperationException("Usage monitor is not configured.");
|
||||
return _usageMonitor.RefreshNowAsync(Context.ConnectionAborted);
|
||||
});
|
||||
|
||||
public Task<IReadOnlyList<ModelUsageRowDto>> GetModelUsage(DateOnly from, DateOnly to) => HubGuard(async () =>
|
||||
{
|
||||
if (_usageReader is null)
|
||||
|
||||
@@ -209,7 +209,10 @@ builder.Services.AddHttpClient<IUsageClient, ClaudeOAuthUsageClient>(client =>
|
||||
});
|
||||
builder.Services.AddSingleton<IUsageGate, UsageGate>();
|
||||
builder.Services.AddSingleton<UsageSnapshotBuilder>();
|
||||
builder.Services.AddHostedService<UsageMonitorService>();
|
||||
builder.Services.AddSingleton<IRunningTaskProbe, RunningTaskProbe>();
|
||||
// Singleton + hosted service (same instance) so WorkerHub.RefreshUsage can drive a manual poll.
|
||||
builder.Services.AddSingleton<UsageMonitorService>();
|
||||
builder.Services.AddHostedService(sp => sp.GetRequiredService<UsageMonitorService>());
|
||||
|
||||
// Loopback-only bind. Firewall is irrelevant for 127.0.0.1.
|
||||
builder.WebHost.UseUrls($"http://127.0.0.1:{cfg.SignalRPort}");
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
using System.Net;
|
||||
using System.Net.Http.Headers;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Nodes;
|
||||
@@ -36,6 +37,8 @@ public sealed class ClaudeOAuthUsageClient : IUsageClient
|
||||
request.Headers.Add("anthropic-beta", "oauth-2025-04-20");
|
||||
|
||||
using var response = await _http.SendAsync(request, ct);
|
||||
if (response.StatusCode == HttpStatusCode.TooManyRequests)
|
||||
throw new UsageRateLimitedException(ReadRetryAfter(response));
|
||||
if (!response.IsSuccessStatusCode)
|
||||
throw new InvalidOperationException($"Usage request failed with status {(int)response.StatusCode}.");
|
||||
|
||||
@@ -43,6 +46,27 @@ public sealed class ClaudeOAuthUsageClient : IUsageClient
|
||||
return Parse(body);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads the <c>Retry-After</c> header in either form (delta-seconds or HTTP-date).
|
||||
/// A missing/past value returns null — the caller then uses its own backoff.
|
||||
/// </summary>
|
||||
internal static TimeSpan? ReadRetryAfter(HttpResponseMessage response)
|
||||
{
|
||||
var header = response.Headers.RetryAfter;
|
||||
if (header is null) return null;
|
||||
|
||||
if (header.Delta is { } delta)
|
||||
return delta > TimeSpan.Zero ? delta : null;
|
||||
|
||||
if (header.Date is { } date)
|
||||
{
|
||||
var remaining = date - DateTimeOffset.UtcNow;
|
||||
return remaining > TimeSpan.Zero ? remaining : null;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private string ReadAccessToken()
|
||||
{
|
||||
if (!File.Exists(_credentialsPath))
|
||||
@@ -162,3 +186,20 @@ public sealed class ClaudeOAuthUsageClient : IUsageClient
|
||||
? dto
|
||||
: null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The usage endpoint answered 429. Carries the server's <c>Retry-After</c> when it sent one so
|
||||
/// <see cref="UsageMonitorService"/> can honour it instead of guessing a backoff.
|
||||
/// </summary>
|
||||
public sealed class UsageRateLimitedException : InvalidOperationException
|
||||
{
|
||||
public UsageRateLimitedException(TimeSpan? retryAfter)
|
||||
: base(retryAfter is { } r
|
||||
? $"Usage request was rate-limited (429); retry after {(int)r.TotalSeconds}s."
|
||||
: "Usage request was rate-limited (429).")
|
||||
{
|
||||
RetryAfter = retryAfter;
|
||||
}
|
||||
|
||||
public TimeSpan? RetryAfter { get; }
|
||||
}
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
namespace ClaudeDo.Worker.Usage.Interfaces;
|
||||
|
||||
/// <summary>
|
||||
/// Tells the usage monitor whether ClaudeDo is currently burning tokens, so it can poll
|
||||
/// often while work is in flight and back off to a slow heartbeat while idle.
|
||||
/// </summary>
|
||||
public interface IRunningTaskProbe
|
||||
{
|
||||
Task<bool> AnyRunningAsync(CancellationToken ct = default);
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
using ClaudeDo.Data;
|
||||
using ClaudeDo.Worker.Usage.Interfaces;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
|
||||
namespace ClaudeDo.Worker.Usage;
|
||||
|
||||
/// <summary>
|
||||
/// Answers "is anything running?" from the task table rather than the in-memory queue slots,
|
||||
/// so override-slot runs, continued runs, and runs still marked Running after a worker restart
|
||||
/// all count. A read failure reports idle — the usage monitor must never poll harder because
|
||||
/// its activity probe broke.
|
||||
/// </summary>
|
||||
public sealed class RunningTaskProbe : IRunningTaskProbe
|
||||
{
|
||||
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
|
||||
|
||||
public RunningTaskProbe(IDbContextFactory<ClaudeDoDbContext> dbFactory) => _dbFactory = dbFactory;
|
||||
|
||||
public async Task<bool> AnyRunningAsync(CancellationToken ct = default)
|
||||
{
|
||||
try
|
||||
{
|
||||
await using var context = await _dbFactory.CreateDbContextAsync(ct);
|
||||
return await context.Tasks.AnyAsync(t => t.Status == TaskStatus.Running, ct);
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5,25 +5,43 @@ using ClaudeDo.Worker.Usage.Interfaces;
|
||||
namespace ClaudeDo.Worker.Usage;
|
||||
|
||||
/// <summary>
|
||||
/// Polls <see cref="IUsageClient"/> on <see cref="WorkerConfig.UsagePollIntervalSeconds"/> and keeps
|
||||
/// <see cref="UsageState"/> current. Polls once immediately at startup. A failure is logged as a
|
||||
/// warning at most once per distinct error message, to avoid log spam on a persistent outage.
|
||||
/// Broadcasts <see cref="HubBroadcaster.UsageUpdated"/> after every poll cycle, success or failure,
|
||||
/// so the UI can reflect a stale/blocked state as soon as it happens.
|
||||
/// Polls <see cref="IUsageClient"/> and keeps <see cref="UsageState"/> current. Polls once
|
||||
/// immediately at startup, then on an **activity-dependent** interval: while any task is
|
||||
/// Running it uses <see cref="WorkerConfig.UsagePollIntervalActiveSeconds"/>, otherwise the
|
||||
/// slower <see cref="WorkerConfig.UsagePollIntervalIdleSeconds"/>. A 429 adds exponential
|
||||
/// backoff on top (honouring <c>Retry-After</c> when the server sends one) — the endpoint is
|
||||
/// undocumented and rate-limits aggressively. <see cref="RefreshNowAsync"/> gives the UI a
|
||||
/// manual refresh that also resets the schedule, so the slow idle interval never leaves the
|
||||
/// user staring at a stale number.
|
||||
///
|
||||
/// A failure is logged as a warning at most once per distinct error message, to avoid log spam
|
||||
/// on a persistent outage. Broadcasts <see cref="HubBroadcaster.UsageUpdated"/> after every
|
||||
/// poll cycle, success or failure, so the UI can reflect a stale/blocked state as soon as it
|
||||
/// happens.
|
||||
/// </summary>
|
||||
public sealed class UsageMonitorService : BackgroundService
|
||||
{
|
||||
/// <summary>A manual refresh inside this window of the last poll reuses the last result.</summary>
|
||||
internal static readonly TimeSpan ManualRefreshCooldown = TimeSpan.FromSeconds(10);
|
||||
|
||||
private readonly IUsageClient _client;
|
||||
private readonly UsageState _state;
|
||||
private readonly WorkerConfig _config;
|
||||
private readonly ILogger<UsageMonitorService> _logger;
|
||||
private readonly UsageSnapshotBuilder _snapshotBuilder;
|
||||
private readonly HubBroadcaster _broadcaster;
|
||||
private readonly IRunningTaskProbe _runningProbe;
|
||||
|
||||
// Serializes the background loop against a manual refresh so two polls never overlap.
|
||||
private readonly SemaphoreSlim _pollLock = new(1, 1);
|
||||
private string? _lastLoggedError;
|
||||
private int _rateLimitStrikes;
|
||||
private DateTime _lastPollUtc = DateTime.MinValue;
|
||||
private DateTime _nextPollDueUtc = DateTime.MinValue;
|
||||
|
||||
public UsageMonitorService(
|
||||
IUsageClient client, UsageState state, WorkerConfig config, ILogger<UsageMonitorService> logger,
|
||||
UsageSnapshotBuilder snapshotBuilder, HubBroadcaster broadcaster)
|
||||
UsageSnapshotBuilder snapshotBuilder, HubBroadcaster broadcaster, IRunningTaskProbe runningProbe)
|
||||
{
|
||||
_client = client;
|
||||
_state = state;
|
||||
@@ -31,49 +49,135 @@ public sealed class UsageMonitorService : BackgroundService
|
||||
_logger = logger;
|
||||
_snapshotBuilder = snapshotBuilder;
|
||||
_broadcaster = broadcaster;
|
||||
_runningProbe = runningProbe;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
await TickAsync(stoppingToken);
|
||||
// Re-read the due time every iteration: a manual refresh pushes it out, which is
|
||||
// how the loop avoids polling again right after the user hit refresh.
|
||||
var wait = _nextPollDueUtc - DateTime.UtcNow;
|
||||
if (wait > TimeSpan.Zero)
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(wait, stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
return;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromSeconds(_config.UsagePollIntervalSeconds), stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
return;
|
||||
}
|
||||
await TickAsync(stoppingToken);
|
||||
}
|
||||
}
|
||||
|
||||
internal async Task TickAsync(CancellationToken ct)
|
||||
/// <summary>
|
||||
/// Forces a poll now and returns the resulting snapshot DTO. Within
|
||||
/// <see cref="ManualRefreshCooldown"/> of the last poll the API call is skipped and the
|
||||
/// current state is returned instead, so click-spamming the refresh button can't earn a 429.
|
||||
/// </summary>
|
||||
public async Task<UsageSnapshotDto> RefreshNowAsync(CancellationToken ct = default)
|
||||
{
|
||||
await PollAsync(ct, ManualRefreshCooldown);
|
||||
return await _snapshotBuilder.BuildAsync(ct);
|
||||
}
|
||||
|
||||
internal Task TickAsync(CancellationToken ct) => PollAsync(ct, null);
|
||||
|
||||
private async Task PollAsync(CancellationToken ct, TimeSpan? skipIfPolledWithin)
|
||||
{
|
||||
await _pollLock.WaitAsync(ct);
|
||||
TimeSpan? retryAfter = null;
|
||||
try
|
||||
{
|
||||
var snapshot = await _client.GetUsageAsync(ct);
|
||||
_state.ReportSuccess(snapshot);
|
||||
_lastLoggedError = null;
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_state.ReportFailure(ex.Message, DateTime.UtcNow);
|
||||
// Checked under the lock so a manual refresh that queued behind a background poll
|
||||
// sees that poll's timestamp and reuses its result instead of firing a second call.
|
||||
if (skipIfPolledWithin is { } window && DateTime.UtcNow - _lastPollUtc < window)
|
||||
return;
|
||||
|
||||
if (_lastLoggedError != ex.Message)
|
||||
try
|
||||
{
|
||||
_logger.LogWarning(ex, "UsageMonitorService: failed to fetch usage: {Message}", ex.Message);
|
||||
_lastLoggedError = ex.Message;
|
||||
var snapshot = await _client.GetUsageAsync(ct);
|
||||
_state.ReportSuccess(snapshot);
|
||||
_lastLoggedError = null;
|
||||
_rateLimitStrikes = 0;
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (UsageRateLimitedException ex)
|
||||
{
|
||||
_rateLimitStrikes = Math.Min(_rateLimitStrikes + 1, UsagePollSchedule.MaxStrikes);
|
||||
retryAfter = ex.RetryAfter;
|
||||
RecordFailure(ex);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
RecordFailure(ex);
|
||||
}
|
||||
|
||||
_lastPollUtc = DateTime.UtcNow;
|
||||
_nextPollDueUtc = _lastPollUtc + await NextDelayAsync(retryAfter, ct);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_pollLock.Release();
|
||||
}
|
||||
|
||||
var dto = await _snapshotBuilder.BuildAsync(ct);
|
||||
await _broadcaster.UsageUpdated(dto);
|
||||
}
|
||||
|
||||
private void RecordFailure(Exception ex)
|
||||
{
|
||||
_state.ReportFailure(ex.Message, DateTime.UtcNow);
|
||||
|
||||
if (_lastLoggedError == ex.Message) return;
|
||||
_logger.LogWarning(ex, "UsageMonitorService: failed to fetch usage: {Message}", ex.Message);
|
||||
_lastLoggedError = ex.Message;
|
||||
}
|
||||
|
||||
private async Task<TimeSpan> NextDelayAsync(TimeSpan? retryAfter, CancellationToken ct)
|
||||
{
|
||||
var anyRunning = await _runningProbe.AnyRunningAsync(ct);
|
||||
return UsagePollSchedule.NextDelay(
|
||||
anyRunning,
|
||||
_config.UsagePollIntervalActiveSeconds,
|
||||
_config.UsagePollIntervalIdleSeconds,
|
||||
_rateLimitStrikes,
|
||||
retryAfter);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Pure poll-interval arithmetic for <see cref="UsageMonitorService"/>: active-vs-idle base
|
||||
/// interval plus 429 backoff. Kept static and side-effect-free so the schedule is testable
|
||||
/// without a running background service.
|
||||
/// </summary>
|
||||
internal static class UsagePollSchedule
|
||||
{
|
||||
/// <summary>Strike count is capped so the exponent can't run away on a long outage.</summary>
|
||||
internal const int MaxStrikes = 4;
|
||||
|
||||
/// <summary>Nothing ever waits longer than this, not even an absurd <c>Retry-After</c>.</summary>
|
||||
internal static readonly TimeSpan MaxDelay = TimeSpan.FromMinutes(30);
|
||||
|
||||
internal static TimeSpan NextDelay(
|
||||
bool anyTaskRunning, int activeSeconds, int idleSeconds, int rateLimitStrikes, TimeSpan? retryAfter)
|
||||
{
|
||||
var baseDelay = TimeSpan.FromSeconds(Math.Max(1, anyTaskRunning ? activeSeconds : idleSeconds));
|
||||
if (rateLimitStrikes <= 0)
|
||||
return baseDelay;
|
||||
|
||||
var backoff = retryAfter ?? baseDelay * Math.Pow(2, Math.Min(rateLimitStrikes, MaxStrikes));
|
||||
// Never poll *sooner* than the normal cadence just because Retry-After was small.
|
||||
if (backoff < baseDelay) backoff = baseDelay;
|
||||
return backoff > MaxDelay ? MaxDelay : backoff;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -39,7 +39,10 @@ public sealed class UsageSnapshotBuilder
|
||||
|
||||
var decision = await _gate.EvaluateAsync(ct);
|
||||
|
||||
var maxAge = TimeSpan.FromSeconds(_cfg.UsagePollIntervalSeconds * 3);
|
||||
// Measured against the *slowest* cadence — the idle interval — so a genuinely idle
|
||||
// worker on its 15-minute heartbeat isn't reported stale just for not polling.
|
||||
var maxAge = TimeSpan.FromSeconds(
|
||||
Math.Max(_cfg.UsagePollIntervalActiveSeconds, _cfg.UsagePollIntervalIdleSeconds) * 3);
|
||||
var isStale = snapshot is null || lastError is not null || (DateTime.UtcNow - snapshot.FetchedAtUtc) > maxAge;
|
||||
|
||||
var limits = (snapshot?.Limits ?? Array.Empty<UsageLimitRow>())
|
||||
|
||||
Reference in New Issue
Block a user