keep a parallel implementation of the pin loader to validate the new one

lib/pins.js

@@ -7,6 +7,9 @@ const Path = require("path");
 const Util = require("./common-util");
 const Plan = require("./plan");
 
+const Semaphore = require('saferphore');
+const nThen = require('nthen');
+
 /*  Accepts a reference to an object, and...
     either a string describing which log is being processed (backwards compatibility),
     or a function which will log the error with all relevant data
@@ -194,3 +197,63 @@ Pins.list = function (_done, config) {
         }).start();
     });
 };
+
+Pins.load = function (cb, config) {
+    const sema = Semaphore.create(config.workers || 5);
+
+    let dirList;
+    const fileList = [];
+    const pinned = {};
+
+    var pinPath = config.pinPath || './pins';
+    var done = Util.once(cb);
+
+    nThen((waitFor) => {
+        // recurse over the configured pinPath, or the default
+        Fs.readdir(pinPath, waitFor((err, list) => {
+            if (err) {
+                if (err.code === 'ENOENT') {
+                    dirList = [];
+                    return; // this ends up calling back with an empty object
+                }
+                waitFor.abort();
+                return void done(err);
+            }
+            dirList = list;
+        }));
+    }).nThen((waitFor) => {
+        dirList.forEach((f) => {
+            sema.take((returnAfter) => {
+                // iterate over all the subdirectories in the pin store
+                Fs.readdir(Path.join(pinPath, f), waitFor(returnAfter((err, list2) => {
+                    if (err) {
+                        waitFor.abort();
+                        return void done(err);
+                    }
+                    list2.forEach((ff) => {
+                        if (config && config.exclude && config.exclude.indexOf(ff) > -1) { return; }
+                        fileList.push(Path.join(pinPath, f, ff));
+                    });
+                })));
+            });
+        });
+    }).nThen((waitFor) => {
+        fileList.forEach((f) => {
+            sema.take((returnAfter) => {
+                Fs.readFile(f, waitFor(returnAfter((err, content) => {
+                    if (err) {
+                        waitFor.abort();
+                        return void done(err);
+                    }
+                    const hashes = Pins.calculateFromLog(content.toString('utf8'), f);
+                    hashes.forEach((x) => {
+                        (pinned[x] = pinned[x] || {})[f.replace(/.*\/([^/]*).ndjson$/, (x, y)=>y)] = 1;
+                    });
+                })));
+            });
+        });
+    }).nThen(() => {
+        done(void 0, pinned);
+    });
+};
+