desktop reads everything by itself: one background read scheduler (4 per store, starts spaced per account, App Store Connect X-Rate-Limit slows it below 20 % and pauses at 100 left, backoff with jitter on 429 / RESOURCE_EXHAUSTED, visible app first, one read per key), per-method TTLs over the saved reads, Settings auto-read (on) and interval (15 min, paused while hidden), Refresh all, a refresh icon and 'read 3 min ago' / 'couldn't refresh' per card, writes invalidate, status line with progress and quota; quota reaches the window through ToolResult.quota (cli/lib/quota.ts); Health top issues open with versions, sample stack (health --issue / --signature, read health_issue), Open in Play Console, Copy for AI and likely-cause hints (cli/lib/crash-hints.ts); tests/scheduler.test.ts, tests/quota-crash-hints.test.ts, boot check auto-read case
This commit is contained in:
parent
89256f1e80
commit
0fb4ce0001
46 changed files with 1531 additions and 141 deletions
77
apps/desktop/src/lib/freshness.ts
Normal file
77
apps/desktop/src/lib/freshness.ts
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
// How long a saved read stays fresh, which store lanes a read uses, and the reads the background reader does for
|
||||
// every app. Pure (tests/scheduler.test.ts).
|
||||
import type { Lane } from './scheduler';
|
||||
|
||||
const MIN = 60_000;
|
||||
|
||||
/** Time to live of a saved read per method: status and submission state 5 min, reviews 15 min, catalog / prices /
|
||||
* listings 30 min, health and analytics 6 h (the stores update them daily). */
|
||||
export const TTL: Record<string, number> = {
|
||||
'portfolio.status': 5 * MIN,
|
||||
'apple.status': 5 * MIN,
|
||||
'play.status': 5 * MIN,
|
||||
'apple.testflight_status': 5 * MIN,
|
||||
'reviews.list': 15 * MIN,
|
||||
'catalog.diff': 30 * MIN,
|
||||
'prices.current': 30 * MIN,
|
||||
'project.listings': 30 * MIN,
|
||||
'apple.ppo_status': 30 * MIN,
|
||||
validate: 30 * MIN,
|
||||
scan: 30 * MIN,
|
||||
health: 6 * 60 * MIN,
|
||||
'health.issue': 6 * 60 * MIN,
|
||||
'analytics.compare': 6 * 60 * MIN,
|
||||
'aso.audit': 6 * 60 * MIN,
|
||||
};
|
||||
export const DEFAULT_TTL = 30 * MIN;
|
||||
|
||||
export const ttlOf = (method: string): number => TTL[method] ?? DEFAULT_TTL;
|
||||
|
||||
/** True when a read saved at `at` (ISO) is older than its TTL, or there is none. */
|
||||
export function isStale(method: string, at: string | null | undefined, now = Date.now()): boolean {
|
||||
if (!at) return true;
|
||||
const t = Date.parse(at);
|
||||
return !Number.isFinite(t) || now - t >= ttlOf(method);
|
||||
}
|
||||
|
||||
/** The store lanes a read takes a slot in. */
|
||||
export function lanesOf(method: string, o: { ios: boolean; android: boolean }): Lane[] {
|
||||
if (method.startsWith('apple.')) return ['apple'];
|
||||
if (method.startsWith('play.')) return ['play'];
|
||||
if (['project.listings', 'scan', 'analytics.compare', 'aso.audit', 'reviews.drafts', 'project.declarations', 'project.renders', 'project.icon'].includes(method)) return ['local'];
|
||||
const l: Lane[] = [];
|
||||
if (o.ios) l.push('apple');
|
||||
if (o.android) l.push('play');
|
||||
return l.length ? l : ['local'];
|
||||
}
|
||||
|
||||
export interface PlannedRead { method: string; args: Record<string, unknown> }
|
||||
|
||||
/** The reads the background reader does for one app (the ones its tabs show), by what the app has. */
|
||||
export function readsFor(o: { folder: boolean; ios: boolean; android: boolean }): PlannedRead[] {
|
||||
const r: PlannedRead[] = [];
|
||||
const add = (method: string, args: Record<string, unknown> = {}) => r.push({ method, args });
|
||||
if (o.ios) { add('apple.status'); add('apple.testflight_status'); }
|
||||
if (o.android) add('play.status');
|
||||
if (o.ios || o.android) { add('reviews.list', { days: 30 }); add('health'); }
|
||||
if (o.folder) {
|
||||
add('scan'); add('project.listings');
|
||||
if (o.ios || o.android) { add('catalog.diff'); add('prices.current'); add('validate'); }
|
||||
add('analytics.compare'); add('aso.audit');
|
||||
}
|
||||
if (o.ios) add('apple.ppo_status');
|
||||
return r;
|
||||
}
|
||||
|
||||
/** Key of one read in the scheduler and the read bus. */
|
||||
export const readKey = (target: string, method: string, args: Record<string, unknown> = {}): string => `${target}\u001f${method}\u001f${JSON.stringify(args)}`;
|
||||
|
||||
/** "429", "RESOURCE_EXHAUSTED", "rate limit": the store refused for its quota (not the read itself). */
|
||||
export const looksThrottled = (text: string | undefined): boolean => !!text && /\b429\b|RESOURCE_EXHAUSTED|rate.?limit|quota exceeded/i.test(text);
|
||||
|
||||
/** The one-line reason of a failed read, for "couldn't refresh: <reason>". */
|
||||
export function oneLine(text: string, max = 140): string {
|
||||
const l = text.trim().split('\n').map(s => s.trim()).filter(Boolean).at(-1) ?? '';
|
||||
const s = l.replace(/^\S+ \[[^\]]+\] /, '');
|
||||
return s.length > max ? `${s.slice(0, max - 1)}…` : s;
|
||||
}
|
||||
60
apps/desktop/src/lib/issues.ts
Normal file
60
apps/desktop/src/lib/issues.ts
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
// One crash / ANR issue in detail (Health tab): the Play error issue or App Store diagnostic signature, its sample
|
||||
// stack (`health.issue`), and the compact "Copy for AI" markdown. Pure (tests/model.test.ts).
|
||||
import { type CrashHint, crashHint } from '../../../../cli/lib/crash-hints';
|
||||
import type { IssueSample, PlayIssue, AppleSignature } from './model';
|
||||
|
||||
/** Likely cause of a known signature, from the issue text and its sample stack. */
|
||||
export const hintFor = (v: IssueView, sample: IssueSample | null): CrashHint | null => crashHint(v.title, v.where, ...(sample?.stack.slice(0, 30) ?? []));
|
||||
|
||||
export interface IssueView {
|
||||
store: 'play' | 'apple';
|
||||
title: string;
|
||||
/** Exception type / diagnostic type. */
|
||||
kind: string;
|
||||
where?: string;
|
||||
reports?: number;
|
||||
users?: number;
|
||||
weight?: number;
|
||||
versions?: { first?: string; last?: string };
|
||||
os?: { first?: string; last?: string };
|
||||
lastSeen?: string;
|
||||
build?: string;
|
||||
url?: string;
|
||||
/** Args of the `health.issue` read for its sample stack; null when there is no id to ask for. */
|
||||
sampleArgs: Record<string, string> | null;
|
||||
}
|
||||
|
||||
export function playIssueView(i: PlayIssue): IssueView {
|
||||
return {
|
||||
store: 'play', title: i.cause || i.location || '?', kind: i.type === 'ANR' ? 'ANR' : i.type, where: i.location || undefined, reports: i.reports, users: i.users,
|
||||
versions: { first: i.firstVersion, last: i.lastVersion }, os: i.firstOs || i.lastOs ? { first: i.firstOs, last: i.lastOs } : undefined, lastSeen: i.lastSeen, url: i.uri,
|
||||
sampleArgs: i.id ? { issue: i.id } : null,
|
||||
};
|
||||
}
|
||||
|
||||
export function appleSignatureView(s: AppleSignature): IssueView {
|
||||
return { store: 'apple', title: s.signature || s.type, kind: s.type, weight: s.weight, build: s.build, sampleArgs: s.id ? { signature: s.id } : null };
|
||||
}
|
||||
|
||||
/** The top frames worth pasting (the full sample stays in the window). */
|
||||
export const TOP_FRAMES = 15;
|
||||
|
||||
/** Compact markdown for an AI agent: title, type, where, counts, versions, top frames. */
|
||||
export function issueMarkdown(app: string, v: IssueView, sample: IssueSample | null): string {
|
||||
const out = [`## ${v.store === 'play' ? 'Google Play' : 'App Store'} ${v.kind}: ${v.title}`, '', `- App: ${app}`];
|
||||
if (v.where && v.where !== v.title) out.push(`- Location: ${v.where}`);
|
||||
if (v.reports !== undefined) out.push(`- Reports: ${v.reports}${v.users !== undefined ? `, users: ${v.users}` : ''}`);
|
||||
if (v.weight !== undefined) out.push(`- Share of reports: ${Math.round(v.weight * 1000) / 10} %`);
|
||||
if (v.versions?.first || v.versions?.last) out.push(`- versionCode: ${v.versions.first ?? '?'} to ${v.versions.last ?? '?'}`);
|
||||
if (v.build) out.push(`- Build: ${v.build}`);
|
||||
if (v.os) out.push(`- Android API: ${v.os.first ?? '?'} to ${v.os.last ?? '?'}`);
|
||||
if (v.lastSeen) out.push(`- Last report: ${v.lastSeen}`);
|
||||
const hint = hintFor(v, sample);
|
||||
if (hint) out.push(`- Likely cause: ${hint.cause}`, `- Suggested fix: ${hint.fix}`);
|
||||
if (sample) {
|
||||
const meta = [sample.version && `version ${sample.version}`, sample.device, sample.os].filter(Boolean).join(', ');
|
||||
out.push('', `Sample report${meta ? ` (${meta})` : ''}, top frames:`, '', '```', ...sample.stack.slice(0, TOP_FRAMES), ...(sample.stack.length > TOP_FRAMES ? [`... ${sample.stack.length - TOP_FRAMES} more frames`] : []), '```');
|
||||
}
|
||||
if (v.url) out.push('', `Console: ${v.url}`);
|
||||
return out.join('\n') + '\n';
|
||||
}
|
||||
|
|
@ -17,11 +17,17 @@ export interface PlayHealth {
|
|||
crash: MetricRow[];
|
||||
anr: MetricRow[];
|
||||
devices?: { model: string; crash?: number; anr?: number; users?: number }[];
|
||||
issues: { type: string; cause: string; location: string; reports: number; users: number; lastVersion?: string }[];
|
||||
issues: PlayIssue[];
|
||||
issuesError?: string;
|
||||
}
|
||||
/** One Play error issue (errorIssues:search; id / versions / API levels / last report when the API sent them). */
|
||||
export interface PlayIssue { type: string; cause: string; location: string; reports: number; users: number; lastVersion?: string; uri?: string; id?: string; firstVersion?: string; firstOs?: string; lastOs?: string; lastSeen?: string }
|
||||
export interface AppleSignature { build: string; type: string; signature: string; weight: number; id?: string }
|
||||
/** `health --issue / --signature --json`: one sample report's stack, top frame first. */
|
||||
export interface IssueSample { store: 'play' | 'apple'; time?: string; version?: string; os?: string; device?: string; type?: string; stack: string[] }
|
||||
export interface AppleHealth {
|
||||
metrics: { category: string; metric: string; unit: string; version: string; value: number; goal?: string }[];
|
||||
signatures: { build: string; type: string; signature: string; weight: number }[];
|
||||
signatures: AppleSignature[];
|
||||
betaCrashes: { build: string; count: number }[];
|
||||
}
|
||||
export type StoreReadError = import('../../../../cli/lib/health').StoreReadError;
|
||||
|
|
|
|||
|
|
@ -55,7 +55,8 @@ export function normalizeSettings(v: unknown): Settings {
|
|||
if (!isRec(v)) return d;
|
||||
const r = isRec(v.refresh) ? v.refresh : {}, c = isRec(v.cache) ? v.cache : {}, n = isRec(v.notify) ? v.notify : {};
|
||||
return {
|
||||
refresh: { autoMinutes: pick(r.autoMinutes, d.refresh.autoMinutes, isNum), onLaunch: pick(r.onLaunch, d.refresh.onLaunch, isBool), parallel: pick(r.parallel, d.refresh.parallel, isNum), icons: pick(r.icons, d.refresh.icons, isBool) },
|
||||
refresh: { autoMinutes: pick(r.autoMinutes, d.refresh.autoMinutes, isNum), onLaunch: pick(r.onLaunch, d.refresh.onLaunch, isBool), parallel: pick(r.parallel, d.refresh.parallel, isNum), icons: pick(r.icons, d.refresh.icons, isBool),
|
||||
autoRead: pick(r.autoRead, d.refresh.autoRead, isBool), everyMinutes: pick(r.everyMinutes, d.refresh.everyMinutes, (x: unknown) => [0, 5, 15, 30, 60].includes(x as number)) },
|
||||
cache: { keepDays: pick(c.keepDays, d.cache.keepDays, isNum) },
|
||||
notify: { reviewState: pick(n.reviewState, d.notify.reviewState, isBool), lowReviews: pick(n.lowReviews, d.notify.lowReviews, isBool), crash: pick(n.crash, d.notify.crash, isBool) },
|
||||
channel: v.channel === 'beta' ? 'beta' : 'stable',
|
||||
|
|
|
|||
243
apps/desktop/src/lib/scheduler.ts
Normal file
243
apps/desktop/src/lib/scheduler.ts
Normal file
|
|
@ -0,0 +1,243 @@
|
|||
// The background read scheduler: every store read of the window goes through one queue, so reading "everything for
|
||||
// every app" stays far inside the stores' limits. Pure logic with an injectable clock (tests/scheduler.test.ts).
|
||||
//
|
||||
// - Lanes: 'apple' (App Store Connect), 'play' (Google Play Developer + Reporting API), 'local' (offline project
|
||||
// reads). A task takes one slot in each of its lanes; each lane has its own concurrency limit (default 4).
|
||||
// - Credentials: starts on one key are spaced (default 250 ms apart). App Store Connect sends X-Rate-Limit
|
||||
// (user-hour-lim / user-hour-rem, about 3600 per key per hour): below 20 % left, the spacing spreads the rest over the
|
||||
// hour; at the reserve (default 100, kept for the owner's own commands) the key pauses for 10 minutes.
|
||||
// - 429 / RESOURCE_EXHAUSTED: the task waits and runs again with exponential backoff and jitter (and its key waits too),
|
||||
// at most `maxAttempts` times.
|
||||
// - Priority: 0 the visible app and tab, 1 portfolio status, 2 the rest; FIFO within a priority.
|
||||
// - Dedupe: one in-flight (or queued) read per key; enqueueing it again returns the same promise (and may raise its
|
||||
// priority).
|
||||
|
||||
export type Lane = 'apple' | 'play' | 'local';
|
||||
export type Priority = 0 | 1 | 2;
|
||||
|
||||
export interface Clock {
|
||||
now(): number;
|
||||
setTimeout(fn: () => void, ms: number): unknown;
|
||||
clearTimeout(h: unknown): void;
|
||||
random(): number;
|
||||
}
|
||||
|
||||
export const realClock: Clock = {
|
||||
now: () => Date.now(),
|
||||
setTimeout: (fn, ms) => setTimeout(fn, ms),
|
||||
clearTimeout: h => clearTimeout(h as ReturnType<typeof setTimeout>),
|
||||
random: () => Math.random(),
|
||||
};
|
||||
|
||||
export interface QuotaSeen { limit?: number; remaining?: number; throttled?: boolean }
|
||||
|
||||
/** What one finished run says about the store's limits. */
|
||||
export interface Outcome {
|
||||
/** 429 / RESOURCE_EXHAUSTED: run it again later. */
|
||||
throttled?: boolean;
|
||||
/** Quota per credential (App Store Connect X-Rate-Limit). */
|
||||
quota?: { credential: string; limit: number; remaining: number };
|
||||
}
|
||||
|
||||
export interface ReadTask<R> {
|
||||
key: string;
|
||||
lanes: Lane[];
|
||||
/** Keys of the store credentials it uses ("apple:<account>", "play:<account>"). */
|
||||
credentials: string[];
|
||||
priority: Priority;
|
||||
run: () => Promise<R>;
|
||||
}
|
||||
|
||||
export interface SchedulerOptions {
|
||||
limits?: Partial<Record<Lane, number>>;
|
||||
/** Minimum gap between two starts on one credential. */
|
||||
spacingMs?: number;
|
||||
/** Requests per hour kept free on an App Store Connect key for the owner's own commands. */
|
||||
reserve?: number;
|
||||
/** Below this share of the hourly limit left, starts are spread over the rest of the hour. */
|
||||
slowBelow?: number;
|
||||
/** How long a key pauses once it is at the reserve. */
|
||||
pauseMs?: number;
|
||||
backoffBaseMs?: number;
|
||||
backoffMaxMs?: number;
|
||||
maxAttempts?: number;
|
||||
clock?: Clock;
|
||||
/** Reads a result (or the error) for throttling and quota. */
|
||||
classify?: (result: unknown, error: unknown, task: ReadTask<unknown>) => Outcome;
|
||||
}
|
||||
|
||||
export interface SchedulerStats {
|
||||
queued: number;
|
||||
running: number;
|
||||
/** Finished in this round (since the queue was last empty). */
|
||||
done: number;
|
||||
/** done + running + queued of this round. */
|
||||
total: number;
|
||||
failed: number;
|
||||
/** The lowest App Store Connect quota seen per credential this hour. */
|
||||
quota: Record<string, { limit: number; remaining: number; at: number }>;
|
||||
/** Credentials waiting on a backoff or quota pause, with the time they may start again. */
|
||||
paused: Record<string, number>;
|
||||
}
|
||||
|
||||
interface Item<R> {
|
||||
task: ReadTask<R>;
|
||||
seq: number;
|
||||
attempt: number;
|
||||
notBefore: number;
|
||||
resolve: (r: R) => void;
|
||||
reject: (e: unknown) => void;
|
||||
promise: Promise<R>;
|
||||
}
|
||||
|
||||
const HOUR = 3_600_000;
|
||||
|
||||
export class ReadScheduler {
|
||||
private o: Required<Omit<SchedulerOptions, 'limits' | 'clock' | 'classify'>> & { limits: Record<Lane, number> };
|
||||
private clock: Clock;
|
||||
private classify: (result: unknown, error: unknown, task: ReadTask<unknown>) => Outcome;
|
||||
private queue: Item<unknown>[] = [];
|
||||
private running = new Map<string, Item<unknown>>();
|
||||
private busy: Record<Lane, number> = { apple: 0, play: 0, local: 0 };
|
||||
private nextAt = new Map<string, number>();
|
||||
private quotas = new Map<string, { limit: number; remaining: number; at: number }>();
|
||||
private seq = 0;
|
||||
private done = 0;
|
||||
private failed = 0;
|
||||
private timer: unknown = null;
|
||||
private timerAt = 0;
|
||||
private listeners = new Set<() => void>();
|
||||
|
||||
constructor(o: SchedulerOptions = {}) {
|
||||
this.o = {
|
||||
limits: { apple: 4, play: 4, local: 2, ...o.limits },
|
||||
spacingMs: o.spacingMs ?? 250,
|
||||
reserve: o.reserve ?? 100,
|
||||
slowBelow: o.slowBelow ?? 0.2,
|
||||
pauseMs: o.pauseMs ?? 10 * 60_000,
|
||||
backoffBaseMs: o.backoffBaseMs ?? 5_000,
|
||||
backoffMaxMs: o.backoffMaxMs ?? 5 * 60_000,
|
||||
maxAttempts: o.maxAttempts ?? 4,
|
||||
};
|
||||
this.clock = o.clock ?? realClock;
|
||||
this.classify = o.classify ?? (() => ({}));
|
||||
}
|
||||
|
||||
setLimits(limits: Partial<Record<Lane, number>>): void { this.o.limits = { ...this.o.limits, ...limits }; this.pump(); }
|
||||
|
||||
subscribe(fn: () => void): () => void { this.listeners.add(fn); return () => { this.listeners.delete(fn); }; }
|
||||
|
||||
has(key: string): boolean { return this.running.has(key) || this.queue.some(i => i.task.key === key); }
|
||||
|
||||
enqueue<R>(task: ReadTask<R>): Promise<R> {
|
||||
const run = this.running.get(task.key);
|
||||
if (run) return run.promise as Promise<R>;
|
||||
const queued = this.queue.find(i => i.task.key === task.key);
|
||||
if (queued) {
|
||||
if (task.priority < queued.task.priority) { queued.task = { ...queued.task, priority: task.priority }; this.pump(); }
|
||||
return queued.promise as Promise<R>;
|
||||
}
|
||||
let resolve!: (r: R) => void, reject!: (e: unknown) => void;
|
||||
const promise = new Promise<R>((a, b) => { resolve = a; reject = b; });
|
||||
this.queue.push({ task, seq: ++this.seq, attempt: 0, notBefore: 0, resolve, reject, promise } as unknown as Item<unknown>);
|
||||
this.pump();
|
||||
this.emit();
|
||||
return promise;
|
||||
}
|
||||
|
||||
/** Drops queued tasks (not the running ones); their promises reject with "cancelled". */
|
||||
clear(filter: (key: string) => boolean = () => true): void {
|
||||
const drop = this.queue.filter(i => filter(i.task.key));
|
||||
this.queue = this.queue.filter(i => !filter(i.task.key));
|
||||
for (const i of drop) i.reject(new Error('cancelled'));
|
||||
this.pump();
|
||||
this.emit();
|
||||
}
|
||||
|
||||
stats(): SchedulerStats {
|
||||
const now = this.clock.now();
|
||||
const paused: Record<string, number> = {};
|
||||
for (const [c, at] of this.nextAt) if (at - now > this.o.spacingMs) paused[c] = at;
|
||||
const quota: SchedulerStats['quota'] = {};
|
||||
for (const [c, q] of this.quotas) if (now - q.at < HOUR) quota[c] = q;
|
||||
return { queued: this.queue.length, running: this.running.size, done: this.done, total: this.done + this.running.size + this.queue.length, failed: this.failed, quota, paused };
|
||||
}
|
||||
|
||||
/** Earliest time a credential may start again. */
|
||||
private gap(cred: string, now: number): number {
|
||||
const q = this.quotas.get(cred);
|
||||
if (!q || now - q.at >= HOUR) return this.o.spacingMs;
|
||||
if (q.remaining <= this.o.reserve) return this.o.pauseMs;
|
||||
if (q.remaining < q.limit * this.o.slowBelow) return Math.max(this.o.spacingMs, Math.ceil(HOUR / Math.max(1, q.remaining - this.o.reserve)));
|
||||
return this.o.spacingMs;
|
||||
}
|
||||
|
||||
private ready(i: Item<unknown>, now: number): number {
|
||||
// 0 = can start now; otherwise the time it can (Infinity: waits on a lane slot)
|
||||
let at = i.notBefore;
|
||||
for (const c of i.task.credentials) at = Math.max(at, this.nextAt.get(c) ?? 0);
|
||||
if (i.task.lanes.some(l => this.busy[l] >= Math.max(1, this.o.limits[l]))) return Infinity;
|
||||
return at <= now ? 0 : at;
|
||||
}
|
||||
|
||||
private pump(): void {
|
||||
const now = this.clock.now();
|
||||
let wake = Infinity;
|
||||
for (;;) {
|
||||
const order = [...this.queue].sort((a, b) => a.task.priority - b.task.priority || a.seq - b.seq);
|
||||
let started = false;
|
||||
for (const i of order) {
|
||||
const r = this.ready(i, now);
|
||||
if (r === 0) { this.start(i, now); started = true; break; }
|
||||
if (r !== Infinity) wake = Math.min(wake, r);
|
||||
}
|
||||
if (!started) break;
|
||||
}
|
||||
if (this.timer !== null && (wake === Infinity || wake !== this.timerAt)) { this.clock.clearTimeout(this.timer); this.timer = null; }
|
||||
if (wake !== Infinity && this.timer === null) {
|
||||
this.timerAt = wake;
|
||||
this.timer = this.clock.setTimeout(() => { this.timer = null; this.pump(); }, Math.max(0, wake - now));
|
||||
}
|
||||
}
|
||||
|
||||
private start(i: Item<unknown>, now: number): void {
|
||||
this.queue = this.queue.filter(x => x !== i);
|
||||
this.running.set(i.task.key, i);
|
||||
for (const l of i.task.lanes) this.busy[l]++;
|
||||
for (const c of i.task.credentials) this.nextAt.set(c, now + this.gap(c, now));
|
||||
i.attempt++;
|
||||
this.emit();
|
||||
i.task.run().then(r => this.finish(i, r, undefined, true), e => this.finish(i, undefined, e, false));
|
||||
}
|
||||
|
||||
private finish(i: Item<unknown>, result: unknown, error: unknown, ok: boolean): void {
|
||||
const now = this.clock.now();
|
||||
this.running.delete(i.task.key);
|
||||
for (const l of i.task.lanes) this.busy[l]--;
|
||||
let out: Outcome = {};
|
||||
try { out = this.classify(result, error, i.task); } catch { /* a classifier bug never stalls the queue */ }
|
||||
if (out.quota) {
|
||||
const prev = this.quotas.get(out.quota.credential);
|
||||
// the hour is the key's: keep the lowest count seen in it
|
||||
const q = prev && now - prev.at < HOUR && prev.remaining < out.quota.remaining ? prev : { limit: out.quota.limit, remaining: out.quota.remaining, at: now };
|
||||
this.quotas.set(out.quota.credential, q);
|
||||
this.nextAt.set(out.quota.credential, Math.max(this.nextAt.get(out.quota.credential) ?? 0, now + this.gap(out.quota.credential, now)));
|
||||
}
|
||||
if (out.throttled && i.attempt < this.o.maxAttempts) {
|
||||
const exp = Math.min(this.o.backoffMaxMs, this.o.backoffBaseMs * 2 ** (i.attempt - 1));
|
||||
const delay = Math.round(exp / 2 + this.clock.random() * exp / 2); // "equal jitter"
|
||||
i.notBefore = now + delay;
|
||||
for (const c of i.task.credentials) this.nextAt.set(c, Math.max(this.nextAt.get(c) ?? 0, now + delay));
|
||||
this.queue.push(i);
|
||||
} else {
|
||||
this.done++;
|
||||
if (!ok || out.throttled) this.failed++;
|
||||
if (ok) i.resolve(result); else i.reject(error);
|
||||
}
|
||||
if (!this.queue.length && !this.running.size) { this.done = 0; this.failed = 0; }
|
||||
this.pump();
|
||||
this.emit();
|
||||
}
|
||||
|
||||
private emit(): void { for (const f of this.listeners) { try { f(); } catch { /* listeners never break the queue */ } } }
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue