From d9788013e00e214b949a69867ba2ded6585f27d4 Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Mon, 12 Mar 2018 17:52:12 -0700
Subject: [PATCH 01/10] avoid race condition by serializing all Redis updates

---
 redis_feature_store.js          | 170 ++++++++++++++++++--------------
 test/feature_store_test_base.js |  21 ++++
 2 files changed, 118 insertions(+), 73 deletions(-)

diff --git a/redis_feature_store.js b/redis_feature_store.js
index 39fd6a6..3fb017a 100644
--- a/redis_feature_store.js
+++ b/redis_feature_store.js
@@ -13,6 +13,7 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
       store = {},
       items_prefix = (prefix || "launchdarkly") + ":",
       cache = cache_ttl ? new NodeCache({ stdTTL: cache_ttl}) : null,
+      updateQueue = [],
       inited = false,
       checked_init = false;
 
@@ -88,6 +89,27 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
     });
   }
 
+  function queueUpdate(updateFn) {
+    updateQueue.push(updateFn);
+    if (updateQueue.length == 1) {
+      executePendingUpdates();
+    }
+  }
+
+  function executePendingUpdates() {
+    if (updateQueue.length > 0) {
+      updateQueue[0]();
+    }
+  }
+
+  function completedUpdate(cb) {
+    updateQueue.shift();
+    if (updateQueue.length > 0) {
+      setImmediate(executePendingUpdates);
+    }
+    cb && cb();
+  }
+
   store.get = function(kind, key, cb) {
     cb = cb || noop;
     do_get(kind, key, function(item) {
@@ -129,97 +151,99 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
   };
 
   store.init = function(allData, cb) {
-    var multi = client.multi();
-    cb = cb || noop;
+    queueUpdate(function() {
+      var multi = client.multi();
 
-    if (cache_ttl) {
-      cache.flushAll();
-    }
+      if (cache_ttl) {
+        cache.flushAll();
+      }
 
-    for (var kindNamespace in allData) {
-      if (Object.hasOwnProperty.call(allData, kindNamespace)) {
-        var kind = dataKind[kindNamespace];
-        var baseKey = items_key(kind);
-        var items = allData[kindNamespace];
-        var stringified = {};
-        multi.del(baseKey);
-        for (var key in items) {
-          if (Object.hasOwnProperty.call(items, key)) {
-            stringified[key] = JSON.stringify(items[key]);
+      for (var kindNamespace in allData) {
+        if (Object.hasOwnProperty.call(allData, kindNamespace)) {
+          var kind = dataKind[kindNamespace];
+          var baseKey = items_key(kind);
+          var items = allData[kindNamespace];
+          var stringified = {};
+          multi.del(baseKey);
+          for (var key in items) {
+            if (Object.hasOwnProperty.call(items, key)) {
+              stringified[key] = JSON.stringify(items[key]);
+            }
+            if (cache_ttl) {
+              cache.set(cache_key(kind, key), items[key]);
+            }
           }
-          if (cache_ttl) {
-            cache.set(cache_key(kind, key), items[key]);
+          // Redis does not allow hmset() with an empty object
+          if (Object.keys(stringified).length > 0) {
+            multi.hmset(baseKey, stringified);
           }
         }
-        // Redis does not allow hmset() with an empty object
-        if (Object.keys(stringified).length > 0) {
-          multi.hmset(baseKey, stringified);
-        }
       }
-    }
 
-    multi.exec(function(err, replies) {
-      if (err) {
-        logger.error("Error initializing Redis store", err);
-      } else {
-        inited = true;
-      }
-      cb();
+      multi.exec(function(err, replies) {
+        if (err) {
+          logger.error("Error initializing Redis store", err);
+        } else {
+          inited = true;
+        }
+        completedUpdate(cb);
+      });
     });
   };
 
   store.delete = function(kind, key, version, cb) {
-    var multi;
-    var baseKey = items_key(kind);
-    cb = cb || noop;
-    client.watch(baseKey);
-    multi = client.multi();
-
-    do_get(kind, key, function(item) {
-      if (item && item.version >= version) {
-        multi.discard();
-        cb();
-      } else {
-        deletedItem = { version: version, deleted: true };
-        multi.hset(baseKey, key, JSON.stringify(deletedItem));
-        multi.exec(function(err, replies) {
-          if (err) {
-            logger.error("Error deleting key " + key + " in '" + kind.namespace + "'", err);
-          } else if (cache_ttl) {            
-            cache.set(cache_key(kind, key), deletedItem);
-          }
-          cb();
-        });
-      }
+    queueUpdate(function() {
+      var multi;
+      var baseKey = items_key(kind);
+      client.watch(baseKey);
+      multi = client.multi();
+
+      do_get(kind, key, function(item) {
+        if (item && item.version >= version) {
+          multi.discard();
+          completedUpdate(cb);
+        } else {
+          deletedItem = { version: version, deleted: true };
+          multi.hset(baseKey, key, JSON.stringify(deletedItem));
+          multi.exec(function(err, replies) {
+            if (err) {
+              logger.error("Error deleting key " + key + " in '" + kind.namespace + "'", err);
+            } else if (cache_ttl) {            
+              cache.set(cache_key(kind, key), deletedItem);
+            }
+            completedUpdate(cb);
+          });
+        }
+      });
     });
   };
 
   store.upsert = function(kind, item, cb) {
-    var multi;
-    var baseKey = items_key(kind);
-    var key = item.key;
-    cb = cb || noop;
-    client.watch(baseKey);
-    multi = client.multi();
-
-    do_get(kind, key, function(original) {
-      if (original && original.version >= item.version) {
-        cb();
-        return;
-      }
-
-      multi.hset(baseKey, key, JSON.stringify(item));
-      multi.exec(function(err, replies) {
-        if (err) {
-          logger.error("Error upserting key " + key + " in '" + kind.namespace + "'", err);
+    queueUpdate(function() {
+      var multi;
+      var baseKey = items_key(kind);
+      var key = item.key;
+      client.watch(baseKey);
+      multi = client.multi();
+
+      do_get(kind, key, function(original) {
+        if (original && original.version >= item.version) {
+          multi.discard();
+          completedUpdate(cb);
         } else {
-          if (cache_ttl) {
-            cache.set(cache_key(kind, key), item);
-          }
+          multi.hset(baseKey, key, JSON.stringify(item));
+          multi.exec(function(err, replies) {
+            if (err) {
+              logger.error("Error upserting key " + key + " in '" + kind.namespace + "'", err);
+            } else {
+              if (cache_ttl) {
+                cache.set(cache_key(kind, key), item);
+              }
+            }
+            completedUpdate(cb);
+          });
         }
-        cb();
       });
-
     });
   };
 
diff --git a/test/feature_store_test_base.js b/test/feature_store_test_base.js
index 981da51..73bebba 100644
--- a/test/feature_store_test_base.js
+++ b/test/feature_store_test_base.js
@@ -97,6 +97,27 @@ function allFeatureStoreTests(makeStore) {
     });
   });
 
+  it('handles upsert race condition correctly', function(done) {
+    var ver1 = { key: feature1.key, version: feature1.version + 1 };
+    var ver2 = { key: feature1.key, version: feature1.version + 2 };
+    initedStore(function(store) {
+      var counter = 0;
+      var combinedCallback = function() {
+        counter++;
+        if (counter == 2) {
+          store.get(dataKind.features, feature1.key, function(result) {
+            expect(result).toEqual(ver2);
+            done();
+          });
+        }
+      };
+      // Deliberately do not wait for the first upsert to complete before starting the second,
+      // so their transactions will be interleaved unless we're correctly serializing updates
+      store.upsert(dataKind.features, ver2, combinedCallback);
+      store.upsert(dataKind.features, ver1, combinedCallback);
+    });
+  });
+
   it('deletes with newer version', function(done) {
     initedStore(function(store) {
       store.delete(dataKind.features, feature1.key, feature1.version + 1, function(result) {

From 1ae6b12d9bbfc929733803f0aae3f230276639ac Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Mon, 12 Mar 2018 18:10:57 -0700
Subject: [PATCH 02/10] semicolons

---
 redis_feature_store.js | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/redis_feature_store.js b/redis_feature_store.js
index 3fb017a..fe5b97e 100644
--- a/redis_feature_store.js
+++ b/redis_feature_store.js
@@ -43,10 +43,10 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
     }
     initialConnect = false;
     connected = true;
-  })
+  });
   client.on('end', function() {
     connected = false;
-  })
+  });
 
   // Allow driver programs to exit, even if the Redis socket is active
   client.unref();

From 156664eccad792364ecda6e22844e9c43b2528a8 Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Mon, 12 Mar 2018 18:29:50 -0700
Subject: [PATCH 03/10] comments

---
 redis_feature_store.js | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/redis_feature_store.js b/redis_feature_store.js
index fe5b97e..70a5969 100644
--- a/redis_feature_store.js
+++ b/redis_feature_store.js
@@ -89,9 +89,12 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
     });
   }
 
+  // Places an update operation on the queue. Each update operation must call completedUpdate()
+  // before the next one can start.
   function queueUpdate(updateFn) {
     updateQueue.push(updateFn);
     if (updateQueue.length == 1) {
+      // if nothing else is in progress, we can start this one right away
       executePendingUpdates();
     }
   }

From 44ac12797fb585d63d6acdcf630732e54a9d6f1c Mon Sep 17 00:00:00 2001
From: "Andrew S. Brown" <andrew@launchdarkly.com>
Date: Tue, 13 Mar 2018 12:41:40 -0700
Subject: [PATCH 04/10] Alternative organization for serialization

---
 redis_feature_store.js | 198 +++++++++++++++++++++--------------------
 1 file changed, 103 insertions(+), 95 deletions(-)

diff --git a/redis_feature_store.js b/redis_feature_store.js
index 70a5969..0d5e0d6 100644
--- a/redis_feature_store.js
+++ b/redis_feature_store.js
@@ -89,29 +89,31 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
     });
   }
 
-  // Places an update operation on the queue. Each update operation must call completedUpdate()
-  // before the next one can start.
-  function queueUpdate(updateFn) {
-    updateQueue.push(updateFn);
-    if (updateQueue.length == 1) {
-      // if nothing else is in progress, we can start this one right away
-      executePendingUpdates();
-    }
-  }
-
   function executePendingUpdates() {
     if (updateQueue.length > 0) {
-      updateQueue[0]();
+      const entry = updateQueue[0];
+      const fn = entry[0];
+      const args = entry[1];
+      const cb = entry[2];
+      const newCb = function() {
+        updateQueue.shift();
+        if (updateQueue.length > 0) {
+          setImmediate(executePendingUpdates);
+        }
+        cb && cb();
+      };
+      fn.apply(store, args.concat([newCb]));
     }
   }
 
-  function completedUpdate(cb) {
-    updateQueue.shift();
-    if (updateQueue.length > 0) {
-      setImmediate(executePendingUpdates);
+  // Places an update operation on the queue.
+  var serializeFn = function(updateFn, fnArgs, cb) {
+    updateQueue.push([updateFn, fnArgs, cb]);
+    if (updateQueue.length == 1) {
+      // if nothing else is in progress, we can start this one right away
+      executePendingUpdates();
     }
-    cb && cb();
-  }
+  };
 
   store.get = function(kind, key, cb) {
     cb = cb || noop;
@@ -154,99 +156,105 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
   };
 
   store.init = function(allData, cb) {
-    queueUpdate(function() {
-      var multi = client.multi();
+    serializeFn(store._init, [allData], cb);
+  };
 
-      if (cache_ttl) {
-        cache.flushAll();
-      }
+  store._init = function(allData, cb) {
+    var multi = client.multi();
 
-      for (var kindNamespace in allData) {
-        if (Object.hasOwnProperty.call(allData, kindNamespace)) {
-          var kind = dataKind[kindNamespace];
-          var baseKey = items_key(kind);
-          var items = allData[kindNamespace];
-          var stringified = {};
-          multi.del(baseKey);
-          for (var key in items) {
-            if (Object.hasOwnProperty.call(items, key)) {
-              stringified[key] = JSON.stringify(items[key]);
-            }
-            if (cache_ttl) {
-              cache.set(cache_key(kind, key), items[key]);
-            }
+    if (cache_ttl) {
+      cache.flushAll();
+    }
+
+    for (var kindNamespace in allData) {
+      if (Object.hasOwnProperty.call(allData, kindNamespace)) {
+        var kind = dataKind[kindNamespace];
+        var baseKey = items_key(kind);
+        var items = allData[kindNamespace];
+        var stringified = {};
+        multi.del(baseKey);
+        for (var key in items) {
+          if (Object.hasOwnProperty.call(items, key)) {
+            stringified[key] = JSON.stringify(items[key]);
           }
-          // Redis does not allow hmset() with an empty object
-          if (Object.keys(stringified).length > 0) {
-            multi.hmset(baseKey, stringified);
+          if (cache_ttl) {
+            cache.set(cache_key(kind, key), items[key]);
           }
         }
+        // Redis does not allow hmset() with an empty object
+        if (Object.keys(stringified).length > 0) {
+          multi.hmset(baseKey, stringified);
+        }
       }
+    }
 
-      multi.exec(function(err, replies) {
-        if (err) {
-          logger.error("Error initializing Redis store", err);
-        } else {
-          inited = true;
-        }
-        completedUpdate(cb);
-      });
+    multi.exec(function(err, replies) {
+      if (err) {
+        logger.error("Error initializing Redis store", err);
+      } else {
+        inited = true;
+      }
+      cb();
     });
   };
 
   store.delete = function(kind, key, version, cb) {
-    queueUpdate(function() {
-      var multi;
-      var baseKey = items_key(kind);
-      client.watch(baseKey);
-      multi = client.multi();
-
-      do_get(kind, key, function(item) {
-        if (item && item.version >= version) {
-          multi.discard();
-          completedUpdate(cb);
-        } else {
-          deletedItem = { version: version, deleted: true };
-          multi.hset(baseKey, key, JSON.stringify(deletedItem));
-          multi.exec(function(err, replies) {
-            if (err) {
-              logger.error("Error deleting key " + key + " in '" + kind.namespace + "'", err);
-            } else if (cache_ttl) {            
-              cache.set(cache_key(kind, key), deletedItem);
-            }
-            completedUpdate(cb);
-          });
-        }
-      });
+      serializeFn(store._delete, [kind, key, version], cb);
+  };
+
+  store._delete = function(kind, key, version, cb) {
+    var multi;
+    var baseKey = items_key(kind);
+    client.watch(baseKey);
+    multi = client.multi();
+
+    do_get(kind, key, function(item) {
+      if (item && item.version >= version) {
+        multi.discard();
+        cb();
+      } else {
+        deletedItem = { version: version, deleted: true };
+        multi.hset(baseKey, key, JSON.stringify(deletedItem));
+        multi.exec(function(err, replies) {
+          if (err) {
+            logger.error("Error deleting key " + key + " in '" + kind.namespace + "'", err);
+          } else if (cache_ttl) {
+            cache.set(cache_key(kind, key), deletedItem);
+          }
+          cb();
+        });
+      }
     });
   };
 
   store.upsert = function(kind, item, cb) {
-    queueUpdate(function() {
-      var multi;
-      var baseKey = items_key(kind);
-      var key = item.key;
-      client.watch(baseKey);
-      multi = client.multi();
-
-      do_get(kind, key, function(original) {
-        if (original && original.version >= item.version) {
-          multi.discard();
-          completedUpdate(cb);
-        } else {
-          multi.hset(baseKey, key, JSON.stringify(item));
-          multi.exec(function(err, replies) {
-            if (err) {
-              logger.error("Error upserting key " + key + " in '" + kind.namespace + "'", err);
-            } else {
-              if (cache_ttl) {
-                cache.set(cache_key(kind, key), item);
-              }
+    serializeFn(store._upsert, [kind, item], cb);
+  };
+
+  store._upsert = function(kind, item, cb) {
+    var multi;
+    var baseKey = items_key(kind);
+    var key = item.key;
+    client.watch(baseKey);
+    multi = client.multi();
+
+    do_get(kind, key, function(original) {
+      if (original && original.version >= item.version) {
+        multi.discard();
+        cb();
+      } else {
+        multi.hset(baseKey, key, JSON.stringify(item));
+        multi.exec(function(err, replies) {
+          if (err) {
+            logger.error("Error upserting key " + key + " in '" + kind.namespace + "'", err);
+          } else {
+            if (cache_ttl) {
+              cache.set(cache_key(kind, key), item);
             }
-            completedUpdate(cb);
-          });
-        }
-      });
+          }
+          cb();
+        });
+      }
     });
   };
 
@@ -266,7 +274,7 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
       client.exists(items_key(dataKind.features), function(err, obj) {
         if (!err && obj) {
           inited = true;
-        } 
+        }
         checked_init = true;
         cb(inited);
       });

From e3b5ecfbefad7995b4baeb9e9efd83ddc01e6bb6 Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Tue, 13 Mar 2018 14:03:57 -0700
Subject: [PATCH 05/10] better test name

---
 test/feature_store_test_base.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/feature_store_test_base.js b/test/feature_store_test_base.js
index 73bebba..5bb874e 100644
--- a/test/feature_store_test_base.js
+++ b/test/feature_store_test_base.js
@@ -97,7 +97,7 @@ function allFeatureStoreTests(makeStore) {
     });
   });
 
-  it('handles upsert race condition correctly', function(done) {
+  it('handles upsert race condition within same client correctly', function(done) {
     var ver1 = { key: feature1.key, version: feature1.version + 1 };
     var ver2 = { key: feature1.key, version: feature1.version + 2 };
     initedStore(function(store) {

From 136f85a558e3099fb3dafb076e5742a3d09f19fd Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Tue, 13 Mar 2018 15:11:08 -0700
Subject: [PATCH 06/10] fix optimistic locking logic to retry updates as needed

---
 redis_feature_store.js           | 90 ++++++++++++++++----------------
 test/redis_feature_store-test.js | 56 ++++++++++++++++++--
 2 files changed, 97 insertions(+), 49 deletions(-)

diff --git a/redis_feature_store.js b/redis_feature_store.js
index 0d5e0d6..f49724a 100644
--- a/redis_feature_store.js
+++ b/redis_feature_store.js
@@ -203,60 +203,60 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
   };
 
   store._delete = function(kind, key, version, cb) {
-    var multi;
-    var baseKey = items_key(kind);
-    client.watch(baseKey);
-    multi = client.multi();
-
-    do_get(kind, key, function(item) {
-      if (item && item.version >= version) {
-        multi.discard();
-        cb();
-      } else {
-        deletedItem = { version: version, deleted: true };
-        multi.hset(baseKey, key, JSON.stringify(deletedItem));
-        multi.exec(function(err, replies) {
-          if (err) {
-            logger.error("Error deleting key " + key + " in '" + kind.namespace + "'", err);
-          } else if (cache_ttl) {
-            cache.set(cache_key(kind, key), deletedItem);
-          }
-          cb();
-        });
-      }
-    });
-  };
+    var deletedItem = { key: key, version: version, deleted: true };
+    updateItemWithVersioning(kind, deletedItem, cb,
+      function(err) {
+        if (err) {
+          logger.error("Error deleting key " + key + " in '" + kind.namespace + "'", err);
+        }
+      });
+  }
 
   store.upsert = function(kind, item, cb) {
     serializeFn(store._upsert, [kind, item], cb);
   };
 
   store._upsert = function(kind, item, cb) {
-    var multi;
-    var baseKey = items_key(kind);
-    var key = item.key;
-    client.watch(baseKey);
-    multi = client.multi();
-
-    do_get(kind, key, function(original) {
-      if (original && original.version >= item.version) {
-        multi.discard();
-        cb();
-      } else {
-        multi.hset(baseKey, key, JSON.stringify(item));
-        multi.exec(function(err, replies) {
-          if (err) {
-            logger.error("Error upserting key " + key + " in '" + kind.namespace + "'", err);
+    updateItemWithVersioning(kind, item, cb,
+      function(err) {
+        if (err) {
+          logger.error("Error upserting key " + key + " in '" + kind.namespace + "'", err);
+        }
+      });
+  }
+
+  function updateItemWithVersioning(kind, newItem, cb, resultFn) {
+    var attempt = function() {
+      client.watch(items_key(kind));
+      var multi = client.multi();
+      // test_transaction_hook is instrumentation, set only by the unit tests
+      var prepare = store.test_transaction_hook || function(prepareCb) { prepareCb(); };
+      prepare(function() {
+        do_get(kind, newItem.key, function(oldItem) {
+          if (oldItem && oldItem.version >= newItem.version) {
+            multi.discard();
+            cb();
           } else {
-            if (cache_ttl) {
-              cache.set(cache_key(kind, key), item);
-            }
+            multi.hset(items_key(kind), newItem.key, JSON.stringify(newItem));
+            multi.exec(function(err, replies) {
+              if (!err && replies === null) {
+                // This means the EXEC failed because someone modified the watched key
+                logger.debug("Concurrent modification detected, retrying");
+                attempt();
+              } else {
+                resultFn(err);
+                if (!err && cache_ttl) {
+                  cache.set(cache_key(kind, newItem.key), newItem);
+                }
+                cb();
+              }
+            });
           }
-          cb();
         });
-      }
-    });
-  };
+      });
+    }
+    attempt();
+  }
 
   store.initialized = function(cb) {
     cb = cb || noop;
diff --git a/test/redis_feature_store-test.js b/test/redis_feature_store-test.js
index 5d16e0e..49181c2 100644
--- a/test/redis_feature_store-test.js
+++ b/test/redis_feature_store-test.js
@@ -1,9 +1,57 @@
 var RedisFeatureStore = require('../redis_feature_store');
 var allFeatureStoreTests = require('./feature_store_test_base');
+var dataKind = require('../versioned_data_kind');
+var redis = require('redis');
 
 describe('RedisFeatureStore', function() {
-  allFeatureStoreTests(function() {
-    redisOpts = { url: 'redis://localhost:6379' };
-    return new RedisFeatureStore(redisOpts, 30000);
-  })
+  var redisOpts = { url: 'redis://localhost:6379' };
+
+  function makeStore() {
+    return new RedisFeatureStore(redisOpts, 30000);    
+  }
+
+  allFeatureStoreTests(makeStore);
+
+  it('handles upsert race condition against external client correctly', function(done) {
+    var store = makeStore();
+    var otherClient = redis.createClient(redisOpts);
+
+    var feature1 = {
+      key: 'foo',
+      version: 1
+    };
+    var intermediateVer = { key: feature1.key, version: feature1.version };
+    var finalVer = { key: feature1.key, version: 10 };
+    
+    var initData = {};
+    initData[dataKind.features.namespace] = {
+      'foo': feature1
+    };
+    
+    store.init(initData, function() {
+      var tries = 0;
+      // This function will be called in between the WATCH and the update transaction.
+      // We're testing that the store will detect this concurrent modification and will
+      // transparently retry the update.
+      store.test_transaction_hook = function(cb) {
+        if (tries < 3) {
+          tries++;
+          intermediateVer.version++;
+          otherClient.hset("launchdarkly:features", "foo", JSON.stringify(intermediateVer), cb);
+        } else {
+          cb();
+        }
+      };
+      // Deliberately do not wait for the first action (ver3) to complete before starting the
+      // second (ver2), so the WATCH on the first will be triggered by the concurrent modification.
+      // The result should be that the ver3 update is transparently retried and succeeds.
+      store.upsert(dataKind.features, finalVer, function() {
+        store.get(dataKind.features, feature1.key, function(result) {
+          otherClient.quit();
+          expect(result).toEqual(finalVer);
+          done();
+        });
+      });      
+    });
+  });
 });

From d52a6b40a888d60789c247a4435e2fe2f2380ee9 Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Tue, 13 Mar 2018 16:29:58 -0700
Subject: [PATCH 07/10] slightly simpler recursion

---
 redis_feature_store.js | 53 ++++++++++++++++++++----------------------
 1 file changed, 25 insertions(+), 28 deletions(-)

diff --git a/redis_feature_store.js b/redis_feature_store.js
index f49724a..cf3a63a 100644
--- a/redis_feature_store.js
+++ b/redis_feature_store.js
@@ -226,36 +226,33 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
   }
 
   function updateItemWithVersioning(kind, newItem, cb, resultFn) {
-    var attempt = function() {
-      client.watch(items_key(kind));
-      var multi = client.multi();
-      // test_transaction_hook is instrumentation, set only by the unit tests
-      var prepare = store.test_transaction_hook || function(prepareCb) { prepareCb(); };
-      prepare(function() {
-        do_get(kind, newItem.key, function(oldItem) {
-          if (oldItem && oldItem.version >= newItem.version) {
-            multi.discard();
-            cb();
-          } else {
-            multi.hset(items_key(kind), newItem.key, JSON.stringify(newItem));
-            multi.exec(function(err, replies) {
-              if (!err && replies === null) {
-                // This means the EXEC failed because someone modified the watched key
-                logger.debug("Concurrent modification detected, retrying");
-                attempt();
-              } else {
-                resultFn(err);
-                if (!err && cache_ttl) {
-                  cache.set(cache_key(kind, newItem.key), newItem);
-                }
-                cb();
+    client.watch(items_key(kind));
+    var multi = client.multi();
+    // test_transaction_hook is instrumentation, set only by the unit tests
+    var prepare = store.test_transaction_hook || function(prepareCb) { prepareCb(); };
+    prepare(function() {
+      do_get(kind, newItem.key, function(oldItem) {
+        if (oldItem && oldItem.version >= newItem.version) {
+          multi.discard();
+          cb();
+        } else {
+          multi.hset(items_key(kind), newItem.key, JSON.stringify(newItem));
+          multi.exec(function(err, replies) {
+            if (!err && replies === null) {
+              // This means the EXEC failed because someone modified the watched key
+              logger.debug("Concurrent modification detected, retrying");
+              updateItemWithVersioning(kind, newItem, cb, resultFn);
+            } else {
+              resultFn(err);
+              if (!err && cache_ttl) {
+                cache.set(cache_key(kind, newItem.key), newItem);
               }
-            });
-          }
-        });
+              cb();
+            }
+          });
+        }
       });
-    }
-    attempt();
+    });
   }
 
   store.initialized = function(cb) {

From 567d7ce241f7dbd279e917a98f571f4aa2dc044f Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Tue, 13 Mar 2018 16:31:26 -0700
Subject: [PATCH 08/10] indent

---
 redis_feature_store.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/redis_feature_store.js b/redis_feature_store.js
index cf3a63a..e014cea 100644
--- a/redis_feature_store.js
+++ b/redis_feature_store.js
@@ -199,7 +199,7 @@ function RedisFeatureStore(redis_opts, cache_ttl, prefix, logger) {
   };
 
   store.delete = function(kind, key, version, cb) {
-      serializeFn(store._delete, [kind, key, version], cb);
+    serializeFn(store._delete, [kind, key, version], cb);
   };
 
   store._delete = function(kind, key, version, cb) {

From 16b2e91be5a85ba20c586874d88f272716f9c639 Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Tue, 13 Mar 2018 17:55:44 -0700
Subject: [PATCH 09/10] rm obsolete comment that referred to a different test

---
 test/redis_feature_store-test.js | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/test/redis_feature_store-test.js b/test/redis_feature_store-test.js
index 49181c2..282d9dd 100644
--- a/test/redis_feature_store-test.js
+++ b/test/redis_feature_store-test.js
@@ -42,9 +42,6 @@ describe('RedisFeatureStore', function() {
           cb();
         }
       };
-      // Deliberately do not wait for the first action (ver3) to complete before starting the
-      // second (ver2), so the WATCH on the first will be triggered by the concurrent modification.
-      // The result should be that the ver3 update is transparently retried and succeeds.
       store.upsert(dataKind.features, finalVer, function() {
         store.get(dataKind.features, feature1.key, function(result) {
           otherClient.quit();

From 7393a21ced6137d878c8b0943c0b05e7272b0e50 Mon Sep 17 00:00:00 2001
From: Eli Bishop <eli@launchdarkly.com>
Date: Wed, 14 Mar 2018 14:06:30 -0700
Subject: [PATCH 10/10] update version and changelog

---
 CHANGELOG.md | 4 ++++
 package.json | 2 +-
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3e52c31..39d9b23 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 All notable changes to the LaunchDarkly Node.js SDK will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org).
 
+## [4.0.2] = 2018-03-14
+### Fixed
+- In the Redis feature store, fixed synchronization problems that could cause a feature flag update to be missed if several of them happened in rapid succession.
+
 ## [4.0.1] - 2018-03-09
 ### Fixed
 - Any Redis connection failure will now be logged and will trigger reconnection attempts transparently. Previously, it caused an uncaught exception. Note that during a Redis outage, flag evaluations will use the last known value from the in-memory cache if available (if this cache was enabled with the `cache_ttl` parameter to `RedisFeatureStore`), or otherwise the default value.
diff --git a/package.json b/package.json
index ccd7964..9bee3fe 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "ldclient-node",
-  "version": "4.0.1",
+  "version": "4.0.2",
   "description": "LaunchDarkly SDK for Node.js",
   "main": "index.js",
   "scripts": {