Add read-only mode option during offline migration

lib/sequent/configuration.rb

@@ -65,7 +65,8 @@ class Configuration
                   :primary_database_role,
                   :primary_database_key,
                   :time_precision,
-                  :enable_autoregistration
+                  :enable_autoregistration,
+                  :enable_offline_migration_read_only_mode
 
     attr_reader :migrations_class_name,
                 :versions_table_name
@@ -127,6 +128,7 @@ def initialize
       self.time_precision = DEFAULT_TIME_PRECISION
 
       self.enable_autoregistration = false
+      self.enable_offline_migration_read_only_mode = false
     end
 
     def can_use_multiple_databases?

lib/sequent/core/projector.rb

@@ -35,6 +35,8 @@ def manages_no_tables_from_superclass?
         end
       end
 
+      class ReadOnlyModeEnabled < RuntimeError; end
+
       def self.projectors
         Sequent.configuration.event_handlers.select { |x| x.is_a? Migratable }.map(&:class)
       end
@@ -54,6 +56,32 @@ def self.all
       def managed_tables
         self.class.managed_tables
       end
+
+      def dispatch_message(message, handlers)
+        assert_not_readonly! if Sequent.configuration.enable_offline_migration_read_only_mode
+        super
+      end
+
+      def assert_not_readonly!
+        fail ReadOnlyModeEnabled if targets_migrating_projectors? || targets_migrating_tables?
+      end
+
+      def targets_migrating_projectors?
+        running_offline_migration.presence && running_offline_migration.target_projectors.include?(self.class.name)
+      end
+
+      def targets_migrating_tables?
+        running_offline_migration.presence && running_offline_migration.target_records.intersect?(
+          managed_tables.map(&:name),
+        )
+      end
+
+      def running_offline_migration
+        @running_offline_migration ||= Sequent::Migrations::Versions
+          .later_versions
+          .migrate_offline_running_or_done
+          .latest
+      end
     end
 
     # Projectors listen to events and update the view state as they see fit.

lib/sequent/migrations/functions.rb

@@ -30,6 +30,10 @@ def table_name
         @record_class.table_name
       end
 
+      def record_class_name
+        @record_class.name
+      end
+
       def copy(with_version)
         self.class.create(record_class, with_version)
       end

lib/sequent/migrations/versions.rb

@@ -10,7 +10,9 @@ class Versions < Sequent::ApplicationRecord
 
       def self.migration_sql
         <<~SQL.chomp
-          CREATE TABLE IF NOT EXISTS #{table_name} (version integer NOT NULL, CONSTRAINT version_pk PRIMARY KEY(version));
+          CREATE TABLE IF NOT EXISTS #{table_name} (version integer NOT NULL, target_projectors text[] DEFAULT '{}'::text[], target_records text[] DEFAULT '{}'::text[], CONSTRAINT version_pk PRIMARY KEY(version));
+          ALTER TABLE #{table_name} ADD COLUMN IF NOT EXISTS target_projectors text[] DEFAULT '{}'::text[];
+          ALTER TABLE #{table_name} ADD COLUMN IF NOT EXISTS target_records text[] DEFAULT '{}'::text[];
           ALTER TABLE #{table_name} drop constraint if exists only_one_running;
           ALTER TABLE #{table_name} ADD COLUMN IF NOT EXISTS status INTEGER DEFAULT NULL CONSTRAINT only_one_running CHECK (status in (1,2,3));
           ALTER TABLE #{table_name} ADD COLUMN IF NOT EXISTS xmin_xact_id BIGINT;
@@ -25,6 +27,10 @@ def self.migration_sql
               where(status: [MIGRATE_ONLINE_RUNNING, MIGRATE_ONLINE_FINISHED, MIGRATE_OFFLINE_RUNNING])
             }
 
+      scope :later_versions, -> { where('version > ?', Sequent.new_version) }
+      scope :migrate_offline_running, -> { where(status: MIGRATE_OFFLINE_RUNNING) }
+      scope :migrate_offline_running_or_done, -> { migrate_offline_running.or(done) }
+
       def self.current_version
         done.latest_version || 0
       end
@@ -55,7 +61,7 @@ def self.rollback!(new_version)
         running.where(version: new_version).delete_all
       end
 
-      def self.start_offline!(new_version)
+      def self.start_offline!(new_version, target_projectors: [], target_records: [])
         current_migration = find_by(version: new_version)
         fail MigrationNotStarted if current_migration.blank?
 
