employ advisory lock for all update services

app/models/journal_manager.rb

@@ -45,14 +45,6 @@ class JournalManager
       @send_notification = true
     end
 
-    def with_advisory_lock_transaction(container)
-      ActiveRecord::Base.transaction do
-        container.class.with_advisory_lock("create_journal_on_#{container.class.name}_#{container.id}") do
-          yield
-        end
-      end
-    end
-
     private
 
     def merge_reference_journals_by_id(new_journals, old_journals, id_key, value)
@@ -261,6 +253,7 @@ class JournalManager
     journable.journals.select(&:new_record?)
 
     journal.save!
+
     journal
   end
 

app/services/attachments/create_service.rb

@@ -40,7 +40,7 @@ class Attachments::CreateService
   #
   # An ActiveRecord::RecordInvalid error is raised if any record can't be saved.
   def call(uploaded_file:, description:)
-    if container.respond_to? :add_journal
+    if JournalManager.journalized?(container)
       save_journalized(uploaded_file, description)
     else
       save_unjournalized(uploaded_file, description)
@@ -50,7 +50,7 @@ class Attachments::CreateService
   private
 
   def save_journalized(uploaded_file, description)
-    JournalManager.with_advisory_lock_transaction(container) do
+    OpenProject::Mutex.with_advisory_lock_transaction(container) do
       # Get the latest attachments to ensure having them all for journalization.
       # A different worker might have added attachments in the meantime, e.g when bulk uploading
       container.attachments.reload

app/services/base_services/base_contracted.rb

@@ -42,7 +42,7 @@ module BaseServices
     end
 
     def call(params = nil)
-      in_context(false) do
+      in_context(model, true) do
         perform(params)
       end
     end

app/services/base_services/create.rb

@@ -30,6 +30,12 @@
 
 module BaseServices
   class Create < Write
+    def call(params = nil)
+      in_user_context(true) do
+        perform(params)
+      end
+    end
+
     private
 
     def instance(_params)

app/services/relations/create_service.rb

@@ -35,7 +35,7 @@ class Relations::CreateService < Relations::BaseService
   end
 
   def call(relation, send_notifications: true)
-    in_context(send_notifications) do
+    in_user_context(send_notifications) do
       update_relation relation, {}
     end
   end

app/services/shared/service_context.rb

@@ -30,12 +30,28 @@
 
 module Shared
   module ServiceContext
-    def in_context(send_notifications)
+    private
+
+    def in_context(model, send_notifications, &block)
+      if model
+        in_mutex_context(model, send_notifications, &block)
+      else
+        in_user_context(send_notifications, &block)
+      end
+    end
+
+    def in_mutex_context(model, send_notifications, &block)
+      OpenProject::Mutex.with_advisory_lock_transaction(model) do
+        in_user_context(send_notifications, &block)
+      end
+    end
+
+    def in_user_context(send_notifications)
       result = nil
 
       ActiveRecord::Base.transaction do
         User.execute_as user do
-          JournalManager.with_send_notifications send_notifications do
+          JournalManager.with_send_notifications(send_notifications) do
             result = yield
 
             if result.failure?

app/services/time_entries/create_service.rb

@@ -29,7 +29,6 @@
 #++
 
 class TimeEntries::CreateService < ::BaseServices::Create
