fix(usage): stop 429s with an activity-dependent poll cadence + manual refresh

The usage monitor polled the undocumented OAuth usage endpoint every 60s and
earned 429s. It now polls every 5 min while any task is Running and every
15 min while idle (usage_poll_interval_active_seconds / _idle_seconds, both
clamped to >= 60; the old single usage_poll_interval_seconds key is gone).

A 429 comes back as UsageRateLimitedException carrying Retry-After and adds
exponential backoff on top, capped at 30 min and never shorter than the normal
cadence; the strike count resets on the first success. The schedule arithmetic
is the pure static UsagePollSchedule.NextDelay.

Since the idle cadence is slow on purpose, WorkerHub.RefreshUsage drives
UsageMonitorService.RefreshNowAsync behind a Refresh now button in the Usage
Monitor modal: an out-of-band poll that pushes the loop's next-due time out so
no double poll follows, with a 10s cooldown so click-spam can't earn a 429.

Staleness now measures against the slower (idle) interval so an idle worker
isn't flagged stale just for not polling.
This commit is contained in:
mika kuns
2026-08-05 16:40:34 +02:00
parent f6cb8250bb
commit 2700c3d817
24 changed files with 615 additions and 50 deletions
+19 -5
View File
@@ -21,7 +21,7 @@ Worker/
Report/ — ClaudeHistoryReader, WeekReportPromptBuilder, WeekReportService; interfaces in Report/Interfaces/
Prime/ — daily-prep ("Prime Claude"): PrimeScheduler (BackgroundService), PrimeRunner (runs the daily prep), DailyPrepPrompt (fixed prompt + CLI args + LogPath() helper), NextDueCalculator, PrimeScheduleSignal; interfaces in Prime/Interfaces/ (IPrimeRunner, IPrimeClock, IPrimeScheduleSignal, IPrimeBroadcaster)
Online/ — optional Online Inbox sync: OnlineInboxConfig (config record), Dtos (RemoteList/RemoteTask/MirrorTask), IOnlineInboxApi, OnlineInboxApiClient (typed HttpClient, bearer auth, HTTPS guard), OnlineTokenStore (DPAPI refresh-token store, Windows-only), StaticTokenAuthProvider (default/test IOnlineAuthProvider), ZitadelAuthProvider (OIDC discovery + refresh-token flow), OnlineSyncService (BackgroundService: reconcile loop), OnlineBacklog (Idle-backlog filter/query); interface in Online/Interfaces/ (IOnlineAuthProvider)
Usage/ — OAuth usage monitor: UsageModels (UsageBucket/UsageLimitRow/UsageSnapshot), ClaudeOAuthUsageClient (reads the access token Claude Code keeps fresh at `~/.claude/.credentials.json`, calls `GET https://api.anthropic.com/api/oauth/usage`; defensive parsing — missing/null buckets → null, missing `limits` → empty list; never logs the token), UsageState (threadsafe singleton; a failed poll never overwrites the last good snapshot, only sets `LastError`), UsageMonitorService (BackgroundService, polls on `usage_poll_interval_seconds`, one poll at startup, logs a failure at most once per distinct error message, broadcasts `HubBroadcaster.UsageUpdated` after every tick via `UsageSnapshotBuilder`), UsageSnapshotBuilder (builds the Hub-facing `UsageSnapshotDto` from `UsageState` + `IUsageGate` + `AppSettings` thresholds — the one place `WorkerHub.GetUsageSnapshot` and `UsageMonitorService` share the stale/threshold/gate logic), TranscriptUsageReader (aggregates Claude Code transcript token usage from `~/.claude/projects/**/*.jsonl` by date/model/scope (ClaudeDo vs Other), deduped by requestId, with a per-file length+mtime cache; `<synthetic>`-model lines are skipped everywhere, not real API calls; also exposes `ReadSessionTotalsAsync(sessionId)` — cumulative raw input/output/cache-read/cache-write totals for one session, located by `{sessionId}.jsonl`, used by `TaskRunner` to populate `task_runs`' per-run token fields), UsageGate (reads `UsageState` + `AppSettings.UsageGateFiveHourPct`/`UsageGateSevenDayPct`, returns a `UsageGateDecision(IsBlocked, Reason)`; `Utilization` from `UsageBucket` is already a 0100 percent, compared directly against the threshold with `>=`; threshold `0` = that bucket never gates; fail-open — no snapshot yet, a failed last poll, or a settings-read error all resolve to not-blocked), UsageThrottle (pure static `EffectiveSlots(configuredSlots, fiveHourPct, sevenDayPct, softPct, hardPct, gateFiveHourPct, gateSevenDayPct)` — stages parallelism down ahead of the hard gate: whichever bucket is more utilized decides the stage, `0` for a threshold disables that stage, `>= softPct` caps at 2 slots, `>= hardPct` caps at 1, `>= either gate threshold` returns 0 — same 0 as `UsageGate`'s hard block, kept in sync since both read the same gate thresholds); interfaces in Usage/Interfaces/ (IUsageClient, ITranscriptUsageReader, IUsageGate)
Usage/ — OAuth usage monitor: UsageModels (UsageBucket/UsageLimitRow/UsageSnapshot), ClaudeOAuthUsageClient (reads the access token Claude Code keeps fresh at `~/.claude/.credentials.json`, calls `GET https://api.anthropic.com/api/oauth/usage`; defensive parsing — missing/null buckets → null, missing `limits` → empty list; never logs the token; a 429 becomes `UsageRateLimitedException` carrying the parsed `Retry-After`, delta or HTTP-date, null when absent/past), UsageState (threadsafe singleton; a failed poll never overwrites the last good snapshot, only sets `LastError`), UsageMonitorService (singleton + BackgroundService, one poll at startup then an activity-dependent interval with 429 backoff, `RefreshNowAsync` for the UI's manual refresh, logs a failure at most once per distinct error message, broadcasts `HubBroadcaster.UsageUpdated` after every tick via `UsageSnapshotBuilder`) + UsagePollSchedule (pure static `NextDelay`, see below), RunningTaskProbe (`IRunningTaskProbe` — `Tasks.Any(Status == Running)`, so override-slot/continued/post-restart runs all count; a read failure reports idle so a broken probe can never make the monitor poll harder), UsageSnapshotBuilder (builds the Hub-facing `UsageSnapshotDto` from `UsageState` + `IUsageGate` + `AppSettings` thresholds — the one place `WorkerHub.GetUsageSnapshot` and `UsageMonitorService` share the stale/threshold/gate logic), TranscriptUsageReader (aggregates Claude Code transcript token usage from `~/.claude/projects/**/*.jsonl` by date/model/scope (ClaudeDo vs Other), deduped by requestId, with a per-file length+mtime cache; `<synthetic>`-model lines are skipped everywhere, not real API calls; also exposes `ReadSessionTotalsAsync(sessionId)` — cumulative raw input/output/cache-read/cache-write totals for one session, located by `{sessionId}.jsonl`, used by `TaskRunner` to populate `task_runs`' per-run token fields), UsageGate (reads `UsageState` + `AppSettings.UsageGateFiveHourPct`/`UsageGateSevenDayPct`, returns a `UsageGateDecision(IsBlocked, Reason)`; `Utilization` from `UsageBucket` is already a 0100 percent, compared directly against the threshold with `>=`; threshold `0` = that bucket never gates; fail-open — no snapshot yet, a failed last poll, or a settings-read error all resolve to not-blocked), UsageThrottle (pure static `EffectiveSlots(configuredSlots, fiveHourPct, sevenDayPct, softPct, hardPct, gateFiveHourPct, gateSevenDayPct)` — stages parallelism down ahead of the hard gate: whichever bucket is more utilized decides the stage, `0` for a threshold disables that stage, `>= softPct` caps at 2 slots, `>= hardPct` caps at 1, `>= either gate threshold` returns 0 — same 0 as `UsageGate`'s hard block, kept in sync since both read the same gate thresholds); interfaces in Usage/Interfaces/ (IUsageClient, ITranscriptUsageReader, IUsageGate, IRunningTaskProbe)
```
Interfaces (e.g. `IQueueWaker`, `IPrimeClock`, `ITaskStateService`) live in an `Interfaces/` subfolder within their area; the namespace stays the area namespace.
@@ -65,8 +65,22 @@ Interfaces (e.g. `IQueueWaker`, `IPrimeClock`, `ITaskStateService`) live in an `
Source: `GET https://api.anthropic.com/api/oauth/usage`, an **undocumented** Anthropic API,
authenticated with the Bearer access token Claude Code itself keeps fresh at
`~/.claude/.credentials.json` — ClaudeDo reads that token but never refreshes it and never
logs it. `UsageMonitorService` polls on `usage_poll_interval_seconds` (default 60s, one poll
at startup too) and broadcasts `HubBroadcaster.UsageUpdated` after every tick.
logs it. The endpoint **rate-limits (429)**, so `UsageMonitorService` polls on an
activity-dependent cadence: one poll at startup, then `usage_poll_interval_active_seconds`
(default 300) while `IRunningTaskProbe` reports any task `Running`, otherwise
`usage_poll_interval_idle_seconds` (default 900). A 429 (surfaced by the client as
`UsageRateLimitedException`) adds exponential backoff on top — the server's `Retry-After` when
it sends one, else `base × 2^strikes` — clamped to `UsagePollSchedule.MaxDelay` (30 min) and
never shorter than the normal cadence; the strike count resets on the first success. The
schedule arithmetic itself is the pure static `UsagePollSchedule.NextDelay`. Every tick
broadcasts `HubBroadcaster.UsageUpdated`.
Because the idle cadence is deliberately slow, `WorkerHub.RefreshUsage` →
`UsageMonitorService.RefreshNowAsync` gives the UI a manual "refresh now" (the Usage Monitor
modal's button): it polls out of band, pushes the loop's next-due time out so no double poll
follows, and within `ManualRefreshCooldown` (10 s) of the last poll reuses that poll's result
instead of hitting the endpoint — click-spam can't earn a 429. A poll and a manual refresh are
serialized against each other by a semaphore.
The gate (`IUsageGate`, thresholds `usage_gate_five_hour_pct`/`usage_gate_seven_day_pct`)
pauses **only** the queue's slot-fill loop (new tasks don't start) once `five_hour >=
@@ -220,7 +234,7 @@ Each CLI invocation is recorded in the `task_runs` table via `TaskRunRepository`
- Agents/settings/lists: `GetAgents`, `RefreshAgents`, `RestoreDefaultAgents`, `GetAppSettings`, `UpdateAppSettings`, `UpdateList`, `UpdateListConfig`, `GetListConfig`, `UpdateTaskAgentSettings`
- Reports/notes/prep: `GetWeekReport`, `GenerateWeekReport`, `GetDailyNotes`, `AddDailyNote`, `UpdateDailyNote`, `DeleteDailyNote`, `RunDailyPrepNow`, `ClearMyDay`, `GetLastPrepLog`, `ListPrimeSchedules`, `UpsertPrimeSchedule`, `DeletePrimeSchedule`
- Diagnostics: `GetRecentLogs` (last 30 min of buffered log records, all levels, for the Log Visualizer overlay)
- Usage: `GetUsageSnapshot() -> UsageSnapshotDto` (built by `UsageSnapshotBuilder` from `UsageState` + `IUsageGate` + `AppSettings` gate thresholds; percentages/limits/`FetchedAtUtc` null and `IsStale=true` when no snapshot has landed yet; `IsStale` also trips on a failed last poll or a snapshot older than 3× `usage_poll_interval_seconds`), `GetModelUsage(from, to) -> IReadOnlyList<ModelUsageRowDto>` (thin wrapper over `ITranscriptUsageReader.ReadAsync`), `GetTaskUsage(from, to) -> IReadOnlyList<TaskUsageRowDto>` (top consumers from `task_runs` joined to task/list, grouped per task — `Runs`/summed `TokensIn`/`TokensOut` (null token columns count as 0, never dropped), `Model` from that task's most recent run — sorted by total tokens descending, capped at 100)
- Usage: `GetUsageSnapshot() -> UsageSnapshotDto` (built by `UsageSnapshotBuilder` from `UsageState` + `IUsageGate` + `AppSettings` gate thresholds; percentages/limits/`FetchedAtUtc` null and `IsStale=true` when no snapshot has landed yet; `IsStale` also trips on a failed last poll or a snapshot older than 3× the *slower* of the two poll intervals — measured against the idle cadence so an idle worker isn't flagged stale for simply not polling), `RefreshUsage() -> UsageSnapshotDto` (manual out-of-band poll via `UsageMonitorService.RefreshNowAsync`; cooldown-guarded), `GetModelUsage(from, to) -> IReadOnlyList<ModelUsageRowDto>` (thin wrapper over `ITranscriptUsageReader.ReadAsync`), `GetTaskUsage(from, to) -> IReadOnlyList<TaskUsageRowDto>` (top consumers from `task_runs` joined to task/list, grouped per task — `Runs`/summed `TokensIn`/`TokensOut` (null token columns count as 0, never dropped), `Model` from that task's most recent run — sorted by total tokens descending, capped at 100)
**HubBroadcaster** events: `TaskStarted`, `TaskFinished`, `TaskMessage`, `WorktreeUpdated`, `TaskUpdated`, `RunCreated`, `ListUpdated`, `WorkerLog`, `PrimeFired`, `PrepStarted`, `PrepLine`, `PrepFinished`, `PlanningMergeStarted`, `PlanningSubtaskMerged`, `PlanningMergeConflict`, `PlanningMergeAborted`, `PlanningCompleted`, `RefineStarted`, `RefineFinished`, `UsageUpdated` (carries the same `UsageSnapshotDto` as `GetUsageSnapshot`; `UsageMonitorService` fires it after every poll cycle, success or failure, via the shared `UsageSnapshotBuilder`)
@@ -240,7 +254,7 @@ Loaded from `~/.todo-app/worker.config.json`:
- `poll_interval_seconds` (int, default 60)
- `zitadel.authority`, `zitadel.client_id`, `zitadel.scopes` — used by `ZitadelAuthProvider` (OIDC discovery + refresh-token flow)
- The refresh token is NOT in this file — stored encrypted via DPAPI at `~/.todo-app/online-inbox.token`
- `usage_poll_interval_seconds` (default 60, clamped to a minimum of 15 on load) — poll interval for `UsageMonitorService`
- `usage_poll_interval_active_seconds` (default 300) / `usage_poll_interval_idle_seconds` (default 900) — `UsageMonitorService`'s poll interval while at least one task is `Running` vs. while idle. Both clamped to a minimum of 60 on load; the endpoint answers 429 on tighter polling. Replaces the old single `usage_poll_interval_seconds` (silently ignored if still present in an existing config file).
Per-list config (`list_config` in DB) provides defaults for `model`, `system_prompt`, `agent_path`; tasks can override each individually. Task-generating MCP tools (`AddTask`, planning `CreateChildTask`, `SuggestImprovement`) accept an optional `model` (alias-validated via `ModelRegistry.NormalizeAlias` — `haiku`/`sonnet`/`opus`, blank = inherit) so Claude assigns the cheapest capable model at creation time; the planning/system/improvement prompts instruct it to do so (`ModelRegistry.ByCostAscending` = the cost order). Planning's `CreateChildTask` additionally accepts an optional `maxTurns` (positive int; `0`/negative rejected with `ArgumentException`, null = inherit list/global default) so the planner can raise the turn budget for a subtask it knows will run long; `SuggestImprovement`/`AddTask` do not expose it.
+14 -4
View File
@@ -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;
}
+16 -1
View File
@@ -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)
+4 -1
View File
@@ -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;
}
}
}
+134 -30
View File
@@ -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>())