Add optional Active Record feature-state cache

rollout-active_record-adapter/README.md

@@ -84,6 +84,25 @@ Pass an abstract record class that owns the intended connection. The adapter
 defines its own models and does not use application models for features or
 events.
 
+Caching is disabled by default (`cache_ttl_seconds: nil`). Pass
+`cache_ttl_seconds:` to keep an in-process cache of feature state, including
+missing features:
+
+```ruby
+Rollout::Adapters::ActiveRecord.new(
+  base_record_class: ApplicationRecord,
+  cache_ttl_seconds: 10,
+)
+```
+
+The cache belongs to the adapter instance and can serve many requests for as
+long as that instance lives. Separate processes and adapter instances do not
+share it. Cached entries refresh on the next read after `cache_ttl_seconds`;
+reads do not extend expiry. Mutations always use the database. Committed writes
+through the same adapter instance invalidate affected entries. Other processes
+and adapter instances see those writes after the cache TTL expires. Cache reads
+are skipped inside an open database transaction.
+
 Users, groups, metadata, and event payloads are stored as JSON-encoded text.
 
 ## Writes

rollout-active_record-adapter/lib/rollout/active_record/feature_cache.rb

@@ -0,0 +1,94 @@
+# frozen_string_literal: true
+
+class Rollout
+  module ActiveRecord
+    class FeatureCache
+      DEFAULT_MAX_SIZE = 4096
+
+      def initialize(ttl_seconds:, clock: nil, max_size: DEFAULT_MAX_SIZE)
+        unless ttl_seconds.is_a?(Integer) && ttl_seconds > 0
+          raise ArgumentError, "cache_ttl_seconds must be an Integer > 0"
+        end
+        unless max_size.is_a?(Integer) && max_size > 0
+          raise ArgumentError, "max_size must be an Integer > 0"
+        end
+
+        @ttl_seconds = ttl_seconds
+        @max_size = max_size
+        @clock = clock || -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) }
+        @entries = {}
+        @generation = 0
+        @mutex = Mutex.new
+      end
+
+      def generation
+        @mutex.synchronize { @generation }
+      end
+
+      def read(name, context: nil)
+        key = entry_key(name, context)
+        @mutex.synchronize do
+          entry = @entries[key]
+          return nil unless entry
+          if entry[:expires_at] <= now
+            @entries.delete(key)
+            return nil
+          end
+
+          entry[:state].deep_clone
+        end
+      end
+
+      def fill(generation, states, context: nil)
+        clones = states.map { |state| [state.name.to_s, state.deep_clone] }
+        @mutex.synchronize do
+          return if generation != @generation
+
+          clones.each do |name, clone|
+            store_locked(name, clone, context)
+          end
+        end
+      end
+
+      def delete(*names, context: nil)
+        keys = names.map { |name| entry_key(name, context) }
+        @mutex.synchronize do
+          @generation += 1
+          keys.each { |key| @entries.delete(key) }
+        end
+      end
+
+      def clear
+        @mutex.synchronize do
+          @generation += 1
+          @entries.clear
+        end
+      end
+
+      private
+
+      def store_locked(name, clone, context)
+        key = entry_key(name, context)
+        @entries.delete(key)
+        if @entries.size >= @max_size
+          prune_expired_locked
+          @entries.shift while @entries.size >= @max_size
+        end
+        @entries[key] = { state: clone, expires_at: now + @ttl_seconds }
+      end
+
+      def prune_expired_locked
+        t = now
+        @entries.delete_if { |_key, entry| entry[:expires_at] <= t }
+      end
+
+      def entry_key(name, context)
+        [context, name.to_s]
+      end
+
+      def now
+        @clock.call
+      end
+    end
+  end
+end

rollout-active_record-adapter/lib/rollout/adapters/active_record.rb

@@ -5,6 +5,7 @@
 require 'rollout/active_record/schema'
 require 'rollout/active_record/codec'
 require 'rollout/active_record/migration'
+require 'rollout/active_record/feature_cache'
 
 class Rollout
   module Adapters
@@ -15,24 +16,40 @@ class DestinationNotEmpty < ArgumentError
       def initialize(
         base_record_class: ::ActiveRecord::Base,
         features_table_name: "rollout_features",
-        events_table_name: "rollout_events"
+        events_table_name: "rollout_events",
+        cache_ttl_seconds: nil
       )
         @base_record_class = base_record_class
         @features_table_name = features_table_name
         @events_table_name = events_table_name
         @feature_record = build_record_class(@features_table_name)
         @event_record = build_record_class(@events_table_name)
+        @feature_cache = cache_ttl_seconds.nil? ? nil : ::Rollout::ActiveRecord::FeatureCache.new(ttl_seconds: cache_ttl_seconds)
       end
 
       def fetch_feature(name)
-        ::Rollout::ActiveRecord::Codec.feature_state(name, find_feature(name))
+        if use_feature_cache?
+          context = cache_context
+          cached = @feature_cache.read(name, context: context)
+          return cached if cached
+
+          generation = @feature_cache.generation
+          state = load_feature(name)
+          @feature_cache.fill(generation, [state], context: context)
+          return state
+        end
+
+        load_feature(name)
       end
 
       def fetch_features(names)
         return [] if names.empty?
