Fetch Manager

A zero dependency TS wrapper around native Javascript fetch providing management of:

rate limiting
concurrency
paging
retry strategy
response data

For Node, Bun and the browser (and probably Deno too).

Basic usage

Initiate a Fetch Manager instance

This will act upon a unique set of targets.

importFetchManagerfrom"fetch-manager";consttime_period="min";constrpp_max=200;// Maximum Requests Per Period (rpp) ie. the rate limitconstconcurrency_max=3;// Maximum active requests at any given timeconstfm=newFetchManager(rpp_max,concurrency_max,time_period,["foo.domain.com","bar.domain.com",// Targets can be also "URL.host" or "URL.host + URL.pathname""baz.domain.com/api",]);

example (1)

Use it akin to native fetch

constfoo_res=awaitfm.fetch("https://foo.domain.com/api").catch((err)=>{// `err` is one of: // if no `Response` then `Error`if(errinstanceofError)throwerr;// if no `Response.ok` then `Response`console.error(err.status);});

example (2)

Pre-process response data and inform fm.fetch of the expected return type

constresponse_cb: fm.cb.resp=async(resp,_req)=>{returnresp.json().then((data: bar_t)=>data.bar);};constreq=newRequest("https://bar.domain.com/api")// `bar` will be typedconstbar=awaitfm.fetch<bar_t["bar"]>(req,{ response_cb });

example (3)

a) Wait and retry 503 / 429 status's based on response headers.
b) Accept non-standard framework / module RequestInit shapes

