Merge pull request #43 from launchdarkly/eb/ch13390/redis-watch

fix Redis optimistic locking logic to retry updates as needed

Author: eli-darkly

lib/ldclient-rb/redis_store.rb

@@ -103,9 +103,6 @@ def get(kind, key)
             nil
           end
         end
-        if !f.nil?
-          put_cache(kind, key, f)
-        end
       end
       if f.nil?
         @logger.debug { "RedisFeatureStore: #{key} not found in '#{kind[:namespace]}'" }
@@ -138,50 +135,36 @@ def all(kind)
     end
 
     def delete(kind, key, version)
-      with_connection do |redis|
-        f = get_redis(kind, redis, key)
-        if f.nil?
-          put_redis_and_cache(kind, redis, key, { deleted: true, version: version })
-        else
-          if f[:version] < version
-            f1 = f.clone
-            f1[:deleted] = true
-            f1[:version] = version
-            put_redis_and_cache(kind, redis, key, f1)
-          else
-            @logger.warn("RedisFeatureStore: attempted to delete #{key} version: #{f[:version]} \
-  in '#{kind[:namespace]}' with a version that is the same or older: #{version}")
-          end
-        end
-      end
+      update_with_versioning(kind, { key: key, version: version, deleted: true })
     end
 
     def init(all_data)
       @cache.clear
       count = 0
       with_connection do |redis|
         all_data.each do |kind, items|
-          redis.multi do |multi|
-            multi.del(items_key(kind))
-            count = count + items.count
-            items.each { |k, v| put_redis_and_cache(kind, multi, k, v) }
-          end
+          begin
+            redis.multi do |multi|
+              multi.del(items_key(kind))
+              count = count + items.count
+              items.each { |key, item|
+                redis.hset(items_key(kind), key, item.to_json)
+              }
+            end
+            items.each { |key, item|
+              put_cache(kind, key.to_sym, item)
+            }
+          rescue => e
+            @logger.error { "RedisFeatureStore: could not initialize '#{kind[:namespace]}' in Redis, error: #{e}" }
+          end    
         end
       end
       @inited.set(true)
       @logger.info { "RedisFeatureStore: initialized with #{count} items" }
     end
 
     def upsert(kind, item)
-      with_connection do |redis|
-        redis.watch(items_key(kind)) do
-          old = get_redis(kind, redis, item[:key])
-          if old.nil? || (old[:version] < item[:version])
-            put_redis_and_cache(kind, redis, item[:key], item)
-          end
-          redis.unwatch
-        end
-      end
+      update_with_versioning(kind, item)
     end
 
     def initialized?
@@ -195,13 +178,12 @@ def stop
       end
     end
 
+    private
+
     # exposed for testing
-    def clear_local_cache()
-      @cache.clear
+    def before_update_transaction(base_key, key)
     end
 
-    private
-
     def items_key(kind)
       @prefix + ":" + kind[:namespace]
     end
@@ -217,7 +199,13 @@ def with_connection
     def get_redis(kind, redis, key)
       begin
         json_item = redis.hget(items_key(kind), key)
-        JSON.parse(json_item, symbolize_names: true) if json_item
+        if json_item
+          item = JSON.parse(json_item, symbolize_names: true)
+          put_cache(kind, key, item)
+          item
+        else
+          nil
+        end
       rescue => e
         @logger.error { "RedisFeatureStore: could not retrieve #{key} from Redis, error: #{e}" }
         nil
@@ -228,13 +216,39 @@ def put_cache(kind, key, value)
       @cache.store(cache_key(kind, key), value, expires: @expiration_seconds)
     end
 
