Fix multiple writes to branch ref key when using source_root.

lib/consul/index.js

@@ -338,68 +338,62 @@ var get_kvs = function(branch, file, cb) {
  */
 var process_records = function(branch, records, cb) {
 
-  var pending_records = 0;
   var errors_seen = [];
+  var finished = _.after(records.length, cb);
 
   var check_pending = function(err) {
     if (err) {
       errors_seen.push(err);
     }
 
-    --pending_records;
-
     // If there are no pending records, callback with all errors seen, if any.
-    if (pending_records <= 0) {
-      _.once(cb((errors_seen.length > 0) ? errors_seen : null));
-    }
+    finished((errors_seen.length > 0) ? errors_seen : null);
 
     // TODO: Add a watchdog timer?  It's a bit scary that this method may never fire its callback
     // if one of the underlying consul operations hangs, especially since the branch is locked
     // waiting for this update to complete.
   };
 
   records.forEach(function(record) {
-    logger.trace('Handling record %s of type %s', record.path, record.type);
-
     // If we have a source_root set but this file is not within source_root, skip it.
     if (branch.source_root && record.path.indexOf(branch.source_root) !== 0) {
+      logger.trace('Skipping record %s outside of branch root %s of type %s', record.path, branch.source_root, record.type);
       return check_pending();
-    };
+    } else {
+      logger.trace('Handling record %s of type %s', record.path, record.type);
+    }
 
     switch (record.type) {
       // Update files that were Added (A), Modified (M), or had their type (i.e. regular file, symlink, submodule, ...) changed (T)
       case 'M':
       case 'A':
       case 'T':
         // Store added/modified file
-        // FIXME: This will definitely fail in scenarios where record path is not in branch source root.
         if (record.path === branch.common_properties) {
-          ++pending_records;
-
           file_modified(branch, record.path, function(err) {
             branch.listAdditionalPropertyFiles(records, function(err, additionalRecords) {
               if (err) return check_pending(err);
               if (additionalRecords.length == 0) return check_pending();
               process_records(branch, additionalRecords, function(errs) {
-                if (errs) check_pending("Some consul updates failed:\n" + errs.join('\n'));
-
-                check_pending();
+                if (errs) {
+                  check_pending(errs.join('\n'));
+                } else {
+                  check_pending();
+                }
               });
             });
           });
         } else {
-          ++pending_records;
           file_modified(branch, record.path, check_pending);
         }
         break;
       case 'D':
         // Delete file
-        ++pending_records;
         file_deleted(branch, record.path, check_pending);
         break;
       /* istanbul ignore next */
       default:
-        logger.error('Unknown git status %s', record.type);
+        check_pending("Unknown git status " + record.type + " for " + record.path);
     }
   });
 };