title: Do not process next job until previous job is completed (BullJS/Redis)?
body: <p>Basically, each of the clients ---that have a <code>clientId</code> associated with them--- can push messages and <strong>it is important that a second message from the same client isn't processed until the first one is finished processing</strong> (Even though the client can send multiple messages in a row, and they are ordered, and multiple clients sending messages should ideally not interfere with each other). And, importantly, a job shouldn't be processed twice.</p>
  <p>I thought that using Redis I might be able to fix this issue, I started with some quick prototyping using the bull library, but I am clearly not doing it well, I was hoping someone would know how to proceed.</p>
  <h2>This is what I tried so far:</h2>
  <ol>
  <li>Create jobs and add them to the same queue name for one process, using the <code>clientId</code> as the job name.</li>
  <li>Consume jobs while waiting large random amounts of random time on 2 separate process.</li>
  <li>I tried adding the default locking provided by the library that I am using (<code>bull</code>) but it locks on the jobId, which is unique for each job, not on the clientId .</li>
  </ol>
  <h2>What I would want to happen:</h2>
  <ul>
  <li>One of the consumers can't take the job from the same <code>clientId</code> until the previous one is finished processing it.</li>
  <li>They should be able to, however, get items from different <code>clientId</code>s in parallel without problem (asynchronously). (I haven't gotten this far, I am right now simply dealing with only one <code>clientId</code>)</li>
  </ul>
  <h2>What I get:</h2>
  <ul>
  <li>Both consumers consume as many items as they can from the queue without waiting for the previous item for the <code>clientId</code> to be completed.</li>
  </ul>
  <h2><strong>Is Redis even the right tool for this job?</strong></h2>
  <h2>Example code</h2>
  <pre><code>// ./setup.ts
  import Queue from 'bull';
  import * as uuid from 'uuid';
  
  // Check that when a message is taken from a place, no other message is taken
  
  // TO do that test, have two processes that process messages and one that sets messages, and make the job take a long time
  
  // queue for each room https://stackoverflow.com/questions/54178462/how-does-redis-pubsub-subscribe-mechanism-works/54243792#54243792
  // https://groups.google.com/forum/#!topic/redis-db/R09u__3Jzfk
  
  // Make a job not be called stalled, waiting enough time https://github.com/OptimalBits/bull/issues/210#issuecomment-190818353
  
  export async function sleep(ms: number): Promise&lt;void&gt; {
    return new Promise((resolve) =&gt; {
      setTimeout(resolve, ms);
    });
  }
  export interface JobData {
    id: string;
    v: number;
  }
  export const queue = new Queue&lt;JobData&gt;('messages', 'redis://127.0.0.1:6379');
  
  queue.on('error', (err) =&gt; {
    console.error('Uncaught error on queue.', err);
    process.exit(1);
  });
  
  export function clientId(): string {
    return uuid.v4();
  }
  
  export function randomWait(minms: number, maxms: number): Promise&lt;void&gt; {
    const ms = Math.random() * (maxms - minms) + minms;
    return sleep(ms);
  }
  
  // Make a job not be called stalled, waiting enough time https://github.com/OptimalBits/bull/issues/210#issuecomment-190818353
  // eslint-disable-next-line @typescript-eslint/ban-ts-comment
  //@ts-ignore
  queue.LOCK_RENEW_TIME = 5 * 60 * 1000;
  
  </code></pre>
  <pre><code>// ./create.ts
  import { queue, randomWait } from './setup';
  
  const MIN_WAIT = 300;
  const MAX_WAIT = 1500;
  async function createJobs(n = 10): Promise&lt;void&gt; {
    await randomWait(MIN_WAIT, MAX_WAIT);
    // always same Id
    const clientId = Math.random() &gt; 1 ? 'zero' : 'one';
    for (let index = 0; index &lt; n; index++) {
      await randomWait(MIN_WAIT, MAX_WAIT);
      const job = { id: clientId, v: index };
      await queue.add(clientId, job).catch(console.error);
      console.log('Added job', job);
    }
  }
  
  export async function create(nIds = 10, nItems = 10): Promise&lt;void&gt; {
    const jobs = [];
    await randomWait(MIN_WAIT, MAX_WAIT);
    for (let index = 0; index &lt; nIds; index++) {
      await randomWait(MIN_WAIT, MAX_WAIT);
      jobs.push(createJobs(nItems));
      await randomWait(MIN_WAIT, MAX_WAIT);
    }
    await randomWait(MIN_WAIT, MAX_WAIT);
    await Promise.all(jobs)
    process.exit();
  }
  
  (function mainCreate(): void {
    create().catch((err) =&gt; {
      console.error(err);
      process.exit(1);
    });
  })();
  
  </code></pre>
  <pre><code>// ./consume.ts
  import { queue, randomWait, clientId } from './setup';
  
  function startProcessor(minWait = 5000, maxWait = 10000): void {
    queue
      .process('*', 100, async (job) =&gt; {
        console.log('LOCKING: ', job.lockKey());
        await job.takeLock();
        const name = job.name;
        const processingId = clientId().split('-', 1)[0];
        try {
          console.log('START: ', processingId, '\tjobName:', name);
          await randomWait(minWait, maxWait);
          const data = job.data;
          console.log('PROCESSING: ', processingId, '\tjobName:', name, '\tdata:', data);
          await randomWait(minWait, maxWait);
          console.log('PROCESSED: ', processingId, '\tjobName:', name, '\tdata:', data);
          await randomWait(minWait, maxWait);
          console.log('FINISHED: ', processingId, '\tjobName:', name, '\tdata:', data);
        } catch (err) {
          console.error(err);
        } finally {
          await job.releaseLock();
        }
      })
      .catch(console.error); // Catches initialization
  }
  
  startProcessor();
  
  </code></pre>
  <p>This is run using 3 different processes, which you might call like this (Although I use different tabs for a clearer view of what is happening)</p>
  <pre class="lang-sh prettyprint-override"><code>npx ts-node consume.ts &amp; 
  npx ts-node consume.ts &amp;
  npx ts-node create.ts &amp;
  </code></pre>