From 970dc6b618c75ce13c152defc980c7540212c7e7 Mon Sep 17 00:00:00 2001
From: "Reser, Ben" <Ben.Reser@cdk.com>
Date: Fri, 28 Oct 2016 21:19:10 -0700
Subject: [PATCH] Fix multiple writes to branch ref key when using source_root.

When using source_root paths outside that source_root would be processed
and due to errors in the processing loop logic would trigger the
callback that would ultimately trigger multiple writes to the branch ref
key in consul.  It also prematurely releases the lock on the branch.

This happens because the logic for skipping paths outside the
source_path calls check_pending which decrements the pending_record
count.  When the record count is at or below zero then it would call
the upstream callback and trigger the ref key write and the removal
of the locks.

Additionally, the callback was wrapped in _.once() implying an attempt
to prevent the callback from being called multiple times.  But this
fails because the code called the cb() function and passed the return
to _.once, instead of using _.once to create a new variable.

In the handling for the common_properties file, errors in processing
the consul updates were prefixed with "Some consul updates failed:",
but these errors are appended to the existing errors which would then
be prefixed by the same string further up, resulting in the prefix
appearing multiple times in the logs.

Fix removes the pending_records counter entirely and uses _.after()
to wrap the callback so that the callback is only called after all
records are processed.  This means that it is important that calling
check_pending() must be called on every record exactly once. The old
logic technically would allow you to call it multiple times as long
as you incremented the same number of times.

Finally, move the error handler for an unknown git status to also
call check_pending and provide the record.path for easier debugging.

While we're at it make it obvious which records are skipped due to
being outside the source_root when doing trace logging.
---
 lib/consul/index.js | 30 ++++++++++++------------------
 1 file changed, 12 insertions(+), 18 deletions(-)

diff --git a/lib/consul/index.js b/lib/consul/index.js
index 220692c..a6bd5e4 100644
--- a/lib/consul/index.js
+++ b/lib/consul/index.js
@@ -332,20 +332,16 @@ 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
@@ -353,12 +349,13 @@ var process_records = function(branch, records, cb) {
   };
 
   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)
@@ -366,34 +363,31 @@ var process_records = function(branch, records, cb) {
       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);
     }
   });
 };