Fix processing of remote Delete activities (#16084)

* Add tests

* Ensure deleted statuses are marked as such

* Save some redis memory by not storing URIs in delete_upon_arrival values

* Avoid possible race condition when processing incoming Deletes

* Avoid potential duplicate Delete forwards

* Lower lock durations to reduce issues in case of hard crash of the Rails process

* Check for `lock.aquired?` and improve comment

* Refactor RedisLock usage in app/lib/activitypub

* Fix using incorrect or non-existent sender for relaying Deletes

app/lib/activitypub/activity/delete.rb

@@ -20,33 +20,35 @@ class ActivityPub::Activity::Delete < ActivityPub::Activity
   def delete_note
     return if object_uri.nil?
 
-    unless invalid_origin?(object_uri)
-      RedisLock.acquire(lock_options) { |_lock| delete_later!(object_uri) }
-      Tombstone.find_or_create_by(uri: object_uri, account: @account)
+    lock_or_return("delete_status_in_progress:#{object_uri}", 5.minutes.seconds) do
+      unless invalid_origin?(object_uri)
+        # This lock ensures a concurrent `ActivityPub::Activity::Create` either
+        # does not create a status at all, or has finished saving it to the
+        # database before we try to load it.
+        # Without the lock, `delete_later!` could be called after `delete_arrived_first?`
+        # and `Status.find` before `Status.create!`
+        lock_or_fail("create:#{object_uri}") { delete_later!(object_uri) }
+
+        Tombstone.find_or_create_by(uri: object_uri, account: @account)
+      end
+
+      @status   = Status.find_by(uri: object_uri, account: @account)
+      @status ||= Status.find_by(uri: @object['atomUri'], account: @account) if @object.is_a?(Hash) && @object['atomUri'].present?
+
+      return if @status.nil?
+
+      forward! if @json['signature'].present? && @status.distributable?
+      delete_now!
     end
-
-    @status   = Status.find_by(uri: object_uri, account: @account)
-    @status ||= Status.find_by(uri: @object['atomUri'], account: @account) if @object.is_a?(Hash) && @object['atomUri'].present?
-
-    return if @status.nil?
-
-    if @status.distributable?
-      forward_for_reply
-      forward_for_reblogs
-    end
-
-    delete_now!
   end
 
-  def forward_for_reblogs
-    return if @json['signature'].blank?
+  def rebloggers_ids
+    return @rebloggers_ids if defined?(@rebloggers_ids)
+    @rebloggers_ids = @status.reblogs.includes(:account).references(:account).merge(Account.local).pluck(:account_id)
+  end
 
-    rebloggers_ids = @status.reblogs.includes(:account).references(:account).merge(Account.local).pluck(:account_id)
-    inboxes        = Account.where(id: ::Follow.where(target_account_id: rebloggers_ids).select(:account_id)).inboxes - [@account.preferred_inbox_url]
-
-    ActivityPub::LowPriorityDeliveryWorker.push_bulk(inboxes) do |inbox_url|
-      [payload, rebloggers_ids.first, inbox_url]
-    end
+  def inboxes_for_reblogs
+    Account.where(id: ::Follow.where(target_account_id: rebloggers_ids).select(:account_id)).inboxes
   end
 
   def replied_to_status
@@ -58,13 +60,19 @@ class ActivityPub::Activity::Delete < ActivityPub::Activity
     !replied_to_status.nil? && replied_to_status.account.local?
   end
 
-  def forward_for_reply
-    return unless @json['signature'].present? && reply_to_local?
+  def inboxes_for_reply
+    replied_to_status.account.followers.inboxes
+  end
 
-    inboxes = replied_to_status.account.followers.inboxes - [@account.preferred_inbox_url]
+  def forward!
+    inboxes = inboxes_for_reblogs
+    inboxes += inboxes_for_reply if reply_to_local?
+    inboxes -= [@account.preferred_inbox_url]
 
-    ActivityPub::LowPriorityDeliveryWorker.push_bulk(inboxes) do |inbox_url|
-      [payload, replied_to_status.account_id, inbox_url]
+    sender_id = reply_to_local? ? replied_to_status.account_id : rebloggers_ids.first
+
+    ActivityPub::LowPriorityDeliveryWorker.push_bulk(inboxes.uniq) do |inbox_url|
+      [payload, sender_id, inbox_url]
     end
   end
 
@@ -75,8 +83,4 @@ class ActivityPub::Activity::Delete < ActivityPub::Activity
   def payload
     @payload ||= Oj.dump(@json)
   end
-
-  def lock_options
-    { redis: Redis.current, key: "create:#{object_uri}" }
-  end
 end