From 4a73615193bae27001c1441a131c0f0961a421b9 Mon Sep 17 00:00:00 2001
From: ThibG <thib@sitedethib.com>
Date: Thu, 14 Sep 2017 22:26:22 +0200
Subject: [PATCH] Fix race condition when receiving an ActivityPub Create
 multiple times (#4930)

* Fix race condition when receiving an ActivityPub Create multiple times

* Use a RedisLock to avoid concurrent processing of a same Create activity
---
 app/lib/activitypub/activity/create.rb | 37 ++++++++++++++++----------
 1 file changed, 23 insertions(+), 14 deletions(-)

diff --git a/app/lib/activitypub/activity/create.rb b/app/lib/activitypub/activity/create.rb
index 9a34484f5..894759d9a 100644
--- a/app/lib/activitypub/activity/create.rb
+++ b/app/lib/activitypub/activity/create.rb
@@ -4,26 +4,31 @@ class ActivityPub::Activity::Create < ActivityPub::Activity
   def perform
     return if delete_arrived_first?(object_uri) || unsupported_object_type?
 
-    status = find_existing_status
-
-    return status unless status.nil?
-
-    ApplicationRecord.transaction do
-      status = Status.create!(status_params)
-
-      process_tags(status)
-      process_attachments(status)
+    RedisLock.acquire(lock_options) do |lock|
+      if lock.acquired?
+        @status = find_existing_status
+        process_status if @status.nil?
+      end
     end
 
-    resolve_thread(status)
-    distribute(status)
-    forward_for_reply if status.public_visibility? || status.unlisted_visibility?
-
-    status
+    @status
   end
 
   private
 
+  def process_status
+    ApplicationRecord.transaction do
+      @status = Status.create!(status_params)
+
+      process_tags(@status)
+      process_attachments(@status)
+    end
+
+    resolve_thread(@status)
+    distribute(@status)
+    forward_for_reply if @status.public_visibility? || @status.unlisted_visibility?
+  end
+
   def find_existing_status
     status   = status_from_uri(object_uri)
     status ||= Status.find_by(uri: @object['atomUri']) if @object['atomUri'].present?
@@ -182,4 +187,8 @@ class ActivityPub::Activity::Create < ActivityPub::Activity
     return unless @json['signature'].present? && reply_to_local?
     ActivityPub::RawDistributionWorker.perform_async(Oj.dump(@json), replied_to_status.account_id)
   end
+
+  def lock_options
+    { redis: Redis.current, key: "create:#{@object['id']}" }
+  end
 end