Add lock option to configure how an existing lock should be handled

Author: albe

README.md

@@ -413,7 +413,18 @@ leads to every single document being flushed directly.
 
 #### Consistency
 
-Since the storage is append-only, consistency is automatically guaranteed.
+Since the storage is append-only, consistency is automatically guaranteed for all successful writes. Writes that fail in
+the middle, e.g. because the machine crashes before the full write buffer is flushed, will lead to a torn write. This is
+a partial invalid write. To recover from such a state, the storage will detect torn writes and truncate them when an existing
+lock is reclaimed. This can be done by instantiating the store with the following option:
+
+```javascript
+const eventstore = new EventStore('my-event-store', { storageConfig: { lock: EventStore.LOCK_RECLAIM } });
+```
+
+Note that this option will effectively bypass the lock that prevents multiple instances from being created, so you should
+not use this carelessly. Having multiple instances write to the same files will lead to inconsistent data that can not be
+easily recovered from.
 
 #### Isolation
 

src/EventStore.js

@@ -397,4 +397,6 @@ class EventStore extends events.EventEmitter {
 
 module.exports = EventStore;
 module.exports.ExpectedVersion = ExpectedVersion;
-module.exports.OptimisticConcurrencyError = OptimisticConcurrencyError;
+module.exports.OptimisticConcurrencyError = OptimisticConcurrencyError;
+module.exports.LOCK_THROW = Storage.LOCK_THROW;
+module.exports.LOCK_RECLAIM = Storage.LOCK_RECLAIM;

src/Storage/WritableStorage.js

@@ -8,6 +8,9 @@ const { assert, matches, buildMetadataForMatcher, buildMatcherFromMetadata } = r
 
 const DEFAULT_WRITE_BUFFER_SIZE = 16 * 1024;
 
+const LOCK_RECLAIM = 0x1;
+const LOCK_THROW = 0x2;
+
 class StorageLockedError extends Error {}
 
 /**
@@ -37,6 +40,7 @@ class WritableStorage extends ReadableStorage {
      * @param {function(object, number): string} [config.partitioner] A function that takes a document and sequence number and returns a partition name that the document should be stored in. Defaults to write all documents to the primary partition.
      * @param {object} [config.indexOptions] An options object that should be passed to all indexes on construction.
      * @param {string} [config.hmacSecret] A private key that is used to verify matchers retrieved from indexes.
+     * @param {number} [config.lock] One of LOCK_* constants that defines how an existing lock should be handled.
      */
     constructor(storageName = 'storage', config = {}) {
         if (typeof storageName !== 'string') {
@@ -60,6 +64,11 @@ class WritableStorage extends ReadableStorage {
             }
         }
         super(storageName, config);
+
+        this.lockFile = path.resolve(this.dataDirectory, this.storageFile + '.lock');
+        if (config.lock === LOCK_RECLAIM) {
+            this.unlock();
+        }
         this.partitioner = config.partitioner;
     }
 
@@ -105,7 +114,6 @@ class WritableStorage extends ReadableStorage {
         if (this.locked) {
             return false;
         }
-        this.lockFile = path.resolve(this.dataDirectory, this.storageFile + '.lock');
         try {
             fs.mkdirSync(this.lockFile);
             this.locked = true;
@@ -125,10 +133,12 @@ class WritableStorage extends ReadableStorage {
      * Current implementation just deletes a lock file that is named like the storage.
      */
     unlock() {
-        if (!this.locked && fs.existsSync(this.lockFile)) {
-            this.checkTornWrites();
+        if (fs.existsSync(this.lockFile)) {
+            if (!this.locked) {
+                this.checkTornWrites();
+            }
+            fs.rmdirSync(this.lockFile);
         }
-        fs.rmdirSync(this.lockFile);
         this.locked = false;
     }
 
@@ -393,4 +403,6 @@ class WritableStorage extends ReadableStorage {
 }
 
 module.exports = WritableStorage;
-module.exports.StorageLockedError = StorageLockedError;
+module.exports.StorageLockedError = StorageLockedError;
+module.exports.LOCK_THROW = LOCK_THROW;
+module.exports.LOCK_RECLAIM = LOCK_RECLAIM;

test/EventStore.spec.js

@@ -60,6 +60,33 @@ describe('EventStore', function() {
         fs.readdir = originalReaddir;
     });
 
+    it('repairs torn writes', function(done) {
+        eventstore = new EventStore({
+            storageDirectory
+        });
+
+        const events = [{foo: 'bar'.repeat(500)}];
+        eventstore.on('ready', () => {
+            eventstore.commit('foo-bar', events, () => {
+                // Simulate a torn write (but indexes are still written)
+                fs.truncateSync(eventstore.storage.getPartition('foo-bar').fileName, 512);
+
+                // The previous instance was not closed, so the lock still exists
+                eventstore = new EventStore({
+                    storageDirectory,
+                    storageConfig: {
+                        lock: EventStore.LOCK_RECLAIM
+                    }
+                });
+                eventstore.on('ready', () => {
+                    expect(eventstore.length).to.be(0);
+                    expect(eventstore.getStreamVersion('foo-bar')).to.be(0);
+                    done();
+                });
+            });
+        });
+    });
+
     it('throws when trying to open non-existing store read-only', function() {
         expect(() => new EventStore({
             storageDirectory,