+        return load_features(names) unless use_feature_cache?
 
-        records = @feature_record.where(name: names.map(&:to_s).uniq).index_by(&:name)
-        names.map { |name| ::Rollout::ActiveRecord::Codec.feature_state(name, records[name.to_s]) }
+        context = cache_context
+        hits, misses = partition_cached_features(names, context)
+        fill_feature_cache(hits, misses, context) unless misses.empty?
+        names.map { |name| hits[name.to_s].deep_clone }
       end
 
       def feature_names
@@ -46,15 +63,18 @@ def feature_exists?(name)
       def save_feature(state)
         @feature_record.transaction do
           persist_state(locked_feature(state.name), state)
+          invalidate_features_after_commit(state.name)
         end
       end
 
       def delete_feature(name)
         @feature_record.where(name: name.to_s).delete_all
+        invalidate_features_after_commit(name)
       end
 
       def clear_features
         @feature_record.delete_all
+        invalidate_all_features_after_commit
       end
 
       def occupied?
@@ -81,6 +101,7 @@ def import_features(states, history: [])
           states.each { |state| persist_state(nil, state) }
           history.each { |entry| insert_imported_event(entry) }
           yield if block_given?
+          invalidate_all_features_after_commit
         end
       end
 
@@ -97,6 +118,7 @@ def mutate_feature(name)
             raise
           end
           persist_mutation(record, mutation)
+          invalidate_features_after_commit(name)
         end
 
         raise rollback_error if rollback_error
@@ -136,6 +158,89 @@ def delete_feature_events(name)
 
       private
 
+      def partition_cached_features(names, context)
+        hits = {}
+        misses = []
+        names.map(&:to_s).uniq.each do |name|
+          cached = @feature_cache.read(name, context: context)
+          if cached
+            hits[name] = cached
+          else
+            misses << name
+          end
+        end
+        [hits, misses]
+      end
+
+      def fill_feature_cache(hits, misses, context)
+        generation = @feature_cache.generation
+        load_features(misses).each do |state|
+          hits[state.name] = state
+        end
+        @feature_cache.fill(generation, misses.map { |name| hits[name] }, context: context)
+      end
+
+      def load_feature(name)
+        record = with_query_cache_bypass { find_feature(name) }
+        ::Rollout::ActiveRecord::Codec.feature_state(name, record)
+      end
+
+      def load_features(names)
+        return [] if names.empty?
+
+        unique_names = names.map(&:to_s).uniq
+        records = with_query_cache_bypass do
+          @feature_record.where(name: unique_names).index_by(&:name)
+        end
+        names.map { |name| ::Rollout::ActiveRecord::Codec.feature_state(name, records[name.to_s]) }
+      end
+
+      def with_query_cache_bypass
+        return yield unless @feature_cache
+
+        @feature_record.uncached { yield }
+      end
+
+      def use_feature_cache?
+        @feature_cache && !@feature_record.connection.transaction_open?
+      end
+
+      def cache_context
+        @feature_record.connection_pool.object_id
+      end
+
+      def invalidate_features_after_commit(*names)
+        return unless @feature_cache
+
+        context = cache_context
+        after_outer_commit { @feature_cache.delete(*names, context: context) }
+      end
+
+      def invalidate_all_features_after_commit
+        return unless @feature_cache
+
+        after_outer_commit { @feature_cache.clear }
+      end
+
+      def after_outer_commit(&block)
+        txn = outermost_open_transaction
+        if txn.nil?
+          block.call
+        elsif txn.respond_to?(:after_commit)
+          txn.after_commit(&block)
+        else
+          txn.add_record(AfterCommitCallback.new(&block))
+        end
+      end
+
+      def outermost_open_transaction
+        connection = @feature_record.connection
+        return nil unless connection.transaction_open?
+
+        stack = connection.transaction_manager.instance_variable_get(:@stack)
+        stack&.first
+      end
+
       def build_record_class(table_name)
         Class.new(@base_record_class) do
           self.table_name = table_name
@@ -256,6 +361,26 @@ def validate_history_length!(history_length)
           raise ArgumentError, "history_length must be an Integer >= 0"
         end
       end
+
+      class AfterCommitCallback
+        def initialize(&block)
+          @block = block
+        end
+
+        def committed!(*)
+          @block.call
+        end
+
+        def before_committed!
+        end
+
+        def rolledback!(*)
+        end
+
+        def trigger_transactional_callbacks?
+          true
+        end
+      end
     end
   end
 end

rollout-active_record-adapter/spec/backend_contract_spec.rb

@@ -5,4 +5,9 @@
   it_behaves_like "a rollout feature backend" do
     let(:backend) { active_record_adapter }
   end
+
+  it_behaves_like "a rollout feature backend" do
+    let(:backend) { active_record_adapter(cache_ttl_seconds: 10) }
+  end
 end
+