Feature/soft lock

README.md

@@ -83,6 +83,12 @@ module.exports = {
   // Enable the algorithm to create a checksum of the file contents and use that in the comparison to determin
   // if the file should be run.  Requires that scripts are coded to be run multiple times.
   useFileHash: false
+
+  // The mongodb collection where the lock will be created.
+  lockCollectionName: "changelog_lock",
+
+  // The value in seconds for the TTL index that will be used for the lock. Value of 0 will disable the feature.
+  lockTtl: 0
 };
 ````
 

lib/actions/down.js

@@ -1,18 +1,29 @@
 const _ = require("lodash");
 const { promisify } = require("util");
-const fnArgs = require('fn-args');
+const fnArgs = require("fn-args");
 
 const status = require("./status");
 const config = require("../env/config");
 const migrationsDir = require("../env/migrationsDir");
-const hasCallback = require('../utils/has-callback');
+const hasCallback = require("../utils/has-callback");
+const lock = require("../utils/lock");
 
 module.exports = async (db, client) => {
   const downgraded = [];
   const statusItems = await status(db);
   const appliedItems = statusItems.filter(item => item.appliedAt !== "PENDING");
   const lastAppliedItem = _.last(appliedItems);
 
+  if (await lock.exist(db)) {
+    throw new Error("Could not migrate down, a lock is in place.");
+  }
+
+  try {
+    await lock.activate(db);
+  } catch(err) {
+    throw new Error(`Could not create a lock: ${err.message}`);
+  }
+
   if (lastAppliedItem) {
     try {
       const migration = await migrationsDir.loadMigration(lastAppliedItem.fileName);
@@ -26,6 +37,7 @@ module.exports = async (db, client) => {
       }
 
     } catch (err) {
+      await lock.clear(db);
       throw new Error(
         `Could not migrate down ${lastAppliedItem.fileName}: ${err.message}`
       );
@@ -40,5 +52,6 @@ module.exports = async (db, client) => {
     }
   }
 
+  await lock.clear(db);
   return downgraded;
 };

lib/actions/up.js

@@ -1,18 +1,29 @@
 const _ = require("lodash");
 const pEachSeries = require("p-each-series");
 const { promisify } = require("util");
-const fnArgs = require('fn-args');
+const fnArgs = require("fn-args");
 
 const status = require("./status");
 const config = require("../env/config");
 const migrationsDir = require("../env/migrationsDir");
-const hasCallback = require('../utils/has-callback');
+const hasCallback = require("../utils/has-callback");
+const lock = require("../utils/lock");
 
 module.exports = async (db, client) => {
   const statusItems = await status(db);
   const pendingItems = _.filter(statusItems, { appliedAt: "PENDING" });
   const migrated = [];
 
+  if (await lock.exist(db)) {
+    throw new Error("Could not migrate up, a lock is in place.");
+  }
+
+  try {
+    await lock.activate(db);
+  } catch(err) {
+    throw new Error(`Could not create a lock: ${err.message}`);
+  }
+
   const migrateItem = async item => {
     try {
       const migration = await migrationsDir.loadMigration(item.fileName);
@@ -31,6 +42,7 @@ module.exports = async (db, client) => {
       );
       error.stack = err.stack;
       error.migrated = migrated;
+      await lock.clear(db);
       throw error;
     }
 
@@ -49,5 +61,6 @@ module.exports = async (db, client) => {
   };
 
   await pEachSeries(pendingItems, migrateItem);
+  await lock.clear(db);
   return migrated;
 };

lib/utils/lock.js

@@ -0,0 +1,42 @@
+const config = require('../env/config');
+
+async function getLockCollection(db) {
+    const { lockCollectionName, lockTtl } = await config.read();
+    if (lockTtl <= 0) {
+        return null;
+    }
+
+    const lockCollection = db.collection(lockCollectionName);
+    lockCollection.createIndex({ createdAt: 1 }, { expireAfterSeconds: lockTtl });
+    return lockCollection;
+}
+
+async function exist(db) {
+    const lockCollection = await getLockCollection(db);
+    if (!lockCollection) {
+        return false;
+    }
+    const foundLocks = await lockCollection.find({}).toArray();
+
+    return foundLocks.length > 0;
+}
+
+async function activate(db) {
+    const lockCollection = await getLockCollection(db);
+    if (lockCollection) {
+        await lockCollection.insertOne({ createdAt: new Date() });
+    }
+}
+
+async function clear(db) {
+    const lockCollection = await getLockCollection(db);
+    if (lockCollection) {
+        await lockCollection.deleteMany({});
+    }
+}
+
+module.exports = {
+    exist,
+    activate,
+    clear,
+}

samples/migrate-mongo-config.js

@@ -22,6 +22,12 @@ const config = {
   // The mongodb collection where the applied changes are stored. Only edit this when really necessary.
   changelogCollectionName: "changelog",
 
+  // The mongodb collection where the lock will be created.
+  lockCollectionName: "changelog_lock",
+
+  // The value in seconds for the TTL index that will be used for the lock. Value of 0 will disable the feature.
+  lockTtl: 0,
+
   // The file extension to create migrations and search for in migration dir 
   migrationFileExtension: ".js",
 

test/actions/down.test.js

@@ -7,11 +7,13 @@ describe("down", () => {
   let down;
   let status;
   let config;
+  let lock;
   let migrationsDir;
   let db;
   let client;
   let migration;
   let changelogCollection;
+  let changelogLockCollection;
 
   function mockStatus() {
     return sinon.stub().returns(
@@ -31,7 +33,11 @@ describe("down", () => {
   function mockConfig() {
     return {
       shouldExist: sinon.stub().returns(Promise.resolve()),
-      read: sinon.stub().returns({ changelogCollectionName: "changelog" })
+      read: sinon.stub().returns({
+        changelogCollectionName: "changelog",
+        lockCollectionName: "changelog_lock",
+        lockTtl: 10
+      })
     };
   }
 
@@ -45,6 +51,7 @@ describe("down", () => {
     const mock = {};
     mock.collection = sinon.stub();
     mock.collection.withArgs("changelog").returns(changelogCollection);
+    mock.collection.withArgs("changelog_lock").returns(changelogLockCollection);
     return mock;
   }
 
@@ -66,24 +73,48 @@ describe("down", () => {
     };
   }
 
+  function mockChangelogLockCollection() {
+    const findStub = {
+      toArray: () => {
+        return [];
+      }
+    }
+
+    return {
+      insertOne: sinon.stub().returns(Promise.resolve()),
+      createIndex: sinon.stub().returns(Promise.resolve()),
+      find: sinon.stub().returns(findStub),
+      deleteMany: sinon.stub().returns(Promise.resolve()),
+    }
+  }
+
   function loadDownWithInjectedMocks() {
     return proxyquire("../../lib/actions/down", {
       "./status": status,
       "../env/config": config,
-      "../env/migrationsDir": migrationsDir
+      "../env/migrationsDir": migrationsDir,
+      "../utils/lock": lock,
+    });
+  }
+
+  function loadLockWithInjectedMocks() {
+    return proxyquire("../../lib/utils/lock", {
+      "../env/config": config
     });
   }
 
   beforeEach(() => {
     migration = mockMigration();
     changelogCollection = mockChangelogCollection();
+    changelogLockCollection = mockChangelogLockCollection();
 
     status = mockStatus();
     config = mockConfig();
     migrationsDir = mockMigrationsDir();
     db = mockDb();
     client = mockClient();
 
+    lock = loadLockWithInjectedMocks();
     down = loadDownWithInjectedMocks();
   });
 
@@ -179,4 +210,61 @@ describe("down", () => {
     const items = await down(db);
     expect(items).to.deep.equal(["20160609113225-last_migration.js"]);
   });
+
+  it("should lock if feature is enabled", async() => {
+    await down(db);
+    expect(changelogLockCollection.createIndex.called).to.equal(true);
+    expect(changelogLockCollection.find.called).to.equal(true);
+    expect(changelogLockCollection.insertOne.called).to.equal(true);
+    expect(changelogLockCollection.deleteMany.called).to.equal(true);
+  });
+
+  it("should ignore lock if feature is disabled", async() => {
+    config.read = sinon.stub().returns({
+      changelogCollectionName: "changelog",
+      lockCollectionName: "changelog_lock",
+      lockTtl: 0
+    });
+    const findStub = {
+      toArray: () => {
+        return [{ createdAt: new Date() }];
+      }
+    }
+    changelogLockCollection.find.returns(findStub);
+
+    await down(db);
+    expect(changelogLockCollection.createIndex.called).to.equal(false);
+    expect(changelogLockCollection.find.called).to.equal(false);
+  });
+
+  it("should yield an error when unable to create a lock", async() => {
+    changelogLockCollection.insertOne.returns(Promise.reject(new Error("Kernel panic")));
+
+    try {
+      await down(db);
+      expect.fail("Error was not thrown");
+    } catch (err) {
+      expect(err.message).to.deep.equal(
+        "Could not create a lock: Kernel panic"
+      );
+    }
+  });
+
+  it("should yield an error when changelog is locked", async() => {
+    const findStub = {
+      toArray: () => {
+        return [{ createdAt: new Date() }];
+      }
+    }
+    changelogLockCollection.find.returns(findStub);
+
+    try {
+      await down(db);
+      expect.fail("Error was not thrown");
+    } catch (err) {
+      expect(err.message).to.deep.equal(
+        "Could not migrate down, a lock is in place."
+      );
+    }
+  });
 });