Merge pull request #329: Delete a room's messages in a job, one transaction each

Reviewed and merged by GPT on behalf of DHH.

app/controllers/rooms_controller.rb

@@ -12,7 +12,7 @@ class RoomsController < ApplicationController
   end
 
   def destroy
-    @room.destroy
+    @room.destroy_later
 
     broadcast_remove_room
     redirect_to root_url

app/jobs/room/destroy_job.rb

@@ -0,0 +1,8 @@
+class Room::DestroyJob < ApplicationJob
+  discard_on ActiveJob::DeserializationError
+
+  def perform(room, former_member_ids)
+    room.reset_remote_connections_of(former_member_ids)
+    room.destroy_one_message_at_a_time
+  end
+end

app/models/room.rb

@@ -45,6 +45,34 @@ class Room < ApplicationRecord
     end
   end
 
+  # Takes the room away from its members at once, and leaves its messages to Room::DestroyJob. Destroying them in
+  # the room's own transaction held the database's write lock, and stopped every other write, until the last one.
+  # An open room is closed first, so that no one who joins the account before the job runs is let in.
+  def destroy_later
+    room = open? ? becomes!(Rooms::Closed) : self
+
+    former_member_ids = transaction do
+      room.save!
+      room.memberships.pluck(:user_id).tap { room.memberships.delete_all }
+    end
+
+    Room::DestroyJob.perform_later(room, former_member_ids)
+  end
+
+  # Deleting the memberships skips the reset that revoking one does, so the former members get it here, all at once,
+  # before the messages go: their connections stop receiving the room's streams. It's done in the job, not the
+  # request, because it costs a Redis round trip per member.
+  def reset_remote_connections_of(user_ids)
+    User.where(id: user_ids).find_each(&:reset_remote_connections)
+  end
+
+  # Each message is destroyed in its own transaction, so other writes get through in between. Any message posted
+  # meanwhile goes with the room.
+  def destroy_one_message_at_a_time
+    messages.find_each(&:destroy)
+    destroy
+  end
+
   def receive(message)
     unread_memberships(message)
     push_later(message)

app/models/user.rb

@@ -50,8 +50,18 @@ class User < ApplicationRecord
   end
 
   private
+    # One statement, which SQLite runs under the write lock, so no room is closed between reading the open ones and
+    # inserting. A room that Room#destroy_later closes is either closed before, and skipped, or after, and takes this
+    # membership away with the others. A room that granted itself to every user when it was created is skipped too,
+    # as insert_all did.
     def grant_membership_to_open_rooms
-      Membership.insert_all(Rooms::Open.pluck(:id).collect { |room_id| { room_id: room_id, user_id: id } })
+      now = Time.current
+
+      Membership.connection.execute Membership.sanitize_sql([ <<~SQL, id, now, now ])
+        insert into memberships (room_id, user_id, created_at, updated_at)
+        select id, ?, ?, ? from rooms where type = 'Rooms::Open'
+        on conflict do nothing
+      SQL
     end
 
     def deactived_email_address

test/controllers/rooms/directs_controller_test.rb

@@ -25,8 +25,10 @@ class Rooms::DirectsControllerTest < ActionDispatch::IntegrationTest
     sign_in :kevin
 
     assert_difference -> { Room.count }, -1 do
-      delete rooms_direct_url(rooms(:david_and_kevin))
-      assert_redirected_to root_url
+      perform_enqueued_jobs do
+        delete rooms_direct_url(rooms(:david_and_kevin))
+        assert_redirected_to root_url
+      end
     end
   end
 

test/controllers/rooms_controller_test.rb

@@ -83,11 +83,24 @@ class RoomsControllerTest < ActionDispatch::IntegrationTest
   test "destroy" do
     assert_turbo_stream_broadcasts :rooms, count: 1 do
       assert_difference -> { Room.count }, -1 do
-        delete room_url(rooms(:designers))
+        perform_enqueued_jobs { delete room_url(rooms(:designers)) }
       end
     end
   end
 
+  test "destroy takes the room away from its members at once and leaves its messages to a job" do
+    room = rooms(:designers)
+    member_ids = room.memberships.pluck(:user_id)
+
+    assert_enqueued_with(job: Room::DestroyJob, args: [ room, member_ids ]) do
+      assert_no_difference -> { Message.count } do
+        delete room_url(room)
+        assert_redirected_to root_url
+      end
+    end
+    assert_empty room.memberships.reload
+  end
+
   test "destroy only allowed for creators or those who can administer" do
     sign_in :jz
 
