A background task processor focused on reliability and scalability.

| Table of Contents

• Promise Pool
  • Features
  • API
    • Install
    • Usage
      • Simple
      • Advanced
        • Singleton Pattern
        • Shared Pool: AWS Lambda & Middy
        • Shared Pool: Express Example
  • Changelog
    • v1.3 - September 2022

• [x] Configurable.
• [x] Concurrency limit.
• [x] Retries. (Use p-retry for this.)
• [x] Zero dependencies.
• [x] Error handling.
• [x] Singleton mode: Option to auto-reset when .done(). (Added .drain() method.)
• [x] Task scheduling & prioritization.
  • [x] Optionally return hints. (To support APIs like GitHub which return rate limit hints in http headers, e.g. X-RateLimit-Remaining header.)

PromisePool exposes 3 methods:

• .add(...tasks) - add one (or more) tasks for background processing. (A task is a function that wraps a Promise value. e.g. () => Promise.resolve(1)).
• .drain() - Returns a promise that resolves when all tasks have been processed, or when another thread calls .drain() mid-task execution. This is important to maximize performance in shared server environments, so subsequent calls to .drain() will ensure only the last caller to .drain() will await until the completion of all tasks.
• .done() - optionally finalize the pool. Returns a promise that resolves when all tasks are complete. No more tasks can be added after this.

# with npm
npm install @elite-libs/promise-pool
# or using yarn
yarn add @elite-libs/promise-pool

import PromisePool from '@elite-libs/promise-pool';
// 1/3: Either use the default instance or create a new one.
const pool = new PromisePool();

// 2/3: Add task(s) to the pool as needed.
// PromisePool will execute them in parallel as soon as possible.
pool.add(() => saveToS3(data));
pool.add(() => expensiveBackgroundWork(data));

// 3/3: REQUIRED: in order to ensure your tasks are executed, 
// you must await either `pool.drain()` or `pool.done()` at some point in your code (`done` prevents additional tasks from being added).
await pool.done();

In this example, we'll use a singleton pattern to ensure we have only 1 Promise Pool instance per process.

// `./src/services/taskPool.ts`
import PromisePool from '@elite-libs/promise-pool';

export const taskPool = new PromisePool({
  maxWorkers: 6, // Optional. Default is `4`.
});

Then from inside your app (Lambda function, Express app, etc.), you can use the taskPool instance to add tasks to the pool.

// Example Lambda function
// `./src/handlers/user.handler.ts`
import { taskPool } from './services/taskPool';
export const handler = async (event, context) => {
  const data = getDataFromEvent(event);
  taskPool.add(() => saveToS3(data));
  taskPool.add(() => expensiveBackgroundWork(data));
  await taskPool.drain();

  return {
    statusCode: 200,
    body: JSON.stringify({
      message: 'Success',
    }),
  }
}

⚠️ Note the await taskPool.drain() call. This is required to ensure your tasks are executed.

You could also utilize Promise Pool in middy middleware:

// Important: Import a global instance of `Promise Pool`
import { taskPool } from './services/taskPool';

const taskPoolMiddleware = () => ({
  before: (request) => {
    Object.assign(request.context, { taskPool });
  },
  after: async (request) => {
    await request.context.taskPool.drain();
  }
});

export default taskPoolMiddleware;

Now you can use taskPool in your Lambda function:

request.context.taskPool.add(() => saveToS3(data));

Now the drain() method will be called automatically after() every Lambda function returns.

// `./src/middleware/taskPool.ts`
import { taskPool } from "../services/taskPool.mjs";

export const taskPoolMiddleware = (req, res, next) => {
  req.taskPool = taskPool;
  next();
};

Then you can use taskPool in your Express app:

import { taskPool } from "../services/taskPool.mjs";

export const taskPoolMiddleware = (req, res, next) => {
  req.taskPool = taskPool;
  next();
};

See /examples folder for more examples.

• What? Adds Smart .drain() behavior.
  • Why?
    1. Speeds up concurrent server environments!
    2. prevents several types of difficult (ghost) bugs!
       • stack overflows/mem access violations,
       • exceeding rarely hit OS limits, (e.g. max open handle/file/socket limits)
       • exceeding limits of Network hardware (e.g. connections/sec, total open socket limit, etc.)
       • uneven observed wait times.
  • How? Only awaits the latest .drain() caller.
  • Who? Server + Singleton use cases will see a major benefit to this design.
  • Errata
    • Here are several competing (tortured) metaphors, please suggest more & let me know which makes the most sense.
      • Like a relay race's series of baton handoffs. As soon as you hand off, your race is over & the next runner must either run until finish (wait to end) OR hand off the job to someone new.
      • A tiny Queue, with a size of one!
      • A 'Take-a-penny-leave-a-penny' tray, big enough for exactly 1 penny.
      • This is similar to the feature in MongoDB called the Write Threshold/Concern setting, or in Distributed SQL Replication where you can set a minimum # of confirmed 'server writes' before SQL considers it safely written in the cluster.
      • Anectode: A group of friends had a rule that whoever is last to arrive must pay for the first friend (to arrive on time.)