From 6e6d9d2869e09bb4b3329aa5db8006d61d978af1 Mon Sep 17 00:00:00 2001 From: Neal Shah <30693865+ProgrammerIn-wonderland@users.noreply.github.com> Date: Fri, 24 Jul 2026 18:37:09 -0400 Subject: [PATCH] buildXhr, dedup, retry (#3432) * buildXhr, dedup, retry * change retry intervals --- src/puter-js/src/lib/networkUtils.js | 459 ++++++++++++++++------ src/puter-js/src/lib/networkUtils.test.js | 184 ++++++++- src/puter-js/src/lib/utils.js | 456 ++++++--------------- src/puter-js/src/modules/Apps.js | 4 +- src/puter-js/src/modules/Hosting.js | 4 +- src/puter-js/src/modules/ai/tts.js | 2 + src/puter-js/src/modules/kv/get.js | 1 + src/puter-js/src/modules/kv/list.js | 2 +- 8 files changed, 655 insertions(+), 457 deletions(-) diff --git a/src/puter-js/src/lib/networkUtils.js b/src/puter-js/src/lib/networkUtils.js index e1bdb6aab..b0986cbeb 100644 --- a/src/puter-js/src/lib/networkUtils.js +++ b/src/puter-js/src/lib/networkUtils.js @@ -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::` 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} factory - runs the request; called only on a miss. + * @param {{windowMs?: number}} [opts] + * @returns {Promise} 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} */ @@ -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 }; diff --git a/src/puter-js/src/lib/networkUtils.test.js b/src/puter-js/src/lib/networkUtils.test.js index 2f7c2e849..6a1c90558 100644 --- a/src/puter-js/src/lib/networkUtils.test.js +++ b/src/puter-js/src/lib/networkUtils.test.js @@ -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'); + }); +}); diff --git a/src/puter-js/src/lib/utils.js b/src/puter-js/src/lib/utils.js index 1d6ec3ae6..adb5e6f05 100644 --- a/src/puter-js/src/lib/utils.js +++ b/src/puter-js/src/lib/utils.js @@ -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.
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.
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.
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) { diff --git a/src/puter-js/src/modules/Apps.js b/src/puter-js/src/modules/Apps.js index d45daff27..5287954fc 100644 --- a/src/puter-js/src/modules/Apps.js +++ b/src/puter-js/src/modules/Apps.js @@ -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) => { diff --git a/src/puter-js/src/modules/Hosting.js b/src/puter-js/src/modules/Hosting.js index 4c9a6077f..d7d67c45a 100644 --- a/src/puter-js/src/modules/Hosting.js +++ b/src/puter-js/src/modules/Hosting.js @@ -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) => { diff --git a/src/puter-js/src/modules/ai/tts.js b/src/puter-js/src/modules/ai/tts.js index ecfbe9357..acba972b3 100644 --- a/src/puter-js/src/modules/ai/tts.js +++ b/src/puter-js/src/modules/ai/tts.js @@ -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); } diff --git a/src/puter-js/src/modules/kv/get.js b/src/puter-js/src/modules/kv/get.js index ce82e496a..be7f453b6 100644 --- a/src/puter-js/src/modules/kv/get.js +++ b/src/puter-js/src/modules/kv/get.js @@ -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; diff --git a/src/puter-js/src/modules/kv/list.js b/src/puter-js/src/modules/kv/list.js index 516c84cdb..0e529d09e 100644 --- a/src/puter-js/src/modules/kv/list.js +++ b/src/puter-js/src/modules/kv/list.js @@ -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 ) {