consthandlers: {retry_cb: fm.cb.retry;wait_cb: fm.cb.wait;}={// Should a failed request be retried?retry_cb: (resp,_req)=>{returnrespinstanceofError ? false : [503,429].includes(resp.status);},// How long to pause the queue before retrying?wait_cb: (resp,_req)=>{/* We have already filtered out Errors in retry_cb, * but the TS pre-processor doesn't know that. */if(respinstanceofError)return0;constwait_ms=resp.headers.get("Retry-After")||"500";returnNumber(wait_ms);},};constbaz=awaitfm.fetch("https://baz.domain.com/api/endpoint",// If the global `RequestInit` type definition is not compliant, then force TS typing.// Ensure that your global `fetch` is capable of accepting the non-standard shape. {tls: {rejectUnauthorized: false}}asRequestInit,handlers,);// Unless there is some failure outside of 503 or 429, baz is guaranteed a Response.ok

Core concepts

top

Fetch Manager provides class instances acting on a unique set of one or more given targets. A target is one of:

host
host + pathname

Under the hood it manages a request queue driven by a (configurable) heartbeat for each instance. Many instances can be created which will act independantly and in parallel, but targets cannot overlap on a thread, and should not overlap across a distributed system - ie. a target must not be repeated in multiple instances. If it is, then rate calculations will be inacurate. See later documentation on:

FetchManager.stop and FetchManager.kill methods for re-defining targets.
FetchManager.bucket.get and FetchManager.bucket.set methods for orchestration.

Caching and orchestration frameworks are outside the scope of this library, and left to the user to implement their own.

If and when rate / concurrency limits are reached for an instance, it's queue is paused until the limits are once again within allowances. Logic such as paging and adaptive retry strategies are managed by user provided callback handlers:

retry_cb: answers "should I retry this failed request?" with a boolean response,
wait_cb: answers "for how long to pause the queue?" with a number in ms,
pager_cb: answers "is there more data?" with nullish for false or a new request for true,
response_cb: manipulates the response payload into a desired data shape before resolving it.
trace_cb: provides diagnostic data about the state of the queue.

Handlers (excepting trace) are injected with the Response as well as the request in it's given shape: (see the fm.req type). The injected response for retry_cb and wait_cb can be Response | Error - pager_cb and response_cb are guaranteed a Response. The pager_cb is injected also with an (optional) collect function to facilitate flattening the final data return.

Handlers can be set at 3 cascading levels of priority (excepting pager):

per individual request
at class instantiation per individual target
at class instantiation for the whole target group.

Behaviour can thus be set globally, and overidden granularly. Paging is set on a per request basis. Fetch Manager is not very opinionated. It is up to the user to build a re-usable kit of callback handlers to manage their application logic.

Each request has further utility options:

prioritise a request to the front of the queue
retry a request x number of times (independent of retry_cb)
prioritise request retries to the front / back of the queue (independent of retry_cb)

Rate and concurrency rules are set at class instantiation. Further default options can be overidden here:

heartbeat: the rate at which the queue is processed (default 20ms)
default retry wait (independant of wait_cb, default 500ms)

Paging

top

... Continued from previous example

/* We inform the callback of the request shape, `<"req" | "url">` * so that the `req` parameter type is disambiguated. */constpager_cb: fm.cb.pager<"req">=async(resp,req,collect)=>{const{ next, data }=(awaitresp.json())asbar_type;collect(data);// optional - overrides `response_cb` and flattens the resolved resultif(!next)return;// return nullish if no more data is expectedconsturl=newURL(req.url);url.searchParams.set("next",next);// remove a stale timeout signal (if it was set) - see the Aborting section// req = new Request(req, { signal: undefined }); returnnewRequest(url,req);// return a request for more data};constreq=newRequest("https://bar.domain.com/api/paged")constall_data=awaitfm.fetch<bar_type["data"][]>(req,pager_cb);// You now have all your paged data in a single batch.

pager_cb will cause fm.fetch to always return an Array (if no fatal error).

In the above example, we used the collect utility function. It is slightly opinionated. If data is any[], then it will flatten the paged results so that all_data is also any[]. This behaviour can be overidden with collect(data, false) in which case all_data will be any[][]

Usage of collect is optional. If not used, all_data will be one of:

If response_cb is defined: Awaited<ReturnType<response_cb>>[] else
Response[]

Trace

top

If used, the trace callback will be executed at every heartbeat while the queue is not paused or empty, so be mindful of the resources it may consume.

example: Log the amount of request tokens remaining for the period, else a message indicating why the queue is stopped.

consttrace_cb: fm.cb.trace=(trace_data)=>{console.debug(trace_data.message||trace_data.tokens);}constfm=newFetchManager(...,{ trace_cb })

trace_data details

typetrace_data={message?: string;// Information about the queue statepaused?: number;// For how many ms is the queue pausedtokens: number;// How many request tokens remain for the periodmax_rpp: number;// The maximum amount of requests allowed for the periodperiod: fm.period;// "sec" | "min" | "hr" | "day"concurrency: number;// How many requests are currently activemax_concurrency: number;// Maximum concurrent requests allowedqueue: number;// The length of the request queueskip_queue: boolean;// Is the active request skipping the queueforce_retry: number;// The amount of retries remaining for the requesttarget_key: string;// The key of the requests limiter buckethref: string;// The href of the current requesttime: number;// Unix Epoch (ms)};

Aborting and timeouts

top

Native Javascript fetch uses signals to manage user controlled aborts. This is compatible with Fetch Manager, eg.

letcount=0;constcontroller=newAbortController();const{ signal }=controller;signal.addEventListener("abort",()=>console.log("abort",count));constresponse_cb: fm.cb.resp=(resp,_req)=>{count++;if(count>1)controller.abort()returnresp.ok;}constpromises=[...Array(4)].map(()=>fm.fetch("https://foo.com",{ signal },{ response_cb }).catch(()=>false););consttwo_of_four=awaitPromise.all(promises)// returns [true, true, false, false] and logs "abort 2"

However, because Fetch Manager is queing requests for unknown lengths of time, attaching a native AbortSignal.timeout at the initial fm.fetch call may lead to unexpected timing results, depending on the result you want to achieve.

For this purpose, you can set an abort_timeout option (in ms) at three levels of priority:

Per request, in the fm.fetch properties object (priority 1)
Default for a target in it's class initialiser options (priority 2)
Default for all targets in the class initialiser options (priority 3)

Fetch Manager will add the timeout signal just in time before the request is sent. For paged queries, don't forget to unset the timeout signal in your pager_cb, else the first page's timer will cascade into subsequent requests.

For example, to set it for a request:

constcontroller=newAbortController();const{ signal }=controller;constreq=newRequest("https://foo.com",{ signal });// Will abort if the request takes longer than 1000msconstresp=awaitfm.fetch(req,{abort_timeout: 1000}).catch((err)=>{console.error(err.name)// "TimeoutError" or "AbortError"})// ...Do stuff that consumes time...controller.abort();

The user's abort signal will remain available, ie. the timeout signal is added - it does not replace user defined signals.

If no abort_timeout is given, the default system timeout will be used - which can vary.

Options and overloads

top

Class initialiser

constfm=newFetchManager(...[max_rpp: number,// The rate limitmax_concurrency: number,// Maximum concurrencypariod: fm.period,// The period of the rate limit, "sec" | "min" | "hr" | "day"targets: [...],// See section belowoptions?: {wait_ms?: number,// Override the default (500) retry wait in msheartbeat?: number,// Override the default (20) queue heartbeat in ms/* Options defined hereunder are  * fallen back on as priority (3) */abort_timeout?: number,// Number of ms after request is sent until abortretry_cb?: fm.cb.retry,wait_cb?: fm.cb.wait,trace_cb?: fm.cb.trace,},])

If a bucket for the instance is available, that may be used to initialise an instance, thus preserving or migrating the state of limits:

constbucket=FetchManager.bucket.get(fm.uid);save_to_file(...,bucket)// Server restartsconstbucket: fm.bucket=get_from_file(...)constfm=newFetchManager(...[bucket,targets: [...],options?: {...}])

The bucket must match the targets, ie. the uid hash of the bucket must align with the targets definition else an error will be thrown. The user must create their own mechanism to store and retrieve targets and their options

Target definitions

A host is as per specification, the string returned by URL.host A pathname is as per specification, the string returned by URL.pathname

A target is one of two forms:

host
host + pathname

A request url is matched against targets on a startsWith basis, so if a target is foo.com/api then

https://foo.com/api/bar - will hit, but
https://foo.com/bar - will miss

Target sets are defined as an Array in two, or a mix of two shapes as below:

["foo.domain.com","bar.domain.com"]

In the shape below, fallback options are specified for bar.domain.com

["foo.domain.com",{target_key: "bar.domain.com",/* Options defined hereunder are * fallen back on as priority (2) */abort_timeout?: number,response_cb?: fm.cb.resp,retry_cb?: fm.cb.retry,wait_cb?: fm.cb.wait,trace_cb?: fm.cb.trace,}]

It is important that a target is only defined once across all instances. Fetch Manager in a single thread context will throw errors on conflicts - but orchestrators of distributed systems should be mindful of this, and more complex scenarios such as the following:

instance_1 - target defined as:

["foo.com"]

This will catch all API endpoints for foo.com, but foo.com may have endpoints with different rate limits, so we can do:

instance_2 - target defined as:

["foo.com/api/special"]

Now calling instance_1.fetch("https://foo.com/api/special") is an error, because even though instance_1can catch it, the target's limit rules are implemented on instance_2. In a single threaded context, Fetch Manager will throw an error - but in a distributed sytem the conflict will need to be managed externally.

Fetch

In order to cater for non-standard RequestInit forms, a request can be defined as one of two shapes:

<Request> or
<"url", {...} as RequestInit>

fm.fetch(Request) is effectively the same as fm.fetch("url", {...} as RequestInit), except that you have the ability to pass non-standard options which native new Request(...) would otherwise reject. For convenience this overload will be notated below as <fm.req>

The anatomy of a fetch is thus ordered as follows:

fm.fetch(<fm.req>,options?,pager_cb?)

This translates to a number of overloaded shapes:

fm.fetch("url")fm.fetch("url",pager_cb)fm.fetch("url",{...options})fm.fetch("url",{...options},pager_cb)fm.fetch("url",{...}asRequestInit)fm.fetch("url",{...}asRequestInit,{...options})fm.fetch("url",{...}asRequestInit,{...options},pager_cb)fm.fetch(Request)fm.fetch(Request,pager_cb)fm.fetch(Request,{...options})fm.fetch(Request,{...options},pager_cb)

Fetch user options

Options passed directly to fm.fetch are priority (1). They will override any options set at individual target level (2) or target group level (3).

{skip_queue?: boolean,// Send this request to the front of the queueforce_retry?: number,// Retry this request x amount of times. -x to retry from the queue front /* Options defined hereunder  * are first priority (1) */abort_timeout?: number,// Number of ms after request is sent until abortresponse_cb?: fm.cb.resp,retry_cb?: fm.cb.retry,// retry_cb always retries from the front of the queuewait_cb?: fm.cb.wait,trace_cb?: fm.cb.trace,}

Advanced usage

top

The Fetch Manager fetch method has some extra options to control queue priority.

skip_queue adds the request to the front of the queue
force_retry forces a failed request to retry x number of times.

skip_queue

constnormal=fm.fetch("http://foo.com/api/normal")constimportant=fm.fetch("http://foo.com/api/important",{skip_queue: true})

important will now be fetched before normal

force_retry

This option is by nature quirky and opinionated. The examples below will illustrate.

constbuggy_endpoint=fm.fetch("http://foo.com/api/buggy",{force_retry: 5})constimportant_stuff=Promise.all([...Array(10)].map(()=>fm.fetch("https://foo.com/api/stuff")))

If buggy_endpoint fails for any reason, it will be re-tried for up to 5 times from the back of the queue. All of important_stuff will thus be requested before buggy_endpoint retries. There are some quirks to this behaviour:

If the default retry wait is set to 500ms, important_stuff will proceed immediately and not wait on the buggy failure.
If important_stuff consumes 300ms to complete it's requests, then buggy_endpoint will wait the remaining 200ms before retrying.
if the user has provided wait_cb, and it returns eg. 1000ms (maybe a rate penalty is being imposed) - then the whole queue, including important_stuff, will wait for 1000ms and the 500ms wait for buggy_endpoint will have expired.

We can also do the inverse, and prioritise request retries to the front of the queue.

constbuggy_important=fm.fetch("http://foo.com/api/buggy",{force_retry: -5})constnormal_stuff=Promise.all([...Array(10)].map(()=>fm.fetch("https://foo.com/api/stuff")))

Note the negative (-5). buggy_important will now retry up to five times from the front of the queue before normal_stuff is requested. If the buggy endpoint takes longer to respond (or error) than the default retry wait time, then normal_stuff will fill in the gaps with requests.

Consider force_retry as an override to prioritise / de-prioritise a single request. A user provided retry_cb will be by-passed until all the request's force_retry counts have been depleted.

Distributed architectures

top

In a single thread context, Fetch Manager will error and warn on conflicts between instances. In a distributed system, the tracking of rates and limits state will have to be orchestrated at a higher level.

Fetch Manager does not in and of itself provide orchestration tooling, but it does expose helpers to access and set the needed data. A bucket contains the limits and current rates for a target group. It is linked to a class instance (a group of targets) by it's hash uid.

fm.uid - Instance uid (it's a hash of target keys sorted longest to shortest)
FetchManager.buckets - Static map of uid keyed buckets
FetchManager.bucket.get(uid_or_target) - Get an existing bucket
FetchManager.bucket.set(uid_or_target, bucket) - Modify an existing bucket
FetchManager.targets - Static map of uid keyed target key groups
FetchManager.hash(targets: fm.target[]) - Create a uid and sorted array of target keys from a target group.

A naive, illustrative pseudo-example for orchestration could look like this:

// source threadconstall_targets=FetchManager.targetsconstbuckets=FetchManager.bucketsyour_orchestration.publish("update",all_targets,buckets)

// destination thread// Your internal registryconstfm_instances: {[uid: string]: InstanceType<FetchManager>}={...}constfm_targets: {[uid: string]: fm.target[]}={...}constfm_options: {[uid:string]: fm.opts.global<fm.kind>}]={...}your_orchestration.on("update",(all_targets,buckets)=>{Object.entries(buckets).forEach(([uid,bucket])=>{// bucket is unused, so save some resourcesif(!bucket.time)return;consttargets=fm_targets[uid]||all_targets[uid]!;constoptions=fm_options[uid]||{};// No instance yet, so create itif(!fm_instances[uid]){fm_instances[uid]=newFetchManager(bucket,targets,options);return}constex_bucket=FetchManager.bucket.get(uid)!;bucket.tokens=Math.min(bucket.tokens,ex_bucket.tokens);bucket.concurrency=bucket.concurrency+ex_bucket.concurrency;// Update the existing instance bucketFetchManager.bucket.set(uid,bucket);)})}

bucket details

typebucket={uid:string,// (immutable) The uid of the group to which this bucket belongstokens: number;// How many requests remain for the periodconcurrency: number;// How many requests are activeperiod: period;// (immutable) "sec" | "min" | "hr" | "day"max_rpp: number;// (immutable) The maximum requests per periodmax_concurrency: number;// (immutable) The maximum concurrent requests allowedtime: number;// (not editable) The last update epoch in ms - 0 until first request is fetched};

Once a bucket is set, only tokens and concurrency can be updated. time will be 0 until the first request is made on a new bucket.

When orchestrating, it is up to the user to validate:

That new values are less favourable than the existing values (the local limiter will be adjusting them favourably)
That timestamps are within tolerance
That new buckets are created with the same limits across the entire system.

If limit settings across a distributed system differ, then global rate calculations will be corrupted.

Instance destruction

top

It may be neeeded to change rate limits for targets - eg. maybe different rates for different times of day. It is possible to destroy an instance and then re-use it's targets with new limiter rules. Two instance methods are provided:

`FetchManager.kill`

This will immediately stop processing the queue and all awaiting requests will be rejected with the message "Target group was killed".

constfm=newFetchManager(...,hosts: [...]);/* Application does stuff with fm */awaitfm.kill();/* Targets can now be re-used in new instances */

`FetchManager.stop`

This will wait for the queue to drain before killing it. It is up to the user to stop feeding the queue, else it will never drain.

constfm=newFetchManager(...,hosts: [...]);/* Application does stuff with fm */awaitfm.stop();/* Targets can now be re-used in new instances */

To cancel in-flight requests, the user should set up an AbortSignal and call it from their application code.

Installation

top

bun add fetch-manager # NPM
bun add citkane/fetchmanager#v0.1.0 #Github
npm i fetch-manager # NPM
npm i citkane/fetchmanager#v0.1.0 #Github

More resources

top

Types

Fetch Manager types are under the fm namespace. They are annotated with examples, so your IDE should give you helpful documentation.

LibFetch

A library of off the shelf re-usable callbacks:

importLibFetchfrom"fetch-manager/lib";constlib_fetch=newLibFetch();constwait_cb=lib_fetch.wait.backoff_factory()constretry_cb=lib_fetch.retry...
...etc

Tests

Tests are made for the Bun framework. You can examine these to better understand the expectations for various aspects of the library.

git clone https://github.com/citkane/fetchmanager.git
cd fetchmanager
bun install
bun run_tests

Demonstration

top

Try a rather nifty WikiData explorer!

npx wikidata-explore
bunx wikidata-explore

It is a Terminal User Interface that queries the public, rate limited Wikibase API

What about Deno?

I don't use Deno, so I am not familiar with it's ecosystem and haven't tried Fetch Manager on it. It will probably work.

Why another `fetch` library?

Because my attention span is rather limited... While working on an application that aggregates data from a number of API's with different rate rules and paging logic, I wanted something with zero dependencies that is as close to the native fetch syntax as possible. The result is this library. It has dramatically reduced my app's boilerplating.

Name		Name	Last commit message	Last commit date
demo		demo
src		src
test		test
.gitignore		.gitignore
.gitmodules		.gitmodules
.npmignore		.npmignore
LICENSE		LICENSE
README.md		README.md
bunfig.toml		bunfig.toml
package-lock.json		package-lock.json
package.json		package.json
tsconfig.json		tsconfig.json

Navigation Menu

Repository files navigation

Fetch Manager

toc

Basic usage

Initiate a Fetch Manager instance

example (1)

example (2)

example (3)

Core concepts

Paging

Trace

Aborting and timeouts

Options and overloads

Class initialiser

Target definitions

Fetch

Fetch user options

Advanced usage

skip_queue

force_retry

Distributed architectures

Instance destruction

`FetchManager.kill`

`FetchManager.stop`

Installation

More resources

Types

LibFetch

Tests

Demonstration

What about Deno?

Why another `fetch` library?

Similar libraries:

About

Releases

Used by

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

Fetch Manager

toc

Basic usage

Initiate a Fetch Manager instance

example (1)

example (2)

example (3)

Core concepts

Paging

Trace

Aborting and timeouts

Options and overloads

Class initialiser

Target definitions

Fetch

Fetch user options

Advanced usage

skip_queue

force_retry

Distributed architectures

Instance destruction

FetchManager.kill

FetchManager.stop

Installation

More resources

Types

LibFetch

Tests

Demonstration

What about Deno?

Why another fetch library?

Similar libraries:

About

Topics

Resources

Stars

Watchers

Forks

Releases

Used by

Contributors

Languages

`FetchManager.kill`

`FetchManager.stop`

Why another `fetch` library?