A little more safety for threads

Author: jnunemaker

lib/flipper/adapters/http.rb

@@ -10,7 +10,7 @@ module Adapters
     class Http
       include Flipper::Adapter
 
-      attr_reader :client, :last_get_all_response
+      attr_reader :client
 
       def initialize(options = {})
         @client = Client.new(url: options.fetch(:url),
@@ -24,6 +24,8 @@ def initialize(options = {})
                              debug_output: options[:debug_output])
         @last_get_all_etag = nil
         @last_get_all_result = nil
+        @last_get_all_response = nil
+        @get_all_mutex = Mutex.new
       end
 
       def get(feature)
@@ -58,31 +60,30 @@ def get_multi(features)
       end
 
       def get_all(cache_bust: false)
+        options = {}
         path = "/features?exclude_gate_names=true"
         path += "&_cb=#{Time.now.to_i}" if cache_bust
+        etag = @get_all_mutex.synchronize { @last_get_all_etag }
 
-        # Pass If-None-Match header if we have an ETag
-        options = {}
-        if @last_get_all_etag
-          options[:headers] = { if_none_match: @last_get_all_etag }
+        if etag
+          options[:headers] = { if_none_match: etag }
         end
 
         response = @client.get(path, options)
-
-        @last_get_all_response = response
+        @get_all_mutex.synchronize { @last_get_all_response = response }
 
         if response.is_a?(Net::HTTPNotModified)
-          if @last_get_all_result
-            return @last_get_all_result
+          cached_result = @get_all_mutex.synchronize { @last_get_all_result }
+
+          if cached_result
+            return cached_result
           else
             raise Error, response
           end
         end
 
         raise Error, response unless response.is_a?(Net::HTTPOK)
 
-        @last_get_all_etag = response['etag'] if response['etag']
-
         parsed_response = response.body.empty? ? {} : Typecast.from_json(response.body)
         parsed_features = parsed_response['features'] || []
         gates_by_key = parsed_features.each_with_object({}) do |parsed_feature, hash|
@@ -96,11 +97,18 @@ def get_all(cache_bust: false)
           result[feature.key] = result_for_feature(feature, gates_by_key[feature.key])
         end
 
-        # Cache the result for 304 responses
-        @last_get_all_result = result
+        @get_all_mutex.synchronize do
+          @last_get_all_etag = response['etag'] if response['etag']
+          @last_get_all_result = result
+        end
+
         result
       end
 
+      def last_get_all_response
+        @get_all_mutex.synchronize { @last_get_all_response }
+      end
+
       def features
         response = @client.get('/features?exclude_gate_names=true')
         raise Error, response unless response.is_a?(Net::HTTPOK)

lib/flipper/poller.rb

@@ -2,6 +2,7 @@
 require 'concurrent/utility/monotonic_time'
 require 'concurrent/map'
 require 'concurrent/atomic/atomic_fixnum'
+require 'concurrent/atomic/atomic_boolean'
 
 module Flipper
   class Poller
@@ -31,7 +32,7 @@ def initialize(options = {})
       @interval = options.fetch(:interval, 10).to_f
       @last_synced_at = Concurrent::AtomicFixnum.new(0)
       @adapter = Adapters::Memory.new(nil, threadsafe: true)
-      @shutdown_requested = false
+      @shutdown_requested = Concurrent::AtomicBoolean.new(false)
 
       if @interval < MINIMUM_POLL_INTERVAL
         warn "Flipper::Cloud poll interval must be greater than or equal to #{MINIMUM_POLL_INTERVAL} but was #{@interval}. Setting @interval to #{MINIMUM_POLL_INTERVAL}."
@@ -47,7 +48,7 @@ def initialize(options = {})
 
     def start
       reset if forked?
-      return if @shutdown_requested
+      return if @shutdown_requested.true?
       ensure_worker_running
     end
 
@@ -82,7 +83,7 @@ def sync
         ensure
           if @remote_adapter.respond_to?(:last_get_all_response) && @remote_adapter.last_get_all_response
             if Flipper::Typecast.to_boolean(@remote_adapter.last_get_all_response["poll-shutdown"])
-              @shutdown_requested = true
+              @shutdown_requested.make_true
               @instrumenter.instrument("poller.#{InstrumentationNamespace}", {
                 operation: :shutdown_requested,
               })
@@ -130,7 +131,7 @@ def thread_alive?
 
     def reset
       @pid = Process.pid
-      @shutdown_requested = false
+      @shutdown_requested.make_false
       mutex.unlock if mutex.locked?
     end
   end