An in memory queue for async processes in high concurrency code paths.

Wrap any async method in a fetcher process to create a buffer where there will only ever be a single active request for that method at any given time.

A very crude example: given an application that displays public customer information, a common service method would be one that fetches the base customer information.

export const getCustomer = async (id: string) => {
  return await sql.query("SELECT id, name FROM customers WHERE id = ?", [id]);
};

With this function, every request would hit the database directly. Given the data is unlikely to change while multiple requests are active at the same time, the database call can be wrapped inside a BurstValve instance so that only a single concurrent query is ever active for the specified customer.

const valve = new BurstValve<Customer>(async (id: string) => {
  return await sql.query("SELECT id, name FROM customers WHERE id = ?", [id]);
});

export const getCustomer = async (id: string) => {
  return await valve.fetch(id);
};

To better visualize the performance gain, a simple benchmark run was setup to test various levels of concurrency (2022 MacBook Air M2).

Again, this is a very crude example. Adding caching layer in front of the database call would improve the initial performance significantly. Even then, adding BurstValve would still add a layer of improvement as traffic rate increases.

const valve = new BurstValve<Customer>(async (id: string) => {
  const customer = await cache.get(`customer:${id}`);
  if (customer) {
    return customer;
  }

  return await sql.query("SELECT id, name FROM customers WHERE id = ?", [id]);
});

BurstValve comes with a unique batching approach, where requests for multiple unique identifiers can occur individually with parallelism. Consider the following:

const valve = new BurstValve<number, number>({
  batch: async (ids) => {
    await sleep(50);
    return ids.map((id) => id * 2);
  },
});

const [run1, run2, run3, run4] = await Promise.all([
  valve.batch([1, 2, 3]),
  valve.batch([3, 4, 5]),
  valve.fetch(4), // When batch fetcher is defined, all fetch requests route through there
  valve.fetch(8),
]);

run1; // [1, 2, 3] -> [2, 4, 6]
run2; // [3(queued), 4, 5] -> [6, 8, 10]
run3; // [4(queued)] -> 8
run4; // [8] -> 16

In the above example, the valve was able to detect that the identifiers 3 & 4 were already requested (active) by previous batch/fetch calls, which means they are not passed along to the batch fetcher for another query. Only inactive identifiers are requested, all active identifiers are queued to wait for a previous run to complete.

To further the concept of individual queues for batch runs, the batch fetcher process provides an early writing mechanism for broadcasting results as they come in. This gives the ability for queues to be drained as quickly as possible.

const valve = new BurstValve<number, number>({
  batch: async (ids, earlyWrite) => {
    await sleep(50);
    earlyWrite(1, 50);
    await sleep(50);
    earlyWrite(2, 100);
    await sleep(50);
    earlyWrite(3, 150);
  },
});

const [run1, run2, run3] = await Promise.all([
  valve.batch([1, 2, 3]),
  valve.fetch(1),
  valve.fetch(2),
]);

// Resolution Order: run2, run3, run1

Note: While early writing may be used in conjunction with overall batch process returned results, anything early written will take priority over returned results.

Performance for batch fetching will vary depending on the number of overlapping identifiers being requested, but in an optimal scenario (high bursty traffic for specific data), the gains are significant.

And similar to the fetch suite at the top, gains are amplified when putting a memcached layer in front

The unsafeBatch method is for cases where batch fetching will throw errors instead of returning them. This provides a typesafe way to fetch an array of only results and not have to do error checks on each entry. unsafeBatch uses the same internal mechanism as batch, giving it the same performance, just passing a modifier to trigger raising of exceptions instead of returning.

The stream method provides a callback style mechanism to obtain access to data as soon at it is available (anything that leverages early writing). Any identifiers requested through the stream interface will follow the batch paradigm, where overlapping ids will share responses to reduce active requests down to a single concurrency.

const valve = new BurstValve<number, number>({
  batch: async (ids, earlyWrite) => {
    await sleep(50);
    earlyWrite(1, 50);
    await sleep(50);
    earlyWrite(2, 100);
    await sleep(50);
    earlyWrite(3, 150);
  },
});

await valve.stream([1, 2, 3], async (id, result) => {
  response.write({ id, result }); // Some external request/response stream
});