mirror of
https://github.com/HeyPuter/puter.git
synced 2026-08-23 22:47:19 +00:00
buildXhr, dedup, retry (#3432)
* buildXhr, dedup, retry * change retry intervals
This commit is contained in:
Neal Shah
@@ -86,6 +86,66 @@ async function resolveReauth (resp) {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* The one XHR builder both `initXhr` (utils.js) and `fetchUrl` wrap. Opens the
|
||||
* request, applies headers/credentials/responseType, and stashes the whole
|
||||
* `spec` on `xhr._puterReq` as the single replay representation — any attempt
|
||||
* (reauth, permission, transient) rebuilds the request by calling
|
||||
* `buildXhr(spec)` again, which re-reads the live token when `includePuterAuth`.
|
||||
*
|
||||
* @param {Object} spec
|
||||
* @param {string} spec.url - Full request URL.
|
||||
* @param {string} [spec.method='GET']
|
||||
* @param {Object} [spec.headers] - Extra headers (nullish values skipped).
|
||||
* @param {boolean} [spec.includePuterAuth=false] - Add a fresh `Authorization: Bearer`.
|
||||
* @param {boolean} [spec.withCredentials=true]
|
||||
* @param {string} [spec.responseType='']
|
||||
* @param {Object} [spec.logId] - Pre-built apiCallLogger request id.
|
||||
* @returns {XMLHttpRequest}
|
||||
*/
|
||||
function buildXhr (spec) {
|
||||
const {
|
||||
url,
|
||||
method = 'GET',
|
||||
headers = {},
|
||||
includePuterAuth = false,
|
||||
withCredentials = true,
|
||||
responseType = '',
|
||||
} = spec;
|
||||
|
||||
const xhr = new XMLHttpRequest();
|
||||
xhr.open(method, url, true);
|
||||
xhr.withCredentials = withCredentials;
|
||||
xhr.responseType = responseType ?? '';
|
||||
|
||||
if ( includePuterAuth && globalThis.puter?.authToken ) {
|
||||
xhr.setRequestHeader('Authorization', `Bearer ${globalThis.puter.authToken}`);
|
||||
}
|
||||
for ( const [ name, value ] of Object.entries(headers) ) {
|
||||
if ( value !== undefined && value !== null ) {
|
||||
xhr.setRequestHeader(name, value);
|
||||
}
|
||||
}
|
||||
|
||||
xhr._puterReq = spec;
|
||||
const origSend = xhr.send.bind(xhr);
|
||||
xhr.send = function (body) {
|
||||
spec.body = body;
|
||||
return origSend(body);
|
||||
};
|
||||
|
||||
if ( globalThis.puter?.apiCallLogger?.isEnabled() ) {
|
||||
xhr._puterRequestId = spec.logId ?? {
|
||||
method,
|
||||
service: 'xhr',
|
||||
operation: url,
|
||||
params: { url, method, responseType },
|
||||
};
|
||||
}
|
||||
|
||||
return xhr;
|
||||
}
|
||||
|
||||
/**
|
||||
* The single HTTP core for puter.js. `fetchUrl` is an XHR-based replacement for
|
||||
* `fetch()` — every request that used to call `fetch()` directly routes through
|
||||
@@ -181,6 +241,241 @@ async function bodyForLog (xhr) {
|
||||
return `[${contentType || 'binary'}]`;
|
||||
}
|
||||
|
||||
// -- Retry engine --
|
||||
// One loop drives every request: build the XHR from its spec, send, classify
|
||||
// the outcome, and either replay (reauth / permission / transient backoff) or
|
||||
// hand the result to the caller's shaper. A replay just rebuilds from the same
|
||||
// spec, so there are no hand-listed argument lists to get wrong.
|
||||
|
||||
const RETRYABLE_STATUS = new Set([ 429, 502, 503, 504 ]);
|
||||
|
||||
// Fixed retry backoff: a quick ramp to a 2s ceiling, then hold at 2s. Index i is
|
||||
// the wait (ms) after attempt i+1 fails. The array length caps the retries — 8
|
||||
// delays ⇒ 9 attempts total, the 2s ceiling used 5 times — after which the
|
||||
// request is failed.
|
||||
const RETRY_DELAYS_MS = [ 250, 500, 1000, 2000, 2000, 2000, 2000, 2000 ];
|
||||
const RETRY_CEILING_MS = 2000;
|
||||
// If a ceiling-length wait overruns real time by more than this, the clock
|
||||
// jumped (e.g. the laptop slept mid-wait); the request is stale, so give up
|
||||
// rather than fire a very old retry.
|
||||
const MAX_SLEEP_DRIFT_MS = 2000;
|
||||
|
||||
// Kill-switch seam: puter.configure() (deferred) will drive this. Default on.
|
||||
const autoRetryEnabled = () => globalThis.puter?.config?.autoRetry ?? true;
|
||||
|
||||
const sleep = (ms, signal) => new Promise((resolve, reject) => {
|
||||
if ( signal?.aborted ) return reject(signal.reason ?? new DOMException('Aborted', 'AbortError'));
|
||||
const t = setTimeout(resolve, ms);
|
||||
signal?.addEventListener('abort', () => {
|
||||
clearTimeout(t);
|
||||
reject(signal.reason ?? new DOMException('Aborted', 'AbortError'));
|
||||
}, { once: true });
|
||||
});
|
||||
|
||||
const retryDelay = attempt => RETRY_DELAYS_MS[attempt - 1];
|
||||
|
||||
const transientRetry = ctx => {
|
||||
if ( ! (ctx.retrySafe && autoRetryEnabled()) ) return null;
|
||||
const delayMs = retryDelay(ctx.attempt);
|
||||
return delayMs === undefined ? null : { delayMs };
|
||||
};
|
||||
|
||||
/**
|
||||
* Drive the env-specific permission prompt for a denied driver call.
|
||||
* @returns {Promise<{granted: boolean}>}
|
||||
*/
|
||||
async function resolvePermission (permission) {
|
||||
try {
|
||||
const perm = await puter.ui.requestPermission({ permission });
|
||||
return { granted: !! perm?.granted };
|
||||
} catch ( e ) {
|
||||
return { granted: false };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Send one attempt. Resolves with a terminal outcome:
|
||||
* { streamed: true, xhr, lineStream } — NDJSON, resolved at HEADERS_RECEIVED
|
||||
* { xhr, status } — buffered response (any HTTP status)
|
||||
* { networkError: true, xhr } — transport error
|
||||
* Rejects only on abort. The NDJSON line buffering matches the old inline
|
||||
* streaming in fetchUrl and driverCall_; per-line semantics (usage/email/etc.)
|
||||
* belong to the caller's `shapeStream`.
|
||||
*/
|
||||
function sendOnce (spec) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const xhr = buildXhr(spec);
|
||||
|
||||
let streamed = false;
|
||||
let responseComplete = false;
|
||||
let signalStreamUpdate = null;
|
||||
const lines = [];
|
||||
let carry = '';
|
||||
let consumed = 0;
|
||||
|
||||
const lineStream = (async function* () {
|
||||
while ( true ) {
|
||||
while ( lines.length > 0 ) {
|
||||
const line = lines.shift();
|
||||
if ( line.trim() === '' ) continue;
|
||||
yield JSON.parse(line);
|
||||
}
|
||||
if ( responseComplete ) break;
|
||||
const sig = createDeferred();
|
||||
signalStreamUpdate = sig.resolve;
|
||||
await sig.promise;
|
||||
}
|
||||
})();
|
||||
|
||||
xhr.onreadystatechange = () => {
|
||||
if ( xhr.readyState === 2 && isNdjson(xhr.getResponseHeader('Content-Type')) ) {
|
||||
streamed = true;
|
||||
resolve({ streamed: true, xhr, lineStream });
|
||||
}
|
||||
if ( xhr.readyState === 4 && streamed ) {
|
||||
if ( carry.length > 0 ) { lines.push(carry); carry = ''; }
|
||||
responseComplete = true;
|
||||
signalStreamUpdate?.();
|
||||
}
|
||||
};
|
||||
|
||||
xhr.onprogress = () => {
|
||||
if ( ! streamed ) return;
|
||||
const fresh = xhr.responseText.slice(consumed);
|
||||
consumed = xhr.responseText.length;
|
||||
if ( ! fresh ) return;
|
||||
carry += fresh;
|
||||
let nl;
|
||||
while ( (nl = carry.indexOf('\n')) !== -1 ) {
|
||||
lines.push(carry.slice(0, nl));
|
||||
carry = carry.slice(nl + 1);
|
||||
}
|
||||
signalStreamUpdate?.();
|
||||
};
|
||||
|
||||
xhr.addEventListener('load', () => {
|
||||
if ( streamed ) return;
|
||||
resolve({ xhr, status: xhr.status });
|
||||
});
|
||||
xhr.addEventListener('error', () => resolve({ networkError: true, xhr }));
|
||||
xhr.addEventListener('abort', () => reject(spec.signal?.reason ?? new DOMException('Aborted', 'AbortError')));
|
||||
|
||||
if ( spec.signal ) {
|
||||
if ( spec.signal.aborted ) return reject(spec.signal.reason ?? new DOMException('Aborted', 'AbortError'));
|
||||
spec.signal.addEventListener('abort', () => xhr.abort(), { once: true });
|
||||
}
|
||||
|
||||
const body = typeof spec.buildBody === 'function' ? spec.buildBody() : spec.body;
|
||||
xhr.send(body ?? null);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify a completed attempt into a retry decision. Reauth and permission are
|
||||
* one-shot (tracked in `ctx.done`) and apply to any request; transient backoff
|
||||
* applies only to `ctx.retrySafe` requests and honors the autoRetry kill switch.
|
||||
* Memoizes the parsed body on `outcome.parsed` and stashes any reauth error on
|
||||
* `outcome.reauthError` for the shaper.
|
||||
*
|
||||
* @returns {Promise<{delayMs:number}|null>} a delay to retry after, or null to stop.
|
||||
*/
|
||||
async function classifyRetry (outcome, ctx) {
|
||||
if ( outcome.streamed ) return null; // committed stream — never retried
|
||||
|
||||
if ( outcome.networkError ) return transientRetry(ctx);
|
||||
|
||||
const { xhr, status } = outcome;
|
||||
if ( outcome.parsed === undefined ) {
|
||||
outcome.parsed = await bodyJson(xhr).catch(() => null);
|
||||
}
|
||||
const parsed = outcome.parsed;
|
||||
|
||||
// reauth (401 / token_auth_failed) — one-shot, any method, no backoff.
|
||||
if ( status === 401 || parsed?.code === 'token_auth_failed' ) {
|
||||
if ( ! ctx.done.has('reauth') ) {
|
||||
const reauth = await resolveReauth(parsed);
|
||||
if ( reauth?.action === 'replay' ) { ctx.done.add('reauth'); return { delayMs: 0 }; }
|
||||
if ( reauth?.action === 'reject' ) outcome.reauthError = reauth.error;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// permission denied (200 success:false) — one-shot, any method, no backoff.
|
||||
if ( ctx.permission && parsed?.success === false && parsed?.error?.code === 'permission_denied' ) {
|
||||
if ( ! ctx.done.has('permission') ) {
|
||||
const perm = await resolvePermission(ctx.permission);
|
||||
if ( perm.granted ) { ctx.done.add('permission'); return { delayMs: 0 }; }
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// transient status — read-safe only, honors kill switch, fixed schedule.
|
||||
if ( RETRYABLE_STATUS.has(status) ) return transientRetry(ctx);
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* The one retry loop. Sends `spec` (rebuilding per attempt), classifies each
|
||||
* outcome, and retries on reauth / permission / transient causes; otherwise
|
||||
* hands the outcome to `shape`.
|
||||
*
|
||||
* @param {Object} spec - buildXhr spec (+ optional buildBody, signal).
|
||||
* @param {Object} opts
|
||||
* @param {boolean} [opts.retrySafe=false] - eligible for transient backoff retry.
|
||||
* @param {string|null} [opts.permission] - `driver:<iface>:<method>` enables the permission cause.
|
||||
* @param {(lineStream, xhr) => any} opts.shapeStream - wrap an NDJSON stream.
|
||||
* @param {(outcome) => any} opts.shape - shape a buffered outcome (may throw).
|
||||
*/
|
||||
async function sendWithRetry (spec, { retrySafe = false, permission = null, shapeStream, shape }) {
|
||||
const ctx = { attempt: 0, retrySafe, permission, done: new Set() };
|
||||
while ( true ) {
|
||||
ctx.attempt++;
|
||||
const outcome = await sendOnce(spec);
|
||||
if ( outcome.streamed ) return shapeStream(outcome.lineStream, outcome.xhr);
|
||||
const decision = await classifyRetry(outcome, ctx);
|
||||
if ( decision ) {
|
||||
const before = Date.now();
|
||||
await sleep(decision.delayMs, spec.signal);
|
||||
if ( decision.delayMs >= RETRY_CEILING_MS
|
||||
&& (Date.now() - before) - decision.delayMs > MAX_SLEEP_DRIFT_MS ) {
|
||||
return shape(outcome);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
return shape(outcome);
|
||||
}
|
||||
}
|
||||
|
||||
// -- In-flight request dedup --
|
||||
// Coalesce concurrent identical requests: a second caller within `windowMs`
|
||||
// gets the first request's promise (shared resolved value). The entry is
|
||||
// deleted when the request settles. Generalized from the copy-pasted logic in
|
||||
// FileSystem readdir/stat (which keep their own bespoke cache and are untouched).
|
||||
const inflightRequests = new Map();
|
||||
|
||||
/**
|
||||
* @param {string} key - fully-qualified request key (namespace it yourself, e.g.
|
||||
* `${method}:${url}:${bodyKey}`).
|
||||
* @param {() => Promise<any>} factory - runs the request; called only on a miss.
|
||||
* @param {{windowMs?: number}} [opts]
|
||||
* @returns {Promise<any>} shared promise (resolved value shared by reference).
|
||||
*/
|
||||
function dedupe (key, factory, { windowMs = 2000 } = {}) {
|
||||
const existing = inflightRequests.get(key);
|
||||
if ( existing ) {
|
||||
if ( Date.now() - existing.timestamp < windowMs ) return existing.promise;
|
||||
inflightRequests.delete(key); // stale — fall through and re-issue
|
||||
}
|
||||
const promise = factory();
|
||||
inflightRequests.set(key, { promise, timestamp: Date.now() });
|
||||
const cleanup = () => {
|
||||
if ( inflightRequests.get(key)?.promise === promise ) inflightRequests.delete(key);
|
||||
};
|
||||
promise.then(cleanup, cleanup);
|
||||
return promise;
|
||||
}
|
||||
|
||||
/**
|
||||
* XHR-based `fetch()` replacement. Returns a `fetch`-Response-like object.
|
||||
*
|
||||
@@ -201,8 +496,11 @@ async function bodyForLog (xhr) {
|
||||
* @param {AbortSignal} [opts.signal]
|
||||
* @param {{service: string, operation: string, params?: Object}} [opts.logContext]
|
||||
* Semantic context for the centralized API-call log. Omit to log generically.
|
||||
* @param {Object} [opts.retry] - Reserved for a later sprint step (ignored).
|
||||
* @param {Object} [opts.dedupe] - Reserved for a later sprint step (ignored).
|
||||
* @param {boolean} [opts.retry] - Force-enable (`true`) or disable (`false`)
|
||||
* transient-failure auto-retry for this request; omit for the default
|
||||
* (idempotent methods retry, others don't). Never retries a write.
|
||||
* @param {boolean|string} [opts.dedupe] - Coalesce concurrent identical in-flight
|
||||
* requests (reads only): `true` auto-keys by method+url+body, or pass a key.
|
||||
* @param {Object} [opts.paginate] - Reserved for a later sprint step (ignored).
|
||||
* @returns {Promise<PuterResponse>}
|
||||
*/
|
||||
@@ -216,132 +514,49 @@ function fetchUrl (url, opts = {}) {
|
||||
withCredentials = true,
|
||||
signal,
|
||||
logContext,
|
||||
_reauthReplayed = false, // internal one-shot guard for reauth replay
|
||||
retry,
|
||||
dedupe: dedupeOpt,
|
||||
} = opts;
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const xhr = new XMLHttpRequest();
|
||||
xhr.open(method, url, true);
|
||||
xhr.withCredentials = withCredentials;
|
||||
// Text mode keeps NDJSON progress deltas available for stream(); callers
|
||||
// that need binary/parsed bodies pass an explicit responseType.
|
||||
xhr.responseType = responseType;
|
||||
const logId = logContext ?? { service: 'fetchUrl', operation: `${method} ${url}`, params: { url, method } };
|
||||
const spec = { url, method, headers, includePuterAuth, withCredentials, responseType, body, signal, logId };
|
||||
|
||||
if ( includePuterAuth && globalThis.puter?.authToken ) {
|
||||
xhr.setRequestHeader('Authorization', `Bearer ${globalThis.puter.authToken}`);
|
||||
}
|
||||
for ( const [ name, value ] of Object.entries(headers) ) {
|
||||
if ( value !== undefined && value !== null ) {
|
||||
xhr.setRequestHeader(name, value);
|
||||
// Read-safety: idempotent methods auto-retry; a POST read opts in with
|
||||
// `retry:true`; nothing retries when `retry:false` (writes/uploads).
|
||||
const idempotent = method === 'GET' || method === 'HEAD';
|
||||
const retrySafe = retry === false ? false : ( retry === true || idempotent );
|
||||
|
||||
const loggingOn = () => globalThis.puter?.apiCallLogger?.isEnabled();
|
||||
|
||||
const run = () => sendWithRetry(spec, {
|
||||
retrySafe,
|
||||
shapeStream: (lineStream, xhr) => {
|
||||
if ( loggingOn() ) logRequest(logId, { result: '[stream]' });
|
||||
return makeResponse(xhr, lineStream);
|
||||
},
|
||||
shape: async (outcome) => {
|
||||
if ( outcome.networkError ) {
|
||||
if ( loggingOn() ) logRequest(logId, { error: { message: 'Network error occurred' } });
|
||||
throw new TypeError(`Network request to ${url} failed`);
|
||||
}
|
||||
}
|
||||
|
||||
if ( signal ) {
|
||||
if ( signal.aborted ) return reject(signal.reason ?? new DOMException('Aborted', 'AbortError'));
|
||||
signal.addEventListener('abort', () => xhr.abort(), { once: true });
|
||||
}
|
||||
|
||||
let logId = null;
|
||||
if ( globalThis.puter?.apiCallLogger?.isEnabled() ) {
|
||||
logId = logContext ?? { service: 'fetchUrl', operation: `${method} ${url}`, params: { url, method } };
|
||||
}
|
||||
|
||||
// -- NDJSON streaming --
|
||||
// Detect a stream at HEADERS_RECEIVED and resolve early with a response
|
||||
// whose stream() yields parsed lines as onprogress deltas arrive. This
|
||||
// mirrors the streaming in driverCall_ (utils.js) and the xhrshim.
|
||||
let resolvedAsStream = false;
|
||||
let responseComplete = false;
|
||||
let signalStreamUpdate = null;
|
||||
const linesReceived = [];
|
||||
let carry = '';
|
||||
let consumedLength = 0;
|
||||
|
||||
const pushLines = text => {
|
||||
carry += text;
|
||||
let nl;
|
||||
while ( (nl = carry.indexOf('\n')) !== -1 ) {
|
||||
linesReceived.push(carry.slice(0, nl));
|
||||
carry = carry.slice(nl + 1);
|
||||
}
|
||||
};
|
||||
|
||||
xhr.onreadystatechange = () => {
|
||||
if ( xhr.readyState === 2 && ! isNdjson(xhr.getResponseHeader('Content-Type')) ) return;
|
||||
if ( xhr.readyState === 2 ) {
|
||||
resolvedAsStream = true;
|
||||
const stream = (async function* () {
|
||||
while ( true ) {
|
||||
while ( linesReceived.length > 0 ) {
|
||||
const line = linesReceived.shift();
|
||||
if ( line.trim() === '' ) continue;
|
||||
yield JSON.parse(line);
|
||||
}
|
||||
if ( responseComplete ) break;
|
||||
const sig = createDeferred();
|
||||
signalStreamUpdate = sig.resolve;
|
||||
await sig.promise;
|
||||
}
|
||||
})();
|
||||
logRequest(logId, { result: '[stream]' });
|
||||
resolve(makeResponse(xhr, stream));
|
||||
}
|
||||
if ( xhr.readyState === 4 && resolvedAsStream ) {
|
||||
if ( carry.length > 0 ) { linesReceived.push(carry); carry = ''; }
|
||||
responseComplete = true;
|
||||
signalStreamUpdate?.();
|
||||
}
|
||||
};
|
||||
|
||||
xhr.onprogress = () => {
|
||||
if ( ! resolvedAsStream ) return;
|
||||
const fresh = xhr.responseText.slice(consumedLength);
|
||||
consumedLength = xhr.responseText.length;
|
||||
if ( ! fresh ) return;
|
||||
pushLines(fresh);
|
||||
signalStreamUpdate?.();
|
||||
};
|
||||
|
||||
// -- Buffered (non-stream) response --
|
||||
xhr.addEventListener('load', async function () {
|
||||
if ( resolvedAsStream ) return;
|
||||
|
||||
if ( this.status === 401 && ! _reauthReplayed ) {
|
||||
let parsed = null;
|
||||
try { parsed = await bodyJson(this); } catch ( e ) { parsed = null; }
|
||||
const reauth = await resolveReauth(parsed);
|
||||
if ( reauth?.action === 'replay' ) {
|
||||
try {
|
||||
return resolve(await fetchUrl(url, { ...opts, _reauthReplayed: true }));
|
||||
} catch ( e ) {
|
||||
return reject(e);
|
||||
}
|
||||
}
|
||||
// reauth 'reject'/null (incl. token_auth_failed handled inside
|
||||
// resolveReauth): surface the 401 as an ok:false response, the
|
||||
// same way fetch does for any error status.
|
||||
}
|
||||
|
||||
const resp = makeResponse(this);
|
||||
if ( logId ) {
|
||||
const logged = await bodyForLog(this);
|
||||
logRequest(logId, this.status >= 400
|
||||
? { error: logged ?? { message: this.statusText, status: this.status } }
|
||||
const { xhr } = outcome;
|
||||
const resp = makeResponse(xhr);
|
||||
if ( loggingOn() ) {
|
||||
const logged = await bodyForLog(xhr);
|
||||
logRequest(logId, xhr.status >= 400
|
||||
? { error: logged ?? { message: xhr.statusText, status: xhr.status } }
|
||||
: { result: logged });
|
||||
}
|
||||
resolve(resp);
|
||||
});
|
||||
|
||||
xhr.addEventListener('error', function (e) {
|
||||
logRequest(logId, { error: { message: 'Network error occurred', event: e.type } });
|
||||
reject(new TypeError(`Network request to ${url} failed`));
|
||||
});
|
||||
xhr.addEventListener('abort', function () {
|
||||
reject(signal?.reason ?? new DOMException('Aborted', 'AbortError'));
|
||||
});
|
||||
|
||||
xhr.send(body);
|
||||
return resp;
|
||||
},
|
||||
});
|
||||
|
||||
if ( dedupeOpt ) {
|
||||
const bodyKey = body == null ? '' : ( typeof body === 'string' ? body : '[body]' );
|
||||
const key = typeof dedupeOpt === 'string' ? dedupeOpt : `${method}:${url}:${bodyKey}`;
|
||||
return dedupe(key, run);
|
||||
}
|
||||
return run();
|
||||
}
|
||||
|
||||
export { fetchUrl, resolveReauth };
|
||||
export { buildXhr, dedupe, fetchUrl, resolveReauth, sendWithRetry };
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { fetchUrl } from './networkUtils.js';
|
||||
import { dedupe, fetchUrl, sendWithRetry } from './networkUtils.js';
|
||||
|
||||
// -- Controllable fake XMLHttpRequest --
|
||||
// Drives fetchUrl's event handlers deterministically. Each instance plays the
|
||||
@@ -108,9 +108,11 @@ describe('fetchUrl', () => {
|
||||
}
|
||||
});
|
||||
|
||||
it('rejects on a network error', async () => {
|
||||
it('rejects on a network error (write — no retry)', async () => {
|
||||
// A write never auto-retries, so the network error surfaces immediately.
|
||||
// Read retry-then-reject is covered in the transient-retry suite.
|
||||
installFakeXHR(xhr => xhr._networkError());
|
||||
await expect(fetchUrl('https://api.example/x')).rejects.toThrow(/failed/);
|
||||
await expect(fetchUrl('https://api.example/x', { method: 'POST' })).rejects.toThrow(/failed/);
|
||||
});
|
||||
|
||||
it('exposes text(), json(), and blob() accessors', async () => {
|
||||
@@ -220,3 +222,179 @@ describe('fetchUrl', () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
// Play a scripted response per attempt (retries create fresh XHR instances).
|
||||
const sequence = (...steps) => {
|
||||
let i = 0;
|
||||
return xhr => steps[Math.min(i++, steps.length - 1)](xhr);
|
||||
};
|
||||
const netError = () => xhr => xhr._networkError();
|
||||
|
||||
describe('transient retry', () => {
|
||||
beforeEach(() => { globalThis.puter = {}; });
|
||||
|
||||
it('retries a GET on 503 then resolves the success', async () => {
|
||||
vi.useFakeTimers();
|
||||
const xhrs = installFakeXHR(sequence(
|
||||
respond({ status: 503, body: {} }),
|
||||
respond({ status: 200, body: { ok: 1 } }),
|
||||
));
|
||||
const p = fetchUrl('https://api.example/x'); // GET → retry-safe
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
const resp = await p;
|
||||
expect(resp.status).toBe(200);
|
||||
expect(xhrs.length).toBe(2);
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it('does not retry a POST by default', async () => {
|
||||
const xhrs = installFakeXHR(sequence(respond({ status: 503, body: {} })));
|
||||
const resp = await fetchUrl('https://api.example/x', { method: 'POST' });
|
||||
expect(resp.status).toBe(503);
|
||||
expect(xhrs.length).toBe(1);
|
||||
});
|
||||
|
||||
it('retries a POST when retry:true (read-style opt-in)', async () => {
|
||||
vi.useFakeTimers();
|
||||
const xhrs = installFakeXHR(sequence(
|
||||
respond({ status: 503, body: {} }),
|
||||
respond({ status: 200, body: { ok: 1 } }),
|
||||
));
|
||||
const p = fetchUrl('https://api.example/x', { method: 'POST', retry: true });
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect((await p).status).toBe(200);
|
||||
expect(xhrs.length).toBe(2);
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it('retry:false disables retry even for a GET', async () => {
|
||||
const xhrs = installFakeXHR(sequence(respond({ status: 503, body: {} })));
|
||||
const resp = await fetchUrl('https://api.example/x', { retry: false });
|
||||
expect(resp.status).toBe(503);
|
||||
expect(xhrs.length).toBe(1);
|
||||
});
|
||||
|
||||
it('does not retry a non-retryable status (400)', async () => {
|
||||
const xhrs = installFakeXHR(sequence(respond({ status: 400, body: {} })));
|
||||
const resp = await fetchUrl('https://api.example/x'); // GET
|
||||
expect(resp.status).toBe(400);
|
||||
expect(xhrs.length).toBe(1);
|
||||
});
|
||||
|
||||
it('respects the autoRetry kill switch', async () => {
|
||||
globalThis.puter = { config: { autoRetry: false } };
|
||||
const xhrs = installFakeXHR(sequence(respond({ status: 503, body: {} })));
|
||||
const resp = await fetchUrl('https://api.example/x'); // GET, but retry off
|
||||
expect(resp.status).toBe(503);
|
||||
expect(xhrs.length).toBe(1);
|
||||
});
|
||||
|
||||
it('retries a network error for a read, then rejects after the cap', async () => {
|
||||
vi.useFakeTimers();
|
||||
const xhrs = installFakeXHR(netError()); // every attempt fails
|
||||
const p = fetchUrl('https://api.example/x').catch(e => e); // GET
|
||||
await vi.advanceTimersByTimeAsync(60_000); // clears the ~11.75s schedule
|
||||
const err = await p;
|
||||
expect(err).toBeInstanceOf(TypeError);
|
||||
expect(xhrs.length).toBe(9); // 1 initial + 8 scheduled retries
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it('rejects a write network error immediately (no retry)', async () => {
|
||||
const xhrs = installFakeXHR(netError());
|
||||
await expect(fetchUrl('https://api.example/x', { method: 'POST' })).rejects.toThrow(/failed/);
|
||||
expect(xhrs.length).toBe(1);
|
||||
});
|
||||
|
||||
it('gives up on a 2s retry when the clock jumps (sleep/drift guard)', async () => {
|
||||
// Fake only the timers, not Date — a manual `clock` drives Date.now so we
|
||||
// can simulate the machine sleeping during a 2s ceiling wait.
|
||||
vi.useFakeTimers({ toFake: [ 'setTimeout', 'clearTimeout' ] });
|
||||
let clock = 0;
|
||||
vi.spyOn(Date, 'now').mockImplementation(() => clock);
|
||||
const xhrs = installFakeXHR(respond({ status: 503, body: {} })); // always retryable
|
||||
const p = fetchUrl('https://api.example/x'); // GET → read-safe
|
||||
|
||||
// Ramp (250+500+1000) → 4 attempts, then paused in the first 2s wait.
|
||||
// Sub-2s waits aren't drift-guarded, so the clock needn't move here.
|
||||
await vi.advanceTimersByTimeAsync(1750);
|
||||
// Simulate the laptop sleeping through the 2s wait: the clock leaps ahead.
|
||||
clock += 2000 + 60_000;
|
||||
await vi.advanceTimersByTimeAsync(2000);
|
||||
|
||||
const resp = await p;
|
||||
expect(resp.status).toBe(503); // failed with the last outcome — no further retry
|
||||
expect(xhrs.length).toBe(4); // stopped after the drifted 2s wait, before attempt 5
|
||||
vi.useRealTimers();
|
||||
});
|
||||
});
|
||||
|
||||
describe('dedupe', () => {
|
||||
it('coalesces concurrent identical requests into one call', async () => {
|
||||
let calls = 0;
|
||||
const factory = () => { calls++; return new Promise(r => setTimeout(() => r({ v: calls }), 5)); };
|
||||
const [ a, b ] = await Promise.all([ dedupe('k', factory), dedupe('k', factory) ]);
|
||||
expect(calls).toBe(1);
|
||||
expect(a).toBe(b); // shared resolved value by reference
|
||||
});
|
||||
|
||||
it('re-issues after the in-flight request settles', async () => {
|
||||
let calls = 0;
|
||||
const factory = () => Promise.resolve(++calls);
|
||||
await dedupe('k2', factory);
|
||||
await dedupe('k2', factory);
|
||||
expect(calls).toBe(2);
|
||||
});
|
||||
|
||||
it('does not collide across distinct keys', async () => {
|
||||
let calls = 0;
|
||||
const factory = () => new Promise(r => setTimeout(() => r(++calls), 5));
|
||||
await Promise.all([ dedupe('a', factory), dedupe('b', factory) ]);
|
||||
expect(calls).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe('driver permission-grant replay (regression)', () => {
|
||||
it('replays exactly once after a grant, preserving the rebuilt request', async () => {
|
||||
// Before the fix, the driver permission replay dropped arguments; here we
|
||||
// assert one prompt, one replay, and the same body on the retry.
|
||||
const requestPermission = vi.fn(async () => ({ granted: true }));
|
||||
globalThis.puter = { ui: { requestPermission } };
|
||||
const xhrs = installFakeXHR(sequence(
|
||||
respond({ status: 200, body: { success: false, error: { code: 'permission_denied' } } }),
|
||||
respond({ status: 200, body: { success: true, result: 'ok' } }),
|
||||
));
|
||||
const spec = {
|
||||
url: 'https://api.example/drivers/call',
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'text/plain;actually=json' },
|
||||
buildBody: () => JSON.stringify({ interface: 'iface', method: 'm', args: { a: 1 } }),
|
||||
};
|
||||
const result = await sendWithRetry(spec, {
|
||||
permission: 'driver:iface:m',
|
||||
shapeStream: () => {},
|
||||
shape: outcome => outcome.parsed,
|
||||
});
|
||||
expect(requestPermission).toHaveBeenCalledTimes(1);
|
||||
expect(requestPermission).toHaveBeenCalledWith({ permission: 'driver:iface:m' });
|
||||
expect(xhrs.length).toBe(2);
|
||||
expect(xhrs[1].reqBody).toBe(JSON.stringify({ interface: 'iface', method: 'm', args: { a: 1 } }));
|
||||
expect(result).toEqual({ success: true, result: 'ok' });
|
||||
});
|
||||
|
||||
it('does not loop when the grant still yields permission_denied', async () => {
|
||||
const requestPermission = vi.fn(async () => ({ granted: true }));
|
||||
globalThis.puter = { ui: { requestPermission } };
|
||||
const denied = respond({ status: 200, body: { success: false, error: { code: 'permission_denied' } } });
|
||||
const xhrs = installFakeXHR(sequence(denied, denied, denied));
|
||||
const spec = { url: 'https://api.example/drivers/call', method: 'POST', headers: {}, buildBody: () => '{}' };
|
||||
const result = await sendWithRetry(spec, {
|
||||
permission: 'driver:iface:m',
|
||||
shapeStream: () => {},
|
||||
shape: outcome => outcome.parsed,
|
||||
});
|
||||
expect(requestPermission).toHaveBeenCalledTimes(1); // one-shot
|
||||
expect(xhrs.length).toBe(2);
|
||||
expect(result.error.code).toBe('permission_denied');
|
||||
});
|
||||
});
|
||||
|
||||
+129
-327
@@ -1,5 +1,5 @@
|
||||
import { FileReaderPoly } from './polyfills/fileReaderPoly.js';
|
||||
import { resolveReauth } from './networkUtils.js';
|
||||
import { buildXhr, resolveReauth, sendWithRetry } from './networkUtils.js';
|
||||
import { showUsageLimitDialog } from '../modules/UsageLimitDialog.js';
|
||||
import { showEmailConfirmationDialog } from '../modules/EmailConfirmationDialog.js';
|
||||
|
||||
@@ -91,68 +91,42 @@ const createDeferred = () => {
|
||||
* @returns {XMLHttpRequest} The initialized XMLHttpRequest object.
|
||||
*/
|
||||
function initXhr (endpoint, APIOrigin, authToken, method = 'post', contentType = 'text/plain;actually=json', responseType = undefined) {
|
||||
const xhr = new XMLHttpRequest();
|
||||
xhr.open(method, APIOrigin + endpoint, true);
|
||||
xhr.withCredentials = true;
|
||||
if ( authToken )
|
||||
{
|
||||
xhr.setRequestHeader('Authorization', `Bearer ${ authToken}`);
|
||||
}
|
||||
xhr.setRequestHeader('Content-Type', contentType);
|
||||
xhr.responseType = responseType ?? '';
|
||||
|
||||
// Capture enough request shape to replay this XHR after a
|
||||
// reauth_required. The body is captured below by intercepting send().
|
||||
xhr._puterReq = {
|
||||
endpoint,
|
||||
APIOrigin,
|
||||
return buildXhr({
|
||||
url: APIOrigin + endpoint,
|
||||
method,
|
||||
contentType,
|
||||
responseType,
|
||||
};
|
||||
const origSend = xhr.send.bind(xhr);
|
||||
xhr.send = function (body) {
|
||||
xhr._puterReq.body = body;
|
||||
return origSend(body);
|
||||
};
|
||||
|
||||
// Add API call logging if available
|
||||
if ( globalThis.puter?.apiCallLogger?.isEnabled() ) {
|
||||
xhr._puterRequestId = {
|
||||
headers: { 'Content-Type': contentType },
|
||||
// `includePuterAuth` re-reads the live token at build time, so a replay
|
||||
// picks up a freshly-minted token (same as the old replay path, which
|
||||
// passed `globalThis.puter.authToken`).
|
||||
includePuterAuth: !! authToken,
|
||||
withCredentials: true,
|
||||
responseType: responseType ?? '',
|
||||
logId: {
|
||||
method,
|
||||
service: 'xhr',
|
||||
operation: endpoint.replace(/^\//, ''),
|
||||
params: { endpoint, contentType, responseType },
|
||||
};
|
||||
}
|
||||
|
||||
return xhr;
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-issue an XHR after the reauth coordinator resolves. Uses the request
|
||||
* shape captured on `_puterReq` and the fresh `puter.authToken` to build a
|
||||
* new XHR; the new response is then routed back through the same callbacks.
|
||||
* Returns true if a replay was scheduled, false otherwise.
|
||||
* Re-issue an XHR after the reauth coordinator resolves. Rebuilds the request
|
||||
* from the captured `_puterReq` spec (with a fresh token via `buildXhr`) and
|
||||
* routes the new response back through the same callbacks. Returns true if a
|
||||
* replay was scheduled, false otherwise.
|
||||
*/
|
||||
function replayXhrAfterReauth (response, success_cb, error_cb, resolve_func, reject_func) {
|
||||
const xhr = response.target ?? response;
|
||||
const req = xhr?._puterReq;
|
||||
if ( ! req ) return false;
|
||||
const spec = xhr?._puterReq;
|
||||
if ( ! spec ) return false;
|
||||
// Already a replay attempt — don't loop into reauth a second time if
|
||||
// even the fresh token comes back rejected. The retry path is one-shot.
|
||||
if ( req._replayed ) return false;
|
||||
const newXhr = initXhr(
|
||||
req.endpoint,
|
||||
req.APIOrigin,
|
||||
globalThis.puter?.authToken,
|
||||
req.method,
|
||||
req.contentType,
|
||||
req.responseType,
|
||||
);
|
||||
newXhr._puterReq._replayed = true;
|
||||
if ( spec._replayed ) return false;
|
||||
const newSpec = { ...spec, _replayed: true };
|
||||
const newXhr = buildXhr(newSpec);
|
||||
setupXhrEventHandlers(newXhr, success_cb, error_cb, resolve_func, reject_func);
|
||||
newXhr.send(req.body);
|
||||
newXhr.send(spec.body);
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -405,311 +379,139 @@ async function driverCall_ (
|
||||
|
||||
const success_cb = Valid.callback(options.success) ?? NOOP;
|
||||
const error_cb = Valid.callback(options.error) ?? NOOP;
|
||||
// create xhr object
|
||||
const xhr = initXhr('/drivers/call', puter.APIOrigin, undefined, 'POST', contentType);
|
||||
|
||||
// Store request info for later logging
|
||||
if ( requestInfo ) {
|
||||
xhr._puterDriverRequestInfo = requestInfo;
|
||||
}
|
||||
|
||||
if ( settings.responseType ) {
|
||||
xhr.responseType = settings.responseType;
|
||||
// Keep _puterReq in sync with the live XHR config so any replay
|
||||
// path (driver or generic) builds the retry with the correct
|
||||
// responseType rather than the stale value captured by initXhr.
|
||||
if ( xhr._puterReq ) xhr._puterReq.responseType = settings.responseType;
|
||||
}
|
||||
|
||||
// ===============================================
|
||||
// TO UNDERSTAND THIS CODE, YOU MUST FIRST
|
||||
// UNDERSTAND THE FOLLOWING TEXT:
|
||||
//
|
||||
// Everything between here and the comment reading
|
||||
// "=== END OF STREAMING ===" is ONLY for handling
|
||||
// requests with content type "application/x-ndjson"
|
||||
// ===============================================
|
||||
|
||||
let is_stream = false;
|
||||
let signal_stream_update = null;
|
||||
let lastLength = 0;
|
||||
let response_complete = false;
|
||||
|
||||
let buffer = '';
|
||||
|
||||
// NOTE: linked-list technically would perform better,
|
||||
// but in practice there are at most 2-3 lines
|
||||
// buffered so this does not matter.
|
||||
const lines_received = [];
|
||||
|
||||
xhr.onreadystatechange = () => {
|
||||
if ( xhr.readyState === 2 ) {
|
||||
if ( xhr.getResponseHeader('Content-Type') !==
|
||||
'application/x-ndjson'
|
||||
) return;
|
||||
is_stream = true;
|
||||
const Stream = async function* Stream () {
|
||||
while ( !response_complete ) {
|
||||
const signal = createDeferred();
|
||||
signal_stream_update = signal.resolve;
|
||||
await signal.promise;
|
||||
if ( response_complete ) break;
|
||||
while ( lines_received.length > 0 ) {
|
||||
const line = lines_received.shift();
|
||||
if ( line.trim() === '' ) continue;
|
||||
const lineObject = (JSON.parse(line));
|
||||
|
||||
// Check for usage limit errors in streaming responses
|
||||
if ( lineObject?.error?.code === 'insufficient_funds' || lineObject?.metadata?.usage_limited === true ) {
|
||||
if ( puter.env === 'web' ) {
|
||||
showUsageLimitDialog('You have reached your usage limit for this account.<br>Please upgrade to continue.');
|
||||
} else if ( puter.env === 'app' ) {
|
||||
await puter.ui.requestUpgrade();
|
||||
}
|
||||
}
|
||||
// Check for email confirmation required (e.g. AI calls)
|
||||
if ( lineObject?.error?.code === 'email_must_be_confirmed' && puter.env === 'web' ) {
|
||||
showEmailConfirmationDialog(lineObject?.error?.message || 'Email confirmation required. Go to Puter.com to confirm your email address.');
|
||||
}
|
||||
|
||||
if ( typeof (lineObject.text) === 'string' ) {
|
||||
Object.defineProperty(lineObject, 'toString', {
|
||||
enumerable: false,
|
||||
value: () => lineObject.text,
|
||||
});
|
||||
}
|
||||
yield lineObject;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const startedStream = Stream();
|
||||
Object.defineProperty(startedStream, 'start', {
|
||||
enumerable: false,
|
||||
value: async (controller) => {
|
||||
const texten = new TextEncoder();
|
||||
for await ( const part of startedStream ) {
|
||||
controller.enqueue(texten.encode(part));
|
||||
}
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
return resolve_func(startedStream);
|
||||
}
|
||||
if ( xhr.readyState === 4 ) {
|
||||
response_complete = true;
|
||||
if ( is_stream ) {
|
||||
signal_stream_update?.();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
xhr.onprogress = function () {
|
||||
if ( ! signal_stream_update ) return;
|
||||
|
||||
const newText = xhr.responseText.slice(lastLength);
|
||||
lastLength = xhr.responseText.length; // Update lastLength to the current length
|
||||
|
||||
let hasUpdates = false;
|
||||
for ( let i = 0; i < newText.length; i++ ) {
|
||||
buffer += newText[i];
|
||||
if ( newText[i] === '\n' ) {
|
||||
hasUpdates = true;
|
||||
lines_received.push(buffer);
|
||||
buffer = '';
|
||||
}
|
||||
}
|
||||
|
||||
if ( hasUpdates ) {
|
||||
signal_stream_update();
|
||||
}
|
||||
};
|
||||
|
||||
// ========================
|
||||
// === END OF STREAMING ===
|
||||
// ========================
|
||||
|
||||
// load: success or error
|
||||
xhr.addEventListener('load', async function (response) {
|
||||
if ( is_stream ) {
|
||||
return;
|
||||
}
|
||||
const resp = await parseResponse(response.target);
|
||||
|
||||
// Log driver call response
|
||||
if ( this._puterDriverRequestInfo && globalThis.puter?.apiCallLogger?.isEnabled() ) {
|
||||
const logDriver = fields => {
|
||||
if ( requestInfo && globalThis.puter?.apiCallLogger?.isEnabled() ) {
|
||||
globalThis.puter.apiCallLogger.logRequest({
|
||||
service: 'drivers',
|
||||
operation: `${this._puterDriverRequestInfo.interface}::${this._puterDriverRequestInfo.method}`,
|
||||
params: { interface: this._puterDriverRequestInfo.interface, driver: this._puterDriverRequestInfo.driver, method: this._puterDriverRequestInfo.method, args: this._puterDriverRequestInfo.args },
|
||||
result: response.status >= 400 || resp?.success === false ? null : resp,
|
||||
error: response.status >= 400 || resp?.success === false ? resp : null,
|
||||
operation: `${driverInterface}::${driverMethod}`,
|
||||
params: { interface: driverInterface, driver: driverName, method: driverMethod, args: driverArgs },
|
||||
...fields,
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
// Check for usage limit errors and show upgrade dialog
|
||||
const isInsufficientFunds = (response.target?.status === 402) ||
|
||||
// The request spec is rebuilt per attempt by the retry engine, so a reauth
|
||||
// replay carries a freshly-minted token in the body.
|
||||
const spec = {
|
||||
url: puter.APIOrigin + '/drivers/call',
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': contentType },
|
||||
withCredentials: true,
|
||||
responseType: settings.responseType || '',
|
||||
buildBody: () => JSON.stringify({
|
||||
interface: driverInterface,
|
||||
driver: driverName,
|
||||
test_mode: settings?.test_mode,
|
||||
method: driverMethod,
|
||||
args: driverArgs,
|
||||
auth_token: puter.authToken,
|
||||
}),
|
||||
};
|
||||
|
||||
// Wrap the engine's raw parsed-line NDJSON stream with the driver's per-line
|
||||
// semantics (usage-limit / email-confirmation dialogs, `toString`) and the
|
||||
// ReadableStream `.start` adapter, then resolve with it.
|
||||
const shapeStream = lineStream => {
|
||||
const startedStream = (async function* () {
|
||||
for await ( const lineObject of lineStream ) {
|
||||
if ( lineObject?.error?.code === 'insufficient_funds' || lineObject?.metadata?.usage_limited === true ) {
|
||||
if ( puter.env === 'web' ) {
|
||||
showUsageLimitDialog('You have reached your usage limit for this account.<br>Please upgrade to continue.');
|
||||
} else if ( puter.env === 'app' ) {
|
||||
await puter.ui.requestUpgrade();
|
||||
}
|
||||
}
|
||||
if ( lineObject?.error?.code === 'email_must_be_confirmed' && puter.env === 'web' ) {
|
||||
showEmailConfirmationDialog(lineObject?.error?.message || 'Email confirmation required. Go to Puter.com to confirm your email address.');
|
||||
}
|
||||
if ( typeof (lineObject.text) === 'string' ) {
|
||||
Object.defineProperty(lineObject, 'toString', {
|
||||
enumerable: false,
|
||||
value: () => lineObject.text,
|
||||
});
|
||||
}
|
||||
yield lineObject;
|
||||
}
|
||||
})();
|
||||
Object.defineProperty(startedStream, 'start', {
|
||||
enumerable: false,
|
||||
value: async (controller) => {
|
||||
const texten = new TextEncoder();
|
||||
for await ( const part of startedStream ) {
|
||||
controller.enqueue(texten.encode(part));
|
||||
}
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
return resolve_func(startedStream);
|
||||
};
|
||||
|
||||
// Interpret the final (non-stream) driver response. Reauth, permission-grant,
|
||||
// and transient retries have already been handled by the engine, so anything
|
||||
// reaching here is terminal.
|
||||
const shape = async outcome => {
|
||||
if ( outcome.networkError ) {
|
||||
logDriver({ error: { message: 'Network error occurred' } });
|
||||
return handle_error(error_cb, reject_func, outcome.xhr);
|
||||
}
|
||||
|
||||
const xhr = outcome.xhr;
|
||||
const status = xhr.status;
|
||||
const resp = await parseResponse(xhr);
|
||||
|
||||
logDriver({
|
||||
result: status >= 400 || resp?.success === false ? null : resp,
|
||||
error: status >= 400 || resp?.success === false ? resp : null,
|
||||
});
|
||||
|
||||
const isInsufficientFunds = (status === 402) ||
|
||||
(resp?.error?.code === 'insufficient_funds') ||
|
||||
(resp?.error?.status === 402);
|
||||
const isUsageLimited = resp?.metadata?.usage_limited === true;
|
||||
|
||||
if ( (isInsufficientFunds || isUsageLimited) && puter.env === 'web' ) {
|
||||
showUsageLimitDialog('Your account has not enough funding to complete this request.<br>Please upgrade to continue.');
|
||||
} else if ( (isInsufficientFunds || isUsageLimited) && puter.env === 'app' ) {
|
||||
await puter.ui.requestUpgrade();
|
||||
}
|
||||
|
||||
// Check for email confirmation required (e.g. AI calls) - web only
|
||||
if ( resp?.error?.code === 'email_must_be_confirmed' && puter.env === 'web' ) {
|
||||
showEmailConfirmationDialog(resp?.error?.message || 'Email confirmation required. Go to Puter.com to confirm your email address.');
|
||||
}
|
||||
|
||||
// HTTP Error - unauthorized
|
||||
if ( response.target.status === 401 || resp?.code === 'token_auth_failed' ) {
|
||||
// v2 reauth signal. Replay the driver call by re-entering
|
||||
// driverCall_ rather than using the
|
||||
// generic replayXhrAfterReauth helper: the generic helper
|
||||
// wires the retried XHR through setupXhrEventHandlers, which
|
||||
// resolves with the parsed response and skips driverCall_'s
|
||||
// streaming detection, usage-limit / email-confirmation
|
||||
// handling, settings.transform, and `resp.result` unwrapping
|
||||
// — i.e. it would silently change the driver call API
|
||||
// contract on retry. One-shot via `settings._reauthReplayed`
|
||||
// so a fresh-token rejection bubbles up instead of looping.
|
||||
if ( resp?.code === 'reauth_required' ) {
|
||||
try {
|
||||
await puter.triggerReauth({
|
||||
reason: resp.reason,
|
||||
auth_id: resp.auth_id,
|
||||
});
|
||||
if ( ! settings._reauthReplayed ) {
|
||||
return driverCall_(
|
||||
options,
|
||||
resolve_func,
|
||||
reject_func,
|
||||
driverInterface,
|
||||
driverName,
|
||||
driverMethod,
|
||||
driverArgs,
|
||||
method,
|
||||
contentType,
|
||||
{ ...settings, _reauthReplayed: true },
|
||||
);
|
||||
}
|
||||
// Already replayed once — the fresh token is still
|
||||
// rejected. Bubble the reauth error.
|
||||
const err = {
|
||||
status: 401,
|
||||
code: 'reauth_required',
|
||||
reason: resp.reason,
|
||||
auth_id: resp.auth_id,
|
||||
message: 'Reauthentication still required after retry',
|
||||
};
|
||||
if ( error_cb && typeof error_cb === 'function' ) error_cb(err);
|
||||
return reject_func(err);
|
||||
} catch ( e ) {
|
||||
const err = {
|
||||
status: 401,
|
||||
code: 'reauth_required',
|
||||
reason: resp.reason,
|
||||
auth_id: resp.auth_id,
|
||||
message: e?.message || 'Reauthentication required',
|
||||
};
|
||||
if ( error_cb && typeof error_cb === 'function' ) error_cb(err);
|
||||
return reject_func(err);
|
||||
}
|
||||
}
|
||||
if ( resp?.code === 'token_auth_failed' && puter.env === 'web' ) {
|
||||
try {
|
||||
puter.resetAuthToken();
|
||||
await puter.ui.authenticateWithPuter();
|
||||
} catch (e) {
|
||||
return reject_func({
|
||||
error: {
|
||||
code: 'auth_canceled', message: 'Authentication canceled',
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
// if error callback is provided, call it
|
||||
if ( error_cb && typeof error_cb === 'function' )
|
||||
{
|
||||
error_cb({ status: 401, message: 'Unauthorized' });
|
||||
}
|
||||
// reject promise
|
||||
// Unauthorized — the reauth / token_auth_failed flows already ran in the
|
||||
// engine's classifier; a leftover 401 here is terminal.
|
||||
if ( status === 401 || resp?.code === 'token_auth_failed' ) {
|
||||
error_cb({ status: 401, message: 'Unauthorized' });
|
||||
return reject_func({ status: 401, message: 'Unauthorized' });
|
||||
}
|
||||
// HTTP Error - other
|
||||
else if ( response.target.status && response.target.status !== 200 ) {
|
||||
// if error callback is provided, call it
|
||||
// Other HTTP error
|
||||
if ( status && status !== 200 ) {
|
||||
error_cb(resp);
|
||||
// reject promise
|
||||
return reject_func(resp);
|
||||
}
|
||||
// HTTP Success
|
||||
else {
|
||||
// Driver Error: permission denied
|
||||
if ( resp.success === false && resp.error?.code === 'permission_denied' ) {
|
||||
let perm = await puter.ui.requestPermission({ permission: `driver:${ driverInterface }:${ driverMethod}` });
|
||||
// try sending again if permission was granted
|
||||
if ( perm.granted ) {
|
||||
// repeat request with permission granted
|
||||
return driverCall_(options, resolve_func, reject_func, driverInterface, driverMethod, driverArgs, method, contentType, settings);
|
||||
} else {
|
||||
// if error callback is provided, call it
|
||||
error_cb(resp);
|
||||
// reject promise
|
||||
return reject_func(resp);
|
||||
}
|
||||
}
|
||||
// Driver Error: other
|
||||
else if ( resp.success === false ) {
|
||||
// if error callback is provided, call it
|
||||
error_cb(resp);
|
||||
// reject promise
|
||||
return reject_func(resp);
|
||||
}
|
||||
|
||||
let result = resp.result !== undefined ? resp.result : resp;
|
||||
if ( settings.transform ) {
|
||||
result = await settings.transform(result);
|
||||
}
|
||||
|
||||
// Success: if callback is provided, call it
|
||||
if ( resolve_func.success )
|
||||
{
|
||||
success_cb(result);
|
||||
}
|
||||
// Success: resolve with the result
|
||||
return resolve_func(result);
|
||||
// Driver-level error (incl. a permission_denied the engine couldn't clear)
|
||||
if ( resp.success === false ) {
|
||||
error_cb(resp);
|
||||
return reject_func(resp);
|
||||
}
|
||||
});
|
||||
|
||||
// error
|
||||
xhr.addEventListener('error', function (e) {
|
||||
// Log driver call error
|
||||
if ( this._puterDriverRequestInfo && globalThis.puter?.apiCallLogger?.isEnabled() ) {
|
||||
globalThis.puter.apiCallLogger.logRequest({
|
||||
service: 'drivers',
|
||||
operation: `${this._puterDriverRequestInfo.interface}::${this._puterDriverRequestInfo.method}`,
|
||||
params: { interface: this._puterDriverRequestInfo.interface, driver: this._puterDriverRequestInfo.driver, method: this._puterDriverRequestInfo.method, args: this._puterDriverRequestInfo.args },
|
||||
error: { message: 'Network error occurred', event: e.type },
|
||||
});
|
||||
let result = resp.result !== undefined ? resp.result : resp;
|
||||
if ( settings.transform ) {
|
||||
result = await settings.transform(result);
|
||||
}
|
||||
return handle_error(error_cb, reject_func, this);
|
||||
});
|
||||
|
||||
// send request
|
||||
xhr.send(JSON.stringify({
|
||||
interface: driverInterface,
|
||||
driver: driverName,
|
||||
test_mode: settings?.test_mode,
|
||||
method: driverMethod,
|
||||
args: driverArgs,
|
||||
auth_token: puter.authToken,
|
||||
}));
|
||||
if ( resolve_func.success ) {
|
||||
success_cb(result);
|
||||
}
|
||||
return resolve_func(result);
|
||||
};
|
||||
|
||||
sendWithRetry(spec, {
|
||||
// Read-style driver methods opt into transient retry via settings.readonly.
|
||||
retrySafe: !! settings.readonly,
|
||||
permission: `driver:${ driverInterface }:${ driverMethod }`,
|
||||
shapeStream,
|
||||
shape,
|
||||
}).catch(reject_func);
|
||||
}
|
||||
|
||||
async function blob_to_url (blob) {
|
||||
|
||||
@@ -80,7 +80,7 @@ class Apps {
|
||||
const { limit, offset, cursor, includeTotal, stream, ...params } = opts;
|
||||
const hasCursor = Object.prototype.hasOwnProperty.call(opts, 'cursor');
|
||||
|
||||
const select = utils.make_driver_method(['uid'], 'puter-apps', 'es:app', 'select');
|
||||
const select = utils.make_driver_method(['uid'], 'puter-apps', 'es:app', 'select', { readonly: true });
|
||||
const base = { predicate: ['user-can-edit'] };
|
||||
if ( isObjectForm ) base.params = params;
|
||||
if ( limit !== undefined ) base.limit = limit;
|
||||
@@ -230,7 +230,7 @@ class Apps {
|
||||
if ( typeof args[0] === 'object' && args[0] !== null ) {
|
||||
options.params = args[0];
|
||||
}
|
||||
return this.#addUserIterationToApp(await utils.make_driver_method(['uid'], 'puter-apps', 'es:app', 'read').call(this, options));
|
||||
return this.#addUserIterationToApp(await utils.make_driver_method(['uid'], 'puter-apps', 'es:app', 'read', { readonly: true }).call(this, options));
|
||||
};
|
||||
|
||||
delete = async (...args) => {
|
||||
|
||||
@@ -48,7 +48,7 @@ class Hosting {
|
||||
|
||||
// todo document the `Subdomain` object.
|
||||
list = (...args) => {
|
||||
const select = utils.make_driver_method([], 'puter-subdomains', undefined, 'select');
|
||||
const select = utils.make_driver_method([], 'puter-subdomains', undefined, 'select', { readonly: true });
|
||||
|
||||
const opts = (typeof args[0] === 'object' && args[0] !== null) ? args[0] : {};
|
||||
const { limit, offset, cursor, includeTotal, stream, success, error, ...rest } = opts;
|
||||
@@ -171,7 +171,7 @@ class Hosting {
|
||||
|
||||
options = { id: { subdomain: args[0] } };
|
||||
}
|
||||
return utils.make_driver_method(['uid'], 'puter-subdomains', undefined, 'read').call(this, options);
|
||||
return utils.make_driver_method(['uid'], 'puter-subdomains', undefined, 'read', { readonly: true }).call(this, options);
|
||||
};
|
||||
|
||||
delete = async (...args) => {
|
||||
|
||||
@@ -231,6 +231,7 @@ export async function listEngines (options = {}) {
|
||||
|
||||
return await utils.make_driver_method(['source'], 'puter-tts', ttsDriverName(provider), 'list_engines', {
|
||||
puter,
|
||||
readonly: true,
|
||||
responseType: 'text',
|
||||
})(params);
|
||||
}
|
||||
@@ -277,6 +278,7 @@ export async function listVoices (options) {
|
||||
|
||||
return utils.make_driver_method(['source'], 'puter-tts', ttsDriverName(provider), 'list_voices', {
|
||||
puter,
|
||||
readonly: true,
|
||||
responseType: 'text',
|
||||
})(params);
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import { assertKeySize } from './lib/validate.js';
|
||||
const getDriverCall = (puter, args) =>
|
||||
utils.make_driver_method(['key'], 'puter-kvstore', undefined, 'get', {
|
||||
puter,
|
||||
readonly: true,
|
||||
preprocess: (driverArgs) => {
|
||||
assertKeySize(driverArgs.key);
|
||||
return driverArgs;
|
||||
|
||||
@@ -213,7 +213,7 @@ export function list (patternOrOptions, returnValuesOrOptConfig, maybeOptConfig)
|
||||
nudgeOnce(this, 'includeTotal', '`includeTotal` runs a metered count over every key matching the query, so its cost grows with the store. Request the total once — on the first page — and avoid it in hot paths; to know whether more pages exist, check for `cursor` instead.');
|
||||
}
|
||||
|
||||
const callList = utils.make_driver_method([], 'puter-kvstore', undefined, 'list', { puter: this.puter });
|
||||
const callList = utils.make_driver_method([], 'puter-kvstore', undefined, 'list', { puter: this.puter, readonly: true });
|
||||
|
||||
if ( stream ) {
|
||||
if ( options.offset !== undefined ) {
|
||||
|
||||
Reference in New Issue
Block a user