Merge pull request #948 from flippercloud/poll-shutdown-based-on-header

Shut down polling and adjust poll interval based on headers

Author: jnunemaker

examples/cloud/poll_interval/README.md

@@ -0,0 +1,111 @@
+# Poll Interval Dynamic Adjustment Demo
+
+This demo shows how the Flipper poller dynamically adjusts its polling interval based on the `poll-interval` header from the server, and how it responds to the `poll-shutdown` header.
+
+## Files
+
+- `server.rb` - Test server that responds with configurable headers
+- `client.rb` - Client that polls the server and logs interval changes
+- `README.md` - This file
+
+## How to Run
+
+### Terminal 1: Start the Server
+
+```bash
+bundle exec ruby examples/cloud/poll_interval/server.rb
+```
+
+The server will start on http://localhost:3000 and show a prompt where you can control what headers to send.
+
+### Terminal 2: Start the Client
+
+```bash
+bundle exec ruby examples/cloud/poll_interval/client.rb
+```
+
+The client will start polling the server every 10 seconds (the minimum) and log all activity.
+
+## Testing Scenarios
+
+### 1. Change Poll Interval
+
+In the **server terminal**, type a number to set the poll interval:
+
+```
+> 20
+```
+
+In the **client terminal**, you'll see:
+
+```
+[HH:MM:SS] WARN: ⚠️  INTERVAL CHANGED: 10.0s → 20.0s
+```
+
+The client will now poll every 20 seconds instead of 10.
+
+### 2. Try an Invalid Interval (Below Minimum)
+
+In the **server terminal**:
+
+```
+> 5
+```
+
+In the **client terminal**, you'll see a warning:
+
+```
+Flipper::Cloud poll interval must be greater than or equal to 10 but was 5.0. Setting interval to 10.
+```
+
+The interval will remain at 10 seconds (the minimum).
+
+### 3. Trigger Shutdown
+
+In the **server terminal**:
+
+```
+> shutdown
+```
+
+In the **client terminal**, you'll see:
+
+```
+[HH:MM:SS] WARN: Shutdown requested by server via poll-shutdown header
+[HH:MM:SS] WARN: Poller stopped
+[HH:MM:SS] WARN: Poller thread is no longer running
+```
+
+The poller will stop gracefully.
+
+### 4. Reset Headers
+
+In the **server terminal**:
+
+```
+> reset
+```
+
+The server will stop sending special headers. The client will continue with its current interval.
+
+## What You'll Learn
+
+- How `poll-interval` header dynamically adjusts polling frequency
+- How `poll-shutdown` header gracefully stops the poller
+- How minimum interval enforcement works (10 seconds minimum)
+- How the poller continues working even if the server returns errors
+- Real-time logging of poller events via instrumentation
+
+## Implementation Details
+
+The poller checks response headers in the `ensure` block of the `sync` method, which means:
+
+- Interval adjustments happen even if the sync fails with an error
+- Shutdown signals are never missed, even during failures
+- The poller is resilient to network issues
+
+The `interval=` setter handles all validation:
+
+- Type conversion via `Flipper::Typecast.to_float`
+- Minimum enforcement (10 seconds)
+- Warning messages for invalid values

examples/cloud/poll_interval/client.rb

@@ -0,0 +1,108 @@
+# Example showing poll interval being dynamically adjusted via poll-interval header
+#
+# Usage:
+#   1. Terminal 1: bundle exec ruby examples/cloud/poll_interval/server.rb
+#   2. Terminal 2: bundle exec ruby examples/cloud/poll_interval/client.rb
+
+require 'bundler/setup'
+require 'flipper'
+require 'flipper/adapters/http'
+require 'flipper/poller'
+require 'logger'
+
+# Setup logging to show what's happening
+logger = Logger.new(STDOUT)
+logger.level = Logger::INFO
+logger.formatter = proc do |severity, datetime, progname, msg|
+  "[#{datetime.strftime('%H:%M:%S')}] #{severity}: #{msg}\n"
+end
+
+# Create HTTP adapter pointing to localhost:3000
+http_adapter = Flipper::Adapters::Http.new(url: 'http://localhost:3000/flipper')
+
+# Create instrumenter to log poller events
+instrumenter = Module.new do
+  def self.instrument(name, payload = {})
+    case payload[:operation]
+    when :poll
+      logger.info "Polling remote adapter..."
+    when :shutdown_requested
+      logger.warn "Shutdown requested by server via poll-shutdown header"
+    when :stop
+      logger.warn "Poller stopped"
+    when :thread_start
+      logger.info "Poller thread started"
+    end
+
+    result = yield if block_given?
+
+    if payload[:operation] == :poll && result
+      logger.info "Poll completed successfully"
+    end
+
+    result
+  end
+
+  def self.logger=(l)
+    @logger = l
+  end
+
+  def self.logger
+    @logger
+  end
+end
+instrumenter.logger = logger
+
+# Create poller with custom instrumenter and short initial interval
+poller = Flipper::Poller.new(
+  remote_adapter: http_adapter,
+  interval: 5,  # Start with 5 second interval (will be enforced to 10 minimum)
+  instrumenter: instrumenter,
+  start_automatically: false,
+  shutdown_automatically: false
+)
+
+logger.info "Starting poller with interval: #{poller.interval} seconds"
+logger.info "Minimum allowed interval: #{Flipper::Poller::MINIMUM_POLL_INTERVAL} seconds"
+logger.info ""
+logger.info "Server can control polling via response headers:"
+logger.info "  - poll-interval: <seconds>  (adjust poll frequency)"
+logger.info "  - poll-shutdown: true       (stop polling)"
+logger.info ""
+
+# Track interval changes
+last_interval = poller.interval
+
+# Start the poller
+poller.start
+
+# Monitor for interval changes and log them
+logger.info "Monitoring poller... (Ctrl+C to exit)"
+logger.info ""
+
+begin
+  loop do
+    sleep 2
+
+    current_interval = poller.interval
+
+    # Highlight when it changes
+    if current_interval != last_interval
+      logger.warn "⚠️  INTERVAL CHANGED: #{last_interval}s → #{current_interval}s"
+      last_interval = current_interval
+    end
+
+    # Check if poller thread is still alive
+    unless poller.thread&.alive?
+      logger.warn "Poller thread is no longer running"
+      break
+    end
+  end
+rescue Interrupt
+  logger.info ""
+  logger.info "Interrupted by user"
+ensure
+  logger.info "Stopping poller..."
+  poller.stop
+  logger.info "Final interval: #{poller.interval} seconds"
+end

