From 3917353645b91dae04f7d9b81162fead6f73072a Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Thu, 28 Apr 2022 17:47:34 +0200
Subject: Fix single Redis connection being used across all threads (#18135)

* Fix single Redis connection being used across all Sidekiq threads

* Fix tests
---
 app/lib/access_token_extension.rb   |  4 +++-
 app/lib/activitypub/activity.rb     |  2 +-
 app/lib/delivery_failure_tracker.rb | 18 ++++++++------
 app/lib/redis_configuration.rb      | 47 +++++++++++++++++++++++++++++++++++++
 4 files changed, 62 insertions(+), 9 deletions(-)
 create mode 100644 app/lib/redis_configuration.rb

(limited to 'app/lib')

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
-- 
cgit v1.2.3