A quick note about myself

Scaling my Apps

• Resource generation (images, pdf, JSON, CSV)
• Automated React templated emailing
• Automated Database backup to an external server
• External services such as payment and delivery providers
• Static React templating generation
• Search engine management
• Sitemap management

• Both async and sync jobs
• Both in-memory and mongoDB job queues
• scheduled and recurring jobs
• Job prioritization
• IPC messaging between master and worker
• Event listeners

A quick worker pool summary

• A job queue, which stores all the jobs in order
• A Master, which supervise the queue and the workers
• Workers, which execute the tasks sent by the Master on individual processes

• It enables you to offload your main event loop (which will result in more responsiveness)
• The offloaded tasks are executed more efficiently since they are not deferred by other events
• It enables you to write fewer, clearer, and modular code because you’ll need to break it down into subroutines
• It becomes painless to design scheduled and recurring events
• It’s easier to manage your application growth (both in code, and performance)

A real-world usage

• As soon as my bank receives a payment, it sends a POST request to my server to notify me that a payment was made.
• As soon as the payment is made on the bank website, the customer is redirected to a page saying “thank you”, “sorry, the payment was refused”, or “checking payment”. If the status is “checking payment”, the page will pull from the server at regular intervals until the status is updated to either success or failure.

• Gets the payment status from my bank through their API
• Updates the order in my database
• Updates the products stocks
• Sends an email to the customer with the order summary

// gets the payment status from my bank through their API
// update the order in my database

async function onNewOrder(job) {
  const order = Orders.findOne({ _id: job.data._id })

  // abort if order has already been handled before
  if (order.status !== 'WAITING_PAYMENT') {
    return
  }

  const paymentStatus = await getBankPaymentStatus(order.paymentId)
  order.paymentStatus = paymentStatus

  if (paymentStatus === 'accepted') {
    order.status = 'WAITING_ACCEPT'
    order.save()

/* technically this is not called here,
    but in a hook after the save event */
    handleStocks(order.products)
    TaskQueue.addTask({
      taskType: 'sendMail',
      data: { orderId: order._id, mailType: 'onNewOrder' },
      priority: 5
    })
  } else if (paymentStatus === 'refused') {
    order.status = 'REFUSED_PAYMENT'
    order.save()
  }
  // if neither of the 2, paymentStatus === 'waiting_payment'
}

// updates the products stocks

function handleStocks(products) {
  products.forEach(p => {
    const product = Products.findOne({ _id: p._id })
    product.stock -= p.qty
    product.save()
  })
}

// sends an email to the customer with the order summary

function sendMail(job) {
  const { orderId, mailType } = job.data
  const order = Orders.findOne({ _id: orderId })
  const html = getReactMailTemplate({ order, mailType })
  Email.send({
    html,
    to: order.customer.email,
    from: 'somemail@somedomain',
    subject: getMailSubject({ order, mailType })
  })
}

// called by the client / bank requests

function getOrderStatus(_id) {
  const order = Orders.findOne({ _id })
  if (order === undefined) {
    throw new Error(`order${_id}:NOT_FOUND`)
  }
  
  if (order.status === 'WAITING_PAYMENT') {
    const taskId = `onNewOrder${order._id}`
    const exists = TaskQueue.findOne({
      taskType: 'onNewOrder',
      _id: taskId
    })
    if (exists === undefined) {
      TaskQueue.addTask({
        _id: taskId,
        taskType: 'onNewOrder',
        data: { _id },
        priority: 11
      })
    }
  }
  return order.paymentStatus
}

• You can only have one task with the _id : onNewOrder${order._id} at the same time.
• Even if the task is called a 2nd time (there’s a small window), it will abort because order.statuswas previously modified.

import { add } from 'date-fns'

function onTaskError({ value, task }) {
  if (task.taskType === 'sendMail') {
    // notify the admins through a Notification Collection
    Notifications.insert({
      type: 'MAIL_ERROR',
      task: task._id,
      ...someData
    })
    // retry in 30 minutes
    const dueDate = add(new Date(), { minutes: 30 })
    TaskQueue.update({ _id: task._id }, { $set: {
      dueDate, onGoing: false 
    }})
  }
}

const taskMap = {
  onNewOrder,
  sendMail,
}

const cluster = new Cluster(taskMap, { maxAvailableWorkers: 4 })
Meteor.startup(() => {
  if (Cluster.isMaster()) {
    TaskQueue.addEventListener('error', onTaskError)
  }
})

What’s next?

• Using UNIX sockets instead of TCP for IPC which is already available in meteor
• Adding a settable keepAlive field for each individual worker to reduce the cost of worker startup/shutdown routine
• Sending back a job to a worker as soon as a job is finished instead of waiting for the next cycle
• Putting up a meteor core pull request to fully disable the TCP server with an environment variable to reduce the cost and delay of the workers' startup

• Multiple “filterable” event listeners
• Support for load balancing (to be discussed) :

• Meteor core integration for build phases :