Merge pull request #510 from zilverline/do-not-default-to-nested-transactions

Do not default to using nested transactions

Author: erikrozendaal

db/migrate/20260220122600_sequent_avoid_exception_handling_in_update_unique_keys.rb

@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+class SequentAvoidExceptionHandlingInUpdateUniqueKeys < ActiveRecord::Migration[7.2]
+  def up
+    Sequent::Support::Database.with_search_path(Sequent.configuration.event_store_schema_name) do
+      execute_sql_file 'update_unique_keys', version: 4
+    end
+  end
+
+  def down
+    Sequent::Support::Database.with_search_path(Sequent.configuration.event_store_schema_name) do
+      execute_sql_file 'update_unique_keys', version: 3
+    end
+  end
+
+  private
+
+  def execute_sql_file(filename, version:)
+    say "Applying '#{filename}' version #{version}", true
+    suppress_messages do
+      execute File.read(
+        File.join(
+          File.dirname(__FILE__),
+          format('sequent/%s_v%02d.sql', filename, version),
+        ),
+      )
+    end
+  end
+end

db/migrate/sequent/update_unique_keys_v04.sql

@@ -0,0 +1,46 @@
+CREATE OR REPLACE PROCEDURE update_unique_keys(_stream_records jsonb)
+LANGUAGE plpgsql SET search_path FROM CURRENT AS $$
+DECLARE
+  _aggregate jsonb;
+  _aggregate_id aggregates.aggregate_id%TYPE;
+  _aggregate_version events.sequence_number%TYPE;
+  _unique_keys jsonb;
+BEGIN
+  FOR _aggregate IN SELECT aggregate FROM jsonb_array_elements(_stream_records) AS aggregate ORDER BY aggregate->>'aggregate_id' LOOP
+    _aggregate_id = _aggregate->>'aggregate_id';
+    _aggregate_version = _aggregate->>'aggregate_version';
+    IF _aggregate_version <> (
+      SELECT e.sequence_number AS aggregate_version
+        FROM aggregates a LEFT JOIN events e ON (a.aggregate_id, a.events_partition_key) = (e.aggregate_id, e.partition_key)
+       WHERE a.aggregate_id = _aggregate_id
+       ORDER BY e.sequence_number DESC
+       LIMIT 1
+         FOR NO KEY UPDATE OF a
+    ) THEN
+      RAISE EXCEPTION 'update_unique_keys: aggregate version is not equal to latest event sequence number for aggregate %', _aggregate_id;
+    END IF;
+  END LOOP;
+
+  FOR _aggregate IN SELECT aggregate FROM jsonb_array_elements(_stream_records) AS aggregate ORDER BY aggregate->>'aggregate_id' LOOP
+    _aggregate_id = _aggregate->>'aggregate_id';
+    _unique_keys = COALESCE(_aggregate->'unique_keys', '{}'::jsonb);
+
+    DELETE FROM aggregate_unique_keys AS target
+     WHERE target.aggregate_id = _aggregate_id
+       AND NOT (_unique_keys ? target.scope);
+  END LOOP;
+
+  FOR _aggregate IN SELECT aggregate FROM jsonb_array_elements(_stream_records) AS aggregate ORDER BY aggregate->>'aggregate_id' LOOP
+    _aggregate_id = _aggregate->>'aggregate_id';
+    _unique_keys = COALESCE(_aggregate->'unique_keys', '{}'::jsonb);
+
+    INSERT INTO aggregate_unique_keys AS target (aggregate_id, scope, key)
+    SELECT _aggregate_id, key, value
+      FROM jsonb_each(_unique_keys) AS x
+     ORDER BY 1, 2
+        ON CONFLICT (aggregate_id, scope) DO UPDATE
+       SET key = EXCLUDED.key
+     WHERE target.key <> EXCLUDED.key;
+  END LOOP;
+END;
+$$;

db/structure.sql

@@ -734,10 +734,6 @@ BEGIN
        SET key = EXCLUDED.key
      WHERE target.key <> EXCLUDED.key;
   END LOOP;
