Add version-aware sync to skip redundant local writes

lib/flipper/adapter.rb

@@ -28,6 +28,21 @@ def read_only?
       false
     end
 
+    # Public: Read a named integer value from the adapter, or nil if absent.
+    # Adapters that support typed integer storage override this; the default
+    # is a no-op so unaware adapters degrade to today's behavior.
+    def read_integer(key)
+      nil
+    end
+
+    # Public: Atomically set a named integer to value if and only if the new
+    # value is strictly greater than the currently stored value. Returns true
+    # if the write happened, false if rejected or unsupported. Adapters that
+    # support typed integer storage override this.
+    def set_integer_if_greater(key, value)
+      false
+    end
+
     # Public: Get all features and gate values in one call. Defaults to one call
     # to features and another to get_multi. Feel free to override per adapter to
     # make this more efficient.
@@ -36,6 +51,14 @@ def get_all(**kwargs)
       get_multi(instances)
     end
 
+    # Public: Get all features and the version that describes that exact result.
+    # Adapters with native snapshot/version support should override this. The
+    # default is intentionally unversioned because independent get_all and
+    # read_integer calls cannot prove the version describes the returned data.
+    def get_all_snapshot(**kwargs)
+      Flipper::Snapshot.new(features: get_all(**kwargs))
+    end
+
     # Public: Get multiple features in one call. Defaults to one get per
     # feature. Feel free to override per adapter to make this more efficient and
     # reduce network calls.
@@ -94,5 +117,6 @@ def adapter_stack
 
 require "set"
 require "flipper/exporter"
+require "flipper/snapshot"
 require "flipper/feature"
 require "flipper/adapters/sync/synchronizer"

lib/flipper/adapters/active_record.rb

@@ -5,6 +5,7 @@
 require_relative 'active_record/model'
 require_relative 'active_record/feature'
 require_relative 'active_record/gate'
+require_relative 'active_record/kv_integer'
 
 module Flipper
   module Adapters
@@ -32,6 +33,7 @@ def initialize(options = {})
         @name = options.fetch(:name, :active_record)
         @feature_class = options.fetch(:feature_class) { Flipper::Adapters::ActiveRecord::Feature }
         @gate_class = options.fetch(:gate_class) { Flipper::Adapters::ActiveRecord::Gate }
+        @kv_integer_class = options.fetch(:kv_integer_class) { Flipper::Adapters::ActiveRecord::KvInteger }
       end
 
       # Public: The set of known features.
@@ -180,13 +182,54 @@ def disable(feature, gate, thing)
         true
       end
 
+      def read_integer(key)
+        return nil unless kv_integer_table_present?
+        with_connection(@kv_integer_class) do
+          @kv_integer_class.where(key: key.to_s).limit(1).pluck(:value).first
+        end
+      end
+
+      def set_integer_if_greater(key, value)
+        return false unless kv_integer_table_present?
+        value = value.to_i
+        key = key.to_s
+        with_write_connection(@kv_integer_class) do
+          updated = @kv_integer_class
+            .where(key: key)
+            .where("value < ?", value)
+            .update_all(value: value, updated_at: Time.current)
+          return true if updated > 0
+
+          begin
+            @kv_integer_class.create!(key: key, value: value)
+            return true
+          rescue ::ActiveRecord::RecordNotUnique
+            # Row exists. Either stored >= ours (steady-state rejection) or a
+            # concurrent insert raced us with a lower value. Retry UPDATE once;
+            # if it still matches nothing, stored is provably >= ours.
+          end
+
+          @kv_integer_class
+            .where(key: key)
+            .where("value < ?", value)
+            .update_all(value: value, updated_at: Time.current) > 0
+        end
+      end
+
       # Private
       def unsupported_data_type(data_type)
         raise "#{data_type} is not supported by this adapter"
       end
 
       private
 
+      def kv_integer_table_present?
+        return @kv_integer_table_present if defined?(@kv_integer_table_present)
+        @kv_integer_table_present = with_connection(@kv_integer_class) { @kv_integer_class.table_exists? }
+      rescue ::ActiveRecord::StatementInvalid
+        false
+      end
+
       def set(feature, gate, thing, options = {})
         clear_feature = options.fetch(:clear, false)
         json_feature = options.fetch(:json, false)

