Bound native fragment caching and collapse concurrent page renders

README.md

@@ -52,6 +52,8 @@ If you'd rather run the Docker image yourself, you can read more about that in t
 Authenticated room, message, sidebar and search pages use a bounded 64 MiB cache per worker.
 Set `CAMPFIRE_RESPONSE_CACHE_MB=0` to disable it. Every request still checks authentication
 and room access; commits from any SQLite writer invalidate pages, and CSRF masks stay fresh.
+Native HTML, JSON and stream fragments have a separate 64 MiB memory limit per worker;
+shared rate limits retain their existing store.
 
 ## Other implementations
 

app/controllers/concerns/cached_responses.rb

@@ -9,14 +9,28 @@ module CachedResponses
   end
 
   def perform_caching
-    super && !@rendering_uncached_response
+    super && @response_cache_version.present? && !ActiveRecord::Base.connection.transaction_open?
+  end
+
+  # Keep the class store (including shared rate limits) and Rails.cache unchanged.
+  def cache_store
+    FragmentCache.store
+  end
+
+  def combined_fragment_cache_key(key)
+    @fragment_cache_namespace ||= [
+      @response_cache_version, request.base_url, request.script_name, request.format.to_s, I18n.locale,
+      Current.user&.id, Current.session&.token,
+      (Digest::SHA256.hexdigest(real_csrf_token) if request.format.html? || request.format.turbo_stream?)
+    ].freeze
+    super([ @fragment_cache_namespace, key ])
   end
 
   private
     def capture_response_cache_version
-      if request.get? && ResponseCache.instance.budget.positive?
-        @response_cache_version = ResponseCache.instance.version
-      end
+      # Capture for native HTML/JSON/stream renders too, even with page reuse off.
+      # Detached renderers do not run callbacks and therefore render uncached.
+      @response_cache_version = ResponseCache.instance.version
     end
 
     # Register after room authorization, but before presentation queries.
@@ -26,27 +40,47 @@ module CachedResponses
         key = response_cache_key
         original_session = session.to_hash.deep_dup
 
-        if key.bytesize <= ResponseCache::MAX_KEY_BYTES && (entry = ResponseCache.instance.read(key, @response_cache_version))
-          response.headers.merge!(entry[:headers])
-          self.response_body = entry[:body].gsub(entry[:marker], token)
-        else
-          render_fresh_response { yield }
-          if response.status == 200 && response.media_type == "text/html" && session.to_hash == original_session
-            marker = "campfire-csrf-#{SecureRandom.hex(32)}"
-            # Replace only framework token attributes, never a matching token in
-            # message text. Postprocessing also leaves fragment caches untouched.
-            body = csrf_neutral_body(response.body, marker)
-            headers = response.headers.slice(*CACHE_HEADERS).to_h.freeze
-            ResponseCache.instance.write(key, @response_cache_version, { body: body.freeze, marker: marker.freeze, headers: headers }.freeze)
+        return yield if key.bytesize > ResponseCache::MAX_KEY_BYTES
+
+        entry = ResponseCache.instance.read(key, @response_cache_version)
+        rendered = false
+        unless entry
+          ResponseCache.instance.synchronize_render(key, @response_cache_version) do
+            entry = ResponseCache.instance.read(key, @response_cache_version)
+            if !entry && ResponseCache.instance.version == @response_cache_version
+              yield
+              rendered = true
+              cache_completed_response(key, original_session)
+            end
           end
         end
+
+        if entry
+          response.headers.merge!(entry[:headers])
+          self.response_body = entry[:body].gsub(entry[:marker], token)
+        elsif !rendered
+          # A queued request retains its pre-auth snapshot. If it has expired,
+          # render outside the stripe instead of blocking the next generation.
+          yield
+        end
       else
