Эволюция схемы базы данных: принципы безопасной эволюции БД в Rails

1class OptimizeProductSearch < ActiveRecord::Migration[7.0]
2  def up
3    add_index :products, [:category_id, :price, :created_at],
4              name: 'idx_products_category_price_date'
5  end
6
7  def down
8    remove_index :products, name: 'idx_products_category_price_date'
9  end
10end

1class NormalizeUserLocations < ActiveRecord::Migration[7.0]
2  def up
3    # Нормализуем адреса пользователей
4    User.transaction do
5      User.where.not(address: nil).find_each do |user|
6        location = extract_location_data(user.address)
7        user.update!(
8          city: location[:city],
9          country: location[:country],
10          postal_code: location[:postal_code]
11        )
12      end
13
14      remove_column :users, :address
15    end
16  end
17
18  def down
19    add_column :users, :address, :text
20
21    # Как восстановить составной адрес из нормализованных полей?
22    User.find_each do |user|
23      # Формат/порядок/детали потеряны навсегда
24      reconstructed = "#{user.city}, #{user.country} #{user.postal_code}"
25      user.update!(address: reconstructed)
26    end
27  end
28end

1# Паттерн: Expand–Contract
2
3# Фаза 1: EXPAND — добавляем новые структуры
4class AddNewUserStatusSystem < ActiveRecord::Migration[7.0]
5  def change
6    create_table :user_statuses do |t|
7      t.references :user, foreign_key: true, null: false
8      t.integer :status_type, null: false, default: 0
9      t.json :metadata, default: {}
10      t.timestamp :effective_from, null: false, default: -> { 'CURRENT_TIMESTAMP' }
11      t.timestamps
12    end
13
14    add_index :user_statuses, [:user_id, :effective_from]
15    add_index :user_statuses, :status_type
16  end
17end
18
19# Фаза 2: MIGRATE — переносим данные и обновляем код
20class MigrateToNewStatusSystem < ActiveRecord::Migration[7.0]
21  def up
22    User.includes(:user_statuses).find_each do |user|
23      next if user.user_statuses.any?
24
25      legacy_status = case user.status
26                      when 'active' then 0
27                      when 'inactive' then 1
28                      when 'suspended' then 2
29                      else 0
30                      end
31
32      user.user_statuses.create!(
33        status_type: legacy_status,
34        metadata: { migrated_from: 'legacy_status' },
35        effective_from: user.created_at
36      )
37    end
38  end
39
40  def down
41    raise ActiveRecord::IrreversibleMigration, 'Use application rollback'
42  end
43end
44
45# Фаза 3: CONTRACT — удаляем легаси‑поле (следующий релиз)
46class RemoveLegacyStatusField < ActiveRecord::Migration[7.0]
47  def up
48    safety_assured { remove_column :users, :status }
49  end
50
51  def down
52    raise ActiveRecord::IrreversibleMigration, 'Legacy field permanently removed'
53  end
54end

1class User < ApplicationRecord
2  has_many :user_statuses, -> { order(:effective_from) }
3
4  # Период совместимости — поддерживаем оба интерфейса
5  def status
6    return read_attribute(:status) if has_attribute?(:status)
7
8    current_user_status&.status_type_name || 'inactive'
9  end
10
11  def status=(new_status)
12    if has_attribute?(:status)
13      # Легаси‑поле существует — пишем в него тоже (dual‑write)
14      write_attribute(:status, new_status)
15    end
16
17    # Всегда поддерживаем новую систему
18    user_statuses.create!(
19      status_type: UserStatus.status_types[new_status] || 0,
20      effective_from: Time.current
21    )
22  end
23
24  private
25
26  def current_user_status
27    user_statuses.where('effective_from <= ?', Time.current).last
28  end
29end

1class AddColumnWithoutDowntime < ActiveRecord::Migration[7.0]
2  disable_ddl_transaction!
3
4  def up
5    # Добавляем колонку без дефолта (быстро)
6    add_column :large_table, :new_field, :string
7
8    # Бэкфилл в фоновых батчах
9    queue_background_migration('FillNewFieldJob')
10
11    # NOT NULL — позже, отдельной миграцией
12    # change_column_null :large_table, :new_field, false
13  end
14end
15
16# Фоновая задача для безопасного бэкфилла
17class FillNewFieldJob < ApplicationJob
18  def perform(start_id = nil, batch_size = 1000)
19    scope = LargeTable.where(new_field: nil)
20    scope = scope.where('id >= ?', start_id) if start_id
21
22    batch = scope.order(:id).limit(batch_size)
23    return if batch.empty?
24
25    batch.find_each do |record|
26      record.update_column(:new_field, calculate_new_field_value(record))
27    end
28
29    # Планируем следующий батч
30    last_id = batch.maximum(:id)
31    self.class.perform_later(last_id + 1, batch_size) if batch.count == batch_size
32  end
33end

