async hook stack has become corrupted

Am getting this strange error and have been unable to trace it.

Setup: Node V8.10.0, expressjs, mongodb

I don't believe it to be part of the API endpoints but rather an importer script.

You can see the error I am getting here: https://dpaste.org/zs2S

It's always a

Error: async hook stack has become corrupted

The below is the code I am running. Its purpose is to use chokidar to watch a directory. (Previously used fs.watch but changed to chokidar due to thinking it was an fs error but still getting it) And when a file is changed, use csv2json to get the file and return it as a json object.

Once this object has been retrieved, insert it into mongo, first by adding it with the name _Update as a new collection. Drop the current live one, then rename the new, removing _Update, therefore making it live.

I use pm2 to run the app. It runs for a while but will always crash at some point with the above error.

//require the mongojs package to setup db connection
var mongojs = require("mongojs");
//require the csvtojson package to convert the CSV file
var csv = require("csvtojson");
require("log-timestamp");
const chokidar = require("chokidar");
var config = 
require("../../config/config").get(process.env.NODE_ENV);
var db = mongojs(config.database);

exports.import = function() {
  // Initialize watcher.
  const watcher = chokidar.watch(config.watchfolder, {
    ignored: /(^|[\/\\])\../, // ignore dotfiles
    persistent: true,
    awaitWriteFinish: true
  });

  // Something to use when events are received.
  const log = console.log.bind(console);

  watcher.on("change", path => {
    log(`File ${path} has been changed`);

    let requiredCollection = config.sources.find(o => o.csvFile === path);

    csv()
      .fromFile(path)
      .then(jsonArrObj => {
        if (requiredCollection.collectionName.includes("RiverLevels")) {
          for (var item in jsonArrObj) {
            if (jsonArrObj[item].Movement == "S") {
              jsonArrObj[item].Movement = "Steady";
            }
            if (jsonArrObj[item].Movement == "R") {
              jsonArrObj[item].Movement = "Rising";
            }
            if (jsonArrObj[item].Movement == "F") {
              jsonArrObj[item].Movement = "Falling";
            }
            if (jsonArrObj[item].PrimaryStation == "1") {
              db.collection("riverOverview").replaceOne(
                { StationName: jsonArrObj[item].StationName },
                jsonArrObj[item]
              );
              console.log("Primary Station updated into riverOverview");
            }
          }
        }

        //Create new collection with _update
        db.createCollection(
          requiredCollection.collectionName + "_Update",
          function(err) {
            //Insert data fulled from file
            db.collection(requiredCollection.collectionName + "_Update").insert(
              jsonArrObj,
              function(err, result) {
                if (err) {
                  console.log(err);
                }
                console.log(
                  requiredCollection.collectionName + "_Update added to mongo"
                );

                //Drop the current DB
                db.collection(requiredCollection.collectionName).drop(function(
                  err
                ) {
                  if (err) {
                    console.log(err);
                  }
                  console.log(
                    "Old " + requiredCollection.collectionName + " deleted."
                  );
                  //rename update DB to make it the live one.
                  db.collection(
                    requiredCollection.collectionName + "_Update"
                  ).rename(requiredCollection.collectionName, function(
                    err,
                    collection
                  ) {
                    if (err) {
                      console.log(err);
                    } else {
                      console.log("rename Results: " + collection.namespace);
                    }
                    console.log(
                      requiredCollection.collectionName +
                        "_Update renamed to: " +
                        requiredCollection.collectionName
                    );
                  });
                });
              }
            );
          }
        );
      });
  });
};