-        render_fresh_response { yield }
+        yield
+      end
+    end
+
+    def cache_completed_response(key, original_session)
+      if response.status == 200 && response.media_type == "text/html" && session.to_hash == original_session
+        marker = "campfire-csrf-#{SecureRandom.hex(32)}"
+        # Replace only framework token attributes, never a matching token in
+        # message text. Postprocessing also leaves fragment caches untouched.
+        body = csrf_neutral_body(response.body, marker)
+        headers = response.headers.slice(*CACHE_HEADERS).to_h.freeze
+        ResponseCache.instance.write(key, @response_cache_version, { body: body.freeze, marker: marker.freeze, headers: headers }.freeze)
       end
     end
 
     def cacheable_read_request?
-      @response_cache_version && request.get? && request.format.html? && Current.session &&
+      ResponseCache.instance.budget.positive? && @response_cache_version && request.get? && request.format.html? && Current.session &&
         !authenticated_by.bot_key? && flash.empty? &&
         !request.headers["If-None-Match"] && !request.headers["If-Modified-Since"] &&
         !Rails.application.config.content_security_policy_nonce_generator &&
@@ -64,16 +98,6 @@ module CachedResponses
       ])
     end
 
-    def render_fresh_response
-      # Foreign SQL can change content without bumping fragment timestamps.
-      # A whole-page miss renders fresh instead of creating unbounded Redis
-      # fragment namespaces for every database commit.
-      @rendering_uncached_response = true
-      yield
-    ensure
-      @rendering_uncached_response = false
-    end
-
     def csrf_neutral_body(body, marker)
       body.gsub(CSRF_TAG) do |tag|
         attribute = tag.start_with?("<meta") ? "content" : "value"

app/models/fragment_cache.rb

@@ -0,0 +1,9 @@
+# Native view caches share one byte budget across every database generation.
+class FragmentCache
+  STORE = ActiveSupport::Cache::MemoryStore.new(size: 64.megabytes)
+  private_constant :STORE
+
+  def self.store
+    STORE
+  end
+end

app/models/response_cache.rb

@@ -7,13 +7,14 @@ class ResponseCache
   MAX_KEY_BYTES = 2.kilobytes
 
   def self.instance
-    @instance ||= new
+    INSTANCE
   end
 
   def initialize
     @mutex = Mutex.new
     @entries = {}
     @bytes = 0
+    @render_locks = Array.new(16) { Mutex.new }
   end
 
   def budget
@@ -36,6 +37,12 @@ class ResponseCache
     nil
   end
 
+  # Collapse cold renders without retaining a mutex for every viewer or URL.
+  # Rendering must never hold the observer/entry mutex: controllers read it too.
+  def synchronize_render(key, version, &block)
+    @render_locks[[ key, version ].hash % @render_locks.length].synchronize(&block)
+  end
+
   def write(key, version, entry)
     size = key.bytesize + entry[:body].bytesize + entry[:headers].sum { |name, value| name.bytesize + value.bytesize } + 256
     return if key.bytesize > MAX_KEY_BYTES || size > [ budget, MAX_ENTRY_BYTES ].min
@@ -86,4 +93,6 @@ class ResponseCache
       end
       [ @database, @namespace, @version ]
     end
+
+  INSTANCE = new
 end

config/initializers/fragment_cache.rb

@@ -0,0 +1,24 @@
+require "jbuilder/jbuilder_template"
+
+# Jbuilder 2.14.1 uses Rails.cache directly; use the same bounded store as ERB
+# while retaining its native keys and controller instrumentation.
+module BoundedJbuilderFragments
+  private
+    def _read_fragment_cache(key, options = nil)
+      @context.controller.instrument_fragment_cache :read_fragment, key do
+        @context.controller.cache_store.read(key, options)
+      end
+    end
+
+    def _write_fragment_cache(key, options = nil)
+      @context.controller.instrument_fragment_cache :write_fragment, key do
+        yield.tap { |value| @context.controller.cache_store.write(key, value, options) }
+      end
+    end
+end
+
+JbuilderTemplate.prepend(BoundedJbuilderFragments)
+
+Rails.application.config.to_prepare do
+  ActionView::PartialRenderer.collection_cache = FragmentCache.store
+end