1class SetupDataSynchronization < ActiveRecord::Migration[7.0]
2  def up
3    # Триггер для авто‑синхронизации в период перехода
4    execute <<-SQL
5      CREATE OR REPLACE FUNCTION sync_user_status()
6      RETURNS TRIGGER AS $$
7      BEGIN
8        -- При изменении легаси‑поля обновляем новую таблицу
9        IF TG_OP = 'UPDATE' AND OLD.status IS DISTINCT FROM NEW.status THEN
10          INSERT INTO user_statuses (user_id, status_type, effective_from, created_at, updated_at)
11          VALUES (
12            NEW.id,
13            CASE NEW.status
14              WHEN 'active' THEN 0
15              WHEN 'inactive' THEN 1
16              WHEN 'suspended' THEN 2
17              ELSE 0
18            END,
19            NOW(),
20            NOW(),
21            NOW()
22          );
23        END IF;
24
25        RETURN NEW;
26      END;
27      $$ LANGUAGE plpgsql;
28
29      CREATE TRIGGER user_status_sync_trigger
30        AFTER UPDATE OF status ON users
31        FOR EACH ROW EXECUTE FUNCTION sync_user_status();
32    SQL
33  end
34
35  def down
36    execute 'DROP TRIGGER IF EXISTS user_status_sync_trigger ON users'
37    execute 'DROP FUNCTION IF EXISTS sync_user_status()'
38  end
39end

1# lib/schema_change_tracker.rb
2class SchemaChangeTracker
3  def self.track_migration_performance
4    ActiveSupport::Notifications.subscribe('sql.active_record') do |*args|
5      event = ActiveSupport::Notifications::Event.new(*args)
6
7      if migration_in_progress? && event.duration > 1000 # > 1 сек
8        Rails.logger.warn(
9          'Замечен медленный запрос миграции',
10          {
11            sql: event.payload[:sql],
12            duration: event.duration,
13            migration: current_migration_name
14          }
15        )
16
17        # Отправка в мониторинг
18        StatsD.histogram('migration.query_duration', event.duration)
19      end
20    end
21  end
22
23  def self.estimate_migration_time(migration_class)
24    affected_tables = extract_affected_tables(migration_class)
25
26    total_estimate = affected_tables.sum do |table_name|
27      row_count = connection.select_value("SELECT reltuples FROM pg_class WHERE relname = '#{table_name}'")
28      estimate_operation_time(table_name, row_count)
29    end
30
31    Rails.logger.info("Оценка времени миграции: #{total_estimate} сек")
32    total_estimate
33  end
34end

1# config/initializers/deployment_checks.rb
2class DeploymentHealthCheck
3  def self.verify_schema_compatibility
4    checks = [
5      :verify_backward_compatibility,
6      :check_index_creation_strategy,
7      :validate_migration_reversibility,
8      :estimate_performance_impact
9    ]
10
11    results = checks.map { |check| send(check) }
12
13    if results.any?(&:failure?)
14      Rails.logger.error('Проверка совместимости схемы провалена')
15      raise 'Деплой заблокирован из-за проблем совместимости схемы'
16    end
17  end
18
19  private
20
21  def self.verify_backward_compatibility
22    # Убедитесь, что новые NOT NULL‑колонки имеют дефолты ИЛИ добавляются только nullable‑колонки
23    recent_migrations = get_recent_migrations
24
25    incompatible_changes = recent_migrations.select do |migration|
26      has_breaking_changes?(migration)
27    end
28
29    HealthCheckResult.new(
30      incompatible_changes.empty?,
31      "Обнаружены ломающие изменения: #{incompatible_changes}"
32    )
33  end
34
35  def self.check_index_creation_strategy
36    # Все индексы должны создаваться CONCURRENTLY
37    migrations_with_blocking_indexes = recent_migrations.select do |migration|
38      creates_blocking_index?(migration)
39    end
40
41    HealthCheckResult.new(
42      migrations_with_blocking_indexes.empty?,
43      "Блокирующие операции с индексами: #{migrations_with_blocking_indexes}"
44    )
45  end
46end

1# lib/tasks/schema_evolution.rake
2namespace :db do
3  desc 'Безопасная эволюция схемы базы данных'
4  task evolve: :environment do
5    SchemaEvolutionManager.new.execute_safe_evolution
6  end
7end
8
9class SchemaEvolutionManager
10  def execute_safe_evolution
11    validate_environment
12    estimate_migration_impact
13    execute_migrations_with_monitoring
14    verify_post_migration_health
15  end
16
17  private
18
19  def validate_environment
20    unless Rails.env.production?
21      puts ' Не production — пропускаем жёсткие проверки'
22      return
23    end
24
25    # Убедимся, что нет долгих активных запросов
26    long_queries = connection.select_all(<<-SQL)
27      SELECT pid, query, state, query_start
28      FROM pg_stat_activity
29      WHERE state = 'active'
30        AND query_start < NOW() - INTERVAL '5 minutes'
31        AND query NOT LIKE '%pg_stat_activity%'
32    SQL
33
34    if long_queries.any?
35      raise 'Обнаружены долгие запросы. Дождитесь завершения или разберитесь.'
36    end
37  end
38
39  def estimate_migration_impact
40    pending_migrations = ActiveRecord::Base.connection.migration_context.open.pending_migrations
41
42    total_estimated_time = 0
43    pending_migrations.each do |migration|
44      estimated_time = SchemaChangeTracker.estimate_migration_time(migration)
45      total_estimated_time += estimated_time
46      puts "#{migration.filename}: ~#{estimated_time}s"
47    end
48
49    if total_estimated_time > 300 # 5 минут
50      puts " Суммарная оценка: #{total_estimated_time}s"
51      puts 'Подумайте о выполнении в окно обслуживания'
52
53      unless ENV['FORCE_LONG_MIGRATION'] == 'true'
54        puts 'Установите FORCE_LONG_MIGRATION=true, чтобы продолжить'
55        exit 1
56      end
57    end
58  end
59end

