Fix single Redis connection being used across all threads (#18135)

* Fix single Redis connection being used across all Sidekiq threads

* Fix tests

44 files changed, 243 insertions, 124 deletions

diff --git a/app/controllers/admin/dashboard_controller.rb b/app/controllers/admin/dashboard_controller.rb
index e376baab22a..da9c6dd1622 100644
--- a/app/controllers/admin/dashboard_controller.rb
+++ b/app/controllers/admin/dashboard_controller.rb
@@ -2,6 +2,8 @@
 
 module Admin
   class DashboardController < BaseController
+    include Redisable
+
     def index
       @system_checks         = Admin::SystemCheck.perform
       @time_period           = (29.days.ago.to_date...Time.now.utc.to_date)
@@ -15,10 +17,10 @@ module Admin
 
     def redis_info
       @redis_info ||= begin
-        if Redis.current.is_a?(Redis::Namespace)
-          Redis.current.redis.info
+        if redis.is_a?(Redis::Namespace)
+          redis.redis.info
         else
-          Redis.current.info
+          redis.info
         end
       end
     end
diff --git a/app/controllers/media_proxy_controller.rb b/app/controllers/media_proxy_controller.rb
index 5596e92d183..d2a4cb207ad 100644
--- a/app/controllers/media_proxy_controller.rb
+++ b/app/controllers/media_proxy_controller.rb
@@ -3,6 +3,7 @@
 class MediaProxyController < ApplicationController
   include RoutingHelper
   include Authorization
+  include Redisable
 
   skip_before_action :store_current_location
   skip_before_action :require_functional!
@@ -45,7 +46,7 @@ class MediaProxyController < ApplicationController
   end
 
   def lock_options
-    { redis: Redis.current, key: "media_download:#{params[:id]}", autorelease: 15.minutes.seconds }
+    { redis: redis, key: "media_download:#{params[:id]}", autorelease: 15.minutes.seconds }
   end
 
   def reject_media?
diff --git a/app/controllers/settings/exports_controller.rb b/app/controllers/settings/exports_controller.rb
index 30138d29ed6..1638d3412f0 100644
--- a/app/controllers/settings/exports_controller.rb
+++ b/app/controllers/settings/exports_controller.rb
@@ -2,6 +2,7 @@
 
 class Settings::ExportsController < Settings::BaseController
   include Authorization
+  include Redisable
 
   skip_before_action :require_functional!
 
@@ -28,6 +29,6 @@ class Settings::ExportsController < Settings::BaseController
   end
 
   def lock_options
-    { redis: Redis.current, key: "backup:#{current_user.id}" }
+    { redis: redis, key: "backup:#{current_user.id}" }
   end
 end
diff --git a/app/lib/access_token_extension.rb b/app/lib/access_token_extension.rb
index 2cafaaa20ff..f51bde49273 100644
--- a/app/lib/access_token_extension.rb
+++ b/app/lib/access_token_extension.rb
@@ -4,6 +4,8 @@ module AccessTokenExtension
   extend ActiveSupport::Concern
 
   included do
+    include Redisable
+
     after_commit :push_to_streaming_api
   end
 
@@ -16,6 +18,6 @@ module AccessTokenExtension
   end
 
   def push_to_streaming_api
-    Redis.current.publish("timeline:access_token:#{id}", Oj.dump(event: :kill)) if revoked? || destroyed?
+    redis.publish("timeline:access_token:#{id}", Oj.dump(event: :kill)) if revoked? || destroyed?
   end
 end
diff --git a/app/lib/activitypub/activity.rb b/app/lib/activitypub/activity.rb
index f599e1b58fd..3c51a7a51df 100644
--- a/app/lib/activitypub/activity.rb
+++ b/app/lib/activitypub/activity.rb
@@ -164,7 +164,7 @@ class ActivityPub::Activity
   end
 
   def lock_or_fail(key, expire_after = 15.minutes.seconds)
-    RedisLock.acquire({ redis: Redis.current, key: key, autorelease: expire_after }) do |lock|
+    RedisLock.acquire({ redis: redis, key: key, autorelease: expire_after }) do |lock|
       if lock.acquired?
         yield
       else
diff --git a/app/lib/delivery_failure_tracker.rb b/app/lib/delivery_failure_tracker.rb
index 7b800fc0b74..7c4e28eb79f 100644
--- a/app/lib/delivery_failure_tracker.rb
+++ b/app/lib/delivery_failure_tracker.rb
@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
 class DeliveryFailureTracker
+  include Redisable
+
   FAILURE_DAYS_THRESHOLD = 7
 
   def initialize(url_or_host)
@@ -8,21 +10,21 @@ class DeliveryFailureTracker
   end
 
   def track_failure!
-    Redis.current.sadd(exhausted_deliveries_key, today)
+    redis.sadd(exhausted_deliveries_key, today)
     UnavailableDomain.create(domain: @host) if reached_failure_threshold?
   end
 
   def track_success!
-    Redis.current.del(exhausted_deliveries_key)
+    redis.del(exhausted_deliveries_key)
     UnavailableDomain.find_by(domain: @host)&.destroy
   end
 
   def clear_failures!
-    Redis.current.del(exhausted_deliveries_key)
+    redis.del(exhausted_deliveries_key)
   end
 
   def days
-    Redis.current.scard(exhausted_deliveries_key) || 0
+    redis.scard(exhausted_deliveries_key) || 0
   end
 
   def available?
@@ -30,12 +32,14 @@ class DeliveryFailureTracker
   end
 
   def exhausted_deliveries_days
-    @exhausted_deliveries_days ||= Redis.current.smembers(exhausted_deliveries_key).sort.map { |date| Date.new(date.slice(0, 4).to_i, date.slice(4, 2).to_i, date.slice(6, 2).to_i) }
+    @exhausted_deliveries_days ||= redis.smembers(exhausted_deliveries_key).sort.map { |date| Date.new(date.slice(0, 4).to_i, date.slice(4, 2).to_i, date.slice(6, 2).to_i) }
   end
 
   alias reset! track_success!
 
   class << self
+    include Redisable
+
     def without_unavailable(urls)
       unavailable_domains_map = Rails.cache.fetch('unavailable_domains') { UnavailableDomain.pluck(:domain).index_with(true) }
 
@@ -54,7 +58,7 @@ class DeliveryFailureTracker
     end
 
     def warning_domains
-      domains = Redis.current.keys(exhausted_deliveries_key_by('*')).map do |key|
+      domains = redis.keys(exhausted_deliveries_key_by('*')).map do |key|
         key.delete_prefix(exhausted_deliveries_key_by(''))
       end
 
@@ -62,7 +66,7 @@ class DeliveryFailureTracker
     end
 
     def warning_domains_map
-      warning_domains.index_with { |domain| Redis.current.scard(exhausted_deliveries_key_by(domain)) }
+      warning_domains.index_with { |domain| redis.scard(exhausted_deliveries_key_by(domain)) }
     end
 
     private
diff --git a/app/lib/redis_configuration.rb b/app/lib/redis_configuration.rb
new file mode 100644
index 00000000000..fc8cf2f80f8
--- /dev/null
+++ b/app/lib/redis_configuration.rb
@@ -0,0 +1,47 @@
+# frozen_string_literal: true
+
+class RedisConfiguration
+  class << self
+    def with
+      pool.with { |redis| yield redis }
+    end
+
+    def pool
+      @pool ||= ConnectionPool.new(size: pool_size) { new.connection }
+    end
+
+    def pool_size
+      if Sidekiq.server?
+        Sidekiq.options[:concurrency]
+      else
+        ENV['MAX_THREADS'] || 5
+      end
+    end
+  end
+
+  def connection
+    if namespace?
+      Redis::Namespace.new(namespace, redis: raw_connection)
+    else
+      raw_connection
+    end
+  end
+
+  def namespace?
+    namespace.present?
+  end
+
+  def namespace
+    ENV.fetch('REDIS_NAMESPACE', nil)
+  end
+
+  def url
+    ENV['REDIS_URL']
+  end
+
+  private
+
+  def raw_connection
+    Redis.new(url: url, driver: :hiredis)
+  end
+end
diff --git a/app/models/account_conversation.rb b/app/models/account_conversation.rb
index 56fd13543c8..45e74bbeb31 100644
--- a/app/models/account_conversation.rb
+++ b/app/models/account_conversation.rb
@@ -14,6 +14,8 @@
 #
 
 class AccountConversation < ApplicationRecord
+  include Redisable
+
   after_commit :push_to_streaming_api
 
   belongs_to :account
@@ -109,7 +111,7 @@ class AccountConversation < ApplicationRecord
   end
 
   def subscribed_to_timeline?
-    Redis.current.exists?("subscribed:#{streaming_channel}")
+    redis.exists?("subscribed:#{streaming_channel}")
   end
 
   def streaming_channel
diff --git a/app/models/account_suggestions/global_source.rb b/app/models/account_suggestions/global_source.rb
index 7bca530d409..651041d6751 100644
--- a/app/models/account_suggestions/global_source.rb
+++ b/app/models/account_suggestions/global_source.rb
@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
 class AccountSuggestions::GlobalSource < AccountSuggestions::Source
+  include Redisable
+
   def key
     :global
   end
@@ -28,7 +30,7 @@ class AccountSuggestions::GlobalSource < AccountSuggestions::Source
   end
 
   def account_ids_for_locale(locale)
-    Redis.current.zrevrange("follow_recommendations:#{locale}", 0, -1).map(&:to_i)
+    redis.zrevrange("follow_recommendations:#{locale}", 0, -1).map(&:to_i)
   end
 
   def to_ordered_list_key(account)
diff --git a/app/models/concerns/redisable.rb b/app/models/concerns/redisable.rb
index c6cf9735960..cce8efb862a 100644
--- a/app/models/concerns/redisable.rb
+++ b/app/models/concerns/redisable.rb
@@ -6,6 +6,6 @@ module Redisable
   private
 
   def redis
-    Redis.current
+    Thread.current[:redis] ||= RedisConfiguration.new.connection
   end
 end
diff --git a/app/models/custom_filter.rb b/app/models/custom_filter.rb
index 9d0f3729b34..8e347679414 100644
--- a/app/models/custom_filter.rb
+++ b/app/models/custom_filter.rb
@@ -24,6 +24,7 @@ class CustomFilter < ApplicationRecord
   ).freeze
 
   include Expireable