test/controllers/fragment_cache_test.rb

@@ -0,0 +1,125 @@
+require "test_helper"
+
+class FragmentRenderingTest < ActionDispatch::IntegrationTest
+  self.use_transactional_tests = false
+
+  setup do
+    host! "once.campfire.test"
+    sign_in :david
+    @previous_caching = ActionController::Base.perform_caching
+    @previous_message_caching = MessagesController.perform_caching
+    @class_store = ApplicationController.cache_store
+    @global_store = Rails.cache
+    ActionController::Base.perform_caching = true
+    MessagesController.perform_caching = true
+    FragmentCache.store.clear
+    ResponseCache.instance.clear
+    @room = rooms(:watercooler)
+    @message = @room.messages.ordered.last
+  end
+
+  teardown do
+    ActionController::Base.perform_caching = @previous_caching
+    MessagesController.perform_caching = @previous_message_caching
+    FragmentCache.store.clear
+    ResponseCache.instance.clear
+  end
+
+  test "native collection caching uses the bounded store without replacing shared stores" do
+    assert_same FragmentCache.store, ApplicationController.new.cache_store
+    assert_same FragmentCache.store, ActionView::PartialRenderer.collection_cache
+    assert_same @class_store, ApplicationController.cache_store
+    assert_same @global_store, Rails.cache
+    ResponseCache.instance.stubs(:budget).returns(0)
+    get room_messages_url(@room)
+    assert_response :success
+    hits = []
+    ActiveSupport::Notifications.subscribed(->(event) { hits.concat(event.payload[:hits]) }, "cache_read_multi.active_support") do
+      get room_messages_url(@room)
+    end
+    assert_response :success
+    assert_not_empty hits
+  end
+
+  test "disabled page caching retains fresh native fragments after foreign leaf writes" do
+    ResponseCache.instance.stubs(:budget).returns(0)
+    get room_messages_url(@room)
+    assert_response :success
+    foreign_write("UPDATE action_text_rich_texts SET body = ? WHERE record_type = 'Message' AND record_id = ?", "foreign disabled body", @message.id)
+    get room_messages_url(@room)
+    assert_response :success
+    assert_includes response.body, "foreign disabled body"
+  end
+
+  test "conditional pagination invalidates warmed native fragments without timestamp changes" do
+    get room_messages_url(@room), headers: { "If-None-Match" => "unmatched" }
+    assert_response :success
+    etag = response.headers["ETag"]
+    foreign_write("UPDATE action_text_rich_texts SET body = ? WHERE record_type = 'Message' AND record_id = ?", "foreign conditional body", @message.id)
+    get room_messages_url(@room), headers: { "If-None-Match" => etag }
+    assert_response :success
+    assert_includes response.body, "foreign conditional body"
+  end
+
+  test "refresh streams recheck native cached creator body and boost presentation" do
+    message = messages(:first)
+    room = message.room
+    get room_refresh_url(room, format: :turbo_stream), params: { since: 0 }
+    assert_response :success
+    foreign_write("UPDATE action_text_rich_texts SET body = ? WHERE record_type = 'Message' AND record_id = ?", "foreign refreshed body", message.id)
+    foreign_write("UPDATE users SET name = ? WHERE id = ?", "Foreign refreshed creator", message.creator_id)
+    foreign_write("UPDATE boosts SET content = ? WHERE message_id = ?", "Foreign refreshed boost", message.id)
+    get room_refresh_url(room, format: :turbo_stream), params: { since: 0 }
+    assert_response :success
+    assert_includes response.body, "foreign refreshed body"
+    assert_includes response.body, "Foreign refreshed creator"
+    assert_includes response.body, "Foreign refreshed boost"
+  end
+
+  test "Jbuilder fragments use the bounded store and do not create HTML token state" do
+    api = open_session { |session| session.host! "once.campfire.test" }
+    path = room_bot_messages_path(@room, users(:bender).bot_key)
+    api.get path
+    assert_equal 200, api.response.status
+    assert_not api.cookies["_campfire_session"]
+    hits = []
+    ActiveSupport::Notifications.subscribed(->(event) { hits << event.payload[:key] if event.payload[:hit] }, "cache_read.active_support") do
+      api.get path
+    end
+    assert_equal 200, api.response.status
+    assert_not_empty hits
+    foreign_write("UPDATE action_text_rich_texts SET body = ? WHERE record_type = 'Message' AND record_id = ?", "foreign JSON body", @message.id)
+    foreign_write("UPDATE users SET name = ? WHERE id = ?", "Foreign JSON creator", @message.creator_id)
+    api.get path
+    json = JSON.parse(api.response.body).find { |message| message["id"] == @message.id }
+    assert_includes json.dig("body", "html"), "foreign JSON body"
+    assert_equal "Foreign JSON creator", json.dig("creator", "name")
+    api.host! "another.campfire.test:8081"
+    api.get path
+    json = JSON.parse(api.response.body).find { |message| message["id"] == @message.id }
+    assert_includes json["url"], "another.campfire.test:8081"
+    assert_same @global_store, Rails.cache
+  end
+
+  test "detached broadcasts bypass fragments without a pre-render model snapshot" do
+    get room_messages_url(@room)
+    foreign_write("UPDATE action_text_rich_texts SET body = ? WHERE record_type = 'Message' AND record_id = ?", "foreign broadcast body", @message.id)
+    FragmentCache.store.expects(:read).never
+    FragmentCache.store.expects(:write).never
+    @message.reload.broadcast_create
+    assert_rendered_turbo_stream_broadcast @room, :messages, action: "append", target: [ @room, :messages ] do |stream|
+      assert_includes stream.to_html, "foreign broadcast body"
+    end
+    body = ApplicationController.render(partial: "messages/message", locals: { message: @message.reload })
+    assert_includes body, "foreign broadcast body"
+    assert_not ApplicationController.new.perform_caching
+  end
+
+  private
+    def foreign_write(sql, *bindings)
+      SQLite3::Database.new(ActiveRecord::Base.connection_db_config.database) do |database|
+        database.execute(sql, bindings)
+      end
+      ActiveRecord::Base.clear_query_caches_for_current_thread
+    end
+end