examples/cloud/poll_interval/server.rb

@@ -0,0 +1,98 @@
+# Simple test server for demonstrating poll interval changes
+#
+# Usage:
+#   1. Terminal 1: bundle exec ruby examples/cloud/poll_interval/server.rb
+#   2. Terminal 2: bundle exec ruby examples/cloud/poll_interval/client.rb
+#
+# Commands in server terminal:
+#   - Type a number (e.g., "15") to set poll-interval header to that value
+#   - Type "shutdown" to send poll-shutdown: true header
+#   - Type "reset" to stop sending special headers
+#   - Ctrl+C to exit
+
+require 'bundler/setup'
+require 'webrick'
+require 'json'
+
+# State for what headers to send
+$poll_interval = nil
+$poll_shutdown = false
+
+# Thread to handle user input for changing headers
+input_thread = Thread.new do
+  puts ""
+  puts "=" * 60
+  puts "Server Controls:"
+  puts "  Type a number (e.g., '15') to set poll-interval"
+  puts "  Type 'shutdown' to trigger poll shutdown"
+  puts "  Type 'reset' to clear all special headers"
+  puts "=" * 60
+  puts ""
+
+  loop do
+    print "> "
+    input = gets&.chomp
+    break if input.nil?
+
+    case input
+    when /^\d+$/
+      $poll_interval = input.to_i
+      puts "✓ Will send poll-interval: #{$poll_interval}"
+    when "shutdown"
+      $poll_shutdown = true
+      puts "✓ Will send poll-shutdown: true"
+    when "reset"
+      $poll_interval = nil
+      $poll_shutdown = false
+      puts "✓ Cleared all special headers"
+    else
+      puts "Unknown command. Use a number, 'shutdown', or 'reset'"
+    end
+  end
+end
+
+# Setup WEBrick server
+server = WEBrick::HTTPServer.new(
+  Port: 3000,
+  Logger: WEBrick::Log.new($stdout, WEBrick::Log::INFO),
+  AccessLog: [[
+    $stdout,
+    WEBrick::AccessLog::COMMON_LOG_FORMAT
+  ]]
+)
+
+# Handle GET /flipper/features
+server.mount_proc '/flipper/features' do |req, res|
+  # Build response
+  response_body = {
+    features: []
+  }
+
+  res.status = 200
+  res['Content-Type'] = 'application/json'
+  res.body = JSON.generate(response_body)
+
+  # Add special headers if configured
+  if $poll_interval
+    res['poll-interval'] = $poll_interval.to_s
+    puts "→ Sent poll-interval: #{$poll_interval}"
+  end
+
+  if $poll_shutdown
+    res['poll-shutdown'] = 'true'
+    puts "→ Sent poll-shutdown: true"
+  end
+end
+
+# Trap interrupt and shutdown gracefully
+trap('INT') do
+  puts "\nShutting down server..."
+  server.shutdown
+  input_thread.kill
+end
+
+puts "Server starting on http://localhost:3000"
+puts "Endpoint: GET http://localhost:3000/flipper/features"
+puts ""
+
+server.start

lib/flipper/adapters/http.rb

@@ -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,29 +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)
+        @get_all_mutex.synchronize { @last_get_all_response = response }
 
-        # Handle 304 Not Modified - return cached result
         if response.is_a?(Net::HTTPNotModified)
-          return @last_get_all_result if @last_get_all_result
-          # If we somehow got 304 without a cached result, treat as error
-          raise Error, response
+          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)
 
-        # Store ETag from response for future requests
-        @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|
@@ -94,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)