Dead-lock on two parallel "SELECT FOR UPDATE" executions with the same condition.

dead-lock.rb

require "mysql2"
require "pp"

def connect(iso: nil)
  db = Mysql2::Client.new(
    host: "127.0.0.1", 
    username: "root", 
    password: '', 
    database: 'test'
  )
  begin
    db.query("set transaction isolation level #{iso}") if iso
    db.query("start transaction")
    yield db
    db.query("commit")
  rescue
    db.query("rollback")
    raise
  end
end

# setup
connect { |db|
  db.query("drop table user") rescue nil
  db.query(<<-SQL)
    create table user
    (
      `id`    serial  primary key,
      `name`  int not null,
      `group` int not null,
      `fixed` int not null default 0,

      index `index_1` (`group`)
    ) ENGINE=InnoDB
  SQL

  db.query("insert into `user` (`name`, `group`) values (0, 7)")
}

s1_group = 7
s2_group = 7

Thread.report_on_exception = false

s1 = Thread.new {
  connect(iso: "READ COMMITTED") { |db|

    sleep 2

    puts("[s1] insert")
    db.query("insert into user (`name`, `group`) values (1, #{s1_group})")

    id = db.last_id
    puts("[s1] inserted: id=#{id}")

    rows = db.query("select id from user where `group` = #{s1_group} and fixed = 0 and id < #{id} for update").each.to_a

    puts("[s1] start updating: rows=#{rows.size}")
    rows.each { |row|
      db.query("update user set fixed = true where id = #{row['id']}") if row['id'] != id
    }

    puts("[s1] done.")
  }
}

s2 = Thread.new {
  connect(iso: "READ COMMITTED") { |db|

    sleep 1

    puts("[s2] insert")
    db.query("insert into user (`name`, `group`) values (2, #{s2_group})")

    id = db.last_id
    puts("[s2] inserted: id=#{id}")

    sleep 2
    rows = db.query("select id from user where `group` = #{s2_group} and fixed = 0 and id < #{id} for update").each.to_a

    puts("[s2] start updating: rows=#{rows.size}")
    rows.each { |row|
      db.query("update user set fixed = true where id = #{row['id']}") if row['id'] != id
    }

    puts("[s2] done.")
  }
}

s1.join()
s2.join()

connect() { |db|
  rows = db.query(<<-SQL)
    select count(*) as c from user where fixed = 0 group by `group`
  SQL
  if rows.any? { |e| e["c"] > 1 }
    throw "Invariant violated!"
  end
}