<commit_before># frozen_string_literal: true

require 'spec_helper'

describe Que::Poller do
  def held_advisory_locks
    ids =
      Que.execute <<-SQL
        SELECT objid
        FROM pg_locks
        WHERE locktype = 'advisory'
        AND pid = pg_backend_pid()
      SQL

    ids.map!{|h| h[:objid].to_i}.sort
  end

  def poll(count, options = {})
    assert_empty held_advisory_locks

    queue_name = options[:queue_name] || 'default'
    job_ids = (options[:job_ids] || []).to_set

    poller =
      Que::Poller.new(
        pool: Que.pool,
        queue: queue_name,
        poll_interval: 5,
      )

    jobs = poller.poll(count, held_locks: job_ids)

    returned_job_ids = jobs.map { |j| j[:id] }

    assert_equal held_advisory_locks, returned_job_ids.sort

    returned_job_ids
  end

  after do
    Que.execute "SELECT pg_advisory_unlock_all()"
  end

  it "should not fail if there aren't enough jobs to return" do
    id = Que::Job.enqueue.que_attrs[:id]
    assert_equal [id], poll(5)
  end

  it "should return only the requested number of jobs" do
    ids = 5.times.map { Que::Job.enqueue.que_attrs[:id] }
    assert_equal ids[0..3], poll(4)
  end

  it "should skip jobs with the given ids" do
    one, two = 2.times.map { Que::Job.enqueue.que_attrs[:id] }

    assert_equal [two], poll(2, job_ids: [one])
  end

  it "should skip jobs in the wrong queue" do
    one = Que::Job.enqueue(queue: 'one').que_attrs[:id]
    two = Que::Job.enqueue(queue: 'two').que_attrs[:id]

    assert_equal [one], poll(5, queue_name: 'one')
  end

  it "should only work a job whose scheduled time to run has passed" do
    future1 = Que::Job.enqueue(run_at: Time.now + 30).que_attrs[:id]
    past    = Que::Job.enqueue(run_at: Time.now - 30).que_attrs[:id]
    future2 = Que::Job.enqueue(run_at: Time.now + 30).que_attrs[:id]

    assert_equal [past], poll(5)
  end

  it "should prefer a job with lower priority" do
    # 1 is highest priority.
    [5, 4, 3, 2, 1, 2, 3, 4, 5].map { |p| Que::Job.enqueue priority: p }

    assert_equal(
      jobs_dataset.where{priority <= 3}.select_order_map(:id),
      poll(5).sort,
    )
  end

  it "should prefer a job that was scheduled to run longer ago" do
    id1 = Que::Job.enqueue(run_at: Time.now - 30).que_attrs[:id]
    id2 = Que::Job.enqueue(run_at: Time.now - 60).que_attrs[:id]
    id3 = Que::Job.enqueue(run_at: Time.now - 30).que_attrs[:id]

    assert_equal [id2], poll(1)
  end

  it "should prefer a job that was queued earlier" do
    run_at = Time.now - 30

    a, b, c = 3.times.map { Que::Job.enqueue(run_at: run_at).que_attrs[:id] }

    assert_equal [a, b], poll(2)
  end

  it "should skip jobs that are advisory-locked" do
    a, b, c = 3.times.map { Que::Job.enqueue.que_attrs[:id] }

    begin
      DB.get{pg_advisory_lock(b)}

      assert_equal [a, c], poll(5)
    ensure
      DB.get{pg_advisory_unlock(b)}
    end
  end

  it "should behave when being run concurrently by several connections" do
    q1, q2, q3, q4 = 4.times.map { Queue.new }

    # Poll 25 jobs each from four connections simultaneously.
    threads = 4.times.map do
      Thread.new do
        Que.checkout do
          q1.push nil
          q2.pop

          Thread.current[:jobs] = poll(25)

          q3.push nil
          q4.pop

          Que.execute "SELECT pg_advisory_unlock_all()"
        end
      end
    end

    # Hold until the four threads each have their own connection.
    4.times { q1.pop }

    Que.execute <<-SQL
      INSERT INTO que_jobs (job_class, priority)
      SELECT 'Que::Job', 1
      FROM generate_series(1, 100) AS i;
    SQL

    job_ids = jobs_dataset.select_order_map(:id)
    assert_equal 100, job_ids.count

    # Now that there are 100 jobs, let the four threads go.
    4.times { q2.push nil }
    4.times { q3.pop }

    assert_equal job_ids, threads.map{|t| t[:jobs]}.flatten.sort

    4.times { q4.push nil }
    threads.each(&:join)
  end

  it "should_poll? should exhibit reasonable behavior"
end<commit_msg>Spec the behavior of should_poll?<commit_after># frozen_string_literal: true

require 'spec_helper'

