<commit_before># frozen_string_literal: true

require 'spec_helper'

describe "An insertion into que_jobs" do
  it "should notify a locker if one is listening" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  4,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        notify_pid =
          Que.execute("SELECT pg_backend_pid()").first[:pg_backend_pid].to_i

        conn.async_exec "LISTEN que_locker_1"

        Que::Job.enqueue
        job = jobs.first

        conn.wait_for_notify do |channel, pid, payload|
          assert_equal 'que_locker_1', channel
          assert_equal notify_pid, pid

          json = JSON.load(payload)
          assert_equal %w(id priority run_at), json.keys.sort
          assert_equal job[:id], json['id']
          assert_equal 100, json['priority']
          assert_in_delta Time.parse(json['run_at']), Time.now, 3
        end

        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end

  it "should not notify a locker if run_at is in the future" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  4,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        conn.async_exec "LISTEN que_locker_1"

        Que::Job.enqueue run_at: Time.now + 60

        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end

  it "should cycle between different lockers weighted by their worker_counts" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  1,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        DB[:que_lockers].insert pid:           2,
                                worker_count:  2,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        notify_pid =
          Que.execute("SELECT pg_backend_pid()").first[:pg_backend_pid].to_i

        conn.async_exec "LISTEN que_locker_1; LISTEN que_locker_2"

        channels = 6.times.map { Que::Job.enqueue; conn.wait_for_notify }
        assert_equal \
          (['que_locker_1'] * 2 + ['que_locker_2'] * 4),
          channels.sort

        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end

  it "should ignore lockers that are marked as not listening" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  4,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     false

        notify_pid =
          Que.execute("SELECT pg_backend_pid()").first[:pg_backend_pid].to_i

        conn.async_exec "LISTEN que_locker_1"

        Que::Job.enqueue
        job = jobs.first
        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end
end<commit_msg>Add a spec that only lockers of the appropriate queue get the relevant notifications.<commit_after># frozen_string_literal: true

require 'spec_helper'

describe "An insertion into que_jobs" do
  it "should notify a locker if one is listening" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  4,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        notify_pid =
          Que.execute("SELECT pg_backend_pid()").first[:pg_backend_pid].to_i

        conn.async_exec "LISTEN que_locker_1"

        Que::Job.enqueue
        job = jobs.first

        conn.wait_for_notify do |channel, pid, payload|
          assert_equal 'que_locker_1', channel
          assert_equal notify_pid, pid

          json = JSON.load(payload)
          assert_equal %w(id priority run_at), json.keys.sort
          assert_equal job[:id], json['id']
          assert_equal 100, json['priority']
          assert_in_delta Time.parse(json['run_at']), Time.now, 3
        end

        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end

  it "should not notify a locker if run_at is in the future" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  4,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        conn.async_exec "LISTEN que_locker_1"

        Que::Job.enqueue run_at: Time.now + 60

        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end

  it "should cycle between different lockers weighted by their worker_counts" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  1,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        DB[:que_lockers].insert pid:           2,
                                worker_count:  2,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        notify_pid =
          Que.execute("SELECT pg_backend_pid()").first[:pg_backend_pid].to_i

        conn.async_exec "LISTEN que_locker_1; LISTEN que_locker_2"

        channels = 6.times.map { Que::Job.enqueue; conn.wait_for_notify }
        assert_equal \
          (['que_locker_1'] * 2 + ['que_locker_2'] * 4),
          channels.sort

        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end

  it "should ignore lockers that are marked as not listening" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  4,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     false

        notify_pid =
          Que.execute("SELECT pg_backend_pid()").first[:pg_backend_pid].to_i

        conn.async_exec "LISTEN que_locker_1"

        Que::Job.enqueue
        job = jobs.first
        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end

  it "should ignore lockers that aren't listening to that queue" do
    DB.synchronize do |conn|
      begin
        DB[:que_lockers].insert pid:           1,
                                worker_count:  1,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['']),
                                listening:     true

        DB[:que_lockers].insert pid:           2,
                                worker_count:  2,
                                ruby_pid:      Process.pid,
                                ruby_hostname: Socket.gethostname,
                                queues:        Sequel.pg_array(['other_queue']),
                                listening:     true

        notify_pid =
          Que.execute("SELECT pg_backend_pid()").first[:pg_backend_pid].to_i

        conn.async_exec "LISTEN que_locker_1; LISTEN que_locker_2"

        channels = 6.times.map { Que::Job.enqueue; conn.wait_for_notify }
        assert_equal \
          (['que_locker_1'] * 6),
          channels

        assert_nil conn.wait_for_notify(0.01)
      ensure
        conn.async_exec "UNLISTEN *"
      end
    end
  end
end
{'fraction_non_alphanumeric': 0.07807008201086378, 'fraction_numerical': 0.00958568537650442, 'mean_word_length': 1.2780203784570596, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4117436', 'n_tokens_mistral': 2695, 'n_tokens_neox': 2566, 'n_words': 561}