test/models/response_cache_test.rb

@@ -44,12 +44,43 @@ class ResponseCacheTest < ActiveSupport::TestCase
     assert_nil @cache.read("k" * 2049, version)
   end
 
-  test "observer namespaces cannot reuse old shared fragment keys after restart" do
+  test "clearing the observer gives later pages a distinct namespace" do
     version = @cache.version
     @cache.clear
     assert_not_equal version, @cache.version
   end
 
+  test "cold renders coalesce without holding the observer mutex" do
+    version = @cache.version
+    entered = Queue.new
+    release = Queue.new
+    first = Thread.new do
+      @cache.synchronize_render("page", version) do
+        entered << true
+        release.pop
+        @cache.write("page", version, entry("completed render"))
+      end
+    end
+    entered.pop
+    second_started = Queue.new
+    second = Thread.new do
+      second_started << true
+      @cache.synchronize_render("page", version) { @cache.read("page", version) }
+    end
+    second_started.pop
+    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 2
+    Thread.pass until second.status == "sleep" || Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline
+    assert_equal "sleep", second.status
+    assert_nil @cache.read("page", version)
+    assert_equal version, @cache.version
+    release << true
+    first.value
+    assert_equal "completed render", second.value[:body]
+  ensure
+    release << true if release
+    [ first, second ].compact.each { |thread| thread.join(2) }
+  end
+
   private
     def entry(body)
       { body: body, marker: "unexposed-marker", headers: {} }