-
   def after_perform(call)
     OpenProject::Notifications.send(
       OpenProject::Events::NEW_TIME_ENTRY_CREATED,

app/services/work_packages/create_service.rb

@@ -43,7 +43,7 @@ class WorkPackages::CreateService
   def call(work_package: WorkPackage.new,
            send_notifications: true,
            **attributes)
-    in_context(send_notifications) do
+    in_user_context(send_notifications) do
       create(attributes, work_package)
     end
   end

app/services/work_packages/update_service.rb

@@ -43,7 +43,7 @@ class WorkPackages::UpdateService
   end
 
   def call(send_notifications: true, **attributes)
-    in_context(send_notifications) do
+    in_context(model, send_notifications) do
       update(attributes)
     end
   end

lib/open_project/info.rb

@@ -1,4 +1,5 @@
 #-- encoding: UTF-8
+
 #-- copyright
 # OpenProject is an open source project management software.
 # Copyright (C) 2012-2020 the OpenProject GmbH

lib/open_project/mutex.rb

@@ -0,0 +1,89 @@
+#-- encoding: UTF-8
+
+#-- copyright
+# OpenProject is an open source project management software.
+# Copyright (C) 2012-2020 the OpenProject GmbH
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License version 3.
+#
+# OpenProject is a fork of ChiliProject, which is a fork of Redmine. The copyright follows:
+# Copyright (C) 2006-2017 Jean-Philippe Lang
+# Copyright (C) 2010-2013 the ChiliProject Team
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+#
+# See docs/COPYRIGHT.rdoc for more details.
+#++
+
+module OpenProject
+  module Mutex
+    module_function
+
+    # Adds a mutex on database level via advisory locks:
+    # https://www.postgresql.org/docs/12/explicit-locking.html
+    # employing the with_advisory_lock gem.
+    # This leads to requests received in parallel by two different workers, which might be in different OS processes
+    # on different hosts, to be executed sequentially.
+    # The locks are, from an abstract perspective, on row level as the entry's id is taken to create the lock.
+    # Other entries of the same class can still be created/updated.
+    # Relying on explicit row level locking (e.g. ROW EXCLUSIVE FOR UPDATE) would not work, I believe, as it would still
+    # allow reading. Allowing reading is desirable in some cases, e.g. when fetching for the work package list, but would not
+    # prevent the creation of journals. The journals transaction could not be committed, but the faulty view of the entries
+    # current state could be committed anyway afterwards. See below for details.
+    #
+    # We in particular have to wait until the transaction is complete.
+    # Otherwise the following is e.g. possible:
+    # Action 1 writes an attachment and releases the lock.
+    # Action 2, running in parallel, writes an attachment and releases the lock.
+    # Action 3 updates something, some time later.
+    # Both have not yet committed their transaction so, given that we have the "READ COMMITTED" transaction level,
+    # the two actions cannot yet see the attachment added by the respective other action.
+    # Both actions now proceed to write their journals, which will create attachable journals based on the attachments
+    # at that time. But given that both do not see the attachment added by the other action yet, they only add
+    # attachable journals for the attachment added by themselves. To the user this will look as if one of the actions
+    # deleted the other attachment. The next action, Action 3,  will then seem to have readded the attachment,
+    # seemingly removed before.
+    def with_advisory_lock_transaction(entry, &block)
+      ActiveRecord::Base.transaction do
+        with_advisory_lock(entry, &block)
+      end
+    end
+
+    def with_advisory_lock(entry)
+      lock_name = "mutex_on_#{entry.class.name}_#{entry.id}"
+
+      debug_log("Attempting to fetched advisory lock", lock_name)
+      result = entry.class.with_advisory_lock(lock_name, transaction: true) do
+        debug_log("Fetched advisory lock", lock_name)
+        yield
+      end
+      debug_log("Released advisory lock", lock_name)
+
+      result
+    end
+
+    def debug_log(action, lock_name)
+      message = <<~MESSAGE
+        #{action}:
+          * lockname: #{lock_name}
+          * thread:  #{Thread.current.object_id}
+          * held locks: #{WithAdvisoryLock::Base.lock_stack.map(&:name).join(', ')}
+      MESSAGE
+
+      Rails.logger.debug { message }
+    end
+  end
+end

lib/plugins/acts_as_attachable/lib/acts_as_attachable.rb

@@ -157,6 +157,8 @@ module Redmine
               (persisted? && allowed_to_on_attachment?(user, self.class.attachable_options[:add_on_persisted_permission]))
           end
 
+          # TODO: Check if we can remove this.
+          # If the attachments where only added via the Attachments::CreateService, it would be thread safe.
           # Bulk attaches a set of files to an object
           def attach_files(attachments)
             return unless attachments&.is_a?(Hash)

lib/plugins/acts_as_journalized/lib/redmine/acts/journalized/save_hooks.rb

@@ -74,9 +74,9 @@ module Redmine::Acts::Journalized
 
       add_journal = journals.empty? || JournalManager.changed?(self) || !@journal_notes.empty?
 
-      journal = JournalManager.add_journal! self, @journal_user, @journal_notes if add_journal
-
       if add_journal
+        journal = JournalManager.add_journal!(self, @journal_user, @journal_notes)
+
         OpenProject::Notifications.send('journal_created',
                                         journal: journal,
                                         send_notification: JournalManager.send_notification)