lib/flipper/adapters/active_record/kv_integer.rb

@@ -0,0 +1,18 @@
+require 'flipper/adapters/active_record/model'
+
+module Flipper
+  module Adapters
+    class ActiveRecord
+      # Private: Do not use outside of this adapter.
+      class KvInteger < Model
+        self.table_name = [
+          Model.table_name_prefix,
+          "flipper_kv_integers",
+          Model.table_name_suffix,
+        ].join
+
+        validates :key, presence: true
+      end
+    end
+  end
+end

lib/flipper/adapters/cache_base.rb

@@ -33,11 +33,13 @@ def initialize(adapter, cache, ttl = 300, prefix: nil)
         @namespace = @namespace.prepend(prefix) if prefix
         @features_cache_key = "#{@namespace}/features"
         @get_all_cache_key = "#{@namespace}/get_all"
+        @get_all_snapshot_cache_key = "#{@namespace}/get_all_snapshot"
       end
 
       # Public: Expire the cache for the set of all features with gates.
       def expire_get_all_cache
         cache_delete @get_all_cache_key
+        cache_delete @get_all_snapshot_cache_key
       end
 
       # Public: Expire the cache for the set of known feature names.
@@ -98,6 +100,21 @@ def get_all(**kwargs)
         }
       end
 
+      def get_all_snapshot(**kwargs)
+        cache_fetch(@get_all_snapshot_cache_key) {
+          snapshot = @adapter.get_all_snapshot(**kwargs)
+          cacheable_snapshot = Flipper::Snapshot.new(features: snapshot.features, version: snapshot.version)
+          cache_write @get_all_cache_key, snapshot.features
+          cache_write @features_cache_key, snapshot.features.keys.to_set
+          if snapshot.version
+            cache_write integer_cache_key(:sync_version), snapshot.version
+          else
+            cache_delete integer_cache_key(:sync_version)
+          end
+          cacheable_snapshot
+        }
+      end
+
       # Public
       def enable(feature, gate, thing)
         result = @adapter.enable(feature, gate, thing)
@@ -112,13 +129,33 @@ def disable(feature, gate, thing)
         result
       end
 
+      # Public
+      def read_integer(key)
+        cache_fetch(integer_cache_key(key)) { @adapter.read_integer(key) }
+      end
+
+      # Public
+      def set_integer_if_greater(key, value)
+        @adapter.set_integer_if_greater(key, value).tap do
+          cache_delete(integer_cache_key(key))
+          cache_delete(@get_all_snapshot_cache_key)
+        end
+      end
+
       # Public: Generate the cache key for a given feature.
       #
       # key - The String or Symbol feature key.
       def feature_cache_key(key)
         "#{@namespace}/feature/#{key}"
       end
 
+      # Public: Generate the cache key for a given integer.
+      #
+      # key - The String or Symbol integer key.
+      def integer_cache_key(key)
+        "#{@namespace}/integer/#{key}"
+      end
+
       private
 
       def read_all_features(**kwargs)

lib/flipper/adapters/dual_write.rb

@@ -58,6 +58,16 @@ def disable(feature, gate, thing)
           @local.disable(feature, gate, thing)
         end
       end
+
+      def read_integer(key)
+        @local.read_integer(key)
+      end
+
+      def set_integer_if_greater(key, value)
+        accepted = @remote.set_integer_if_greater(key, value)
+        @local.set_integer_if_greater(key, value) if accepted
+        accepted
+      end
     end
   end
 end

lib/flipper/adapters/failover.rb

@@ -80,6 +80,18 @@ def disable(feature, gate, thing)
           @secondary.disable(feature, gate, thing) if @dual_write
         end
       end
+
+      def read_integer(key)
+        @primary.read_integer(key)
+      rescue *@errors
+        @secondary.read_integer(key)
+      end
+
+      def set_integer_if_greater(key, value)
+        accepted = @primary.set_integer_if_greater(key, value)
+        @secondary.set_integer_if_greater(key, value) if accepted && @dual_write
+        accepted
+      end
     end
   end
 end

lib/flipper/adapters/failsafe.rb

@@ -67,6 +67,18 @@ def disable(feature, gate, thing)
       rescue *@errors
         false
       end