@@ -64,7 +70,7 @@ def self.start_offline!(new_version)
           fail MigrationDone if current_migration.status.nil?
           fail ConcurrentMigration if current_migration.status != MIGRATE_ONLINE_FINISHED
 
-          current_migration.update(status: MIGRATE_OFFLINE_RUNNING)
+          current_migration.update(status: MIGRATE_OFFLINE_RUNNING, target_projectors:, target_records:)
         end
       rescue ActiveRecord::LockWaitTimeout
         raise ConcurrentMigration

lib/sequent/migrations/view_schema.rb

@@ -214,6 +214,7 @@ def migrate_online
         rollback_migration
         raise e
       end
+
       ##
       # Last part of a view schema migration
       #
@@ -237,7 +238,13 @@ def migrate_offline
         return if Sequent.new_version == current_version
 
         ensure_version_correct!
-        in_view_schema { Versions.start_offline!(Sequent.new_version) }
+        in_view_schema do
+          Versions.start_offline!(
+            Sequent.new_version,
+            target_projectors: plan.projectors.map(&:name),
+            target_records: plan.alter_tables.map(&:record_class_name),
+          )
+        end
         Sequent.logger.info("Start migrate_offline for version #{Sequent.new_version}")
 
         executor.set_table_names_to_new_version(plan)

spec/lib/sequent/core/projector_spec.rb

@@ -7,8 +7,106 @@
     class TestProjector1 < Sequent::Core::Projector
       self.skip_autoregister = true
     end
+
     expect do
       Sequent.configuration.event_handlers << TestProjector1.new
     end.to raise_error(/A Projector must manage at least one table/)
   end
+
+  context 'given enable_offline_migration_read_only_mode set to true' do
+    class Versions < Sequent::Migrations::Projectors
+      def self.version
+        1
+      end
+    end
+
+    class MigrationTestRecord; end
+
+    class MigrationTestEvent < Sequent::Core::Event; end
+
+    class MigrationTestProjector < Sequent::Core::Projector
+      manages_tables MigrationTestRecord
+
+      on MigrationTestEvent do
+      end
+    end
+
+    before do
+      Sequent.configuration.migrations_class_name = Versions.name
+      Sequent.configuration.enable_offline_migration_read_only_mode = true
+    end
+
+    subject(:handle_message) do
+      MigrationTestProjector.new.handle_message(
+        MigrationTestEvent.new(aggregate_id: Sequent.new_uuid, sequence_number: 1),
+      )
+    end
+
+    context 'and no migration' do
+      it 'succeeds' do
+        expect { subject }.to_not raise_error
+      end
+    end
+
+    context 'and a migration for projector' do
+      before do
+        Sequent::Migrations::Versions.create!(version: 2, status:, target_projectors: [MigrationTestProjector.name])
+      end
+
+      after do
+        Sequent::Migrations::Versions.delete_all
+      end
+
+      context 'online migration' do
+        context 'running' do
+          let(:status) { Sequent::Migrations::Versions::MIGRATE_ONLINE_RUNNING }
+          it 'succeeds' do
+            expect { subject }.to_not raise_error
+          end
+        end
+
+        context 'finished' do
+          let(:status) { Sequent::Migrations::Versions::MIGRATE_ONLINE_FINISHED }
+          it 'succeeds' do
+            expect { subject }.to_not raise_error
+          end
+        end
+      end
+
+      context 'offline migration' do
+        context 'running' do
+          let(:status) { Sequent::Migrations::Versions::MIGRATE_OFFLINE_RUNNING }
+          it 'raises ReadOnlyModeEnabled ' do
+            expect { subject }.to raise_error(Sequent::Core::Projector::ReadOnlyModeEnabled)
+          end
+        end
+
+        context 'finished' do
+          let(:status) { Sequent::Migrations::Versions::DONE }
+          it 'raises ReadOnlyModeEnabled' do
+            expect { subject }.to raise_error(Sequent::Core::Projector::ReadOnlyModeEnabled)
+          end
+        end
+      end
+    end
+
+    context 'and a migration for managed_table' do
+      before do
+        Sequent::Migrations::Versions.create!(version: 2, status:, target_records: [MigrationTestRecord.name])
+      end
+
+      after do
+        Sequent::Migrations::Versions.delete_all
+      end
+
+      context 'offline migration' do
+        context 'running' do
+          let(:status) { Sequent::Migrations::Versions::MIGRATE_OFFLINE_RUNNING }
+          it 'raises ReadOnlyModeEnabled ' do
+            expect { subject }.to raise_error(Sequent::Core::Projector::ReadOnlyModeEnabled)
+          end
+        end
+      end
+    end
+  end
 end