1class MigrationCircuitBreaker
2  def self.with_circuit_breaker(operation_name)
3    circuit = CircuitBreaker.new(operation_name, {
4      failure_threshold: 3,
5      timeout: 300,      # 5 минут
6      recovery_time: 600 # 10 минут
7    })
8
9    circuit.call do
10      yield
11    end
12  rescue CircuitBreaker::OpenCircuitError
13    Rails.logger.error("Открыт предохранитель для #{operation_name}")
14    SlackNotifier.alert("Сработал предохранитель миграций: #{operation_name}")
15    raise 'Миграция прервана из‑за повторных ошибок'
16  end
17end
18
19# Использование в миграциях
20class SafeLargeDataMigration < ActiveRecord::Migration[7.0]
21  def up
22    MigrationCircuitBreaker.with_circuit_breaker('large_data_migration') do
23      migrate_user_preferences_in_batches
24    end
25  end
26end

1# lib/tasks/schema_testing.rake
2namespace :db do
3  namespace :test do
4    desc 'Прогон миграций на продоподобном датасете'
5    task production_scale: :environment do
6      unless Rails.env.staging?
7        puts 'Эта задача должна выполняться только в стейджинге'
8        exit 1
9      end
10
11      # Сгенерируем продоподобный датасет
12      DatasetGenerator.create_production_scale_data
13
14      # Тестируем ожидающие миграции
15      migration_tester = MigrationTester.new
16      results = migration_tester.test_pending_migrations
17
18      # Генерируем отчёт
19      generate_migration_report(results)
20    end
21  end
22end
23
24class MigrationTester
25  def test_pending_migrations
26    pending_migrations = get_pending_migrations
27
28    results = pending_migrations.map do |migration|
29      test_result = test_single_migration(migration)
30
31      {
32        migration: migration.filename,
33        duration: test_result[:duration],
34        memory_usage: test_result[:memory_usage],
35        blocking_queries: test_result[:blocking_queries],
36        success: test_result[:success]
37      }
38    end
39
40    results
41  end
42
43  private
44
45  def test_single_migration(migration)
46    start_time = Time.current
47    start_memory = get_memory_usage
48
49    # Мониторим блокирующие запросы во время миграции
50    blocking_monitor = start_blocking_query_monitor
51
52    begin
53      # Запускаем миграцию в транзакции, чтобы откатить после
54      ActiveRecord::Base.transaction(requires_new: true) do
55        migration.migrate(:up)
56        raise ActiveRecord::Rollback # Откатываем для повторяемости
57      end
58      success = true
59    rescue => e
60      Rails.logger.error("Тест миграции упал: #{e.message}")
61      success = false
62    ensure
63      blocking_monitor.stop
64    end
65
66    {
67      duration: Time.current - start_time,
68      memory_usage: get_memory_usage - start_memory,
69      blocking_queries: blocking_monitor.detected_blocks,
70      success: success
71    }
72  end
73end

1class MigrationSafetyNet
2  def self.create_recovery_point(migration_name)
3    return unless Rails.env.production?
4
5    # Логические бэкапы критичных таблиц
6    affected_tables = extract_affected_tables(migration_name)
7    backup_name = "pre_migration_#{migration_name}_#{Time.current.to_i}"
8
9    affected_tables.each do |table|
10      create_table_snapshot(table, backup_name)
11    end
12
13    Rails.logger.info("Создана точка восстановления: #{backup_name}")
14    backup_name
15  end
16
17  def self.validate_recovery_point(backup_name)
18    # Проверка целостности бэкапа
19    backup_tables = connection.tables.select { |t| t.start_with?("backup_#{backup_name}") }
20
21    backup_tables.all? do |backup_table|
22      original_table = backup_table.gsub("backup_#{backup_name}_", '')
23      validate_backup_integrity(original_table, backup_table)
24    end
25  end
26
27  private
28
29  def self.create_table_snapshot(table_name, backup_name)
30    snapshot_name = "backup_#{backup_name}_#{table_name}"
31
32    connection.execute(<<-SQL)
33      CREATE TABLE #{snapshot_name} AS
34      SELECT * FROM #{table_name}
35    SQL
36
37    # Метаданные бэкапа
38    BackupMetadata.create!(
39      backup_name: backup_name,
40      table_name: table_name,
41      snapshot_name: snapshot_name,
42      row_count: connection.select_value("SELECT COUNT(*) FROM #{table_name}"),
43      created_at: Time.current
44    )
45  end
46end