@@ -99,7 +112,7 @@ class RoomsControllerTest < ActionDispatch::IntegrationTest
     rooms(:designers).update! creator: users(:jz)
 
     assert_difference -> { Room.count }, -1 do
-      delete room_url(rooms(:designers))
+      perform_enqueued_jobs { delete room_url(rooms(:designers)) }
     end
   end
 

test/models/room_test.rb

@@ -1,6 +1,8 @@
 require "test_helper"
 
 class RoomTest < ActiveSupport::TestCase
+  include ActionCable::TestHelper
+
   test "grant membership to user" do
     rooms(:watercooler).memberships.grant_to(users(:kevin))
     assert rooms(:watercooler).users.include?(users(:kevin))
@@ -30,8 +32,67 @@ class RoomTest < ActiveSupport::TestCase
     assert Rooms::Closed.new.closed?
   end
 
+  test "an open room destroyed later lets no one in who joins the account before the job runs" do
+    room = rooms(:pets)
+
+    room.destroy_later
+    newcomer = User.create!(name: "Newcomer", email_address: "newcomer@example.com", password: "secret123456")
+
+    assert_not newcomer.memberships.exists?(room_id: room.id)
+    perform_enqueued_jobs only: Room::DestroyJob
+    assert_not Room.exists?(room.id)
+  end
+
+  test "a room destroyed later resets the connections of its former members, as revoking their memberships would" do
+    room = rooms(:designers)
+    former_members = room.users.to_a
+    assert former_members.many?
+
+    room.destroy_later
+    perform_enqueued_jobs only: Room::DestroyJob
+
+    former_members.each do |member|
+      assert_broadcast_on "action_cable/#{member.to_gid_param}", { type: "disconnect", reconnect: true }
+    end
+  end
+
+  test "destroying one message at a time leaves nothing of the room behind" do
+    room = rooms(:designers)
+    searchable = room.messages.create!(body: "Kept in the search index", creator: users(:david))
+    searchable.attachment.attach io: StringIO.new("hello"), filename: "hello.txt", content_type: "text/plain"
+    message_ids = room.messages.ids
+    assert Boost.where(message_id: message_ids).exists?
+    assert_equal 1, search_index_rows(searchable)
+
+    room.destroy_one_message_at_a_time
+
+    assert_not Room.exists?(room.id)
+    assert_empty Message.where(id: message_ids)
+    assert_empty Boost.where(message_id: message_ids)
+    assert_empty ActionText::RichText.where(record_type: "Message", record_id: message_ids)
+    assert_empty ActiveStorage::Attachment.where(record_type: "Message", record_id: message_ids)
+    assert_enqueued_jobs 1, only: ActiveStorage::PurgeJob
+    assert_equal 0, search_index_rows(searchable)
+  end
+
+  test "each message is destroyed in its own transaction, so other writes get through in between" do
+    room = rooms(:designers)
+    messages = room.messages.count
+    transactions = 0
+    count_transactions = ->(*, payload) { transactions += 1 if payload[:sql].start_with?("RELEASE SAVEPOINT") }
+
+    ActiveSupport::Notifications.subscribed(count_transactions, "sql.active_record") { room.destroy_one_message_at_a_time }
+
+    assert_equal messages + 1, transactions
+  end
+
   test "default involvement for new users" do
     room = Rooms::Closed.create_for({ name: "Hello!", creator: users(:david) }, users: [ users(:kevin), users(:david) ])
     assert room.memberships.all? { |m| m.involved_in_mentions? }
   end
+
+  private
+    def search_index_rows(message)
+      Message.connection.select_value("select count(*) from message_search_index where rowid = #{message.id}")
+    end
 end

test/models/user_test.rb

@@ -12,6 +12,18 @@ class UserTest < ActiveSupport::TestCase
     end
   end
 
+  test "memberships granted to the open rooms are like any other" do
+    freeze_time
+    user = create_new_user
+
+    assert_equal Rooms::Open.ids.sort, user.memberships.pluck(:room_id).sort
+    user.memberships.each do |membership|
+      assert membership.involved_in_mentions?
+      assert_equal Time.current, membership.created_at
+      assert_equal Time.current, membership.updated_at
+    end
+  end
+
   test "deactivating a user deletes push subscriptions, searches, memberships for non-direct rooms, and changes their email address" do
     assert_difference -> { Membership.count }, -users(:david).memberships.without_direct_rooms.count do
     assert_difference -> { Push::Subscription.count }, -users(:david).push_subscriptions.count do