+  include Redisable
 
   belongs_to :account
 
@@ -51,7 +52,7 @@ class CustomFilter < ApplicationRecord
 
   def remove_cache
     Rails.cache.delete("filters:#{account_id}")
-    Redis.current.publish("timeline:#{account_id}", Oj.dump(event: :filters_changed))
+    redis.publish("timeline:#{account_id}", Oj.dump(event: :filters_changed))
   end
 
   def context_must_be_valid
diff --git a/app/models/encrypted_message.rb b/app/models/encrypted_message.rb
index aa4182b4e12..7b4e32283fa 100644
--- a/app/models/encrypted_message.rb
+++ b/app/models/encrypted_message.rb
@@ -19,6 +19,7 @@ class EncryptedMessage < ApplicationRecord
   self.inheritance_column = nil
 
   include Paginable
+  include Redisable
 
   scope :up_to, ->(id) { where(arel_table[:id].lteq(id)) }
 
@@ -38,7 +39,7 @@ class EncryptedMessage < ApplicationRecord
   end
 
   def subscribed_to_timeline?
-    Redis.current.exists?("subscribed:#{streaming_channel}")
+    redis.exists?("subscribed:#{streaming_channel}")
   end
 
   def streaming_channel
diff --git a/app/models/follow_recommendation_filter.rb b/app/models/follow_recommendation_filter.rb
index acf03cd8440..53133261435 100644
--- a/app/models/follow_recommendation_filter.rb
+++ b/app/models/follow_recommendation_filter.rb
@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
 class FollowRecommendationFilter