-    def put_redis_and_cache(kind, redis, key, item)
-      begin
-        redis.hset(items_key(kind), key, item.to_json)
-      rescue => e
-        @logger.error { "RedisFeatureStore: could not store #{key} in Redis, error: #{e}" }
+    def update_with_versioning(kind, new_item)
+      base_key = items_key(kind)
+      key = new_item[:key]
+      try_again = true
+      while try_again
+        try_again = false
+        with_connection do |redis|
+          redis.watch(base_key) do
+            old_item = get_redis(kind, redis, key)
+            before_update_transaction(base_key, key)
+            if old_item.nil? || old_item[:version] < new_item[:version]
+              begin
+                result = redis.multi do |multi|
+                  multi.hset(base_key, key, new_item.to_json)
+                end
+                if result.nil?
+                  @logger.debug { "RedisFeatureStore: concurrent modification detected, retrying" }
+                  try_again = true
+                else
+                  put_cache(kind, key.to_sym, new_item)
+                end
+              rescue => e
+                @logger.error { "RedisFeatureStore: could not store #{key} in Redis, error: #{e}" }
+              end
+            else
+              action = new_item[:deleted] ? "delete" : "update"
+              @logger.warn { "RedisFeatureStore: attempted to #{action} #{key} version: #{old_item[:version]} \
+      in '#{kind[:namespace]}' with a version that is the same or older: #{new_item[:version]}" }
+            end
+            redis.unwatch
+          end
+        end
       end
-      put_cache(kind, key.to_sym, item)
     end
 
     def query_inited

spec/redis_feature_store_spec.rb

@@ -1,5 +1,6 @@
 require "feature_store_spec_base"
 require "json"
+require "redis"
 require "spec_helper"
 
 
@@ -21,23 +22,63 @@ def create_redis_store_uncached()
 describe LaunchDarkly::RedisFeatureStore do
   subject { LaunchDarkly::RedisFeatureStore }
 
-  let(:feature0_with_higher_version) do
-    f = feature0.clone
-    f[:version] = feature0[:version] + 10
-    f
-  end
-
   # These tests will all fail if there isn't a Redis instance running on the default port.
 
   context "real Redis with local cache" do
-
     include_examples "feature_store", method(:create_redis_store)
-
   end
 
   context "real Redis without local cache" do
-
     include_examples "feature_store", method(:create_redis_store_uncached)
+  end
+
+  def add_concurrent_modifier(store, other_client, flag, start_version, end_version)
+    version_counter = start_version
+    expect(store).to receive(:before_update_transaction) { |base_key, key|
+      if version_counter <= end_version
+        new_flag = flag.clone
+        new_flag[:version] = version_counter
+        other_client.hset(base_key, key, new_flag.to_json)
+        version_counter = version_counter + 1
+      end
+    }.at_least(:once)
+  end
+
+  it "handles upsert race condition against external client with lower version" do
+    store = create_redis_store
+    other_client = Redis.new({ url: "redis://localhost:6379" })
+    
+    begin
+      flag = { key: "foo", version: 1 }
+      store.init(LaunchDarkly::FEATURES => { flag[:key] => flag })
+
+      add_concurrent_modifier(store, other_client, flag, 2, 4)
+
+      my_ver = { key: "foo", version: 10 }
+      store.upsert(LaunchDarkly::FEATURES, my_ver)
+      result = store.get(LaunchDarkly::FEATURES, flag[:key])
+      expect(result[:version]).to eq 10
+    ensure
+      other_client.close
+    end
+  end
+
+  it "handles upsert race condition against external client with higher version" do
+    store = create_redis_store
+    other_client = Redis.new({ url: "redis://localhost:6379" })
+    
+    begin
+      flag = { key: "foo", version: 1 }
+      store.init(LaunchDarkly::FEATURES => { flag[:key] => flag })
+
+      add_concurrent_modifier(store, other_client, flag, 3, 3)
 
+      my_ver = { key: "foo", version: 2 }
+      store.upsert(LaunchDarkly::FEATURES, my_ver)
+      result = store.get(LaunchDarkly::FEATURES, flag[:key])
+      expect(result[:version]).to eq 3
+    ensure
+      other_client.close
+    end
   end
 end