-EXCEPTION
-  WHEN unique_violation THEN
-    RAISE unique_violation
-    USING MESSAGE = 'duplicate unique key value for aggregate ' || (_aggregate->>'aggregate_type') || ' ' || _aggregate_id || ' (' || SQLERRM || ')';
 END;
 $$;
 
@@ -1588,6 +1584,7 @@ ALTER TABLE ONLY sequent_schema.snapshot_records
 SET search_path TO public,view_schema,sequent_schema;
 
 INSERT INTO "schema_migrations" (version) VALUES
+('20260220122600'),
 ('20260129130000'),
 ('20250815103000'),
 ('20250630113000'),

lib/sequent/core/active_projectors_event_publisher.rb

@@ -21,7 +21,7 @@ def process_events(event_handlers, ...)
         # (active or not) and updating the projector tables. Normally a transaction is already
         # active due to using the `CommandService#execute_command`, but if this event publisher is
         # used directly it is also important to run inside a transaction.
-        Sequent.configuration.transaction_provider.transactional do
+        Sequent.configuration.transaction_provider.transaction do
           ensure_no_unknown_active_projectors!(event_handlers)
           active_event_handlers = event_handlers.select { |x| active?(x) }
           super(active_event_handlers, ...)

lib/sequent/core/command_service.rb

@@ -44,7 +44,7 @@ def execute_commands(*commands, validation_context: nil)
 
       def process_commands(validation_context)
         Sequent::Util.skip_if_already_processing(:command_service_process_commands) do
-          transaction_provider.transactional do
+          transaction_provider.transaction do
             until command_queue.empty?
               command = command_queue.pop
               command_middleware.invoke(command) do

lib/sequent/core/event_store.rb

@@ -10,24 +10,8 @@
 module Sequent
   module Core
     class AggregateKeyNotUniqueError < RuntimeError
-      attr_reader :aggregate_type, :aggregate_id
-
       def self.unique_key_error_message?(message)
-        message =~ /duplicate unique key value for aggregate/
-      end
-
-      def initialize(message)
-        super
-
-        match = message.match(
-          # rubocop:disable Layout/LineLength
-          /aggregate (\p{Upper}\p{Alnum}*(?:::\p{Upper}\p{Alnum}*)*) (\p{XDigit}{8}-\p{XDigit}{4}-\p{XDigit}{4}-\p{XDigit}{4}-\p{XDigit}{12})/,
-          # rubocop:enable Layout/LineLength
-        )
-        if match
-          @aggregate_type = match[1]
-          @aggregate_id = match[2]
-        end
+        message =~ /duplicate key value violates unique constraint \"aggregate_unique_keys_scope_key_key\"/
       end
     end
 

lib/sequent/core/projectors.rb

