diff --git a/Appraisals b/Appraisals
index 303c1cc76..6933820c8 100644
--- a/Appraisals
+++ b/Appraisals
@@ -2,6 +2,8 @@
 
 appraise "rails-7-1" do
   gem "railties", "~> 7.1.0"
+  # json 3 rejects the quirks_mode option this Active Support passes.
+  gem "json", "< 3"
 end
 
 appraise "rails-7-2" do
@@ -10,6 +12,8 @@ end
 
 appraise "rails-8-0" do
   gem "railties", "~> 8.0.0"
+  # json 3 rejects the quirks_mode option this Active Support passes.
+  gem "json", "< 3"
 end
 
 appraise "rails-8-1" do
diff --git a/README.md b/README.md
index 4367c6b07..ac408f7a6 100644
--- a/README.md
+++ b/README.md
@@ -401,6 +401,8 @@ The supervisor is in charge of managing these processes, and it responds to the
 
 When receiving a `QUIT` signal, if workers still have jobs in-flight, these will be returned to the queue when the processes are deregistered.
 
+In `async` mode, workers, dispatchers and the scheduler are threads inside the supervisor's process, so they can't be signaled one by one. On `TERM` or `INT`, the supervisor stops them and waits up to `SolidQueue.shutdown_timeout`. If any are still running by then, it deregisters itself, which returns their in-flight jobs to the queue, and exits. On `QUIT`, it exits immediately without deregistering anything, so in-flight jobs remain claimed until another supervisor prunes these processes and marks the jobs as failed with a `SolidQueue::Processes::ProcessPrunedError`.
+
 On Windows, the `QUIT` signal can't be trapped, so the supervisor only responds to `TERM` and `INT` there.
 
 If processes have no chance of cleaning up before exiting (e.g. if someone pulls a cable somewhere), in-flight jobs might remain claimed by the processes executing them. Processes send heartbeats, and the supervisor checks and prunes processes with expired heartbeats. Jobs that were claimed by processes with an expired heartbeat will be marked as failed with a `SolidQueue::Processes::ProcessPrunedError`. You can configure both the frequency of heartbeats and the threshold to consider a process dead. See the section below for this.
@@ -927,6 +929,8 @@ It's possible to run multiple schedulers with the same `recurring_tasks` configu
 
 **Note**: a single recurring schedule is supported, so you can have multiple schedulers using the same schedule, but not multiple schedulers using different configurations.
 
