Dop 2258 (#455)

* Fix sigterm pass-through to worker instances

* Ensure container pruning in integration

* Re-enqueue jobs on graceful shutdown

* Add re-enqueue tests, fix gracefulShutdown tests

* Fix typo in mongo.test.js for new reenqueue tests

Author: casthewiz

.github/workflows/deploy-integration-ec2.yml

@@ -27,5 +27,6 @@ jobs:
           sudo docker stop $(sudo docker ps -a -q)
           sudo sh rundocker.sh
           sudo sh rundocker.sh
+          sudo docker container prune -f
           '
           

Dockerfile

@@ -72,4 +72,4 @@ RUN mkdir repos && chmod 755 repos
 
 # entry to kick-off the worker
 EXPOSE 3000
-CMD ["npm", "start"]
+CMD ["node", "index.js"]

worker/tests/unit/mongo.test.js

@@ -263,6 +263,32 @@ describe('Mongo Tests', () => {
     await mongo.logMessageInMongo(job2, 'message 1');
   }, 5000);
 
+    /** ******************************************************************
+   *                       resetJobForReenqueue()                     *
+   ******************************************************************* */
+     it('resetJobForReenqueue(queueCollection, job) works properly', async () => {
+      const jobsColl = db.collection('jobs');
+  
+      await mongo.resetJobForReenqueue(jobsColl, job2);
+      const currJob = await jobsColl.findOne({ _id: job2._id });
+      expect(currJob).toBeTruthy();
+      expect(currJob.status).toEqual('inQueue');
+      expect(currJob.startTime).toBeNull();
+      expect(currJob.logs[0]).toEqual('Job restarted due to server shutdown.');
+    }, 5000);
+  
+    it('resetJobForReenqueue() fails on incorrect jobId', async () => {
+      const jobsColl = db.collection('jobs');
+      expect.assertions(1);
+      await expect(
+        mongo.resetJobForReenqueue(
+          jobsColl,
+          { _id: 'notRealId', numFailures: 0 },
+          'a'
+        )
+      ).rejects.toBeTruthy();
+    }, 5000);
+
   /** ******************************************************************
    *                       reportStatus()                               *
    ******************************************************************* */

worker/tests/unit/worker.test.js

@@ -36,10 +36,10 @@ describe('Test Class', () => {
   it('onSignal()', async () => {
     worker.setCurrentJob({ job: 'doesnt matter' });
     workerUtils.promiseTimeoutS = jest.fn().mockResolvedValue();
-    mongo.finishJobWithFailure = jest.fn().mockResolvedValue();
+    mongo.resetJobForReenqueue = jest.fn().mockResolvedValue();
 
     await expect(worker.gracefulShutdown()).resolves.toBeUndefined();
-    expect(mongo.finishJobWithFailure).toHaveBeenCalledTimes(1);
+    expect(mongo.resetJobForReenqueue).toHaveBeenCalledTimes(1);
 
     worker.setCurrentJob(null);
   });

worker/utils/mongo.js

@@ -130,6 +130,26 @@ module.exports = {
     }
   },
 
+  // Updates the status to be inQueue
+  async resetJobForReenqueue(queueCollection, job) {
+    const query = { _id: job._id };
+    const reenqueueMessage = 'Job restarted due to server shutdown.';
+
+    const update = {
+      $set: {
+        startTime: null,
+        status: 'inQueue',
+        error: {},
+        logs: [reenqueueMessage],
+      }
+    };
+
+    const updateResult = await queueCollection.updateOne(query, update);
+    if (updateResult.result.n < 1) {
+      throw new Error(`Failed to update job (${job._id}) in queue during re-enqueue operation`);
+    }
+  },
+
   async updateJobWithPurgedURLs(currentJob, urlArray) {
     const queueCollection = module.exports.getCollection();
     if (queueCollection) {

worker/worker.js

@@ -96,17 +96,17 @@ module.exports = {
       workerUtils.resetDirectory('work/');
       await workerUtils.promiseTimeoutS(
         MONGO_TIMEOUT_S,
-        await mongo.finishJobWithFailure(
+        await mongo.resetJobForReenqueue(
           queueCollection,
           currentJob,
-          'Server is being shutdown'
         ),
-        `Mongo Timeout Error: Timed out finishing failed job with jobId: ${currentJob._id}`
+        `Mongo Timeout Error: Timed out finishing re-enqueueing job with jobId: ${currentJob._id}`
       );
     }
     if (mongoClient) {
       monitorInstance.reportStatus('closed connection');
       mongoClient.close();
+      console.log('\nServer has closed mongo client connection');
     }
   },