Q70: provider cooldowns survive restarts
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
c6a0460035
commit
bb279ff85a
6 files changed
+101
-54
No files matched your search
@@ -1,3 +1,5 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
|
||||
/** Without a provider-given wait, parks double from this base with each failure in a row, up to the cap. */
|
||||
const BACKOFF_BASE_MS = 2 * 60_000;
|
||||
const BACKOFF_CAP_MS = 30 * 60_000;
|
||||
@@ -5,78 +7,73 @@ const BACKOFF_CAP_MS = 30 * 60_000;
|
||||
const RETRY_AFTER_MIN_MS = 30_000;
|
||||
const RETRY_AFTER_MAX_MS = 60 * 60_000;
|
||||
|
||||
/** In-memory parks for providers that reported a transient failure. */
|
||||
interface CooldownRow {
|
||||
provider_id: string;
|
||||
label: string;
|
||||
until_ms: number;
|
||||
strikes: number;
|
||||
interactive_probed_at_ms: number | null;
|
||||
}
|
||||
|
||||
/** Parks for providers that reported a transient failure, stored so a restart keeps waiting instead of hammering them. */
|
||||
export class ProviderCooldowns {
|
||||
private readonly parks = new Map<string, { until: number; label: string; interactiveProbedAt?: number }>();
|
||||
/** Transient failures in a row per provider, reset by a served request. */
|
||||
private readonly strikes = new Map<string, number>();
|
||||
constructor(private readonly db: DatabaseSync, private readonly now: () => number) {}
|
||||
|
||||
constructor(private readonly now: () => number) {}
|
||||
|
||||
/**
|
||||
* Parks a provider after it reported a transient failure (usage limit, rate
|
||||
* limit, overload), so the queue stops feeding it doomed runs every tick.
|
||||
* The park is in-memory by design: after a restart the first probe run
|
||||
* either succeeds or re-arms the cooldown at the cost of one request.
|
||||
*/
|
||||
/** Parks a provider after a transient failure, for the wait it gave or a growing backoff. */
|
||||
park(provider: { id: string; label: string }, message: string, retryAfterMs: number | null): void {
|
||||
const active = this.parks.get(provider.id);
|
||||
const wasParked = active !== undefined && active.until > this.now();
|
||||
const strikes = (this.strikes.get(provider.id) ?? 0) + 1;
|
||||
this.strikes.set(provider.id, strikes);
|
||||
const row = this.row(provider.id);
|
||||
const wasParked = row !== undefined && row.until_ms > this.now();
|
||||
const strikes = (row?.strikes ?? 0) + 1;
|
||||
const waitMs = retryAfterMs !== null
|
||||
? Math.min(Math.max(retryAfterMs, RETRY_AFTER_MIN_MS), RETRY_AFTER_MAX_MS)
|
||||
: Math.min(BACKOFF_BASE_MS * 2 ** (strikes - 1), BACKOFF_CAP_MS);
|
||||
this.parks.set(provider.id, {
|
||||
until: this.now() + waitMs,
|
||||
label: provider.label,
|
||||
// A failed probe must not unlock endless probes: re-arming the park
|
||||
// keeps the spent-probe mark until the window expires naturally.
|
||||
interactiveProbedAt: wasParked ? active?.interactiveProbedAt : undefined,
|
||||
});
|
||||
// A failed probe must not unlock endless probes: re-arming the park keeps the spent-probe mark.
|
||||
const probedAt = wasParked ? row.interactive_probed_at_ms : null;
|
||||
this.db.prepare(`
|
||||
insert into provider_cooldowns(provider_id, label, until_ms, strikes, interactive_probed_at_ms) values (?, ?, ?, ?, ?)
|
||||
on conflict(provider_id) do update set label = excluded.label, until_ms = excluded.until_ms, strikes = excluded.strikes,
|
||||
interactive_probed_at_ms = excluded.interactive_probed_at_ms
|
||||
`).run(provider.id, provider.label, this.now() + waitMs, strikes, probedAt);
|
||||
console.warn(`[task-queue] provider "${provider.label}" reported a transient failure ("${message}"); its tasks are parked for ${Math.round(waitMs / 1000)}s (failure ${strikes} in a row).`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Providers currently inside a transient-failure cooldown. Expired parks are
|
||||
* pruned (with a resume log line) so the next tick probes the provider again.
|
||||
*/
|
||||
/** Providers currently parked; expired parks end (with a log line) but keep their strike count until a request is served. */
|
||||
cooledProviderIds(): string[] {
|
||||
const rows = this.db.prepare("select provider_id, label, until_ms from provider_cooldowns where until_ms > 0").all() as Array<{ provider_id: string; label: string; until_ms: number }>;
|
||||
const active: string[] = [];
|
||||
for (const [id, park] of this.parks) {
|
||||
if (park.until > this.now()) {
|
||||
active.push(id);
|
||||
} else {
|
||||
this.parks.delete(id);
|
||||
console.info(`[task-queue] provider "${park.label}" cooldown ended; its tasks resume.`);
|
||||
for (const row of rows) {
|
||||
if (row.until_ms > this.now()) {
|
||||
active.push(row.provider_id);
|
||||
continue;
|
||||
}
|
||||
this.db.prepare("update provider_cooldowns set until_ms = 0, interactive_probed_at_ms = null where provider_id = ?").run(row.provider_id);
|
||||
console.info(`[task-queue] provider "${row.label}" cooldown ended; its tasks resume.`);
|
||||
}
|
||||
return active;
|
||||
}
|
||||
|
||||
/**
|
||||
* Cooled providers that even the interactive lane must skip: their one
|
||||
* founder-probe for this park window is already spent. A cooled provider
|
||||
* not yet probed stays interactively claimable — one human message is
|
||||
* allowed through to learn whether the outage was a blip, instead of the
|
||||
* founder sitting out the whole cooldown for a single failed background
|
||||
* request.
|
||||
*/
|
||||
/** Parked providers whose one founder-probe for this park is already spent; the interactive lane skips only these. */
|
||||
interactiveCooledProviderIds(): string[] {
|
||||
return this.cooledProviderIds().filter((id) => this.parks.get(id)?.interactiveProbedAt !== undefined);
|
||||
const active = new Set(this.cooledProviderIds());
|
||||
const probed = this.db.prepare("select provider_id from provider_cooldowns where interactive_probed_at_ms is not null").all() as Array<{ provider_id: string }>;
|
||||
return probed.map((row) => row.provider_id).filter((id) => active.has(id));
|
||||
}
|
||||
|
||||
/** Marks the cooled provider's founder-probe as spent (no-op when healthy). */
|
||||
/** Marks the parked provider's founder-probe as spent (no-op when healthy). */
|
||||
noteInteractiveProbe(providerId: string): void {
|
||||
const park = this.parks.get(providerId);
|
||||
if (park) park.interactiveProbedAt = this.now();
|
||||
this.db.prepare("update provider_cooldowns set interactive_probed_at_ms = ? where provider_id = ? and until_ms > ?").run(this.now(), providerId, this.now());
|
||||
}
|
||||
|
||||
/** A served request is proof of health: the park ends for everyone at once. */
|
||||
/** A served request is proof of health: the park and the failure streak end. */
|
||||
clear(provider: { id: string; label: string }): void {
|
||||
this.strikes.delete(provider.id);
|
||||
if (this.parks.delete(provider.id)) {
|
||||
console.info(`[task-queue] provider "${provider.label}" served a request; cooldown lifted early.`);
|
||||
}
|
||||
const row = this.row(provider.id);
|
||||
if (!row) return;
|
||||
this.db.prepare("delete from provider_cooldowns where provider_id = ?").run(provider.id);
|
||||
if (row.until_ms > this.now()) console.info(`[task-queue] provider "${provider.label}" served a request; cooldown lifted early.`);
|
||||
}
|
||||
|
||||
private row(providerId: string): CooldownRow | undefined {
|
||||
return this.db.prepare("select provider_id, label, until_ms, strikes, interactive_probed_at_ms from provider_cooldowns where provider_id = ?")
|
||||
.get(providerId) as CooldownRow | undefined;
|
||||
}
|
||||
}
|
||||
@@ -83,7 +83,7 @@ export class TaskExecutionService {
|
||||
this.audit = new AuditService(db);
|
||||
this.organization = new OrganizationService(db);
|
||||
this.chat = new ChatService(db);
|
||||
this.cooldowns = new ProviderCooldowns(now);
|
||||
this.cooldowns = new ProviderCooldowns(db, now);
|
||||
this.heartbeat = new HeartbeatScheduler(db, this.cooldowns, now, notifyDataChanged);
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user