Xử lý transaction trong multithread Ruby on Rails

Nếu bạn đã có kinh nghiệm làm việc với Node.js, bạn có thể nhận thấy việc sử dụng Promise và Promise.all là một giải pháp hiệu quả cho các công việc mất nhiều thời gian. Trong khi đó, nếu bạn đang làm việc với Ruby on Rails và muốn áp dụng tính năng tương tự, gem concurrent-ruby có thể là sự lựa chọn cho bạn. Dưới đây là một giải pháp cho vấn đề khi sử dụng concurrent-ruby và cần duy trì tính toàn vẹn của transaction.

Vấn đề

Sự khác biệt cơ bản giữa Javascript và Ruby là JavaScript là ngôn ngữ chạy đơn luồng (single-threaded), trong khi Ruby có thể hỗ trợ chạy đa luồng (multi-threaded). Mỗi promise trong Promise.all sẽ tương ứng với một microtask và chúng sẽ chạy đồng thời bất đồng bộ trên cùng một thread. Ngược lại, concurrent-ruby cho phép chúng ta chạy nhiều task đồng thời trên các thread riêng biệt.

def delete_user(user)
  user.destroy!
end

def delete_user_setting(user)
  user.user_setting.destroy!
end

user = User.find(69)

promises = [
  Concurrent::Promise.execute { delete_user_setting(user) },
  Concurrent::Promise.execute { delete_user(user) }
]

Concurrent::Promise.zip(*promises).value!

Tuy nhiên, một thách thức xuất hiện khi chúng ta muốn đảm bảo rằng cả delete_user_setting và delete_user đều thành công hoặc cả hai đều thất bại cùng một lúc.

Giải pháp

Đề xuất một giải pháp bằng cách sử dụng một class PromiseTransaction để quản lý các task và đảm bảo toàn vẹn của transaction. Class này sẽ xác định trạng thái của từng task (đang chờ, đã hoàn thành, có lỗi), và khi một task hoàn thành, nó sẽ đợi cho đến khi tất cả các task đã hoàn thành hoặc một task nào đó gặp lỗi.

class PromiseTransaction
  attr_accessor :tasks, :flags, :error 

  def initialize
    @tasks = []
    @flags = {}
    @error = false
  end

  def perform
    promises.each(&:execute)
    Concurrent::Promise.zip(*promises).value!
  end

  def queue_task(&task)
    self.flags[task.object_id] = false
    self.tasks.push(wrap_task_in_transaction(task))
  end

  private

  def promises
    @promises ||= tasks.map { |task| Concurrent::Promise.new { task.call } }
  end

  def wrap_task_in_transaction(task)
    Proc.new do
      ActiveRecord::Base.transaction do
        begin
          return if error
          result = task.call
          self.flags[task.object_id] = true
          pending_until_done!
          result
        rescue
          self.error = true
          raise ActiveRecord::Rollback
        end
      end
    end
  end

  def pending_until_done!
    while pending?
      raise ActiveRecord::Rollback if error
    end
  end

  def pending?
    flags.values.any?(&:blank?)
  end
end

Class PromiseTransaction này giữ track trạng thái của từng task, đảm bảo rằng khi một task gặp lỗi, toàn bộ transaction sẽ được rollback. Để sử dụng, bạn chỉ cần khởi tạo và thêm các task vào đó:

user = User.find(69)
promise_transaction = PromiseTransaction.new

promise_transaction.queue_task { delete_user_setting(user) }
promise_transaction.queue_task { delete_user(user) }
promise_transaction.queue_task do
  sleep 5
  raise Exception 
end

promise_transaction.perform

Kết quả là tất cả các task sẽ được thực hiện đồng thời và transaction sẽ được quản lý một cách an toàn.