diff --git a/CHANGELOG.md b/CHANGELOG.md index a9e2ccd..3291bc6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Fixed + +- Job class search and the queue filter now match any part of the name, ignoring case (previously class search matched only from the start, case-sensitively, and the queue filter needed the exact name). + ## [1.2.0] - 2026-09-04 ### Added diff --git a/app/controllers/flightdeck/jobs/actions_controller.rb b/app/controllers/flightdeck/jobs/actions_controller.rb index b1896ac..032ca64 100644 --- a/app/controllers/flightdeck/jobs/actions_controller.rb +++ b/app/controllers/flightdeck/jobs/actions_controller.rb @@ -14,12 +14,16 @@ class ActionsController < Flightdeck::ApplicationController include Toasts def create - if params[:id].present? - act_on_single - elsif params[:scope] == "all" - act_on_all_matching - else - act_on_selected + # Fresh names, or "all matching" could skip a job with a brand-new class + # or queue. + KnownNames.fresh do + if params[:id].present? + act_on_single + elsif params[:scope] == "all" + act_on_all_matching + else + act_on_selected + end end respond_to do |format| diff --git a/app/models/flightdeck/jobs_query.rb b/app/models/flightdeck/jobs_query.rb index 712193c..eb928c1 100644 --- a/app/models/flightdeck/jobs_query.rb +++ b/app/models/flightdeck/jobs_query.rb @@ -204,8 +204,11 @@ def apply_filters(relation) relation = relation.joins(:job) if join_jobs? relation = relation.where(jobs_table[:class_name].eq(class_name)) if class_name - relation = relation.where(queue_table[:queue_name].eq(queue_name)) if queue_name - relation = relation.where(jobs_table[:class_name].matches("#{escape_like(q)}%", nil, true)) if q + # Fragments resolve to exact names (see KnownNames): an indexed IN list. + if queue_name + relation = relation.where(queue_table[:queue_name].in(KnownNames.matching_queues(queue_name))) + end + relation = relation.where(jobs_table[:class_name].in(KnownNames.matching_classes(q))) if q relation end @@ -225,9 +228,5 @@ def jobs_table = SolidQueue::Job.arel_table def queue_table own_queue_column? ? model.arel_table : jobs_table end - - def escape_like(value) - ActiveRecord::Base.sanitize_sql_like(value.to_s) - end end end diff --git a/app/models/flightdeck/known_names.rb b/app/models/flightdeck/known_names.rb new file mode 100644 index 0000000..e518f71 --- /dev/null +++ b/app/models/flightdeck/known_names.rb @@ -0,0 +1,119 @@ +# frozen_string_literal: true + +module Flightdeck + # The distinct queue and job class names Solid Queue has seen, used to turn a + # search fragment into an exact list of names. + # + # `LIKE '%frag%'` can't use an index, so every list, tab count and bulk + # action would scan the jobs table. Matching here, in Ruby, leaves SQL an + # indexed `IN (...)` and folds case the same on every adapter. + # + # Memoized per process rather than in Rails.cache, so a page's list and seven + # tab counts cost a couple of DISTINCTs per poll interval whatever the host's + # cache store. Freshness: + # * entries live one poll interval; + # * a fragment matching nothing re-reads first (at most every MISS_REFRESH); + # * `fresh { }` ignores entries read before it began. + module KnownNames + MISS_REFRESH = 1.0 # seconds + FRESH_KEY = :flightdeck_known_names_fresh_after + + # `fresh` compares `sequence` (equal timestamps can't fool it), the TTL runs + # from `started_at`, and MISS_REFRESH from `finished_at`, so a slow read + # isn't due again the moment it lands. + Entry = Struct.new(:names, :sequence, :started_at, :finished_at) + + COLUMNS = %i[queue_name class_name].freeze + + @memo = {} + @sequence = 0 + @lock = Mutex.new + # Held across a read, so concurrent threads share one DISTINCT. + @read_locks = COLUMNS.index_with { Mutex.new } + + class << self + def matching_queues(fragment) = matching(:queue_name, fragment) + def matching_classes(fragment) = matching(:class_name, fragment) + + def fresh + previous = Thread.current[FRESH_KEY] + Thread.current[FRESH_KEY] ||= @lock.synchronize { @sequence } + yield + ensure + Thread.current[FRESH_KEY] = previous + end + + def reset! + @lock.synchronize { @memo.clear } + end + + # Public so tests can pin it. + def now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + + private + # Case-insensitive substring match. + def matching(column, fragment) + needle = fragment.to_s.downcase + entry = current(column) + found = select(entry.names, needle) + return found if found.any? || cooling_down?(entry) + + # The name may be newer than the entry. + select(refresh(column) { |latest| cooling_down?(latest) }.names, needle) + end + + def select(names, needle) + names.select { |name| name.downcase.include?(needle) } + end + + def current(column) + entry = stored(column) + valid?(entry) ? entry : refresh(column) { |latest| valid?(latest) } + end + + def valid?(entry) + return false unless entry + + fresh_after = Thread.current[FRESH_KEY] + return entry.sequence > fresh_after if fresh_after + + now - entry.started_at < ttl + end + + def cooling_down?(entry) = now - entry.finished_at < MISS_REFRESH + + def stored(column) = @lock.synchronize { @memo[column] } + + # A thread that waited takes the entry the one ahead of it read, if + # `good_enough`. Connection before lock: the reverse order deadlocks + # once the pool is exhausted. + def refresh(column, &good_enough) + SolidQueue::Job.connection_pool.with_connection do + @read_locks.fetch(column).synchronize do + latest = stored(column) + return latest if latest && good_enough.call(latest) + + read(column) + end + end + end + + def read(column) + sequence = @lock.synchronize { @sequence += 1 } + started_at = now + # Uncached, or a stale cached result would pass as a fresh read. + names = SolidQueue::Job.uncached { SolidQueue::Job.distinct.pluck(column) }.compact.sort + entry = Entry.new(names, sequence, started_at, now) + + @lock.synchronize do + # Never let an older read replace a newer one. + existing = @memo[column] + @memo[column] = entry if existing.nil? || existing.sequence < entry.sequence + end + entry + end + + def ttl = [ Flightdeck.config.poll_interval.to_f, 1.0 ].max + end + end +end diff --git a/test/integration/job_actions_test.rb b/test/integration/job_actions_test.rb index c129bb3..9c9c6a0 100644 --- a/test/integration/job_actions_test.rb +++ b/test/integration/job_actions_test.rb @@ -120,6 +120,35 @@ class Flightdeck::JobActionsTest < FlightdeckIntegrationTest assert in_table?(SolidQueue::FailedExecution, other), "a job outside the filter must not be touched" end + # Pinned clock: a slow run mustn't outlast the TTL and pass without `fresh`. + test "applying to all matching sees job classes newer than the search memo" do + create_failed_job(class_name: "Screenshots::OldJob") + + with_known_names_clock do + get_fd "/flightdeck/jobs?state=failed&q=screenshots" # warms the names memo + newcomer = create_failed_job(class_name: "Screenshots::NewJob") + + post_fd "/flightdeck/jobs/retry", params: { scope: "all", state: "failed", q: "screenshots" } + + assert in_table?(SolidQueue::ReadyExecution, newcomer), "a job whose class is newer than the memo was skipped" + assert_equal 0, SolidQueue::FailedExecution.count + end + end + + test "applying to all matching sees queues newer than the search memo" do + create_failed_job(queue_name: "within_5_minutes") + + with_known_names_clock do + get_fd "/flightdeck/jobs?state=failed&queue_name=within" # warms the names memo + newcomer = create_failed_job(queue_name: "within_1_minute") + + post_fd "/flightdeck/jobs/discard", params: { scope: "all", state: "failed", queue_name: "within" } + + refute SolidQueue::Job.exists?(newcomer.id), "a job whose queue is newer than the memo was skipped" + assert_equal 0, SolidQueue::FailedExecution.count + end + end + test "applying to all matching stops at the configured limit and reports the remainder" do Flightdeck.config.bulk_action_limit = 2 5.times { create_failed_job } diff --git a/test/integration/jobs_test.rb b/test/integration/jobs_test.rb index 428d5a1..b3eb779 100644 --- a/test/integration/jobs_test.rb +++ b/test/integration/jobs_test.rb @@ -56,11 +56,11 @@ class Flightdeck::JobsTest < FlightdeckIntegrationTest assert_select ".fd-chip", text: /clear filters/ end - test "search narrows the list by job class prefix" do + test "search narrows the list by a fragment of the job class" do match = create_ready_job(class_name: "Billing::ChargeJob") create_ready_job(class_name: "WebhookJob") - get_fd "/flightdeck/jobs?state=ready&q=Billing" + get_fd "/flightdeck/jobs?state=ready&q=charge" assert_equal [ match.id ], job_ids_in_response end diff --git a/test/jobs_query_test.rb b/test/jobs_query_test.rb index 1638215..86e045e 100644 --- a/test/jobs_query_test.rb +++ b/test/jobs_query_test.rb @@ -52,19 +52,65 @@ class Flightdeck::JobsQueryTest < ActiveSupport::TestCase assert_equal 1, query.count end - test "search is a prefix match on the job class" do + test "search is a case-insensitive substring match on the job class" do match = create_ready_job(class_name: "Billing::ChargeJob") create_ready_job(class_name: "WebhookJob") assert_equal [ match.id ], Flightdeck::JobsQuery.new(state: :ready, q: "Billing").rows.map(&:id) - assert_empty Flightdeck::JobsQuery.new(state: :ready, q: "ChargeJob").rows + assert_equal [ match.id ], Flightdeck::JobsQuery.new(state: :ready, q: "chargejob").rows.map(&:id) + assert_empty Flightdeck::JobsQuery.new(state: :ready, q: "Refund").rows end - test "search escapes LIKE wildcards instead of treating them as a pattern" do - create_ready_job(class_name: "AlphaJob") + test "queue filter is a case-insensitive substring match" do + a = create_failed_job(queue_name: "within_5_minutes") + b = create_ready_job(queue_name: "within_1_minute") + create_ready_job(queue_name: "default") + + assert_equal [ a.id ], Flightdeck::JobsQuery.new(state: :failed, queue_name: "5_MIN").rows.map(&:id) + assert_equal [ b.id ], Flightdeck::JobsQuery.new(state: :ready, queue_name: "within").rows.map(&:id) + end + + test "search treats wildcard characters literally" do + create_ready_job(class_name: "AlphaJob", queue_name: "low") + create_ready_job(class_name: "Under_scoreJob", queue_name: "a_b") assert_empty Flightdeck::JobsQuery.new(state: :ready, q: "%").rows assert_empty Flightdeck::JobsQuery.new(state: :ready, q: "_lphaJob").rows + assert_equal 1, Flightdeck::JobsQuery.new(state: :ready, q: "r_s").count + assert_empty Flightdeck::JobsQuery.new(state: :ready, queue_name: "l_w").rows + assert_empty Flightdeck::JobsQuery.new(state: :ready, queue_name: "\\").rows + end + + test "search folds case beyond ASCII" do + job = create_ready_job(class_name: "ÉmailJob", queue_name: "Ünicode") + + assert_equal [ job.id ], Flightdeck::JobsQuery.new(state: :ready, q: "émail").rows.map(&:id) + assert_equal [ job.id ], Flightdeck::JobsQuery.new(state: :ready, queue_name: "ünI").rows.map(&:id) + end + + test "fragments resolve to an exact IN list, never a LIKE scan" do + create_ready_job(class_name: "Billing::ChargeJob", queue_name: "within_5_minutes") + create_ready_job(class_name: "WebhookJob", queue_name: "default") + + sql = capture_sql do + query = Flightdeck::JobsQuery.new(state: :ready, q: "charge", queue_name: "5_min") + query.rows + query.count + end + + filtered = sql.grep(/solid_queue_ready_executions/) + assert filtered.any? + filtered.each { |statement| refute_match(/\bLIKE\b/i, statement) } + assert filtered.any? { |statement| statement.include?("Billing::ChargeJob") } + end + + test "a fragment matching no known name returns nothing in every state" do + create_failed_job(class_name: "AlphaJob", queue_name: "critical") + + Flightdeck::JobsQuery::STATES.each do |state| + assert_equal 0, Flightdeck::JobsQuery.new(state: state, q: "Nope").count, state + assert_equal 0, Flightdeck::JobsQuery.new(state: state, queue_name: "nope").count, state + end end test "keyset pagination walks pages without overlap or gaps" do diff --git a/test/known_names_test.rb b/test/known_names_test.rb new file mode 100644 index 0000000..c8cc075 --- /dev/null +++ b/test/known_names_test.rb @@ -0,0 +1,173 @@ +# frozen_string_literal: true + +require "test_helper" + +class KnownNamesTest < ActiveSupport::TestCase + test "a filtered page reads each column's names once, even without a cache store" do + create_failed_job(class_name: "Billing::ChargeJob", queue_name: "critical") + + reads = with_cache_store(nil) do + count_distinct_reads do + Flightdeck::JobsQuery.new(state: :failed, q: "charge", queue_name: "crit").rows + Flightdeck::JobsQuery.state_counts(q: "charge", queue_name: "crit") + end + end + + assert_equal 2, reads + end + + test "names are re-read once the poll interval passes" do + create_ready_job(class_name: "OldJob") + + with_known_names_clock do |clock| + assert_equal [ "OldJob" ], matching_classes("job") + create_ready_job(class_name: "NewJob") + + clock.advance(0.5) + assert_equal [ "OldJob" ], matching_classes("job"), "still inside the interval" + + clock.advance(Flightdeck.config.poll_interval) + assert_equal %w[NewJob OldJob], matching_classes("job") + end + end + + test "a fragment that matches nothing re-reads before answering no matches" do + create_ready_job(class_name: "OldJob") + + with_known_names_clock do |clock| + matching_classes("old") + create_ready_job(class_name: "DeployedJustNowJob") + + clock.advance(Flightdeck::KnownNames::MISS_REFRESH) + assert_equal [ "DeployedJustNowJob" ], matching_classes("deployed") + end + end + + test "repeated no-match searches re-read at most once per MISS_REFRESH" do + create_ready_job(class_name: "OldJob") + + with_known_names_clock do |clock| + reads = count_distinct_reads do + matching_classes("nope") + clock.advance(Flightdeck::KnownNames::MISS_REFRESH / 4) + 3.times { matching_classes("nope") } + end + + assert_equal 1, reads + end + end + + # The cooldown runs from when a read returned, or a slow read is due again at once. + test "a slow read does not immediately trigger a second read" do + create_ready_job(class_name: "OldJob") + + with_known_names_clock do |clock| + reads = count_distinct_reads(on_read: -> { clock.advance(Flightdeck::KnownNames::MISS_REFRESH * 1.5) }) do + assert_empty matching_classes("nope") + assert_empty matching_classes("nope") + end + + assert_equal 1, reads + end + end + + # The first read is held open until the other threads are parked on the lock. + test "threads that find the names expired together share one read" do + create_ready_job(class_name: "OldJob") + + with_known_names_clock do |clock| + matching_classes("old") + clock.advance(Flightdeck.config.poll_interval + 1) + + waiters = [] + reading = Queue.new + hold_first_read = lambda do + reading << true + wait_until { waiters.size == 7 && waiters.all? { |thread| thread.status == "sleep" } } + end + + results = nil + reads = count_distinct_reads(on_read: hold_first_read) do + first = Thread.new { matching_classes("old") } + wait_until { !reading.empty? || !first.alive? } + 7.times { waiters << Thread.new { matching_classes("old") } } + results = [ first, *waiters ].map { |thread| thread.join(10)&.value } + ensure + # Join after kill, so teardown never reclaims the connection mid-unwind. + threads = [ first, *waiters ].compact + threads.each(&:kill) + threads.each { |thread| thread.join(5) } + end + + assert_equal 1, reads + assert_equal [ [ "OldJob" ] ] * 8, results + end + end + + test "a read inside fresh is never served from the query cache" do + create_ready_job(class_name: "OldJob") + + reads = SolidQueue::Job.cache do + count_distinct_reads do + matching_classes("job") + Flightdeck::KnownNames.fresh { matching_classes("job") } + end + end + + assert_equal 2, reads, "the second DISTINCT must reach the database, not the query cache" + end + + test "fresh ignores names read before the block began" do + create_ready_job(class_name: "OldJob") + matching_classes("job") + create_ready_job(class_name: "NewJob") + + Flightdeck::KnownNames.fresh do + assert_equal %w[NewJob OldJob], matching_classes("job") + end + end + + test "fresh restores the outer state, also when the block raises" do + assert_raises(RuntimeError) { Flightdeck::KnownNames.fresh { raise "boom" } } + assert_nil Thread.current[Flightdeck::KnownNames::FRESH_KEY] + + Flightdeck::KnownNames.fresh do + outer = Thread.current[Flightdeck::KnownNames::FRESH_KEY] + Flightdeck::KnownNames.fresh { assert_equal outer, Thread.current[Flightdeck::KnownNames::FRESH_KEY] } + assert_equal outer, Thread.current[Flightdeck::KnownNames::FRESH_KEY] + end + assert_nil Thread.current[Flightdeck::KnownNames::FRESH_KEY] + end + + private + def matching_classes(fragment) = Flightdeck::KnownNames.matching_classes(fragment) + + # Real clock (KnownNames' may be pinned); gives up rather than hang. + def wait_until(seconds: 5) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + seconds + sleep 0.01 until yield || Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + end + + def count_distinct_reads(on_read: nil) + reads = 0 + lock = Mutex.new + subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") do |*, payload| + next unless payload[:sql].match?(/SELECT DISTINCT/i) && !payload[:cached] + + lock.synchronize { reads += 1 } + on_read&.call + end + yield + reads + ensure + ActiveSupport::Notifications.unsubscribe(subscriber) + end + + def with_cache_store(store) + previous = Rails.cache + Rails.cache = store + yield + ensure + Rails.cache = previous + end +end diff --git a/test/test_helper.rb b/test/test_helper.rb index 1525b39..eb2108f 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -51,6 +51,26 @@ def with_env_basic_auth(username: "pilot", password: "correct-horse-battery") def basic_auth_header(username, password) ActionController::HttpAuthentication::Basic.encode_credentials(username, password) end + + # A hand-moved stand-in for KnownNames' monotonic clock. + class FakeClock + attr_reader :now + + def initialize(now) = @now = now + def advance(seconds) = @now += seconds + end + + # Pins KnownNames' clock (minitest 6 has no `stub`). Not nestable. + def with_known_names_clock + clock = FakeClock.new(Flightdeck::KnownNames.now) + names = Flightdeck::KnownNames.singleton_class + names.alias_method(:__real_now, :now) + names.define_method(:now) { clock.now } + yield clock + ensure + names.alias_method(:now, :__real_now) + names.remove_method(:__real_now) + end end end @@ -64,6 +84,7 @@ def before_setup # Counts and series are cached for a few seconds. Without this, one test's # numbers would still be warm when the next one runs. Rails.cache.clear + Flightdeck::KnownNames.reset! end end