Transport-agnostic request queue with steady-state pacing, a hard sliding-window rate budget, a shared rate-limit cooldown breaker, linear retry backoff, and a per-attempt watchdog. @require-able from other userscripts.
لا ينبغي أن لا يتم تثبيت هذا السكريت مباشرة. هو مكتبة لسكبتات لتشمل مع التوجيه الفوقية // @require https://update.greasyfork.org/scripts/589318/1890407/Request%20Queue%20%28backoff%29%20Library.js
// ==UserScript==
// @name Request Queue (backoff) Library
// @namespace 861ddd094884eac5bea7a3b12e074f34
// @description Transport-agnostic request queue with steady-state pacing, a hard sliding-window rate budget, a shared rate-limit cooldown breaker, linear retry backoff, and a per-attempt watchdog. Optional hooks let several queues share one budget across tabs. @require-able from other userscripts.
// @author Anonymous
// @version 1.1.1
// @license MIT-0
// ==/UserScript==
/* OVERVIEW
***********
A request queue/pump interface.
- the transport is caller-supplied via an async `perform(job)` -- so it works
with GM_xmlhttpRequest, fetch, $.ajax, etc.; caching/parsing live in perform
- it is exposed as a global (`RequestQueue.create`) for @require
What the queue does:
- steady-state pacing: a constant inter-request floor, DERIVED as
ceil(rateWindow / rateBudget), enforced between successive dispatches
- hard sliding-window backstop: never more than `rateBudget` dispatches per
rolling `rateWindow`; the queue defers (without claiming a slot) until the
oldest in-window send ages out
- shared rate-limit cooldown breaker: a `perform()` rejection flagged
`.rateLimited` pauses the WHOLE queue (not just that job) until a cooldown
elapses, honouring a caller-supplied `.retryAfter` when larger and otherwise
escalating exponentially with the run of consecutive hits; a clean success
resets the streak
- coalesced single-chain pacing: every pump (re)start funnels through one
shared timer, so concurrent triggers can't each spawn a self-sustaining loop
and multiply the effective request rate
- linear retry backoff for ordinary failures (retryDelay * attempt)
- per-attempt watchdog that recovers the queue if a transport hangs without
ever resolving/rejecting (the classic silent stall); if the killed
transport later settles anyway, the caller is resolved and the scheduled
retry is dropped rather than re-dispatched
- priority jobs jump ahead of the queued backlog
- optional observer/gate hooks (externalDelay, onDispatch, onRateLimit,
onSuccess) that let independent queue instances -- separate browser tabs,
say -- pool one budget and one cooldown through a store the caller owns
All state above is per-instance. A queue paces ITSELF; it knows nothing of
any other queue against the same remote. Where several instances share an
endpoint (one per tab is the usual case), the caller pools that state via
the hooks: publish dispatches and cooldowns with onDispatch/onRateLimit, and
gate on the pooled result with externalDelay.
This is NOT a deduplicating cache: concurrent submits are independent jobs.
Callers that want single-flight coalescing do it above the queue.
*/
// RequestQueue.create(config) -> { submit, size, clear }
//
// config.perform
// REQUIRED. async (job) => result. The job is whatever was handed to
// submit(); it is opaque to the queue and forwarded here untouched.
// Resolve with the value to deliver to the submit() promise.
// Reject/throw to fail the attempt (triggers retry/backoff). To trip the
// shared cooldown breaker instead of a normal retry, reject with an error
// whose `.rateLimited` is truthy (optionally `.retryAfter` in ms); this
// pauses the whole queue. Map HTTP 429 + Retry-After to this shape here.
// config.rateWindow
// sliding-window span for the hard backstop (default: 60000)
// config.rateLimit
// hard req/window cap the remote enforces (default: 60). `rateBudget`
// (the hard backstop's dispatch ceiling) is DERIVED from this as
// rateLimit * 0.8 -- an 80% headroom margin, so other applications
// sharing the same endpoint's hard limit always have some of it free --
// and is not separately configurable. rateBudget in turn DERIVES the
// steady-state floor ceil(rateWindow / rateBudget), so pacing alone
// spreads the budget evenly across the whole window instead of front-
// loading it and then stalling; neither knob is separately configurable.
// config.maxRetries
// attempts past the first before giving up (default: 3)
// config.retryDelay
// linear backoff unit; wait = retryDelay * attempt (default: 2000)
// config.cooldownBase
// base rate-limit cooldown; base * 2^streak (default: 60000)
// config.cooldownMax
// ceiling for the escalated cooldown (default: 960000)
// config.maxRateLimitRetries
// rate-limit re-queues before giving up (default: 5)
// config.timeout
// per-attempt budget the watchdog assumes (default: 5000)
// config.watchdogGrace
// extra ms past `timeout` before deeming a call hung (default: 3000)
// config.name
// log prefix (default: 'request-queue')
// config.debug
// when truthy, emits lifecycle logs via console.debug
//
// -- Shared-budget hooks (all optional; absent = a self-contained queue) --
//
// Together these let N independent instances against one endpoint pool their
// pacing through a store the caller owns (localStorage across tabs, say).
// The library holds no opinion on that store: it asks how long to wait and
// reports what it did. All four are called synchronously and their return
// values, other than externalDelay's, are ignored; a throw is caught and
// logged rather than being allowed to wedge the pump.
//
// config.externalDelay
// () => ms. A third dispatch gate, consulted alongside the cooldown
// breaker and the sliding window, in the same defer-without-claiming-a-slot
// phase: return >0 and the queue reschedules itself that far out and
// re-asks; return 0/falsy to proceed. Because it is consulted BEFORE the
// attempt is committed, a wait here is free -- it does not occupy the
// single-flight slot and the watchdog is not yet armed, so returning a
// multi-minute pooled cooldown is safe (whereas sleeping inside perform()
// for the same span would be killed by the watchdog).
// Called on every pump tick, including its own reschedules: keep it cheap
// and memoise anything expensive behind a change event.
// config.onDispatch
// (at) => void. Fires as an attempt is committed, `at` being the epoch-ms
// dispatch stamp just recorded in this instance's sliding window. Publish
// it so peers' externalDelay can count it against a pooled budget.
// config.onRateLimit
// (until, wait, streak) => void. Fires when the cooldown breaker opens:
// `until` is the epoch-ms deadline, `wait` its span, `streak` the run of
// consecutive hits after this one. Publish so peers back off too.
// config.onSuccess
// () => void. Fires on a clean success, the same event that resets this
// instance's escalation streak. Publish so a pooled streak can clear.
//
// submit(job) -> Promise resolves with perform()'s value, rejects on terminal
// failure. A truthy `job.priority` jumps the queue.
// size() not-yet-started job count
// clear() drop the backlog and reset pacing/cooldown state
// (an outstanding in-flight attempt still settles)
(function(root) {
'use strict';
function createRequestQueue(config) {
config = config || {};
const perform = config.perform;
if (typeof perform !== 'function') {
throw new Error('RequestQueue.create: `perform` (async job => result)'
+ ' is required');
}
const rateWindow = config.rateWindow != null ? config.rateWindow : 60000;
const rateLimit = config.rateLimit != null ? config.rateLimit : 60;
const maxRetries = config.maxRetries != null ? config.maxRetries : 3;
const retryDelay = config.retryDelay != null ? config.retryDelay : 2000;
const cooldownBase = config.cooldownBase != null ? config.cooldownBase : 60000;
const cooldownMax = config.cooldownMax != null ? config.cooldownMax : 960000;
const maxRateLimitRetries = config.maxRateLimitRetries != null
? config.maxRateLimitRetries : 5;
const timeout = config.timeout != null ? config.timeout : 5000;
const watchdogGrace = config.watchdogGrace != null ? config.watchdogGrace : 3000;
const name = config.name || 'request-queue';
// Shared-budget hooks. Optional throughout: absent means a queue that
// paces only itself, which is the 1.0 behaviour.
const externalDelay = typeof config.externalDelay === 'function'
? config.externalDelay : null;
const onDispatch = typeof config.onDispatch === 'function' ? config.onDispatch : null;
const onRateLimit = typeof config.onRateLimit === 'function' ? config.onRateLimit : null;
const onSuccess = typeof config.onSuccess === 'function' ? config.onSuccess : null;
// Hard backstop ceiling, DERIVED from rateLimit with a fixed 80%
// headroom margin (clock jitter, unrelated in-flight retries). Not
// separately configurable.
const rateBudgetPct = 0.8;
const rateBudget = Math.floor(rateLimit * rateBudgetPct);
// Steady-state inter-request floor, DERIVED from rateWindow/rateBudget.
// Not separately configurable. This is what makes pacing alone spread
// rateBudget dispatches evenly across rateWindow -- e.g. at the
// defaults (rateLimit 60 -> rateBudget 48, rateWindow 60000) paced
// continuously it lands exactly 48 dispatches per 60s, leaving the
// other 20% of the hard limit free throughout the window, rather than
// exhausting the budget early and then stalling for the remainder.
const minInterval = Math.ceil(rateWindow / rateBudget);
const debug = config.debug
? function() { console.debug.apply(console,
[`[${name}]`].concat([].slice.call(arguments))); }
: function() {};
const warn = function() { console.warn.apply(console,
[`[${name}]`].concat([].slice.call(arguments))); };
// Caller hooks run inside the pump, so a throwing one would strand the
// queue mid-dispatch (or, from the gate, spin it). Swallow and carry on
// with the fallback: the caller's bookkeeping is not worth the pump.
function safeHook(hook, what, fallback, args) {
if (!hook) return fallback;
try {
const v = hook.apply(null, args || []);
return v;
} catch (e) {
warn(`${what} hook threw; ignoring`, e);
return fallback;
}
}
const queue = []; // [{ job, retries, rl, finished, resolve, reject }]
let processing = false; // single-flight: at most one perform() outstanding
// Single shared timer for the next pump tick. EVERY (re)start of the pump
// -- external submits, the post-request resume in done(), and the
// cooldown/window-full deferrals -- funnels through schedulePump() so at
// most one pending tick ever exists. `processing` only blocks two requests
// being outstanding at the same instant; it does NOT stop N independent
// paced loops from interleaving (submit a job in the gap between done() and
// its scheduled resume and each would spawn its own self-sustaining chain,
// multiplying the request rate). Coalescing all pumps into one timer is
// what prevents that second chain from forming.
let pumpTimer = null;
// Rate-limit cooldown breaker. A flagged rejection pauses the *entire*
// queue (single shared gate, not per-job) until `cooldownUntil`, honouring
// the caller's `retryAfter` when larger and otherwise escalating
// exponentially with the run of consecutive hits. A clean success resets
// the streak.
let cooldownUntil = 0; // epoch ms; queue is gated while now < this
let cooldownStreak = 0; // consecutive rate-limit hits, drives escalation
// Sliding-window dispatch log: timestamps of recent sends, trimmed to
// `rateWindow`. Drives the hard backstop (windowCount).
const sendTimes = [];
function pruneSendTimes(now) {
while (sendTimes.length && now - sendTimes[0] > rateWindow) {
sendTimes.shift();
}
}
function recordSend(now) { sendTimes.push(now); pruneSendTimes(now); }
function windowCount(now) { pruneSendTimes(now); return sendTimes.length; }
function label(job) {
return (job && (job.label != null ? job.label : job.url)) || 'job';
}
// priority jobs jump the backlog
function enqueue(rec) {
if (rec.job && rec.job.priority) queue.unshift(rec);
else queue.push(rec);
}
// submit(job): resolves with perform()'s value once the job runs.
function submit(job) {
return new Promise((resolve, reject) => {
enqueue({ job, retries: 0, rl: 0, finished: false, resolve, reject });
debug(`enqueued "${label(job)}" (queue depth ${queue.length},`
+ ` processing=${processing})`);
schedulePump(0);
});
}
// The sole entry point for (re)starting the pump. If a tick is already
// pending we coalesce: the existing timer wins and this call is a no-op, so
// concurrent triggers can never each own a chain. A dropped pump (one that
// fires while a request is still outstanding and bails at the `processing`
// guard) is always recovered -- the outstanding request's done() reschedules.
function schedulePump(delay) {
if (pumpTimer != null) return; // a tick is already pending; coalesce
pumpTimer = setTimeout(() => {
pumpTimer = null;
pump();
}, Math.max(0, delay || 0));
}
function pump() {
if (processing) {
debug(`pump: a request is already in flight`
+ ` (queue depth ${queue.length})`);
return;
}
if (queue.length === 0) {
debug('pump: queue drained, going idle');
return;
}
// cooldown breaker: hold the whole queue during a rate-limit pause
const cdRemaining = cooldownUntil - Date.now();
if (cdRemaining > 0) {
debug(`pump: rate-limit cooldown, ${Math.round(cdRemaining)}ms`
+ ` remaining (queue depth ${queue.length})`);
schedulePump(cdRemaining);
return;
}
// hard sliding-window backstop: never exceed `rateBudget` per
// `rateWindow`. Defer (don't claim the slot) until the oldest in-window
// request ages out.
const now = Date.now();
const used = windowCount(now);
if (used >= rateBudget) {
const wait = rateWindow - (now - sendTimes[0]) + 1;
debug(`pump: window full (${used}/${rateBudget} in ${rateWindow}ms);`
+ ` deferring ${Math.round(wait)}ms (queue depth ${queue.length})`);
schedulePump(wait);
return;
}
// pooled gate: the same defer-without-claiming-a-slot treatment for
// whatever budget/cooldown the caller shares with other instances.
// Deliberately the last gate and deliberately before the slot is
// claimed -- nothing here is committed yet, so a wait of any length
// costs nothing and never meets the watchdog.
const external = safeHook(externalDelay, 'externalDelay', 0) || 0;
if (external > 0) {
debug(`pump: external gate holding ${Math.round(external)}ms`
+ ` (queue depth ${queue.length})`);
schedulePump(external);
return;
}
processing = true;
const rec = queue.shift();
// A record whose submit() promise already settled (a killed
// transport that came back between the watchdog's re-enqueue and
// this dispatch) must not run again: drop it before a slot is
// claimed or a dispatch recorded.
if (rec.finished) {
debug(`dropping already-settled "${label(rec.job)}" before`
+ ` dispatch`);
processing = false;
schedulePump(0);
return;
}
const startedAt = Date.now();
recordSend(startedAt);
safeHook(onDispatch, 'onDispatch', undefined, [startedAt]);
debug(`-> "${label(rec.job)}" (attempt ${rec.retries + 1}, queue depth now`
+ ` ${queue.length}; ${sendTimes.length}/${rateBudget} used in`
+ ` ${rateWindow}ms window)`);
// `done` is idempotent: the watchdog and a (late) perform() settlement
// may both fire; whichever lands first unblocks the queue. A rate-limit
// hit supplies its cooldown as `overrideDelay` and bypasses the steady
// interval entirely. `rec.finished` is the record-level analogue: once
// the submit() promise settles, the retry a watchdog kill scheduled
// must not re-dispatch the job -- the caller already has its result,
// so the second run would be an invisible duplicate request. Both
// re-entry points (the retry timer and the pump dispatch) check it.
let settled = false;
let watchdog = null;
const done = (success, overrideDelay) => {
if (settled) {
debug(`ignored duplicate settle for "${label(rec.job)}"`
+ ` (success=${success})`);
return;
}
settled = true;
if (watchdog) { clearTimeout(watchdog); watchdog = null; }
processing = false;
const delay = (overrideDelay != null) ? overrideDelay : minInterval;
debug(`<- "${label(rec.job)}" settled in ${Date.now() - startedAt}ms`
+ ` (success=${success}); next request in ${Math.round(delay)}ms,`
+ ` ${queue.length} still queued`);
schedulePump(delay);
};
// Recover from a transport that never settles its promise.
watchdog = setTimeout(() => {
warn(`"${label(rec.job)}" produced no result after`
+ ` ${timeout + watchdogGrace}ms; treating as failed to unstick`
+ ` the queue`);
fail(rec, done, new Error('hung'));
}, timeout + watchdogGrace);
Promise.resolve()
.then(() => perform(rec.job))
.then(
result => {
cooldownStreak = 0; // a clean success clears the breaker
safeHook(onSuccess, 'onSuccess', undefined);
rec.finished = true;
rec.resolve(result); done(true);
},
err => {
if (err && err.rateLimited) return rateLimited(rec, err, done);
debug(`perform failed for "${label(rec.job)}"`, err);
fail(rec, done, err);
});
}
function fail(rec, done, err) {
if (rec.retries < maxRetries) {
rec.retries++;
const wait = retryDelay * rec.retries; // linear backoff
debug(`retry ${rec.retries}/${maxRetries} for "${label(rec.job)}"`
+ ` in ${wait}ms`);
setTimeout(() => {
if (rec.finished) {
debug(`skipping retry for "${label(rec.job)}":`
+ ` settled while waiting`);
return;
}
enqueue(rec); schedulePump(0);
}, wait);
done(false);
} else {
debug(`giving up on "${label(rec.job)}" after ${maxRetries} retries`);
rec.finished = true;
rec.reject(err || new Error('request failed'));
done(false);
}
}
// Rate-limit hit: open the cooldown breaker. Pause the whole queue until
// the cooldown elapses; the job is re-queued (without spending a normal
// retry) unless it has been rate-limited too many times, in which case we
// give up on it.
function rateLimited(rec, err, done) {
const headerWait = (typeof err.retryAfter === 'number')
? err.retryAfter : null;
const escalated = Math.min(cooldownMax,
cooldownBase * Math.pow(2, cooldownStreak));
cooldownStreak++;
const wait = Math.max(headerWait != null ? headerWait : 0, escalated);
cooldownUntil = Date.now() + wait;
safeHook(onRateLimit, 'onRateLimit', undefined,
[cooldownUntil, wait, cooldownStreak]);
rec.rl++;
warn(`rate-limited on "${label(rec.job)}"; pausing queue`
+ ` ${Math.round(wait)}ms (retryAfter`
+ ` ${headerWait != null ? Math.round(headerWait) + 'ms' : 'absent'},`
+ ` consecutive hits ${cooldownStreak}, ${queue.length} queued)`);
if (rec.rl > maxRateLimitRetries) {
debug(`giving up on "${label(rec.job)}" after ${maxRateLimitRetries}`
+ ` rate-limit retries`);
rec.finished = true;
rec.reject(err || new Error('rate limited'));
} else {
enqueue(rec); // not the job's fault: re-queue, keep its retries
}
done(false, wait);
}
// Drop the backlog and reset pacing/cooldown state. An attempt already in
// flight is left alone: its done() will settle it and find the queue idle.
// Only this instance's state is reset: anything published through the
// hooks belongs to the caller's store, which may still be gating peers
// (and will still gate this queue through externalDelay).
function clear() {
if (pumpTimer != null) { clearTimeout(pumpTimer); pumpTimer = null; }
queue.length = 0;
sendTimes.length = 0;
cooldownUntil = 0;
cooldownStreak = 0;
}
return {
submit,
size: () => queue.length,
clear,
};
}
const api = { create: createRequestQueue };
if (typeof module !== 'undefined' && module.exports) module.exports = api;
root.RequestQueue = api;
})(typeof unsafeWindow !== 'undefined' ? unsafeWindow
: typeof globalThis !== 'undefined' ? globalThis
: this);
/* Usage: rule34 tag-type classifier (caching + GM_xmlhttpRequest transport)
***************************************************************************
// @require this library, then:
// Parse a `Retry-After` header (delta-seconds or HTTP-date) -> ms, or null. This
// is HTTP-specific, so it lives in the caller's perform, not in the library.
function parseRetryAfter(headers) {
if (!headers) return null;
const m = /^retry-after:[ \t]*(.+?)[ \t]*$/im.exec(headers);
if (!m) return null;
if (/^\d+$/.test(m[1])) return parseInt(m[1], 10) * 1000;
const when = Date.parse(m[1]);
return Number.isNaN(when) ? null : Math.max(0, when - Date.now());
}
const UNKNOWN = -1;
const queue = RequestQueue.create({
name: 'r34vi-tags',
// The tag API allows ~60 requests per rolling 60s (NOT 60/s). rateLimit is
// that hard cap; the library derives an 80%-headroom budget (48 dispatches)
// and, from that, the steady ~1250ms floor as ceil(rateWindow / rateBudget)
// -- spreading the budget evenly across the window instead of front-
// loading it, so other apps sharing the endpoint always have headroom.
rateWindow: 60000,
rateLimit: 60,
debug: DEBUG,
perform: (job) => new Promise((resolve, reject) => {
GM_xmlhttpRequest({
method: job.method, url: job.url, timeout: 5000,
onload: xhr => {
if (xhr.status === 429) { // trip the shared cooldown breaker
const e = new Error('HTTP 429');
e.rateLimited = true;
e.retryAfter = parseRetryAfter(xhr.responseHeaders); // ms|null
return reject(e);
}
if (xhr.status !== 200) return reject(new Error('HTTP ' + xhr.status));
resolve(xhr);
},
onerror: () => reject(new Error('network')),
ontimeout: () => reject(new Error('timeout')),
});
}),
});
// app-level single-flight + caching layered above the queue
const known = new Map();
const listeners = new Map();
function classify(name, cb) {
if (known.has(name)) return cb(known.get(name));
if (listeners.has(name)) { listeners.get(name).push(cb); return; }
listeners.set(name, [cb]);
queue.submit({ method: 'GET', label: name,
url: `${TAG_API_BASE}&name=${encodeURIComponent(name)}${creds}` })
.then(xhr => {
const type = parseTagType(xhr.responseText);
const resolved = (type == null) ? UNKNOWN : type;
known.set(name, resolved);
(listeners.get(name) || []).forEach(fn => fn(resolved));
listeners.delete(name);
})
.catch(() => {
known.set(name, UNKNOWN);
(listeners.get(name) || []).forEach(fn => fn(UNKNOWN));
listeners.delete(name);
});
}
*/