A simple pattern for concurrent workers using observers from Meteor collections

worker.js

import { Meteor } from 'meteor/meteor';
import { Tasks } from '/imports/common/collections';

// The workerId should be unique, and at the same time it should persist
// across server restarts, because otherwise we may end up with unfinished tasks falling into limbo.
const workerId = process.env.WORKER_ID || 'theOnlyWorker';

const tasksInProgress = {};
const makeTaskProcessor = execute => (task) => {
  if (!task.workerId) {
    Meteor.defer(() => {
      Tasks.update({
        _id: task._id,
        workerId: null,
      }, {
        $set: {
          workerId,
        },
      });
    });
  } else if (!tasksInProgress[task._id]) {
    tasksInProgress[task._id] = true;
    Meteor.defer(() => {
      execute(task);
      delete tasksInProgress[task._id];
    });
  }
};

const processTask = makeTaskProcessor((task) => {
  // Here is where the task execution happens.
  // ...

  // At the end we need to remove it somehow from tasks queue.
  // Removing from the collection is the simplest option, but not the only one.
  Tasks.remove({ _id: task._id });
});

Meteor.startup(() => {
  // On each restart we clear assignment of tasks that could not be accomplished for some reason.
  Tasks.update({ workerId }, { $unset: { workerId: 1 } }, { multi: true });

  // With this setup, the observer ensures that there are always at most 10 items
  Tasks.find({
    $or: [
      { workerId: null },
      { workerId },
    ],
  }, {
    limit: 10,
    // With "workerId: -1" nulls come last, which is exactly what we want.
    sort: { workerId: -1 },
  }).observe({
    added:   processTask,
    changed: processTask,
  });
});