+  include Redisable
+
   KEYS = %i(
     language
     status
@@ -17,7 +19,7 @@ class FollowRecommendationFilter
     if params['status'] == 'suppressed'
       Account.joins(:follow_recommendation_suppression).order(FollowRecommendationSuppression.arel_table[:id].desc).to_a
     else
-      account_ids = Redis.current.zrevrange("follow_recommendations:#{@language}", 0, -1).map(&:to_i)
+      account_ids = redis.zrevrange("follow_recommendations:#{@language}", 0, -1).map(&:to_i)
       accounts    = Account.where(id: account_ids).index_by(&:id)
 
       account_ids.map { |id| accounts[id] }.compact
diff --git a/app/models/user.rb b/app/models/user.rb
index 9da7b2639ee..ab832bcd080 100644
--- a/app/models/user.rb
+++ b/app/models/user.rb
@@ -52,6 +52,7 @@ class User < ApplicationRecord
 
   include Settings::Extend
   include UserRoles
+  include Redisable
 
   # The home and list feeds will be stored in Redis for this amount
   # of time, and status fan-out to followers will include only people
@@ -456,7 +457,7 @@ class User < ApplicationRecord
   end
 
   def regenerate_feed!
-    RegenerationWorker.perform_async(account_id) if Redis.current.set("account:#{account_id}:regeneration", true, nx: true, ex: 1.day.seconds)
+    RegenerationWorker.perform_async(account_id) if redis.set("account:#{account_id}:regeneration", true, nx: true, ex: 1.day.seconds)
   end
 
   def needs_feed_update?
diff --git a/app/services/activitypub/process_account_service.rb b/app/services/activitypub/process_account_service.rb
index ec5140720fe..5649153ee92 100644
--- a/app/services/activitypub/process_account_service.rb
+++ b/app/services/activitypub/process_account_service.rb
@@ -3,6 +3,7 @@
 class ActivityPub::ProcessAccountService < BaseService
   include JsonLdHelper
   include DomainControlHelper
+  include Redisable
 
   # Should be called with confirmed valid JSON
   # and WebFinger-resolved username and domain
@@ -289,7 +290,7 @@ class ActivityPub::ProcessAccountService < BaseService
   end
 
   def lock_options
-    { redis: Redis.current, key: "process_account:#{@uri}", autorelease: 15.minutes.seconds }
+    { redis: redis, key: "process_account:#{@uri}", autorelease: 15.minutes.seconds }
   end
 
   def process_tags
diff --git a/app/services/activitypub/process_status_update_service.rb b/app/services/activitypub/process_status_update_service.rb
index 3d9d9cb8462..fb6e44c6d95 100644
--- a/app/services/activitypub/process_status_update_service.rb
+++ b/app/services/activitypub/process_status_update_service.rb
@@ -2,6 +2,7 @@
 
 class ActivityPub::ProcessStatusUpdateService < BaseService
   include JsonLdHelper
+  include Redisable
 
   def call(status, json)
     raise ArgumentError, 'Status has unsaved changes' if status.changed?
@@ -241,7 +242,7 @@ class ActivityPub::ProcessStatusUpdateService < BaseService
   end
 
   def lock_options
-    { redis: Redis.c