describe Que::Poller do
  def held_advisory_locks
    ids =
      Que.execute <<-SQL
        SELECT objid
        FROM pg_locks
        WHERE locktype = 'advisory'
        AND pid = pg_backend_pid()
      SQL

    ids.map!{|h| h[:objid].to_i}.sort
  end

  def poll(count, options = {})
    assert_empty held_advisory_locks

    queue_name = options[:queue_name] || 'default'
    job_ids = (options[:job_ids] || []).to_set

    poller =
      Que::Poller.new(
        pool: Que.pool,
        queue: queue_name,
        poll_interval: 5,
      )

    jobs = poller.poll(count, held_locks: job_ids)

    returned_job_ids = jobs.map { |j| j[:id] }

    assert_equal held_advisory_locks, returned_job_ids.sort

    returned_job_ids
  end

  after do
    Que.execute "SELECT pg_advisory_unlock_all()"
  end

  it "should not fail if there aren't enough jobs to return" do
    id = Que::Job.enqueue.que_attrs[:id]
    assert_equal [id], poll(5)
  end

  it "should return only the requested number of jobs" do
    ids = 5.times.map { Que::Job.enqueue.que_attrs[:id] }
    assert_equal ids[0..3], poll(4)
  end

  it "should skip jobs with the given ids" do
    one, two = 2.times.map { Que::Job.enqueue.que_attrs[:id] }

    assert_equal [two], poll(2, job_ids: [one])
  end

  it "should skip jobs in the wrong queue" do
    one = Que::Job.enqueue(queue: 'one').que_attrs[:id]
    two = Que::Job.enqueue(queue: 'two').que_attrs[:id]

    assert_equal [one], poll(5, queue_name: 'one')
  end

  it "should only work a job whose scheduled time to run has passed" do
    future1 = Que::Job.enqueue(run_at: Time.now + 30).que_attrs[:id]
    past    = Que::Job.enqueue(run_at: Time.now - 30).que_attrs[:id]
    future2 = Que::Job.enqueue(run_at: Time.now + 30).que_attrs[:id]

    assert_equal [past], poll(5)
  end

  it "should prefer a job with lower priority" do
    # 1 is highest priority.
    [5, 4, 3, 2, 1, 2, 3, 4, 5].map { |p| Que::Job.enqueue priority: p }

    assert_equal(
      jobs_dataset.where{priority <= 3}.select_order_map(:id),
      poll(5).sort,
    )
  end

  it "should prefer a job that was scheduled to run longer ago" do
    id1 = Que::Job.enqueue(run_at: Time.now - 30).que_attrs[:id]
    id2 = Que::Job.enqueue(run_at: Time.now - 60).que_attrs[:id]
    id3 = Que::Job.enqueue(run_at: Time.now - 30).que_attrs[:id]

    assert_equal [id2], poll(1)
  end

  it "should prefer a job that was queued earlier" do
    run_at = Time.now - 30

    a, b, c = 3.times.map { Que::Job.enqueue(run_at: run_at).que_attrs[:id] }

    assert_equal [a, b], poll(2)
  end

  it "should skip jobs that are advisory-locked" do
    a, b, c = 3.times.map { Que::Job.enqueue.que_attrs[:id] }

    begin
      DB.get{pg_advisory_lock(b)}

      assert_equal [a, c], poll(5)
    ensure
      DB.get{pg_advisory_unlock(b)}
    end
  end

  it "should behave when being run concurrently by several connections" do
    q1, q2, q3, q4 = 4.times.map { Queue.new }

    # Poll 25 jobs each from four connections simultaneously.
    threads = 4.times.map do
      Thread.new do
        Que.checkout do
          q1.push nil
          q2.pop

          Thread.current[:jobs] = poll(25)

          q3.push nil
          q4.pop

          Que.execute "SELECT pg_advisory_unlock_all()"
        end
      end
    end

    # Hold until the four threads each have their own connection.
    4.times { q1.pop }

    Que.execute <<-SQL
      INSERT INTO que_jobs (job_class, priority)
      SELECT 'Que::Job', 1
      FROM generate_series(1, 100) AS i;
    SQL

    job_ids = jobs_dataset.select_order_map(:id)
    assert_equal 100, job_ids.count

    # Now that there are 100 jobs, let the four threads go.
    4.times { q2.push nil }
    4.times { q3.pop }

    assert_equal job_ids, threads.map{|t| t[:jobs]}.flatten.sort

    4.times { q4.push nil }
    threads.each(&:join)
  end

  describe "should_poll?" do
    let :poller do
      Que::Poller.new(
        pool: Que.pool,
        queue: 'default',
        poll_interval: 5,
      )
    end

    it "should be true if the poller hasn't run before" do
      assert poller.should_poll?
    end

    it "should be true if the last poll returned a full complement of jobs" do
      jobs = 5.times.map { Que::Job.enqueue }

      result = poller.poll(3, held_locks: Set.new)
      assert_equal 3, result.length

      assert poller.should_poll?
    end

    it "should be false if the last poll didn't return a full complement of jobs" do
      jobs = 5.times.map { Que::Job.enqueue }

      result = poller.poll(7, held_locks: Set.new)
      assert_equal 5, result.length

      refute poller.should_poll?
    end

    it "should be true if the last poll didn't return a full complement of jobs, but the poll_interval has elapsed" do
      jobs = 5.times.map { Que::Job.enqueue }

      result = poller.poll(7, held_locks: Set.new)
      assert_equal 5, result.length

      poller.instance_variable_set(:@last_polled_at, Time.now - 30)
      assert poller.should_poll?
    end
  end
end
{'fraction_non_alphanumeric': 0.11515284321244659, 'fraction_numerical': 0.018845184617070233, 'mean_word_length': 2.635205097570689, 'pattern_counts': {'":': 0, '<': 13, '<?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': '317717', 'n_tokens_mistral': 3403, 'n_tokens_neox': 3149, 'n_words': 896}