@@ -43,7 +43,7 @@ def deactivate_unknown_projectors!(known_projector_classes: Sequent::Core::Migra
             .pluck(:name)
           return if unknown_active_projector_names.empty?
 
-          transaction_provider.transactional do
+          transaction_provider.transaction do
             ProjectorState
               .where.not(name: known_projector_classes.map(&:name))
               .update_all(active_version: nil)
@@ -89,7 +89,7 @@ def projector_states
           cached = Thread.current[PROJECTOR_STATES_KEY]
           return cached unless cached.nil?
 
-          transaction_provider.transactional do
+          transaction_provider.transaction do
             cleanup = -> { Thread.current[PROJECTOR_STATES_KEY] = nil }
             transaction_provider.after_commit(&cleanup)
             transaction_provider.after_rollback(&cleanup)
@@ -116,7 +116,7 @@ def transaction_provider
         def update_projector_state(rows)
           return if rows.empty?
 
-          transaction_provider.transactional do
+          transaction_provider.transaction do
             lock_projector_states_for_update
 
             ProjectorState.upsert_all(rows)

lib/sequent/core/transactions/active_record_transaction_provider.rb

@@ -3,36 +3,51 @@
 module Sequent
   module Core
     module Transactions
-      ##
-      # Always require a new transaction.
-      #
-      # Read:
-      # http://api.rubyonrails.org/classes/ActiveRecord/Transactions/ClassMethods.html
-      #
-      # Without this change, there is a potential bug:
-      #
-      # ```ruby
-      # ActiveRecord::Base.transaction do
-      #   Sequent.configuration.command_service.execute_commands command
-      # end
-      #
-      # on Command do
-      #   do.some.things
-      #   fail ActiveRecord::Rollback
-      # end
-      # ```
-      #
-      # In this example, you might be surprised to find that `do.some.things`
-      # does not get rolled back! This is because AR doesn't automatically make
-      # a "savepoint" for us when we call `.transaction` in a nested manner. In
-      # order to enable this behaviour, we have to call `.transaction` like
-      # this: `.transaction(requires_new: true)`.
-      #
       class ActiveRecordTransactionProvider
-        def transactional(&block)
-          ActiveRecord::Base.transaction(requires_new: true, &block)
+        attr_reader :requires_new
+
+        ##
+        # Configure if save points should be used to simulate nested transactions. This is only
+        # useful in combination with the `ActiveRecord::Rollback` exception.
+        #
+        # Read: https://api.rubyonrails.org/classes/ActiveRecord/Rollback.html and
+        # http://api.rubyonrails.org/classes/ActiveRecord/Transactions/ClassMethods.html
+        #
+        # ```ruby
+        # ActiveRecord::Base.transaction do
+        #   Sequent.configuration.command_service.execute_commands command
+        # end
+        #
+        # on Command do
+        #   do.some.things
+        #   fail ActiveRecord::Rollback
+        # end
+        # ```
+        #
+        # In this example, you might be surprised to find that `do.some.things` does not get rolled
+        # back! This only happens with `ActiveRecord::Rollback`, since it is handled by the inner
+        # transaction. All other exceptions are automatically propagated and will cause the parent
+        # transaction to rollback.
+        #
+        # Note that using save points with PostgreSQL adds additional overhead and is rarely useful,
+        # so our advice is to only use `ActiveRecord::Rollback` directly inside of an
+        # `ActiveRecord::Base.transaction(requires_new: true) do ... end` block so it is clear what
+        # the expected behavior is.
+        def initialize(requires_new: false)
+          @requires_new = requires_new
+          warn <<~EOS if @requires_new
+            [DEPRECATED] avoid using `requires_new: true` globally, use explicit `ActiveRecord::Base.transaction(requires_new: true)`
+            blocks with `ActiveRecord::Rollback` instead if nested transactions are needed.
+          EOS
+        end
+
+        def transaction(requires_new: @requires_new, &block)
+          ActiveRecord::Base.transaction(requires_new:, &block)
         end
 
+        # Deprecated alias
+        alias transactional transaction
+
         def after_commit(&block)
           ActiveRecord::Base.current_transaction.after_commit(&block)
         end

lib/sequent/core/transactions/no_transactions.rb

@@ -9,9 +9,15 @@ module Transactions
       # view state will always be recreated anyway.
       #
       class NoTransactions
-        def transactional
-          yield
+        def transaction
+          yield ActiveRecord::Transaction::NULL_TRANSACTION
         end
+
+        # Deprecated alias
+        alias transactional transaction
+
+        def after_commit = yield
+        def after_rollback = nil
       end
     end
   end

lib/sequent/core/transactions/read_only_active_record_transaction_provider.rb

@@ -8,24 +8,21 @@ def initialize(transaction_provider)
           @transaction_provider = transaction_provider
         end
 
-        def transactional(&block)
+        def transaction(&block)
           register_call
-          @transaction_provider.transactional do
+          @transaction_provider.transaction(requires_new: true) do |transaction|
             Sequent::ApplicationRecord.connection.execute('SET TRANSACTION READ ONLY')
-            block.call
-          ensure
-            deregister_call
-            reset_stack_size if stack_size == 0
+            block.call(transaction)
           end
+        ensure
+          deregister_call
+          reset_stack_size if stack_size == 0
         end
 
-        def after_commit(&block)
-          ActiveRecord::Base.current_transaction.after_commit(&block)
-        end
+        # Deprecated
+        alias transactional transaction
 
-        def after_rollback(&block)
-          ActiveRecord::Base.current_transaction.after_rollback(&block)
-        end
+        delegate :after_commit, :after_rollback, to: :@transaction_provider
 
         private