Merge 8e5fe67 into a53c02d

lib/database/filedown.js

@@ -3,6 +3,8 @@ var AbstractLevelDOWN = require("abstract-leveldown").AbstractLevelDOWN;
 var async = require("async");
 var fs = require("fs");
 var path = require("path");
+var tmp = require("tmp");
+tmp.setGracefulCleanup();
 
 util.inherits(FileDown, AbstractLevelDOWN);
 
@@ -15,17 +17,78 @@ FileDown.prototype._open = function(options, callback) {
   var self = this;
   callback(null, self);
 };
-
+const accessQueue = {
+  next: (lKey) => {
+    const cont = accessQueue.cache[lKey].shift();
+    if (cont) {
+      cont();
+    } else {
+      delete accessQueue.cache[lKey];
+    }
+  },
+  execute: (lKey, callback) => {
+    let cache = accessQueue.cache[lKey];
+    if (cache) {
+      cache.push(callback);
+    } else {
+      accessQueue.cache[lKey] = [];
+      callback();
+    }
+  },
+  cache: {}
+};
 FileDown.prototype._put = function(key, value, options, callback) {
-  fs.writeFile(path.join(this.location, key), value, "utf8", callback);
+  const lKey = path.join(this.location, key);
+  // This fixes an issue caused by writing AND reading the same key multiple times
+  // simultaneously. Sometimes the read operation would end up reading the value as 0 bytes
+  // due to the way writes are performed in node. To fix this, we implemented a queue so only a
+  // single read or write operation can occur at a time for each key; basically an in-memory lock.
+  // Additionally, during testing we found that it was possible for a write operation to fail
+  // due to program termination. This failure would sometimes cause the key to _exist_ but contain
+  // 0 bytes (which is always invalid). To fix this we write to a temp file, and only if it works
+  // do we move this temp file to its correct key location. This prevents early termination from
+  // writing partial/empty values.
+  // Of course, all this will eventually be for nothing as we are migrating the db to actual an
+  // leveldb implementation that doesn't use a separate file for every key Soon(TM).
+  accessQueue.execute(lKey, () => {
+    // get a tmp file to write the contents to...
+    tmp.file((err, path, fd) => {
+      if (err) {
+        callback(err);
+        accessQueue.next(lKey);
+        return;
+      }
+
+      // write the contents to that temp file
+      fs.writeFile(fd, value, "utf8", (err) => {
+        if (err) {
+          callback(err);
+          accessQueue.next(lKey);
+          return;
+        }
+        // move the temp file to its final destination
+        fs.rename(path, lKey, (err) => {
+          callback(err);
+
+          // if there is more work to be done on this key, do it.
+          accessQueue.next(lKey);
+        });
+      });
+    });
+  });
 };
 
 FileDown.prototype._get = function(key, options, callback) {
-  fs.readFile(path.join(this.location, key), "utf8", function(err, data) {
-    if (err) {
-      return callback(new Error("NotFound"));
-    }
-    callback(null, data);
+  const lKey = path.join(this.location, key);
+  accessQueue.execute(lKey, () => {
+    fs.readFile(lKey, "utf8", (err, data) => {
+      if (err) {
+        callback(new Error("NotFound"));
+      } else {
+        callback(null, data);
+      }
+      accessQueue.next(lKey);
+    });
   });
 };