+When a scheduler starts, it removes any static recurring tasks that are no longer in its configuration. If you remove all of them, though, no scheduler starts, so the old tasks stay in the `solid_queue_recurring_tasks` table, and in Mission Control, even though nothing runs them. You can delete them from a console with `SolidQueue::RecurringTask.static.delete_all`.
+
 Finally, it's possible to configure jobs that aren't handled by Solid Queue. That is, you can have a job like this in your app:
 ```ruby
 class MyResqueJob < ApplicationJob
diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb
index 4b39c8c10..8c87cf9f3 100644
--- a/app/models/solid_queue/claimed_execution.rb
+++ b/app/models/solid_queue/claimed_execution.rb
@@ -66,11 +66,11 @@ def fail_all_with(error)
       end
 
       def discard_all_in_batches(*)
-        raise UndiscardableError, "Can't discard jobs in progress"
+        raise SolidQueue::Execution::UndiscardableError, "Can't discard jobs in progress"
       end
 
       def discard_all_from_jobs(*)
-        raise UndiscardableError, "Can't discard jobs in progress"
+        raise SolidQueue::Execution::UndiscardableError, "Can't discard jobs in progress"
       end
     end
 
diff --git a/app/models/solid_queue/execution.rb b/app/models/solid_queue/execution.rb
index e56a6b3e0..938889186 100644
--- a/app/models/solid_queue/execution.rb
+++ b/app/models/solid_queue/execution.rb
@@ -28,13 +28,15 @@ def execution_data_from_jobs(jobs)
       end
 
       def discard_all_in_batches(batch_size: 500)
-        pending = count
+        # Newly unblocked executions must survive later discard batches.
+        last_id = maximum(:id) || 0
+        pending = where(id: ..last_id).count
         discarded = 0
 
         SolidQueue.instrument(:discard_all, batch_size: batch_size, status: type, batches: 0, size: 0) do |payload|
           loop do
             transaction do
-              job_ids = limit(batch_size).order(:job_id).lock.pluck(:job_id)
+              job_ids = where(id: ..last_id).limit(batch_size).order(:job_id).lock.pluck(:job_id)
               discarded = discard_jobs job_ids
 
               where(job_id: job_ids).delete_all
diff --git a/app/models/solid_queue/failed_execution.rb b/app/models/solid_queue/failed_execution.rb
index 352e5224f..1fe2b1a50 100644
--- a/app/models/solid_queue/failed_execution.rb
+++ b/app/models/solid_queue/failed_execution.rb
@@ -13,7 +13,10 @@ class FailedExecution < Execution
     def self.retry_all(jobs)
       SolidQueue.instrument(:retry_all, jobs_size: jobs.size) do |payload|
         transaction do
-          payload[:size] = dispatch_jobs lock_all_from_jobs(jobs)
+          job_ids = lock_all_from_jobs(jobs)
+          Job.where(id: job_ids).find_each(&:reset_execution_counters)
+
+          payload[:size] = dispatch_jobs(job_ids)
         end
       end
     end
@@ -33,11 +36,11 @@ def retry
     end
 
     private
-      JSON_OVERHEAD = 256
+      TRUNCATED_MESSAGE_OMISSION = "... (truncated)"
 
       def expand_error_details_from_exception
         if exception
-          self.error = { exception_class: exception_class_name, message: exception_message, backtrace: exception_backtrace }
+          self.error = fit_into_error_column(exception_class: exception_class_name, message: exception_message, backtrace: exception_backtrace)
         end
       end
 
@@ -50,32 +53,58 @@ def exception_message
       end
 
       def exception_backtrace
-        if (limit = determine_backtrace_size_limit) && exception.backtrace.to_json.bytesize > limit
-          truncate_backtrace(exception.backtrace, limit)
-        else
-          exception.backtrace
+        exception.backtrace
+      end
+
+      # Columns with a size limit, like MySQL's TEXT, reject (or truncate into invalid JSON)
+      # errors that don't fit. Drop backtrace lines first and, if that's not enough, cut the
+      # message, measuring the error as it'll be stored, after JSON escaping.
+      def fit_into_error_column(error)
+        if (limit = error_column_limit) && serialized_error_size(error) > limit
+          error[:backtrace] = truncate_backtrace(error, limit) if error[:backtrace]
+          error[:message] = truncate_message(error, limit) if serialized_error_size(error) > limit
         end
+
+        error
       end
 
-      def determine_backtrace_size_limit
+      def error_column_limit
         column = self.class.connection_pool.with_connection do |connection|
           connection.schema_cache.columns_hash(self.class.table_name)["error"]
         end
 
-        if column && column.limit.present?
-          column.limit - exception_class_name.bytesize - exception_message.bytesize - JSON_OVERHEAD
+        column&.limit
+      end
+
+      def serialized_error_size(error)
+        self.class.type_for_attribute(:error).serialize(error).bytesize
+      end
+
+      def truncate_backtrace(error, limit)
+        # Each kept line adds its own encoded size plus a separating comma
+        available = limit - serialized_error_size(error.merge(backtrace: [])) + 1
+
+        error[:backtrace].take_while do |line|
+          (available -= serialized_error_size(line) + 1) >= 0
         end
       end
 
-      def truncate_backtrace(lines, limit)
-        [].tap do |truncated_backtrace|
-          lines.each do |line|
-            if (truncated_backtrace << line).to_json.bytesize > limit
-              truncated_backtrace.pop
-              break
-            end
+      def truncate_message(error, limit)
+        message = error[:message].to_s
+
+        # Every byte cut from the message shrinks its encoded size by at least one byte,
+        # so cutting the overflow converges in very few iterations, even with escaping
+        while message.present? && (overflow = serialized_error_size(error.merge(message: message)) - limit) > 0
+          truncate_to = message.bytesize - overflow
+
+          message = if truncate_to > TRUNCATED_MESSAGE_OMISSION.bytesize
+            message.truncate_bytes(truncate_to, omission: TRUNCATED_MESSAGE_OMISSION)
+          else
+            ""
           end
         end
+
+        message
       end
   end
 end
diff --git a/app/models/solid_queue/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb
index b59c464ad..4acec12a5 100644
--- a/app/models/solid_queue/job/concurrency_controls.rb
+++ b/app/models/solid_queue/job/concurrency_controls.rb
@@ -14,8 +14,8 @@ module ConcurrencyControls
       end
 
       class_methods do
-        def release_all_concurrency_locks(jobs)
-          Semaphore.signal_all(jobs.select(&:concurrency_limited?))
+        def unblock_next_blocked_jobs(jobs)
+          jobs.find_each(&:unblock_next_blocked_job)
         end
       end
 
@@ -63,7 +63,7 @@ def handle_concurrency_conflict
         end
 
         def block
-          BlockedExecution.create_or_find_by!(job_id: id)
+          BlockedExecution.create_or_find_by!(job: self)
         end
 
         def release_next_blocked_job
diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb
index 75a5d2111..a3860e1fe 100644
--- a/app/models/solid_queue/job/executable.rb
+++ b/app/models/solid_queue/job/executable.rb
@@ -39,8 +39,12 @@ def dispatch_all_at_once(jobs)
             ReadyExecution.create_all_from_jobs jobs
           end
 
+          # Each job acquires its concurrency lock here, and holds it until the
+          # transaction dispatching the whole set ends. Going through them by
+          # concurrency key makes everyone dispatching at the same time acquire
+          # the locks they share in the same order, so they can't deadlock.
           def dispatch_all_one_by_one(jobs)
-            jobs.each(&:dispatch)
+            jobs.sort_by { |job| [ job.concurrency_key, job.id ] }.each(&:dispatch)
           end
 
           def successfully_dispatched(jobs)
@@ -106,7 +110,7 @@ def discard
 
       private
         def ready
-          ReadyExecution.create_or_find_by!(job_id: id)
+          ReadyExecution.create_or_find_by!(job: self)
         end
 
         def execution
diff --git a/app/models/solid_queue/job/schedulable.rb b/app/models/solid_queue/job/schedulable.rb
index 04419aa61..c8bfb0a81 100644
--- a/app/models/solid_queue/job/schedulable.rb
+++ b/app/models/solid_queue/job/schedulable.rb
@@ -38,7 +38,7 @@ def scheduled?
 
       private
         def schedule
-          ScheduledExecution.create_or_find_by!(job_id: id)
+          ScheduledExecution.create_or_find_by!(job: self)
         end
 
         def execution
diff --git a/app/models/solid_queue/queue_selector.rb b/app/models/solid_queue/queue_selector.rb
index d0d61dd14..01a1d6da5 100644
--- a/app/models/solid_queue/queue_selector.rb
+++ b/app/models/solid_queue/queue_selector.rb
@@ -34,7 +34,7 @@ def queue_names
       def eligible_queues
         if include_all_queues? then all_queues
         else
-          in_raw_order(exact_names + prefixed_names)
+          in_raw_order(exact_names | prefixed_names)
         end
       end
 
@@ -53,12 +53,12 @@ def exact_names
       def prefixed_names
         if prefixes.empty? then []
         else
-          relation.where(([ "queue_name LIKE ?" ] * prefixes.count).join(" OR "), *prefixes).distinct_values_of(:queue_name)
+          relation.where(([ "queue_name LIKE ? ESCAPE '!'" ] * prefixes.count).join(" OR "), *prefixes).distinct_values_of(:queue_name)
         end
       end
 
       def prefixes
-        @prefixes ||= raw_queues.select { |queue| prefixed_name?(queue) }.map { |queue| queue.tr("*", "%") }
+        @prefixes ||= raw_queues.select { |queue| prefixed_name?(queue) }.map { |queue| relation.sanitize_sql_like(queue, "!").tr("*", "%") }
       end
 
       def exact_name?(queue)
@@ -74,9 +74,9 @@ def paused_queues
       end
 
       def in_raw_order(queues)
-        # Only need to sort if we have prefixes and more than one queue name.
-        # Exact names are selected in the same order as they're found
-        if queues.one? || prefixes.empty?
+        # Exact names are already in configuration order. Prefix matches always
+        # go through the literal filter so a single false LIKE hit is not returned.
+        if prefixes.empty?
           queues
         else
           queues = queues.dup
@@ -89,9 +89,7 @@ def delete_in_order(raw_queue, queues)
           queues.delete(raw_queue)
         elsif prefixed_name?(raw_queue)
           prefix = raw_queue.tr("*", "")
-          queues.select { |queue| queue.start_with?(prefix) }.tap do |matches|
-            queues -= matches
-          end
+          queues.extract! { |queue| queue.start_with?(prefix) }
         end
       end
   end
diff --git a/app/models/solid_queue/ready_execution.rb b/app/models/solid_queue/ready_execution.rb
index ea14c193a..c0a67ce21 100644
--- a/app/models/solid_queue/ready_execution.rb
+++ b/app/models/solid_queue/ready_execution.rb
@@ -45,7 +45,7 @@ def lock_candidates(executions, process_id)
 
 
         def discard_jobs(job_ids)
-          Job.release_all_concurrency_locks Job.where(id: job_ids)
+          Job.unblock_next_blocked_jobs Job.where(id: job_ids)
           super
         end
     end
diff --git a/app/models/solid_queue/semaphore.rb b/app/models/solid_queue/semaphore.rb
index 7d111ac55..da2ea4c3b 100644
--- a/app/models/solid_queue/semaphore.rb
+++ b/app/models/solid_queue/semaphore.rb
@@ -14,10 +14,6 @@ def signal(job)
         Proxy.new(job).signal
       end
 
-      def signal_all(jobs)
-        Proxy.signal_all(jobs)
-      end
-
       # Requires a unique index on key
       def create_unique_by(attributes)
         if supports_insert_conflict_target?
@@ -31,16 +27,6 @@ def create_unique_by(attributes)
     end
 
     class Proxy
-      def self.signal_all(jobs)
-        # Guard against incrementing a semaphore's value beyond its limit. Jobs can
-        # have different limits, so group them and cap each group with `value < limit`.
-        jobs.group_by { |job| job.concurrency_limit || 1 }.each do |limit, grouped_jobs|
-          Semaphore.where(key: grouped_jobs.map(&:concurrency_key))
-            .where(value: ...limit)
-            .update_all("value = value + 1")
-        end
-      end
-
       def initialize(job)
         @job = job
       end
diff --git a/gemfiles/rails_7_1.gemfile b/gemfiles/rails_7_1.gemfile
index 036d98368..5fd12a49c 100644
--- a/gemfiles/rails_7_1.gemfile
+++ b/gemfiles/rails_7_1.gemfile
@@ -3,5 +3,6 @@
 source "https://rubygems.org"
 
 gem "railties", "~> 7.1.0"
+gem "json", "< 3"
 
 gemspec path: "../"
diff --git a/gemfiles/rails_8_0.gemfile b/gemfiles/rails_8_0.gemfile
index 28bd9f00e..297d3f85c 100644
--- a/gemfiles/rails_8_0.gemfile
+++ b/gemfiles/rails_8_0.gemfile
@@ -3,5 +3,6 @@
 source "https://rubygems.org"
 
 gem "railties", "~> 8.0.0"
+gem "json", "< 3"
 
 gemspec path: "../"
diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb
index fbc2f5e65..14ef5e803 100644
--- a/lib/solid_queue.rb
+++ b/lib/solid_queue.rb
@@ -16,6 +16,8 @@
 loader.ignore("#{__dir__}/solid_queue/tasks.rb")
 loader.ignore("#{__dir__}/generators")
 loader.ignore("#{__dir__}/puma")
+# Only bin/jobs uses the CLI, and requires it. Eager loading it would load Thor in every process.
+loader.do_not_eager_load("#{__dir__}/solid_queue/cli.rb")
 loader.setup
 
 module SolidQueue
diff --git a/lib/solid_queue/async_supervisor.rb b/lib/solid_queue/async_supervisor.rb
index f6ab38342..c22af0144 100644
--- a/lib/solid_queue/async_supervisor.rb
+++ b/lib/solid_queue/async_supervisor.rb
@@ -38,12 +38,26 @@ def perform_graceful_termination
         process_instances.values.each(&:stop)
 
         Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
+
+        # The exit! that follows skips the shutdown callbacks, so release the jobs
+        # of the threads still running now instead of leaving them to be pruned
+        attempt_to_deregister unless all_processes_terminated?
       end
 
       def perform_immediate_termination
         exit!
       end
 
+      def attempt_to_deregister
+        wrap_in_app_executor do
+          # Not our own copy: it shares the threads' registrations in memory, and
+          # those of threads that already deregistered are destroyed and frozen
+          SolidQueue::Process.find_by(id: process_id)&.deregister
+        end
+      rescue StandardError => error
+        handle_thread_error(error)
+      end
+
       def all_processes_terminated?
         process_instances.values.none?(&:alive?)
       end
diff --git a/lib/solid_queue/scheduler/recurring_schedule.rb b/lib/solid_queue/scheduler/recurring_schedule.rb
index 526a8edcd..69bc9e5de 100644
--- a/lib/solid_queue/scheduler/recurring_schedule.rb
+++ b/lib/solid_queue/scheduler/recurring_schedule.rb
@@ -10,6 +10,7 @@ def initialize(static_tasks, dynamic_tasks_enabled: false)
       @static_tasks = Array(static_tasks).map { |task| RecurringTask.wrap(task) }.select(&:valid?)
       @dynamic_tasks_enabled = dynamic_tasks_enabled
 
+      @schedule_lock = Mutex.new
       @scheduled_tasks = Concurrent::Hash.new
     end
 
@@ -28,18 +29,16 @@ def schedule_tasks
         reload_dynamic_tasks
       end
 
-      configured_tasks.each do |task|
-        schedule_task(task)
+      schedule_lock.synchronize do
+        configured_tasks.each { |task| schedule_task(task) }
       end
     end
 
-    def schedule_task(task, run_at: task.next_time)
-      scheduled_tasks[task.key] = schedule(task, run_at: run_at)
-    end
-
     def unschedule_tasks
-      scheduled_tasks.values.each(&:cancel)
-      scheduled_tasks.clear
+      schedule_lock.synchronize do
+        scheduled_tasks.values.each(&:cancel)
+        scheduled_tasks.clear
+      end
     end
 
     def task_keys
@@ -49,13 +48,16 @@ def task_keys
     def reschedule_dynamic_tasks
       wrap_in_app_executor do
         reload_dynamic_tasks
-        schedule_created_dynamic_tasks
-        unschedule_deleted_dynamic_tasks
+
+        schedule_lock.synchronize do
+          schedule_created_dynamic_tasks
+          unschedule_deleted_dynamic_tasks
+        end
       end
     end
 
     private
-      attr_reader :static_tasks
+      attr_reader :static_tasks, :schedule_lock
 
       def static_task_keys
         static_tasks.map(&:key)
@@ -69,19 +71,22 @@ def dynamic_tasks_enabled?
         @dynamic_tasks_enabled
       end
 
+      # Scheduled tasks are tracked by record id, so a dynamic task that's deleted and
+      # created again with the same key is unscheduled and scheduled as a new one.
       def schedule_created_dynamic_tasks
-        RecurringTask.dynamic.where.not(key: scheduled_tasks.keys).each do |task|
-          schedule_task(task)
-        end
+        dynamic_tasks.reject { |task| scheduled_tasks.key?(task.id) }.each { |task| schedule_task(task) }
       end
 
       def unschedule_deleted_dynamic_tasks
-        (scheduled_tasks.keys - RecurringTask.pluck(:key)).each do |key|
-          scheduled_tasks[key].cancel
-          scheduled_tasks.delete(key)
+        (scheduled_tasks.keys - static_tasks.map(&:id) - dynamic_tasks.map(&:id)).each do |id|
+          scheduled_tasks.delete(id).cancel
         end
       end
 
+      def schedule_task(task, run_at: task.next_time)
+        scheduled_tasks[task.id] = schedule(task, run_at: run_at)
+      end
+
       def persist_static_tasks
         RecurringTask.static.where.not(key: static_task_keys).delete_all
         RecurringTask.create_or_update_all static_tasks
@@ -102,8 +107,8 @@ def load_dynamic_tasks
       def schedule(task, run_at: task.next_time)
         delay = [ (run_at - Time.current).to_f, 0.1 ].max
 
-        scheduled_task = Concurrent::ScheduledTask.new(delay, args: [ self, task, run_at ]) do |thread_schedule, thread_task, thread_task_run_at|
-          thread_schedule.schedule_task(thread_task, run_at: thread_task.next_time_after(thread_task_run_at))
+        scheduled_task = Concurrent::ScheduledTask.new(delay, args: [ task, run_at ]) do |thread_task, thread_task_run_at|
+          schedule_next_run(thread_task, run_at: thread_task.next_time_after(thread_task_run_at))
 
           wrap_in_app_executor do
             thread_task.enqueue(at: thread_task_run_at)
@@ -118,5 +123,12 @@ def schedule(task, run_at: task.next_time)
 
         scheduled_task.tap(&:execute)
       end
+
+      # Only schedule the next run if the task wasn't unscheduled in the meantime
+      def schedule_next_run(task, run_at:)
+        schedule_lock.synchronize do
+          schedule_task(task, run_at: run_at) if scheduled_tasks.key?(task.id)
+        end
+      end
   end
 end
diff --git a/test/dummy/app/jobs/large_error_message_job.rb b/test/dummy/app/jobs/large_error_message_job.rb
new file mode 100644
index 000000000..58106d744
--- /dev/null
+++ b/test/dummy/app/jobs/large_error_message_job.rb
@@ -0,0 +1,7 @@
+class LargeErrorMessageJob < ApplicationJob
+  queue_as :background
+
+  def perform(message_size)
+    raise ExpectedTestError, "x" * message_size
+  end
+end
diff --git a/test/integration/async_processes_lifecycle_test.rb b/test/integration/async_processes_lifecycle_test.rb
index dc8e4c7ee..8185835d8 100644
--- a/test/integration/async_processes_lifecycle_test.rb
+++ b/test/integration/async_processes_lifecycle_test.rb
@@ -173,14 +173,11 @@ class AsyncProcessesLifecycleTest < ActiveSupport::TestCase
         "registered processes: #{SolidQueue::Process.all.map { |p| { id: p.id, kind: p.kind, pid: p.pid, last_heartbeat_at: p.last_heartbeat_at } }.inspect}"
     end
 
-    # After shutdown, the pause job may be either:
-    # - claimed (exit! called, no cleanup) OR
-    # - ready (graceful exit, job released back to queue)
-    # Both are valid outcomes depending on the timing race between supervisor and worker timeouts.
-    skip_active_record_query_cache do
-      job = SolidQueue::Job.find_by(active_job_id: pause.job_id)
-      assert job.claimed? || job.ready?, "Expected pause job to be claimed or ready, but was neither"
-    end
+    # Workers were shutdown without a chance to terminate orderly, but
+    # since they're linked to the supervisor, the supervisor deregistering
+    # also deregistered them and released claimed jobs
+    assert_job_status(pause, :ready)
+    assert_clean_termination
   end
 
   test "process some jobs that raise errors" do
diff --git a/test/integration/instrumentation_test.rb b/test/integration/instrumentation_test.rb
index 4fd8a295d..77d9a7270 100644
--- a/test/integration/instrumentation_test.rb
+++ b/test/integration/instrumentation_test.rb
@@ -252,8 +252,8 @@ class InstrumentationTest < ActiveSupport::TestCase
 
     assert_equal 2, events.size
     assert_event events.first, "discard_all", jobs_size: 9, status: :ready, size: 6
-    # Only 3 blocked jobs remaining for the second discard_all_from_jobs call
-    assert_event events.second, "discard_all", jobs_size: 3, status: :ready, size: 0
+    # 1 newly ready job and 2 blocked jobs remain for the second call.
+    assert_event events.second, "discard_all", jobs_size: 3, status: :ready, size: 1
   end
 
   test "discarding jobs in batches emits a discard_all event" do
diff --git a/test/integration/jobs_lifecycle_test.rb b/test/integration/jobs_lifecycle_test.rb
index 4b23de296..ba377bc68 100644
--- a/test/integration/jobs_lifecycle_test.rb
+++ b/test/integration/jobs_lifecycle_test.rb
@@ -87,6 +87,39 @@ class JobsLifecycleTest < ActiveSupport::TestCase
     assert_equal 1, SolidQueue::FailedExecution.count
   end
 
+  test "bulk retries reset the automatic retry attempts of failed jobs" do
+    RaisingJob.perform_later(RaisingJob::DefaultError, "A", 5)
+    RaisingJob.perform_later(RaisingJob::DefaultError, "B", 5)
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_jobs_to_finish_for(3.seconds)
+
+    assert_equal 4, SolidQueue::Job.finished.count
+    assert_equal 2, SolidQueue::FailedExecution.count
+    failed_jobs = SolidQueue::Job.failed.to_a
+    finished_jobs = SolidQueue::Job.finished.to_a
+
+    # Check the reset before workers can claim the retried jobs.
+    SolidQueue::Record.transaction do
+      SolidQueue::FailedExecution.retry_all(SolidQueue::Job.all)
+
+      failed_jobs.each do |job|
+        assert_equal 0, job.reload.arguments["executions"]
+        assert_equal({}, job.arguments["exception_executions"])
+      end
+      finished_jobs.each do |job|
+        assert_equal job.arguments, job.reload.arguments
+      end
+    end
+
+    wait_for_jobs_to_finish_for(3.seconds)
+
+    assert_equal 8, SolidQueue::Job.finished.count
+    assert_equal 2, SolidQueue::FailedExecution.count
+  end
+
   test "enqueue and run jobs that fail and it's discarded" do
     RaisingJob.perform_later(RaisingJob::DiscardableError, "A")
 
diff --git a/test/models/solid_queue/claimed_execution_test.rb b/test/models/solid_queue/claimed_execution_test.rb
index b09fdeb90..dd29b0ceb 100644
--- a/test/models/solid_queue/claimed_execution_test.rb
+++ b/test/models/solid_queue/claimed_execution_test.rb
@@ -187,6 +187,16 @@ class SolidQueue::ClaimedExecutionTest < ActiveSupport::TestCase
     assert_equal "provider_job_id: #{job.provider_job_id}", JobBuffer.last_value
   end
 
+  test "discarding claimed executions in bulk raises UndiscardableError" do
+    assert_raises SolidQueue::Execution::UndiscardableError do
+      SolidQueue::ClaimedExecution.discard_all_in_batches
+    end
+
+    assert_raises SolidQueue::Execution::UndiscardableError do
+      SolidQueue::ClaimedExecution.discard_all_from_jobs([])
+    end
+  end
+
   private
     def prepare_and_claim_job(active_job, process: @process)
       job = SolidQueue::Job.find_by(active_job_id: active_job.job_id)
diff --git a/test/models/solid_queue/failed_execution_test.rb b/test/models/solid_queue/failed_execution_test.rb
index fda90e3ab..e6e3480fb 100644
--- a/test/models/solid_queue/failed_execution_test.rb
+++ b/test/models/solid_queue/failed_execution_test.rb
@@ -23,6 +23,68 @@ class SolidQueue::FailedExecutionTest < ActiveSupport::TestCase
     assert_equal "stack level too deep", SolidQueue::FailedExecution.last.message
   end
 
+  test "run job that fails with an error message larger than the error column" do
+    # 70KB doesn't fit in MySQL's TEXT; other adapters have no limit and store it as is
+    LargeErrorMessageJob.perform_later(70_000)
+    @worker.start
+
+    assert_equal 1, SolidQueue::FailedExecution.count
+    assert SolidQueue::Job.last.failed?
+    assert_fits_error_column SolidQueue::FailedExecution.last
+  end
+
+  test "truncate backtrace to fit the error column" do
+    with_error_column_limit(1_000) do
+      execution = fail_job_with ExpectedTestError.new("boom"), backtrace: Array.new(100) { |i| "app/models/thing.rb:#{i}:in 'do_it'" }
+
+      assert_fits_error_column execution, limit: 1_000
+      assert_equal "boom", execution.message
+      assert_operator execution.backtrace.size, :>, 0
+      assert_operator execution.backtrace.size, :<, 100
+      assert_equal "app/models/thing.rb:0:in 'do_it'", execution.backtrace.first
+    end
+  end
+
+  test "truncate message when it doesn't fit the error column by itself" do
+    with_error_column_limit(1_000) do
+      execution = fail_job_with ExpectedTestError.new("x" * 5_000), backtrace: [ "app/models/thing.rb:1:in 'do_it'" ]
+
+      assert_fits_error_column execution, limit: 1_000
+      assert_equal [], execution.backtrace
+      assert execution.message.start_with?("xxx")
+      assert execution.message.end_with?("... (truncated)")
+    end
+  end
+
+  test "account for JSON escaping when fitting the error column" do
+    with_error_column_limit(1_000) do
+      # Quotes take two bytes once escaped. The error column uses Ruby's JSON, which leaves <, > and & as they are
+      execution = fail_job_with ExpectedTestError.new("\"<>&" * 200), backtrace: [ "app/models/thing.rb:1:in 'do_it'" ]
+
+      assert_fits_error_column execution, limit: 1_000
+    end
+  end
+
+  test "fit errors without a backtrace into the error column" do
+    with_error_column_limit(1_000) do
+      execution = fail_job_with ExpectedTestError.new("x" * 5_000), backtrace: nil
+
+      assert_fits_error_column execution, limit: 1_000
+      assert_nil execution.backtrace
+    end
+  end
+
+  test "keep errors that fit in the error column intact" do
+    backtrace = Array.new(5) { |i| "app/models/thing.rb:#{i}:in 'do_it'" }
+
+    with_error_column_limit(1_000) do
+      execution = fail_job_with ExpectedTestError.new("boom"), backtrace: backtrace
+
+      assert_equal "boom", execution.message
+      assert_equal backtrace, execution.backtrace
+    end
+  end
+
   test "retry failed job" do
     RaisingJob.perform_later(ExpectedTestError, "A")
     @worker.start
@@ -46,4 +108,25 @@ class SolidQueue::FailedExecutionTest < ActiveSupport::TestCase
       end
     end
   end
+
+  private
+    def with_error_column_limit(limit)
+      SolidQueue::FailedExecution.any_instance.stubs(:error_column_limit).returns(limit)
+      yield
+    end
+
+    def fail_job_with(exception, backtrace:)
+      exception.set_backtrace(backtrace) if backtrace
+      job = SolidQueue::Job.find_by!(active_job_id: AddToBufferJob.perform_later(1).job_id)
+
+      job.failed_with(exception)
+      job.reload.failed_execution
+    end
+
+    def assert_fits_error_column(execution, limit: execution.send(:error_column_limit))
+      if limit
+        stored_error = execution.class.connection.select_value(execution.class.where(id: execution.id).select(:error).to_sql)
+        assert_operator stored_error.bytesize, :<=, limit
+      end
+    end
 end
diff --git a/test/models/solid_queue/job_test.rb b/test/models/solid_queue/job_test.rb
index 7e9c1568f..cfd604f63 100644
--- a/test/models/solid_queue/job_test.rb
+++ b/test/models/solid_queue/job_test.rb
@@ -41,6 +41,24 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob
     )
   end
 
+  test "enqueue does not reload the job to create its execution" do
+    job_selects = []
+    callback = ->(*, payload) do
+      # Skip the schema lookups Active Record runs the first time it uses the table
+      job_selects << payload[:sql] if payload[:name] != "SCHEMA" && payload[:sql].match?(/\ASELECT .*solid_queue_jobs/i)
+    end
+
+    ActiveSupport::Notifications.subscribed(callback, "sql.active_record") do
+      AddToBufferJob.perform_later(1)
+      AddToBufferJob.set(wait: 1.minute).perform_later(2)
+      2.times { NonOverlappingJob.perform_later(@result) }
+    end
+
+    assert SolidQueue::BlockedExecution.exists?
+
+    assert_empty job_selects
+  end
+
   test "enqueue active job to be executed right away" do
     active_job = AddToBufferJob.new(1).set(priority: 8, queue: "test")
 
@@ -227,6 +245,21 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob
     assert_not not_enqueued.successfully_enqueued?
   end
 
+  test "enqueue jobs in bulk acquiring their concurrency locks by key" do
+    results = 3.times.map { JobResult.create!(queue_name: "default") }
+
+    acquired = []
+    SolidQueue::Semaphore.stubs(:wait).with { |job| acquired << [ job.concurrency_key, job.id ] }.returns(true)
+
+    # Two jobs for each key, with the keys in descending order
+    active_jobs = (results.reverse + results.reverse).map { |result| NonOverlappingJob.new(result) }
+    ActiveJob.perform_all_later(active_jobs)
+
+    # By key and, within a key, in the order they were enqueued
+    assert_equal 6, acquired.size
+    assert_equal acquired.sort, acquired
+  end
+
   test "discard ready job" do
     AddToBufferJob.perform_later(1)
     job = SolidQueue::Job.last
@@ -315,6 +348,10 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob
       ActiveJob.perform_all_later(active_jobs)
     end
 
+    assert_job_counts(blocked: -2) do
+      SolidQueue::BlockedExecution.discard_all_from_jobs(SolidQueue::Job.all)
+    end
+
     assert_job_counts(ready: -5) do
       SolidQueue::ReadyExecution.discard_all_from_jobs(SolidQueue::Job.all)
     end
@@ -322,10 +359,6 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob
     assert_job_counts(scheduled: -2) do
       SolidQueue::ScheduledExecution.discard_all_from_jobs(SolidQueue::Job.all)
     end
-
-    assert_job_counts(blocked: -2) do
-      SolidQueue::BlockedExecution.discard_all_from_jobs(SolidQueue::Job.all)
-    end
   end
 
   test "raise EnqueueError when there's an ActiveRecordError" do
diff --git a/test/models/solid_queue/ready_execution_test.rb b/test/models/solid_queue/ready_execution_test.rb
index a5f40659a..ae988b1f2 100644
--- a/test/models/solid_queue/ready_execution_test.rb
+++ b/test/models/solid_queue/ready_execution_test.rb
@@ -67,6 +67,34 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     assert @jobs.none?(&:claimed?)
   end
 
+  test "prefixes escape LIKE wildcards so only literal queue names match" do
+    AddToBufferJob.set(queue: "mailbox").perform_later(1)
+    AddToBufferJob.set(queue: "mail-ops").perform_later(1)
+
+    assert_claimed_jobs(0) do
+      SolidQueue::ReadyExecution.claim("mail_*", SolidQueue::Job.count + 1, 42)
+    end
+
+    AddToBufferJob.set(queue: "mail_ops").perform_later(1)
+    AddToBufferJob.set(queue: "100x_done").perform_later(1)
+
+    assert_claimed_jobs(1) do
+      SolidQueue::ReadyExecution.claim("mail_*", SolidQueue::Job.count + 1, 42)
+    end
+    assert_equal "mail_ops", SolidQueue::ClaimedExecution.last.job.queue_name
+
+    assert_claimed_jobs(0) do
+      SolidQueue::ReadyExecution.claim("100%_*", SolidQueue::Job.count + 1, 42)
+    end
+
+    AddToBufferJob.set(queue: "100%_done").perform_later(1)
+
+    assert_claimed_jobs(1) do
+      SolidQueue::ReadyExecution.claim("100%_*", 1, 42)
+    end
+    assert_equal "100%_done", SolidQueue::ClaimedExecution.last.job.queue_name
+  end
+
   test "claim jobs using a wildcard and having paused queues" do
     AddToBufferJob.perform_later("hey")
 
@@ -168,6 +196,13 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
       claimed_jobs.map(&:queue_name)
   end
 
+  test "queues matched by more than one entry in the list are selected once" do
+    AddToBufferJob.set(queue: "queue_b1").perform_later(1)
+
+    assert_equal 1, SolidQueue::ReadyExecution.aggregated_count_across(%w[ queue_* queue_b* ])
+    assert_equal 1, SolidQueue::ReadyExecution.aggregated_count_across(%w[ queue_b* queue_b1 ])
+  end
+
   test "distinct_values_of returns all distinct queue names" do
     AddToBufferJob.perform_later("hey") # goes to background queue
 
@@ -222,6 +257,40 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     end
   end
 
+  test "bulk discard unblocks one job for each released concurrency slot" do
+    result = JobResult.create!
+    7.times { |i| ThrottledUpdateResultJob.perform_later(result, name: i.to_s) }
+    jobs = SolidQueue::Job.where(class_name: ThrottledUpdateResultJob.name).order(:id).to_a
+
+    assert_equal 4, SolidQueue::BlockedExecution.count
+
+    assert_difference -> { SolidQueue::Job.count } => -2, -> { SolidQueue::BlockedExecution.count } => -2 do
+      SolidQueue::ReadyExecution.discard_all_from_jobs(jobs.first(2))
+    end
+
+    assert_equal jobs[2..4].map(&:id), SolidQueue::ReadyExecution.where(job_id: jobs.map(&:id)).order(:job_id).pluck(:job_id)
+    assert jobs.last(2).all? { |job| job.reload.blocked? }
+    assert_equal 0, SolidQueue::Semaphore.find_by!(key: jobs.first.concurrency_key).value
+  end
+
+  test "discard in batches leaves newly unblocked jobs ready" do
+    blocked_jobs = 2.times.map do
+      result = JobResult.create!
+      NonOverlappingUpdateResultJob.perform_later(result, name: "ready")
+      NonOverlappingUpdateResultJob.perform_later(result, name: "blocked")
+      SolidQueue::Job.last
+    end
+
+    assert_difference -> { SolidQueue::Job.count }, -2 do
+      SolidQueue::ReadyExecution.queued_as("default").discard_all_in_batches(batch_size: 1)
+    end
+
+    assert_equal blocked_jobs.map(&:id), SolidQueue::Job.where(queue_name: "default").order(:id).pluck(:id)
+    assert blocked_jobs.all? { |job| job.reload.ready? }
+    assert_equal 0, SolidQueue::BlockedExecution.count
+    assert_equal [ 0, 0 ], SolidQueue::Semaphore.order(:id).pluck(:value)
+  end
+
   private
     def assert_claimed_jobs(count, &block)
       assert_difference -> { SolidQueue::ClaimedExecution.count } => +count, -> { SolidQueue::ReadyExecution.count } => -count do
diff --git a/test/models/solid_queue/semaphore_test.rb b/test/models/solid_queue/semaphore_test.rb
index 1aa9f39a8..2877afd27 100644
--- a/test/models/solid_queue/semaphore_test.rb
+++ b/test/models/solid_queue/semaphore_test.rb
@@ -5,6 +5,13 @@
 class SolidQueue::SemaphoreTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
+  class NonOverlappingByKeyJob < ApplicationJob
+    limits_concurrency key: ->(key) { key }
+
+    def perform(key)
+    end
+  end
+
   setup do
     @result = JobResult.create!(queue_name: "default")
   end
@@ -108,41 +115,73 @@ class SolidQueue::SemaphoreTest < ActiveSupport::TestCase
     assert_equal 0, SolidQueue::BlockedExecution.where(concurrency_key: concurrency_key).count
   end
 
-  test "signal_all does not increment semaphore value beyond limit" do
-    # Create a job with concurrency limit of 1
-    NonOverlappingUpdateResultJob.perform_later(@result)
-    job = SolidQueue::Job.last
-    concurrency_key = job.concurrency_key
+  test "dispatchers with jobs for the same concurrency keys in different order don't deadlock" do
+    skip_on_sqlite
 
-    # Semaphore should exist with value=0 (one slot taken)
-    semaphore = SolidQueue::Semaphore.find_by(key: concurrency_key)
-    assert_equal 0, semaphore.value
+    # By job id: one, two, two, one. In batches of two, a dispatcher gets the
+    # first two jobs and another one, skipping those, the other two.
+    jobs = %w[ one two two one ].map do |key|
+      NonOverlappingByKeyJob.set(wait: 1.minute).perform_later(key)
+      SolidQueue::Job.last
+    end
+    travel_to 2.minutes.from_now
 
-    # Manually set semaphore to its limit (simulating the slot was already returned)
-    semaphore.update!(value: 1)
+    errors = dispatch_in_two_batches_at_once
 
-    # signal_all should NOT increment beyond the limit
-    SolidQueue::Semaphore.signal_all([ job ])
+    assert_empty errors
+    assert_equal 0, SolidQueue::ScheduledExecution.count
 
-    assert_equal 1, semaphore.reload.value, "signal_all should not increment semaphore beyond its limit"
+    # The first job for each key gets the lock
+    assert_equal [ :ready, :ready, :blocked, :blocked ], jobs.map { |job| job.reload.status }
   end
 
-  test "signal_all increments semaphore when below limit" do
-    # Create a job with concurrency limit of 1
-    NonOverlappingUpdateResultJob.perform_later(@result)
-    job = SolidQueue::Job.last
-    concurrency_key = job.concurrency_key
+  private
+    # Two dispatchers, each with a batch of two jobs. Each one, after acquiring
+    # its first lock, gives the other the chance to acquire its first one too,
+    # so that both hold a lock when they go for the second.
+    def dispatch_in_two_batches_at_once
+      acquired_first = [ Concurrent::Event.new, Concurrent::Event.new ]
+
+      pause_after_first_lock(acquired_first) do
+        dispatchers = 2.times.map do |index|
+          Thread.new do
+            Thread.current[:dispatcher] = index
+            # The second one starts once the first has its batch and its first lock
+            acquired_first[0].wait(5.seconds) if index == 1
+
+            SolidQueue::Record.connection_pool.with_connection do
+              SolidQueue::ScheduledExecution.dispatch_next_batch(2)
+            end
+
+            nil
+          rescue => error
+            error
+          end
+        end
 
-    # Semaphore should exist with value=0
-    assert_equal 0, SolidQueue::Semaphore.find_by(key: concurrency_key).value
+        dispatchers.map(&:value).compact
+      end
+    end
 
-    # signal_all should increment from 0 to 1
-    SolidQueue::Semaphore.signal_all([ job ])
+    def pause_after_first_lock(acquired_first)
+      original_wait = SolidQueue::Semaphore.method(:wait)
 
-    assert_equal 1, SolidQueue::Semaphore.find_by(key: concurrency_key).value
-  end
+      SolidQueue::Semaphore.define_singleton_method(:wait) do |job|
+        original_wait.call(job).tap do
+          index = Thread.current[:dispatcher]
+
+          if index && !acquired_first[index].set?
+            acquired_first[index].set
+            acquired_first[1 - index].wait(2.seconds)
+          end
+        end
+      end
+
+      yield
+    ensure
+      SolidQueue::Semaphore.define_singleton_method(:wait, original_wait)
+    end
 
-  private
     def skip_on_sqlite
       skip "Row-level locking not supported on SQLite" if SolidQueue::Record.connection.adapter_name.downcase.include?("sqlite")
     end
diff --git a/test/unit/cli_test.rb b/test/unit/cli_test.rb
index 45657f97a..3a3d896ff 100644
--- a/test/unit/cli_test.rb
+++ b/test/unit/cli_test.rb
@@ -2,8 +2,22 @@
 
 require "test_helper"
 require "solid_queue/cli"
+require "open3"
 
 class CliTest < ActiveSupport::TestCase
+  test "eager loading leaves the CLI to bin/jobs" do
+    script = <<~SCRIPT
+      require #{File.expand_path("../dummy/config/environment", __dir__).dump}
+      Zeitwerk::Loader.eager_load_all
+      print SolidQueue.autoload?(:Cli) ? "not loaded" : "loaded"
+    SCRIPT
+
+    out, err, status = Open3.capture3(RbConfig.ruby, "-e", script)
+
+    assert status.success?, err
+    assert_equal "not loaded", out
+  end
+
   test "mode defaults to fork when no env var or option" do
     with_env("SOLID_QUEUE_SUPERVISOR_MODE" => nil) do
       config = configuration_from_cli
diff --git a/test/unit/queue_test.rb b/test/unit/queue_test.rb
index c89c932c8..cfca74c5c 100644
--- a/test/unit/queue_test.rb
+++ b/test/unit/queue_test.rb
@@ -28,6 +28,26 @@ class QueueTest < ActiveSupport::TestCase
     end
   end
 
+  %w[ default background ].each do |next_queue|
+    test "clear queue unblocks the next job in #{next_queue}" do
+      result = JobResult.create!
+      [ "default", next_queue, next_queue ].each do |queue|
+        NonOverlappingUpdateResultJob.set(queue: queue).perform_later(result, name: "A")
+      end
+      ready_job, next_job, blocked_job = SolidQueue::Job.last(3)
+
+      assert_difference -> { SolidQueue::Job.count }, -1 do
+        @default_queue.clear
+      end
+
+      assert_not SolidQueue::Job.exists?(ready_job.id)
+      assert next_job.reload.ready?
+      assert_not next_job.blocked?
+      assert blocked_job.reload.blocked?
+      assert_equal 0, SolidQueue::Semaphore.find_by!(key: ready_job.concurrency_key).value
+    end
+  end
+
   test "all existing queues" do
     assert_equal [ @background_queue ], SolidQueue::Queue.all
   end
diff --git a/test/unit/scheduler_test.rb b/test/unit/scheduler_test.rb
index e914a23ca..214e2cbb8 100644
--- a/test/unit/scheduler_test.rb
+++ b/test/unit/scheduler_test.rb
@@ -156,4 +156,66 @@ class SchedulerTest < ActiveSupport::TestCase
   ensure
     scheduler&.stop
   end
+
+  test "reschedules a dynamic task recreated with the same key" do
+    recurring_schedule = nil
+    travel_to Time.zone.local(2026, 8, 25, 10, 15) do
+      SolidQueue.schedule_recurring_task(
+        "dynamic_task",
+        class: "AddToBufferJob",
+        schedule: "every hour",
+        args: [ 42 ]
+      )
+      original_task = SolidQueue::RecurringTask.find_by!(key: "dynamic_task")
+      recurring_schedule = SolidQueue::Scheduler::RecurringSchedule.new({}, dynamic_tasks_enabled: true)
+      recurring_schedule.schedule_tasks
+      original_scheduled_task = recurring_schedule.scheduled_tasks.fetch(original_task.id)
+
+      SolidQueue.unschedule_recurring_task("dynamic_task")
+      SolidQueue.schedule_recurring_task(
+        "dynamic_task",
+        class: "AddToBufferJob",
+        schedule: "every day",
+        args: [ 42 ]
+      )
+      replacement_task = SolidQueue::RecurringTask.find_by!(key: "dynamic_task")
+
+      recurring_schedule.reschedule_dynamic_tasks
+      replacement_scheduled_task = recurring_schedule.scheduled_tasks.fetch(replacement_task.id)
+
+      assert_instance_of Concurrent::CancelledOperationError, original_scheduled_task.reason
+      assert_not_same original_scheduled_task, replacement_scheduled_task
+      assert_in_delta replacement_task.next_time - Time.current, replacement_scheduled_task.initial_delay, 0.1
+
+      recurring_schedule.send(:schedule_next_run, original_task, run_at: original_task.next_time)
+      assert_equal [ replacement_task.id ], recurring_schedule.scheduled_tasks.keys
+      assert_same replacement_scheduled_task, recurring_schedule.scheduled_tasks.fetch(replacement_task.id)
+    end
+  ensure
+    recurring_schedule&.unschedule_tasks
+  end
+
+  test "does not reschedule a dynamic task updated in place" do
+    recurring_schedule = nil
+    # Away from the top of the hour, so the task can't run between the two lookups
+    travel_to Time.zone.local(2026, 8, 25, 10, 15) do
+      task = SolidQueue::RecurringTask.create!(
+        key: "dynamic_task",
+        static: false,
+        class_name: "AddToBufferJob",
+        schedule: "every hour",
+        arguments: [ 42 ]
+      )
+      recurring_schedule = SolidQueue::Scheduler::RecurringSchedule.new({}, dynamic_tasks_enabled: true)
+      recurring_schedule.schedule_tasks
+      scheduled_task = recurring_schedule.scheduled_tasks.fetch(task.id)
+
+      task.update!(schedule: "every day")
+      recurring_schedule.reschedule_dynamic_tasks
+
+      assert_same scheduled_task, recurring_schedule.scheduled_tasks.fetch(task.id)
+    end
+  ensure
+    recurring_schedule&.unschedule_tasks
+  end
 end