|
|
@@ -10,9 +10,16 @@
|
|
|
* Features:
|
|
|
* - Batches input in groups of ≤64 (configurable via QMD_EMBED_BATCH_SIZE)
|
|
|
* - Retries 429 / 503 with exponential backoff (1s, 4s, 16s)
|
|
|
+ * - 429 gets its own, larger retry budget and honours `Retry-After`
|
|
|
+ * (header or `"Retry after 29s"` body text) — a shared gateway bucket
|
|
|
+ * refilling in 30s must not exhaust a 1s/4s/16s schedule (i-yghj098h)
|
|
|
+ * - Bulk lane (`embedBatch` with >1 input): a 429 pauses the WHOLE worker
|
|
|
+ * pool for the advertised cooldown and halves in-flight concurrency
|
|
|
+ * (AIMD), so reindex traffic yields the bucket to interactive callers
|
|
|
+ * instead of contending head-on with them
|
|
|
* - 4xx (non-429) → no retry, count as failure
|
|
|
- * - Circuit breaker: >50% failures in a 60s window → OPEN for 5 min,
|
|
|
- * callers receive failures; local fallback is forbidden
|
|
|
+ * - Circuit breaker: >50% failures in a 60s window → OPEN for 5 min;
|
|
|
+ * callers receive failures and no model fallback is selected
|
|
|
* - Per-call timeout via AbortSignal (default QMD_EMBED_TIMEOUT_MS=30000)
|
|
|
* - Healthcheck via `GET /health` if available, else a probe embed call
|
|
|
*/
|
|
|
@@ -57,6 +64,31 @@ export const DEFAULT_TIMEOUT_MS = 30_000;
|
|
|
*/
|
|
|
export const RETRY_BACKOFFS_MS: readonly number[] = [1_000, 4_000, 16_000];
|
|
|
|
|
|
+/**
|
|
|
+ * Rate-limit (429) retry budget, separate from `RETRY_BACKOFFS_MS`.
|
|
|
+ *
|
|
|
+ * A 429 is backpressure, not a fault: the shared ai.mm.mk gateway answers
|
|
|
+ * `{"detail":"Rate limit exceeded (tokens). Retry after 29s."}` with observed
|
|
|
+ * retry-after values of 7s/29s/30s, while the generic schedule waits at most
|
|
|
+ * 1+4+16 = 21s in total. Bulk reindex runs therefore burned all three attempts
|
|
|
+ * inside one bucket refill and reported the chunk as failed (i-yghj098h).
|
|
|
+ * 429s get their own attempt count and always wait at least as long as the
|
|
|
+ * server asked for, capped by `RATE_LIMIT_MAX_BACKOFF_MS`.
|
|
|
+ */
|
|
|
+export const DEFAULT_RATE_LIMIT_RETRIES = 5;
|
|
|
+
|
|
|
+/** Upper bound on a single rate-limit wait, so a bogus Retry-After can't hang a run. */
|
|
|
+export const RATE_LIMIT_MAX_BACKOFF_MS = 60_000;
|
|
|
+
|
|
|
+/** Fallback wait when a 429 carries no parseable Retry-After (doubles per attempt). */
|
|
|
+export const RATE_LIMIT_BASE_BACKOFF_MS = 5_000;
|
|
|
+
|
|
|
+/**
|
|
|
+ * Consecutive successful bulk requests before the lane additively recovers one
|
|
|
+ * unit of concurrency after a 429-triggered halving (AIMD).
|
|
|
+ */
|
|
|
+export const LANE_RECOVERY_STREAK = 8;
|
|
|
+
|
|
|
/**
|
|
|
* Circuit breaker — flips OPEN when error rate exceeds threshold within
|
|
|
* window. While OPEN, every call fails fast so the caller can fall back.
|
|
|
@@ -97,6 +129,12 @@ export type OpenAIProviderConfig = {
|
|
|
fetchImpl?: typeof fetch;
|
|
|
/** Custom retry schedule (for testing). Defaults to RETRY_BACKOFFS_MS. */
|
|
|
retryBackoffsMs?: readonly number[];
|
|
|
+ /**
|
|
|
+ * Extra retry attempts granted to 429 responses on top of the generic
|
|
|
+ * schedule (default DEFAULT_RATE_LIMIT_RETRIES = 5, env
|
|
|
+ * `QMD_EMBED_RATE_LIMIT_RETRIES`). Rate limits are backpressure, not faults.
|
|
|
+ */
|
|
|
+ rateLimitRetries?: number;
|
|
|
/** Custom sleep impl (for testing). Defaults to setTimeout. */
|
|
|
sleep?: (ms: number) => Promise<void>;
|
|
|
/** Custom clock (for testing). Defaults to Date.now. */
|
|
|
@@ -122,12 +160,27 @@ export type OpenAIEmbeddingsResponse = {
|
|
|
*/
|
|
|
export type CircuitState = "closed" | "open" | "half-open";
|
|
|
|
|
|
+/**
|
|
|
+ * Traffic lane for a single `embedBatch` call. `bulk` = reindex traffic
|
|
|
+ * (multi-input batch), which self-throttles via `BulkLaneGate` and is
|
|
|
+ * attributed separately in `X-AI-Caller`. `interactive` = query-time embeds,
|
|
|
+ * which are never made to wait behind a bulk cooldown.
|
|
|
+ */
|
|
|
+export type EmbedLane = "interactive" | "bulk";
|
|
|
+
|
|
|
// ─────────────────────────── Helpers ─────────────────────────────────────────
|
|
|
|
|
|
function defaultSleep(ms: number): Promise<void> {
|
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
|
}
|
|
|
|
|
|
+/** Parse a non-negative integer env value; `undefined` when unset/invalid. */
|
|
|
+function parseNonNegativeInt(raw: string | undefined): number | undefined {
|
|
|
+ if (raw == null || raw.trim() === "") return undefined;
|
|
|
+ const n = Number.parseInt(raw, 10);
|
|
|
+ return Number.isFinite(n) && n >= 0 ? n : undefined;
|
|
|
+}
|
|
|
+
|
|
|
/**
|
|
|
* Build the advisory `X-AI-Caller` attribution header value (Oivo ai.mm.mk
|
|
|
* "Observability-Driven Fleet Self-Improvement" rollout, Thread 1). Format
|
|
|
@@ -208,6 +261,51 @@ export function isRetryableStatus(status: number): boolean {
|
|
|
return status === 429 || status === 503;
|
|
|
}
|
|
|
|
|
|
+/**
|
|
|
+ * Extract the server-advertised cooldown from a rate-limited response.
|
|
|
+ *
|
|
|
+ * Two sources, in priority order:
|
|
|
+ * 1. the standard `Retry-After` header — delta-seconds or an HTTP-date;
|
|
|
+ * 2. the ai.mm.mk body text, which states the cooldown in prose only:
|
|
|
+ * `{"detail":"Rate limit exceeded (tokens). Retry after 29s.", ...}`.
|
|
|
+ *
|
|
|
+ * Returns `undefined` when neither source yields a sane positive duration, so
|
|
|
+ * the caller falls back to its own schedule. Values are clamped to
|
|
|
+ * `RATE_LIMIT_MAX_BACKOFF_MS`.
|
|
|
+ */
|
|
|
+export function parseRetryAfterMs(
|
|
|
+ headerValue: string | null | undefined,
|
|
|
+ bodyPreview?: string,
|
|
|
+ now: () => number = Date.now,
|
|
|
+): number | undefined {
|
|
|
+ const clamp = (ms: number): number | undefined =>
|
|
|
+ Number.isFinite(ms) && ms > 0 ? Math.min(ms, RATE_LIMIT_MAX_BACKOFF_MS) : undefined;
|
|
|
+
|
|
|
+ const header = headerValue?.trim();
|
|
|
+ if (header) {
|
|
|
+ if (/^\d+(\.\d+)?$/.test(header)) {
|
|
|
+ const fromSeconds = clamp(Number(header) * 1000);
|
|
|
+ if (fromSeconds !== undefined) return fromSeconds;
|
|
|
+ }
|
|
|
+ const asDate = Date.parse(header);
|
|
|
+ if (!Number.isNaN(asDate)) {
|
|
|
+ const fromDate = clamp(asDate - now());
|
|
|
+ if (fromDate !== undefined) return fromDate;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (bodyPreview) {
|
|
|
+ const m = /retry\s+after\s+(\d+(?:\.\d+)?)\s*(ms|s|seconds?)?/i.exec(bodyPreview);
|
|
|
+ if (m) {
|
|
|
+ const value = Number(m[1]);
|
|
|
+ const unit = (m[2] ?? "s").toLowerCase();
|
|
|
+ return clamp(unit === "ms" ? value : value * 1000);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return undefined;
|
|
|
+}
|
|
|
+
|
|
|
/**
|
|
|
* Chunk an array into pieces of ≤ size each. `size` MUST be ≥ 1.
|
|
|
*/
|
|
|
@@ -335,7 +433,6 @@ export class CircuitBreaker {
|
|
|
|
|
|
/**
|
|
|
* Raised when the circuit breaker is OPEN and a call is short-circuited.
|
|
|
- * Callers (e.g. fallback wrapper) can catch this to switch to local provider.
|
|
|
*/
|
|
|
export class CircuitOpenError extends Error {
|
|
|
constructor(message = "OpenAIEmbeddingsProvider circuit is OPEN") {
|
|
|
@@ -350,11 +447,115 @@ export class CircuitOpenError extends Error {
|
|
|
export class HttpError extends Error {
|
|
|
readonly status: number;
|
|
|
readonly bodyPreview: string;
|
|
|
- constructor(status: number, bodyPreview: string) {
|
|
|
+ /** Server-advertised cooldown for 429s, when the response stated one. */
|
|
|
+ readonly retryAfterMs?: number;
|
|
|
+ constructor(status: number, bodyPreview: string, retryAfterMs?: number) {
|
|
|
super(`HTTP ${status}: ${bodyPreview.slice(0, 200)}`);
|
|
|
this.name = "HttpError";
|
|
|
this.status = status;
|
|
|
this.bodyPreview = bodyPreview.slice(0, 1024);
|
|
|
+ this.retryAfterMs = retryAfterMs;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// ─────────────────────────── Bulk lane ───────────────────────────────────────
|
|
|
+
|
|
|
+/**
|
|
|
+ * Client-side bulk lane for reindex traffic (i-yghj098h).
|
|
|
+ *
|
|
|
+ * qmd's bulk embedding shares the interactive ai.mm.mk token bucket. Without a
|
|
|
+ * server-side per-caller budget, the only way bulk traffic can stop contending
|
|
|
+ * head-on with interactive callers is to police itself:
|
|
|
+ *
|
|
|
+ * - a 429 on ANY worker pauses the ENTIRE pool for the advertised cooldown
|
|
|
+ * (one shared promise — concurrent 429s coalesce instead of stacking N
|
|
|
+ * cooldowns), so the bucket refills for interactive callers rather than
|
|
|
+ * being re-drained by the remaining workers;
|
|
|
+ * - the in-flight cap halves on each cooldown (floor 1) and recovers one unit
|
|
|
+ * per `LANE_RECOVERY_STREAK` successes — classic AIMD, so a run settles at
|
|
|
+ * whatever share the bucket actually has spare.
|
|
|
+ *
|
|
|
+ * Deliberately clock-free: cooldowns are modelled as a promise produced by the
|
|
|
+ * injected `sleep`, so tests drive them with a fake sleep and no fake clock.
|
|
|
+ */
|
|
|
+export class BulkLaneGate {
|
|
|
+ private readonly maxPermits: number;
|
|
|
+ private readonly sleep: (ms: number) => Promise<void>;
|
|
|
+ private permits: number;
|
|
|
+ private inFlight = 0;
|
|
|
+ private okStreak = 0;
|
|
|
+ private cooldown: Promise<void> | null = null;
|
|
|
+ private waiters: (() => void)[] = [];
|
|
|
+
|
|
|
+ constructor(maxPermits: number, sleep: (ms: number) => Promise<void>) {
|
|
|
+ this.maxPermits = Math.max(1, maxPermits);
|
|
|
+ this.permits = this.maxPermits;
|
|
|
+ this.sleep = sleep;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Current in-flight cap — exported state for tests/diagnostics. */
|
|
|
+ get permitCount(): number {
|
|
|
+ return this.permits;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** True while the lane is serving a rate-limit cooldown. */
|
|
|
+ get isCoolingDown(): boolean {
|
|
|
+ return this.cooldown !== null;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Take a slot, waiting out any cooldown and respecting the current cap. */
|
|
|
+ async acquire(): Promise<void> {
|
|
|
+ while (this.cooldown !== null || this.inFlight >= this.permits) {
|
|
|
+ if (this.cooldown !== null) {
|
|
|
+ await this.cooldown;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ await new Promise<void>((resolve) => this.waiters.push(resolve));
|
|
|
+ }
|
|
|
+ this.inFlight++;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Return a slot. Always call from a `finally`. */
|
|
|
+ release(): void {
|
|
|
+ if (this.inFlight > 0) this.inFlight--;
|
|
|
+ this.wakeOne();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Enter (or join) a cooldown of `waitMs` and halve the in-flight cap.
|
|
|
+ * Returns the shared cooldown promise — the caller awaits it INSTEAD of
|
|
|
+ * sleeping itself, so a burst of 429s costs one cooldown, not one each.
|
|
|
+ */
|
|
|
+ penalize(waitMs: number): Promise<void> {
|
|
|
+ if (this.cooldown !== null) return this.cooldown;
|
|
|
+ this.okStreak = 0;
|
|
|
+ this.permits = Math.max(1, Math.floor(this.permits / 2));
|
|
|
+ const cooldown = this.sleep(waitMs).then(() => {
|
|
|
+ if (this.cooldown === cooldown) this.cooldown = null;
|
|
|
+ this.wakeAll();
|
|
|
+ });
|
|
|
+ this.cooldown = cooldown;
|
|
|
+ return cooldown;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Record a successful bulk request; recovers one permit per success streak. */
|
|
|
+ noteSuccess(): void {
|
|
|
+ if (this.permits >= this.maxPermits) return;
|
|
|
+ if (++this.okStreak >= LANE_RECOVERY_STREAK) {
|
|
|
+ this.okStreak = 0;
|
|
|
+ this.permits++;
|
|
|
+ this.wakeOne();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private wakeOne(): void {
|
|
|
+ this.waiters.shift()?.();
|
|
|
+ }
|
|
|
+
|
|
|
+ private wakeAll(): void {
|
|
|
+ const pending = this.waiters;
|
|
|
+ this.waiters = [];
|
|
|
+ for (const resolve of pending) resolve();
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -372,12 +573,15 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
private readonly timeoutMs: number;
|
|
|
private readonly fetchImpl: typeof fetch;
|
|
|
private readonly retryBackoffsMs: readonly number[];
|
|
|
+ private readonly rateLimitRetries: number;
|
|
|
private readonly sleep: (ms: number) => Promise<void>;
|
|
|
private readonly now: () => number;
|
|
|
|
|
|
private dimensions: number | undefined = undefined;
|
|
|
private lastError: string | undefined = undefined;
|
|
|
readonly breaker: CircuitBreaker;
|
|
|
+ /** Shared bulk-traffic lane — see `BulkLaneGate`. */
|
|
|
+ readonly lane: BulkLaneGate;
|
|
|
|
|
|
constructor(config: OpenAIProviderConfig) {
|
|
|
if (!config.endpoint) {
|
|
|
@@ -392,9 +596,14 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
this.timeoutMs = config.timeoutMs ?? DEFAULT_TIMEOUT_MS;
|
|
|
this.fetchImpl = config.fetchImpl ?? globalThis.fetch;
|
|
|
this.retryBackoffsMs = config.retryBackoffsMs ?? RETRY_BACKOFFS_MS;
|
|
|
+ this.rateLimitRetries =
|
|
|
+ config.rateLimitRetries ??
|
|
|
+ parseNonNegativeInt(process.env.QMD_EMBED_RATE_LIMIT_RETRIES) ??
|
|
|
+ DEFAULT_RATE_LIMIT_RETRIES;
|
|
|
this.sleep = config.sleep ?? defaultSleep;
|
|
|
this.now = config.now ?? Date.now;
|
|
|
this.breaker = new CircuitBreaker({ now: this.now });
|
|
|
+ this.lane = new BulkLaneGate(this.concurrency, this.sleep);
|
|
|
|
|
|
if (!this.fetchImpl) {
|
|
|
throw new Error(
|
|
|
@@ -507,6 +716,12 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
throw new CircuitOpenError();
|
|
|
}
|
|
|
|
|
|
+ // Bulk = anything that isn't a single query-time embed. Reindex runs go
|
|
|
+ // through the self-throttling lane and are attributed as `embeddings-bulk`;
|
|
|
+ // `embed()` (search path) stays interactive and is never parked behind a
|
|
|
+ // bulk cooldown (i-yghj098h).
|
|
|
+ const lane: EmbedLane = texts.length > 1 ? "bulk" : "interactive";
|
|
|
+
|
|
|
const chunks = chunkArray(texts, this.batchSize);
|
|
|
const results: (ProviderEmbedding | null)[] = new Array(texts.length).fill(null);
|
|
|
|
|
|
@@ -561,7 +776,14 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
}
|
|
|
|
|
|
try {
|
|
|
- const embeddings = await this.requestWithRetry(chunk, options);
|
|
|
+ if (lane === "bulk") await this.lane.acquire();
|
|
|
+ let embeddings: number[][];
|
|
|
+ try {
|
|
|
+ embeddings = await this.requestWithRetry(chunk, options, lane);
|
|
|
+ } finally {
|
|
|
+ if (lane === "bulk") this.lane.release();
|
|
|
+ }
|
|
|
+ if (lane === "bulk") this.lane.noteSuccess();
|
|
|
for (let i = 0; i < chunk.length; i++) {
|
|
|
const embedding = embeddings[i];
|
|
|
if (embedding) {
|
|
|
@@ -580,7 +802,12 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
}
|
|
|
this.breaker.recordSuccess();
|
|
|
} catch (err) {
|
|
|
- this.breaker.recordFailure();
|
|
|
+ // A rate limit is backpressure, not a fault. Counting 429s toward the
|
|
|
+ // breaker turns "the bucket is empty for 30s" into a 5-minute hard
|
|
|
+ // OPEN, which is the outage the retry budget above exists to avoid.
|
|
|
+ // The lane cooldown is already the correct control response.
|
|
|
+ const rateLimited = err instanceof HttpError && err.status === 429;
|
|
|
+ if (!rateLimited) this.breaker.recordFailure();
|
|
|
if (err instanceof CircuitOpenError) {
|
|
|
circuitTrippedDuringRun = err;
|
|
|
return;
|
|
|
@@ -639,13 +866,18 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
return `endpoint=${this.endpoint}/v1/embeddings error="${String(err)}"`;
|
|
|
}
|
|
|
|
|
|
- private buildHeaders(): Record<string, string> {
|
|
|
+ private buildHeaders(lane: EmbedLane = "interactive"): Record<string, string> {
|
|
|
const headers: Record<string, string> = {
|
|
|
"Content-Type": "application/json",
|
|
|
"Accept": "application/json",
|
|
|
// Advisory caller attribution for the ai.mm.mk gateway (NULL-safe, never
|
|
|
// auth). Single chokepoint for both /health and /v1/embeddings.
|
|
|
- "X-AI-Caller": aiCallerHeaderValue("embeddings"),
|
|
|
+ // `via` distinguishes bulk reindex traffic from query-time embeds so the
|
|
|
+ // gateway can budget the two separately (i-yghj098h part 1b) — the
|
|
|
+ // gateway-side per-caller bucket keys off this label.
|
|
|
+ "X-AI-Caller": aiCallerHeaderValue(
|
|
|
+ lane === "bulk" ? "embeddings-bulk" : "embeddings",
|
|
|
+ ),
|
|
|
};
|
|
|
if (this.apiKey) {
|
|
|
headers["Authorization"] = `Bearer ${this.apiKey}`;
|
|
|
@@ -661,26 +893,45 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
private async requestWithRetry(
|
|
|
texts: string[],
|
|
|
options: ProviderEmbedOptions,
|
|
|
+ lane: EmbedLane = "interactive",
|
|
|
): Promise<number[][]> {
|
|
|
let lastErr: unknown = null;
|
|
|
- const maxAttempts = this.retryBackoffsMs.length + 1;
|
|
|
+ // 429s draw on their own budget: a token bucket that refills in 30s must
|
|
|
+ // not be able to exhaust a schedule whose total wait is 21s (i-yghj098h).
|
|
|
+ let backoffAttempt = 0;
|
|
|
+ let rateLimitAttempt = 0;
|
|
|
|
|
|
- for (let attempt = 0; attempt < maxAttempts; attempt++) {
|
|
|
+ for (;;) {
|
|
|
// Honor user abort BEFORE issuing the call (avoids wasted network)
|
|
|
if (options.signal?.aborted) {
|
|
|
throw new Error("aborted by caller");
|
|
|
}
|
|
|
|
|
|
try {
|
|
|
- return await this.requestOnce(texts, options);
|
|
|
+ return await this.requestOnce(texts, options, lane);
|
|
|
} catch (err) {
|
|
|
lastErr = err;
|
|
|
- const retryable =
|
|
|
- err instanceof HttpError ? isRetryableStatus(err.status) : false;
|
|
|
- if (!retryable) throw err;
|
|
|
- if (attempt < this.retryBackoffsMs.length) {
|
|
|
- await this.sleep(this.retryBackoffsMs[attempt]!);
|
|
|
+ const status = err instanceof HttpError ? err.status : 0;
|
|
|
+ if (!isRetryableStatus(status)) throw err;
|
|
|
+
|
|
|
+ if (status === 429) {
|
|
|
+ if (rateLimitAttempt >= this.rateLimitRetries) throw err;
|
|
|
+ const waitMs = this.rateLimitWaitMs(err as HttpError, rateLimitAttempt);
|
|
|
+ rateLimitAttempt++;
|
|
|
+ if (lane === "bulk") {
|
|
|
+ // Pause the whole pool, not just this worker: the other workers
|
|
|
+ // hammering the same empty bucket is exactly what turns one 429
|
|
|
+ // into a cascade, and what starves interactive callers.
|
|
|
+ await this.lane.penalize(waitMs);
|
|
|
+ } else {
|
|
|
+ await this.sleep(waitMs);
|
|
|
+ }
|
|
|
+ continue;
|
|
|
}
|
|
|
+
|
|
|
+ if (backoffAttempt >= this.retryBackoffsMs.length) break;
|
|
|
+ await this.sleep(this.retryBackoffsMs[backoffAttempt]!);
|
|
|
+ backoffAttempt++;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -688,12 +939,32 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
throw lastErr ?? new Error("requestWithRetry exhausted");
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * How long to wait after a 429. The server's own `Retry-After` wins when it
|
|
|
+ * gave one; otherwise fall back to the configured schedule (so injected test
|
|
|
+ * schedules stay authoritative) and then to an exponential 5s/10s/20s… ramp.
|
|
|
+ * Always clamped to `RATE_LIMIT_MAX_BACKOFF_MS`.
|
|
|
+ */
|
|
|
+ private rateLimitWaitMs(err: HttpError, rateLimitAttempt: number): number {
|
|
|
+ const scheduled =
|
|
|
+ this.retryBackoffsMs[
|
|
|
+ Math.min(rateLimitAttempt, this.retryBackoffsMs.length - 1)
|
|
|
+ ];
|
|
|
+ const fallback =
|
|
|
+ scheduled ?? RATE_LIMIT_BASE_BACKOFF_MS * Math.pow(2, rateLimitAttempt);
|
|
|
+ return Math.min(
|
|
|
+ Math.max(err.retryAfterMs ?? 0, fallback),
|
|
|
+ RATE_LIMIT_MAX_BACKOFF_MS,
|
|
|
+ );
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* Issue one HTTP attempt to `POST /v1/embeddings`. Does NOT retry.
|
|
|
*/
|
|
|
private async requestOnce(
|
|
|
texts: string[],
|
|
|
options: ProviderEmbedOptions,
|
|
|
+ lane: EmbedLane = "interactive",
|
|
|
): Promise<number[][]> {
|
|
|
const { signal: attemptSig, cleanup } = buildAttemptSignal(options.signal, this.timeoutMs);
|
|
|
try {
|
|
|
@@ -703,14 +974,24 @@ export class OpenAIEmbeddingsProvider implements EmbeddingProvider {
|
|
|
});
|
|
|
const resp = await this.fetchImpl(`${this.endpoint}/v1/embeddings`, {
|
|
|
method: "POST",
|
|
|
- headers: this.buildHeaders(),
|
|
|
+ headers: this.buildHeaders(lane),
|
|
|
body,
|
|
|
signal: attemptSig,
|
|
|
});
|
|
|
|
|
|
if (!resp.ok) {
|
|
|
const text = await resp.text().catch(() => "");
|
|
|
- throw new HttpError(resp.status, text);
|
|
|
+ // ai.mm.mk states the cooldown in the JSON body, not always in a
|
|
|
+ // Retry-After header — read both (i-yghj098h).
|
|
|
+ const retryAfterMs =
|
|
|
+ resp.status === 429
|
|
|
+ ? parseRetryAfterMs(
|
|
|
+ resp.headers?.get?.("retry-after"),
|
|
|
+ text,
|
|
|
+ this.now,
|
|
|
+ )
|
|
|
+ : undefined;
|
|
|
+ throw new HttpError(resp.status, text, retryAfterMs);
|
|
|
}
|
|
|
|
|
|
let parsed: OpenAIEmbeddingsResponse;
|