From b4f467b76c6f982e2119067988c01c4aa0c83194 Mon Sep 17 00:00:00 2001 From: Ryan Rodriguez Date: Wed, 4 Nov 2026 20:47:47 -0600 Subject: [PATCH] fix(jobs): fuzzy job class search and queue filter Class search was a case-sensitive prefix match and the queue filter an exact match, so "Docu" never found UserDocumentScreenshotJob. Both are now a case-insensitive substring match. Fragments are resolved in Ruby against the distinct names Solid Queue holds (Flightdeck::KnownNames), leaving SQL an indexed IN list instead of a leading-wildcard LIKE that would scan the jobs table for every list, tab count and bulk action. Case folding is then the same on every adapter. The names are memoized per process for one poll interval, re-read before answering "no matches", and read fresh inside retry/discard so "all matching" never skips a job with a brand-new class or queue. --- CHANGELOG.md | 4 + .../flightdeck/jobs/actions_controller.rb | 16 +- app/models/flightdeck/jobs_query.rb | 11 +- app/models/flightdeck/known_names.rb | 119 ++++++++++++ test/integration/job_actions_test.rb | 29 +++ test/integration/jobs_test.rb | 4 +- test/jobs_query_test.rb | 54 +++++- test/known_names_test.rb | 173 ++++++++++++++++++ test/test_helper.rb | 21 +++ 9 files changed, 413 insertions(+), 18 deletions(-) create mode 100644 app/models/flightdeck/known_names.rb create mode 100644 test/known_names_test.rb 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