+
+      def read_integer(key)
+        @adapter.read_integer(key)
+      rescue *@errors
+        nil
+      end
+
+      def set_integer_if_greater(key, value)
+        @adapter.set_integer_if_greater(key, value)
+      rescue *@errors
+        false
+      end
     end
   end
 end

lib/flipper/adapters/http.rb

@@ -25,6 +25,7 @@ def initialize(options = {})
         @last_get_all_etag = nil
         @last_get_all_result = nil
         @last_get_all_response = nil
+        @last_sync_version = nil
         @get_all_mutex = Mutex.new
       end
 
@@ -60,6 +61,10 @@ def get_multi(features)
       end
 
       def get_all(cache_bust: false)
+        get_all_snapshot(cache_bust: cache_bust).features
+      end
+
+      def get_all_snapshot(cache_bust: false)
         options = {}
         path = "/features?exclude_gate_names=true"
         path += "&_cb=#{Time.now.to_i}" if cache_bust
@@ -73,10 +78,14 @@ def get_all(cache_bust: false)
         @get_all_mutex.synchronize { @last_get_all_response = response }
 
         if response.is_a?(Net::HTTPNotModified)
-          cached_result = @get_all_mutex.synchronize { @last_get_all_result }
+          cached_result, cached_version = @get_all_mutex.synchronize { [@last_get_all_result, @last_sync_version] }
 
           if cached_result
-            return cached_result
+            return Flipper::Snapshot.new(
+              features: cached_result,
+              version: cached_version,
+              metadata: { response: response }
+            )
           else
             raise Error, response
           end
@@ -97,18 +106,32 @@ def get_all(cache_bust: false)
           result[feature.key] = result_for_feature(feature, gates_by_key[feature.key])
         end
 
+        sync_version = sync_version_from(response)
         @get_all_mutex.synchronize do
           @last_get_all_etag = response['etag'] if response['etag']
           @last_get_all_result = result
+          @last_sync_version = sync_version
         end
 
-        result
+        Flipper::Snapshot.new(
+          features: result,
+          version: sync_version,
+          metadata: { response: response }
+        )
       end
 
       def last_get_all_response
         @get_all_mutex.synchronize { @last_get_all_response }
       end
 
+      def read_integer(key)
+        return nil unless key.to_sym == :sync_version
+        # HTTP exposes sync_version as metadata on /features. This returns the
+        # last observed value from get_all_snapshot rather than making a second
+        # remote read that could race the feature snapshot.
+        @get_all_mutex.synchronize { @last_sync_version }
+      end
+
       def features
         response = @client.get('/features?exclude_gate_names=true')
         raise Error, response unless response.is_a?(Net::HTTPOK)
@@ -167,6 +190,13 @@ def import(source)
 
       private
 
+      def sync_version_from(response)
+        header = response['flipper-sync-version']
+        Integer(header) if header
+      rescue ArgumentError
+        nil
+      end
+
       def request_body_for_gate(gate, value)
         data = case gate.key
         when :boolean

lib/flipper/adapters/instrumented.rb

@@ -112,6 +112,17 @@ def get_all(**kwargs)
         end
       end
 
+      def get_all_snapshot(**kwargs)
+        default_payload = {
+          operation: :get_all_snapshot,
+          adapter_name: @adapter.name,
+        }
+
+        @instrumenter.instrument(InstrumentationName, default_payload) do |payload|
+          payload[:result] = @adapter.get_all_snapshot(**kwargs)
+        end
+      end
+
       # Public
       def enable(feature, gate, thing)
         default_payload = {
@@ -165,6 +176,31 @@ def export(format: :json, version: 1)
           payload[:result] = @adapter.export(format: format, version: version)
         end
       end
+
+      def read_integer(key)
+        default_payload = {
+          operation: :read_integer,
+          adapter_name: @adapter.name,
+          key: key,
+        }
+
+        @instrumenter.instrument(InstrumentationName, default_payload) do |payload|
+          payload[:result] = @adapter.read_integer(key)
+        end
+      end
+
+      def set_integer_if_greater(key, value)
+        default_payload = {
+          operation: :set_integer_if_greater,
+          adapter_name: @adapter.name,
+          key: key,
+          value: value,
+        }
+
+        @instrumenter.instrument(InstrumentationName, default_payload) do |payload|
+          payload[:result] = @adapter.set_integer_if_greater(key, value)
+        end
+      end
     end
   end
 end