Node.js batch processing

Question

Batch processing steps:

1. When a job is posted, push it to the queue.
2. Get the required number of items from the queue and put it in the batch.
3. If a batch was processed before this, wait till BATCH_INTERVAL time has passed.
4. Process the batch.
5. Clear the batch.
6. Monitor the queue. Back to step 2.

batcher.js

const QUEUE_MONITORING_INTERVAL = 200;
const BATCH_SIZE = 3;
const BATCH_INTERVAL = 1000;
const PROCESSING = "processing";
const IDLE = "idle";

const queue = [];
const batch = [];
let queueInterval = null;
let lastBatchProcessing = null;
let batchStatus = IDLE;

function pushJob(fn) {
  const newPromise = new Promise((resolve, reject) => {
    const job = createJob(fn, resolve, reject);
    queue.push(job);
  });

  monitorQueue();

  return newPromise;
}

function createJob(fn, resolve, reject) {
  return {
    fn,
    resolve,
    reject,
  };
}

function monitorQueue() {
  if (queueInterval == null) {
    queueInterval = setInterval(controller, QUEUE_MONITORING_INTERVAL);
  }
}

function controller() {
  handleQueue();
  handleBatch();
}

function handleQueue() {
  if (batchStatus !== IDLE) {
    return;
  } else if (shouldStopMonitoringQueue()) {
    stopMonitoringQueue();
  } else if (queue.length && batch.length < BATCH_SIZE) {
    pushToBatch();
  }
}

function shouldStopMonitoringQueue() {
  return (
    queue.length === 0 &&
    batch.length === 0 &&
    Date.now() - lastBatchProcessing > BATCH_INTERVAL
  );
}

function stopMonitoringQueue() {
  clearInterval(queueInterval);
  queueInterval = null;
}

function pushToBatch() {
  const spaceRemaining = BATCH_SIZE - batch.length;
  const jobsToPop = Math.min(queue.length, spaceRemaining);
  const jobs = queue.splice(0, jobsToPop);
  batch.push(...jobs);
}

function handleBatch() {
  if (shouldProcessBatch()) {
    processBatch();
  }
}

function shouldProcessBatch() {
  if (!(batch.length && batchStatus === IDLE)) {
    return false;
  }

  return (
    lastBatchProcessing == null ||
    Date.now() - lastBatchProcessing > BATCH_INTERVAL
  );
}

async function processBatch() {
  batchStatus = PROCESSING;
  const promises = getPromisesFromBatch();
  const results = await Promise.allSettled(promises);
  processResults(results);

  clearBatch();
  batchStatus = IDLE;
  lastBatchProcessing = Date.now();
}

function getPromisesFromBatch() {
  const promises = [];

  for (const job of batch) {
    const promise = Promise.resolve(job.fn());
    promises.push(promise);
  }

  return promises;
}

function processResults(results) {
  for (let i = 0; i < results.length; i++) {
    const result = results[i];
    const job = batch[i];

    if (result.status === "fulfilled") {
      job.resolve(result.value);
    } else {
      job.reject(result.reason);
    }
  }
}

function clearBatch() {
  batch.splice(0, batch.length);
}

module.exports = {
  pushJob,
};

test.js

const batcher = require("./batcher");

async function main() {
  const start = Date.now();
  const promises = [];

  for (let i = 0; i < 20; i++) {
    const promise = batcher.pushJob(async () => await fn());
    promises.push(promise);
  }

  const results = await Promise.all(promises);
  const timeElapsed = (Date.now() - start) / 1000;

  console.log(results);
  console.log("Time elapsed: ", timeElapsed);
}

async function fn() {
  return new Promise((resolve) => setTimeout(() => resolve(Math.random()), 3000));
}

main();