diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index 998422c8e..07b3475f7 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -13,6 +13,7 @@ jobs:
         with:
           ruby-version: 3.4
           bundler-cache: true
+          cache-version: 1
       - name: Run rubocop
         run: |
           bundle exec rubocop --parallel
@@ -20,6 +21,7 @@ jobs:
   tests:
     name: Tests
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     strategy:
       fail-fast: false
       matrix:
@@ -28,12 +30,17 @@ jobs:
         - 3.2
         - 3.3
         - 3.4
+        - 4.0
         database: [ mysql, postgres, sqlite ]
-        gemfile: [ rails_7_1, rails_7_2, rails_8_0, rails_main ]
+        gemfile: [ rails_7_1, rails_7_2, rails_8_0, rails_8_1, rails_main ]
         exclude:
           - ruby-version: "3.1"
             gemfile: rails_8_0
           - ruby-version: "3.1"
+            gemfile: rails_8_1
+          - ruby-version: "3.1"
+            gemfile: rails_main
+          - ruby-version: "3.2"
             gemfile: rails_main
     services:
       mysql:
@@ -61,6 +68,7 @@ jobs:
         with:
           ruby-version: ${{ matrix.ruby-version }}
           bundler-cache: true
+          cache-version: 1
       - name: Update to latest Rails
         run: |
           bundle update railties
@@ -77,4 +85,4 @@ jobs:
           path: |
             test/dummy/log/test.log
           if-no-files-found: ignore
-          retention-days: 7
+          retention-days: 30
diff --git a/Appraisals b/Appraisals
index 248605287..9f3c8df48 100644
--- a/Appraisals
+++ b/Appraisals
@@ -15,6 +15,10 @@ appraise "rails-8-0" do
   gem "railties", "~> 8.0.0"
 end
 
+appraise "rails-8-1" do
+  gem "railties", "~> 8.1.0"
+end
+
 appraise "rails-main" do
   gem "railties", github: "rails/rails", branch: "main"
 end
diff --git a/Gemfile.lock b/Gemfile.lock
index 32f66b90a..7c4662de5 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    solid_queue (1.2.2)
+    solid_queue (1.4.0)
       activejob (>= 7.1)
       activerecord (>= 7.1)
       concurrent-ruby (>= 1.3.1)
@@ -201,6 +201,7 @@ DEPENDENCIES
   appraisal
   debug (~> 1.9)
   logger
+  minitest (~> 5.0)
   mocha
   mysql2
   pg
diff --git a/README.md b/README.md
index 92a018d4a..f77c4d5a1 100644
--- a/README.md
+++ b/README.md
@@ -14,8 +14,10 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
   - [Dashboard UI Setup](#dashboard-ui-setup)
   - [Incremental adoption](#incremental-adoption)
   - [High performance requirements](#high-performance-requirements)
+- [Workers, dispatchers, and scheduler](#workers-dispatchers-and-scheduler)
+  - [Fork vs. async mode](#fork-vs-async-mode)
 - [Configuration](#configuration)
-  - [Workers, dispatchers, and scheduler](#workers-dispatchers-and-scheduler)
+  - [Optional scheduler configuration](#optional-scheduler-configuration)
   - [Queue order and priorities](#queue-order-and-priorities)
   - [Queues specification and performance](#queues-specification-and-performance)
   - [Threads, processes, and signals](#threads-processes-and-signals)
@@ -30,6 +32,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
 - [Puma plugin](#puma-plugin)
 - [Jobs and transactional integrity](#jobs-and-transactional-integrity)
 - [Recurring tasks](#recurring-tasks)
+  - [Scheduling and unscheduling recurring tasks dynamically](#scheduling-and-unscheduling-recurring-tasks-dynamically)
 - [Inspiration](#inspiration)
 - [License](#license)
 
@@ -92,10 +95,10 @@ development:
 + primary:
     <<: *default
     database: storage/development.sqlite3
-+  queue:
-+    <<: *default
-+    database: storage/development_queue.sqlite3
-+    migrations_paths: db/queue_migrate
++ queue:
++   <<: *default
++   database: storage/development_queue.sqlite3
++   migrations_paths: db/queue_migrate
 ```
 
 Next, add the following to `development.rb`
@@ -127,7 +130,7 @@ In `config/cable.yml`
 ```diff
 development:
 -  adapter: async
-+ adapter: solid_cable
++  adapter: solid_cable
 +  connects_to:
 +    database:
 +      writing: cable
@@ -177,11 +180,9 @@ end
 
 ### High performance requirements
 
-Solid Queue was designed for the highest throughput when used with MySQL 8+ or PostgreSQL 9.5+, as they support `FOR UPDATE SKIP LOCKED`. You can use it with older versions, but in that case, you might run into lock waits if you run multiple workers for the same queue. You can also use it with SQLite on smaller applications.
-
-## Configuration
+Solid Queue was designed for the highest throughput when used with MySQL 8+, MariaDB 10.6+, or PostgreSQL 9.5+, as they support `FOR UPDATE SKIP LOCKED`. You can use it with older versions, but in that case, you might run into lock waits if you run multiple workers for the same queue. You can also use it with SQLite on smaller applications.
 
-### Workers, dispatchers, and scheduler
+## Workers, dispatchers, and scheduler
 
 We have several types of actors in Solid Queue:
 
@@ -190,7 +191,19 @@ We have several types of actors in Solid Queue:
 - The _scheduler_ manages [recurring tasks](#recurring-tasks), enqueuing jobs for them when they're due.
 - The _supervisor_ runs workers and dispatchers according to the configuration, controls their heartbeats, and stops and starts them when needed.
 
-Solid Queue's supervisor will fork a separate process for each supervised worker/dispatcher/scheduler.
+### Fork vs. async mode
+
+By default, Solid Queue runs in `fork` mode. This means the supervisor will fork a separate process for each supervised worker/dispatcher/scheduler. This provides the best isolation and performance, but can have additional memory usage and might not work with some Ruby implementations. As an alternative, you can run all workers, dispatchers and schedulers in the same process as the supervisor, in different threads, with an `async` mode. You can choose this mode by running `bin/jobs` as:
+
+```
+bin/jobs --mode async
+```
+
+Or you can also set the environment variable `SOLID_QUEUE_SUPERVISOR_MODE` to `async`. If you use the `async` mode, the `processes` option in the configuration described below will be ignored.
+
+**The recommended and default mode is `fork`. Only use `async` if you know what you're doing and have strong reasons to**
+
+## Configuration
 
 By default, Solid Queue will try to find your configuration under `config/queue.yml`, but you can set a different path using the environment variable `SOLID_QUEUE_CONFIG` or by using the `-c/--config_file` option with `bin/jobs`, like this:
 
@@ -198,7 +211,7 @@ By default, Solid Queue will try to find your configuration under `config/queue.
 bin/jobs -c config/calendar.yml
 ```
 
-You can also skip all recurring tasks by setting the environment variable `SOLID_QUEUE_SKIP_RECURRING=true`. This is useful for environments like staging, review apps, or development where you don't want any recurring jobs to run. This is equivalent to using the `--skip-recurring` option with `bin/jobs`.
+You can also skip the scheduler process by setting the environment variable `SOLID_QUEUE_SKIP_RECURRING=true`. This is useful for environments like staging, review apps, or development where you don't want any recurring jobs to run. This is equivalent to using the `--skip-recurring` option with `bin/jobs`.
 
 This is what this configuration looks like:
 
@@ -216,6 +229,10 @@ production:
       threads: 5
       polling_interval: 0.1
       processes: 3
+  scheduler:
+    dynamic_tasks_enabled: true
+    polling_interval: 5
+
 ```
 
 Everything is optional. If no configuration at all is provided, Solid Queue will run with one dispatcher and one worker with default settings. If you want to run only dispatchers or workers, you just need to include that section alone in the configuration. For example, with the following configuration:
@@ -247,6 +264,8 @@ Here's an overview of the different options:
   ```
 
   This will create a worker fetching jobs from all queues starting with `staging`. The wildcard `*` is only allowed on its own or at the end of a queue name; you can't specify queue names such as `*_some_queue`. These will be ignored.
+  
+  Also, if a wildcard (*) is included alongside explicit queue names, for example: `queues: [default, backend, *]`, then it would behave like `queues: *`
 
   Finally, you can combine prefixes with exact names, like `[ staging*, background ]`, and the behaviour with respect to order will be the same as with only exact names.
 
@@ -254,10 +273,23 @@ Here's an overview of the different options:
 
 - `threads`: this is the max size of the thread pool that each worker will have to run jobs. Each worker will fetch this number of jobs from their queue(s), at most and will post them to the thread pool to be run. By default, this is `3`. Only workers have this setting.
 It is recommended to set this value less than or equal to the queue database's connection pool size minus 2, as each worker thread uses one connection, and two additional connections are reserved for polling and heartbeat.
-- `processes`: this is the number of worker processes that will be forked by the supervisor with the settings given. By default, this is `1`, just a single process. This setting is useful if you want to dedicate more than one CPU core to a queue or queues with the same configuration. Only workers have this setting.
+- `processes`: this is the number of worker processes that will be forked by the supervisor with the settings given. By default, this is `1`, just a single process. This setting is useful if you want to dedicate more than one CPU core to a queue or queues with the same configuration. Only workers have this setting. **Note**: this option will be ignored if [running in `async` mode](#fork-vs-async-mode).
 - `concurrency_maintenance`: whether the dispatcher will perform the concurrency maintenance work. This is `true` by default, and it's useful if you don't use any [concurrency controls](#concurrency-controls) and want to disable it or if you run multiple dispatchers and want some of them to just dispatch jobs without doing anything else.
 
 
+### Optional scheduler configuration
+
+Optionally, you can configure the scheduler process under the `scheduler` section in your `config/queue.yml` if you'd like to [schedule recurring tasks dynamically](#scheduling-and-unscheduling-recurring-tasks-dynamically).
+
+```yaml
+scheduler:
+  dynamic_tasks_enabled: true
+  polling_interval: 5
+```
+
+- `dynamic_tasks_enabled`: whether the scheduler should poll for [dynamically scheduled recurring tasks](#scheduling-and-unscheduling-recurring-tasks-dynamically). This is `false` by default. When enabled, the scheduler will poll the database at the given `polling_interval` to pick up tasks scheduled via `SolidQueue.schedule_recurring_task`.
+- `polling_interval`: how frequently (in seconds) the scheduler checks for dynamic task changes. Defaults to `5`.
+
 ### Queue order and priorities
 
 As mentioned above, if you specify a list of queues for a worker, these will be polled in the order given, such as for the list `real_time,background`, no jobs will be taken from `background` unless there aren't any more jobs waiting in `real_time`.
@@ -334,7 +366,7 @@ queues: back*
 
 Workers in Solid Queue use a thread pool to run work in multiple threads, configurable via the `threads` parameter above. Besides this, parallelism can be achieved via multiple processes on one machine (configurable via different workers or the `processes` parameter above) or by horizontal scaling.
 
-The supervisor is in charge of managing these processes, and it responds to the following signals:
+The supervisor is in charge of managing these processes, and it responds to the following signals when running in its own process via `bin/jobs` or with [the Puma plugin](#puma-plugin) with the default `fork` mode:
 - `TERM`, `INT`: starts graceful termination. The supervisor will send a `TERM` signal to its supervised processes, and it'll wait up to `SolidQueue.shutdown_timeout` time until they're done. If any supervised processes are still around by then, it'll send a `QUIT` signal to them to indicate they must exit.
 - `QUIT`: starts immediate termination. The supervisor will send a `QUIT` signal to its supervised processes, causing them to exit immediately.
 
@@ -342,7 +374,7 @@ When receiving a `QUIT` signal, if workers still have jobs in-flight, these will
 
 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.
 
-In a similar way, if a worker is terminated in any other way not initiated by the above signals (e.g. a worker is sent a `KILL` signal), jobs in progress will be marked as failed so that they can be inspected, with a `SolidQueue::Processes::Process::ProcessExitError`. Sometimes a job in particular is responsible for this, for example, if it has a memory leak and you have a mechanism to kill processes over a certain memory threshold, so this will help identifying this kind of situation.
+In a similar way, if a worker is terminated in any other way not initiated by the above signals (e.g. a worker is sent a `KILL` signal), jobs in progress will be marked as failed so that they can be inspected, with a `SolidQueue::Processes::ProcessExitError`. Sometimes a job in particular is responsible for this, for example, if it has a memory leak and you have a mechanism to kill processes over a certain memory threshold, so this will help identifying this kind of situation.
 
 
 ### Database configuration
@@ -366,7 +398,7 @@ There are several settings that control how Solid Queue works that you can set a
 
   **This is not used for errors raised within a job execution**. Errors happening in jobs are handled by Active Job's `retry_on` or `discard_on`, and ultimately will result in [failed jobs](#failed-jobs-and-retries). This is for errors happening within Solid Queue itself.
 
-- `use_skip_locked`: whether to use `FOR UPDATE SKIP LOCKED` when performing locking reads. This will be automatically detected in the future, and for now, you only need to set this to `false` if your database doesn't support it. For MySQL, that'd be versions < 8, and for PostgreSQL, versions < 9.5. If you use SQLite, this has no effect, as writes are sequential.
+- `use_skip_locked`: whether to use `FOR UPDATE SKIP LOCKED` when performing locking reads. This will be automatically detected in the future, and for now, you only need to set this to `false` if your database doesn't support it. For MySQL, that'd be versions < 8; for MariaDB, versions < 10.6; and for PostgreSQL, versions < 9.5. If you use SQLite, this has no effect, as writes are sequential.
 - `process_heartbeat_interval`: the heartbeat interval that all processes will follow—defaults to 60 seconds.
 - `process_alive_threshold`: how long to wait until a process is considered dead after its last heartbeat—defaults to 5 minutes.
 - `shutdown_timeout`: time the supervisor will wait since it sent the `TERM` signal to its supervised processes before sending a `QUIT` version to them requesting immediate termination—defaults to 5 seconds.
@@ -449,7 +481,7 @@ class MyJob < ApplicationJob
 - `group` is used to control the concurrency of different job classes together. It defaults to the job class name.
 - `on_conflict` controls behaviour when enqueuing a job that conflicts with the concurrency limits configured. It can be set to one of the following:
   - (default) `:block`: the job is blocked and is dispatched when another job completes and unblocks it, or when the duration expires.
-  - `:discard`: the job is discarded. When you choose this option, bear in mind that if a job runs and fails to remove the concurrency lock (or _semaphore_, read below to know more about this), all jobs conflicting with it will be discarded up to the interval defined by `duration` has elapsed.
+  - `:discard`: the job is discarded. When you choose this option, bear in mind that if a job runs and fails to remove the concurrency lock (or _semaphore_, read below to know more about this), all jobs conflicting with it will be discarded until the interval defined by `duration` has elapsed.
 
 When a job includes these controls, we'll ensure that, at most, the number of jobs (indicated as `to`) that yield the same `key` will be performed concurrently, and this guarantee will last for `duration` for each job enqueued. Note that there's no guarantee about _the order of execution_, only about jobs being performed at the same time (overlapping).
 
@@ -459,7 +491,7 @@ Since something can happen that prevents the first job from releasing the semaph
 
 It's important to note that after one or more candidate jobs are unblocked (either because a job finishes or because `duration` expires and a semaphore is released), the `duration` timer for the still blocked jobs is reset. This happens indirectly via the expiration time of the semaphore, which is updated.
 
-When using `discard` as the behaviour to handle conflicts, you might have jobs discarded for up to the `duration` interval if something happens and a running job fails to release the semaphore.
+When using `discard` as the behaviour to handle conflicts, you might have jobs discarded for until the `duration` interval if something happens and a running job fails to release the semaphore.
 
 
 For example:
@@ -519,11 +551,11 @@ DeliverAnnouncementToContactJob.set(wait: 30.minutes).perform_later(contact)
 
 The 3 jobs will go into the scheduled queue and will wait there until they're due. Then, 10 minutes after, the first two jobs will be enqueued and the second one most likely will be blocked because the first one will be running first. Then, assuming the jobs are fast and finish in a few seconds, when the third job is due, it'll be enqueued normally.
 
-Normally scheduled jobs are enqueued in batches, but with concurrency controls, jobs need to be enqueued one by one. This has an impact on performance, similarly to the impact of concurrency controls in bulk enqueuing. Read below for more details. I'd generally advise against mixing concurrency controls with waiting/scheduling in the future.
+Normally scheduled jobs are enqueued in batches, but with concurrency controls, jobs need to be enqueued one by one. This has an impact on performance, similarly to the impact of concurrency controls in bulk enqueuing. Read below for more details. We generally advise against mixing concurrency controls with waiting/scheduling in the future.
 
 ### Performance considerations
 
-Concurrency controls introduce significant overhead (blocked executions need to be created and promoted to ready, semaphores need to be created and updated) so you should consider carefully whether you need them. For throttling purposes, where you plan to have `limit` significantly larger than 1, I'd encourage relying on a limited number of workers per queue instead. For example:
+Concurrency controls introduce significant overhead (blocked executions need to be created and promoted to ready, semaphores need to be created and updated) so you should consider carefully whether you need them. For throttling purposes, where you plan to have `limit` significantly larger than 1, we encourage relying on a limited number of workers per queue instead. For example:
 
 ```ruby
 class ThrottledJob < ApplicationJob
@@ -603,6 +635,22 @@ that you set in production only. This is what Rails 8's default Puma config look
 
 **Note**: phased restarts are not supported currently because the plugin requires [app preloading](https://github.com/puma/puma?tab=readme-ov-file#cluster-mode) to work.
 
+### Running as a fork or asynchronously
+
+By default, the Puma plugin will fork additional processes for each worker and dispatcher so that they run in different processes. This provides the best isolation and performance, but can have additional memory usage.
+
+Alternatively, workers and dispatchers can be run within the same Puma process(s). To do so just configure the plugin as:
+
+```ruby
+plugin :solid_queue
+solid_queue_mode :async
+```
+
+Note that in this case, the `processes` configuration option will be ignored. See also [Fork vs. async mode](#fork-vs-async-mode).
+
+**The recommended and default mode is `fork`. Only use `async` if you know what you're doing and have strong reasons to**
+
+
 ## Jobs and transactional integrity
 :warning: Having your jobs in the same ACID-compliant database as your application data enables a powerful yet sharp tool: taking advantage of transactional integrity to ensure some action in your app is not committed unless your job is also committed and vice versa, and ensuring that your job won't be enqueued until the transaction within which you're enqueuing it is committed. This can be very powerful and useful, but it can also backfire if you base some of your logic on this behaviour, and in the future, you move to another active job backend, or if you simply move Solid Queue to its own database, and suddenly the behaviour changes under you. Because this can be quite tricky and many people shouldn't need to worry about it, by default Solid Queue is configured in a different database as the main app.
 
@@ -703,6 +751,38 @@ my_periodic_resque_job:
 
 and the job will be enqueued via `perform_later` so it'll run in Resque. However, in this case we won't track any `solid_queue_recurring_execution` record for it and there won't be any guarantees that the job is enqueued only once each time.
 
+### Scheduling and unscheduling recurring tasks dynamically
+
+You can schedule and unschedule recurring tasks at runtime, without editing the configuration file. To enable this, you need to set `dynamic_tasks_enabled: true` in the `scheduler` section of your `config/queue.yml`, [as explained earlier](#optional-scheduler-configuration).
+
+```yaml
+scheduler:
+  dynamic_tasks_enabled: true
+```
+
+Then you can use the following methods to add recurring tasks dynamically:
+
+```ruby
+SolidQueue.schedule_recurring_task(
+  "my_dynamic_task",
+  class: "MyJob",
+  args: [1, 2],
+  schedule: "every 10 minutes"
+)
+```
+
+This accepts the same options as the YAML configuration: `class`, `args`, `command`, `schedule`, `queue`, `priority`, and `description`.
+
+To remove a dynamically scheduled task:
+
+```ruby
+SolidQueue.unschedule_recurring_task("my_dynamic_task")
+```
+
+Only dynamic tasks can be unscheduled at runtime. Attempting to unschedule a static task (defined in `config/recurring.yml`) will raise an `ActiveRecord::RecordNotFound` error.
+
+Tasks scheduled like this persist between Solid Queue's restarts and won't stop running until you manually unschedule them. 
+
 ## Inspiration
 
 Solid Queue has been inspired by [resque](https://github.com/resque/resque) and [GoodJob](https://github.com/bensheldon/good_job). We recommend checking out these projects as they're great examples from which we've learnt a lot.
diff --git a/Rakefile b/Rakefile
index bdb0ea233..412547017 100644
--- a/Rakefile
+++ b/Rakefile
@@ -5,7 +5,9 @@ require "bundler/setup"
 APP_RAKEFILE = File.expand_path("test/dummy/Rakefile", __dir__)
 load "rails/tasks/engine.rake"
 
-load "rails/tasks/statistics.rake"
+if Rails::VERSION::MAJOR < 8
+  load "rails/tasks/statistics.rake"
+end
 
 require "bundler/gem_tasks"
 require "rake/tasklib"
diff --git a/app/models/solid_queue/blocked_execution.rb b/app/models/solid_queue/blocked_execution.rb
index 4ad4d2397..68551a5f5 100644
--- a/app/models/solid_queue/blocked_execution.rb
+++ b/app/models/solid_queue/blocked_execution.rb
@@ -26,7 +26,9 @@ def release_many(concurrency_keys)
 
       def release_one(concurrency_key)
         transaction do
-          if execution = ordered.where(concurrency_key: concurrency_key).limit(1).non_blocking_lock.first
+          if execution = ordered.where(concurrency_key: concurrency_key).limit(1)
+              .use_index(:index_solid_queue_blocked_executions_for_release)
+              .non_blocking_lock.first
             execution.release
           end
         end
diff --git a/app/models/solid_queue/failed_execution.rb b/app/models/solid_queue/failed_execution.rb
index 8bcdc92f0..09a9596ef 100644
--- a/app/models/solid_queue/failed_execution.rb
+++ b/app/models/solid_queue/failed_execution.rb
@@ -6,7 +6,7 @@ class FailedExecution < Execution
 
     serialize :error, coder: JSON
 
-    before_create :expand_error_details_from_exception
+    before_save :expand_error_details_from_exception, if: :exception
 
     attr_accessor :exception
 
diff --git a/app/models/solid_queue/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb
index b7410b085..30d4399ed 100644
--- a/app/models/solid_queue/job/concurrency_controls.rb
+++ b/app/models/solid_queue/job/concurrency_controls.rb
@@ -26,7 +26,7 @@ def unblock_next_blocked_job
       end
 
       def concurrency_limited?
-        concurrency_key.present?
+        concurrency_key.present? && job_class.present?
       end
 
       def blocked?
diff --git a/app/models/solid_queue/job/retryable.rb b/app/models/solid_queue/job/retryable.rb
index 34eb9e4df..e2e2a2d07 100644
--- a/app/models/solid_queue/job/retryable.rb
+++ b/app/models/solid_queue/job/retryable.rb
@@ -16,7 +16,16 @@ def retry
       end
 
       def failed_with(exception)
-        FailedExecution.create_or_find_by!(job_id: id, exception: exception)
+        FailedExecution.transaction(requires_new: true) do
+          FailedExecution.create!(job_id: id, exception: exception)
+        end
+      rescue ActiveRecord::RecordNotUnique
+        if (failed_execution = FailedExecution.find_by(job_id: id))
+          failed_execution.exception = exception
+          failed_execution.save!
+        else
+          retry
+        end
       end
 
       def reset_execution_counters
diff --git a/app/models/solid_queue/ready_execution.rb b/app/models/solid_queue/ready_execution.rb
index 35a11292c..ea14c193a 100644
--- a/app/models/solid_queue/ready_execution.rb
+++ b/app/models/solid_queue/ready_execution.rb
@@ -30,7 +30,8 @@ def select_and_lock(queue_relation, process_id, limit)
         end
 
         def select_candidates(queue_relation, limit)
-          queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id)
+          # Force query execution here with #to_a to avoid unintended FOR UPDATE query executions
+          queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id).to_a
         end
 
         def lock_candidates(executions, process_id)
diff --git a/app/models/solid_queue/record.rb b/app/models/solid_queue/record.rb
index 0a704d2c8..8c7000bfe 100644
--- a/app/models/solid_queue/record.rb
+++ b/app/models/solid_queue/record.rb
@@ -20,6 +20,13 @@ def supports_insert_conflict_target?
           connection.supports_insert_conflict_target?
         end
       end
+
+      # Pass index hints to the query optimizer using SQL comment hints.
+      # Uses MySQL 8 optimizer hint query comments, which SQLite and
+      # PostgreSQL ignore.
+      def use_index(*indexes)
+        optimizer_hints "INDEX(#{quoted_table_name} #{indexes.join(', ')})"
+      end
     end
   end
 end
diff --git a/app/models/solid_queue/recurring_task.rb b/app/models/solid_queue/recurring_task.rb
index e6d4db573..9bb634e6f 100644
--- a/app/models/solid_queue/recurring_task.rb
+++ b/app/models/solid_queue/recurring_task.rb
@@ -6,11 +6,12 @@ module SolidQueue
   class RecurringTask < Record
     serialize :arguments, coder: Arguments, default: []
 
-    validate :supported_schedule
+    validate :ensure_schedule_supported
     validate :ensure_command_or_class_present
-    validate :existing_job_class
+    validate :ensure_existing_job_class
 
     scope :static, -> { where(static: true) }
+    scope :dynamic, -> { where(static: false) }
 
     has_many :recurring_executions, foreign_key: :task_key, primary_key: :key
 
@@ -32,7 +33,15 @@ def from_configuration(key, **options)
           queue_name: options[:queue].presence,
           priority: options[:priority].presence,
           description: options[:description],
-          static: true
+          static: options.fetch(:static, true)
+      end
+
+      def create_dynamic_task(key, **options)
+        from_configuration(key, **options.merge(static: false)).save!
+      end
+
+      def delete_dynamic_task(key)
+        RecurringTask.dynamic.find_by!(key: key).destroy
       end
 
       def create_or_update_all(tasks)
@@ -102,19 +111,28 @@ def attributes_for_upsert
     end
 
     private
-      def supported_schedule
+      def ensure_schedule_supported
         unless parsed_schedule.instance_of?(Fugit::Cron)
           errors.add :schedule, :unsupported, message: "is not a supported recurring schedule"
         end
+      rescue ArgumentError => error
+        message = if error.message.include?("multiple crons")
+          "generates multiple cron schedules. Please use separate recurring tasks for each schedule, " +
+          "or use explicit cron syntax (e.g., '40 0,15 * * *' for multiple times with the same minutes)"
+        else
+          error.message
+        end
+
+        errors.add :schedule, :unsupported, message: message
       end
 
       def ensure_command_or_class_present
         unless command.present? || class_name.present?
-          errors.add :base, :command_and_class_blank, message: "either command or class_name must be present"
+          errors.add :base, :command_and_class_blank, message: "either command or class must be present"
         end
       end
 
-      def existing_job_class
+      def ensure_existing_job_class
         if class_name.present? && job_class.nil?
           errors.add :class_name, :undefined, message: "doesn't correspond to an existing class"
         end
@@ -152,7 +170,7 @@ def arguments_with_kwargs
 
 
       def parsed_schedule
-        @parsed_schedule ||= Fugit.parse(schedule)
+        @parsed_schedule ||= Fugit.parse(schedule, multi: :fail)
       end
 
       def job_class
diff --git a/app/models/solid_queue/semaphore.rb b/app/models/solid_queue/semaphore.rb
index e93b24195..d8caa64ea 100644
--- a/app/models/solid_queue/semaphore.rb
+++ b/app/models/solid_queue/semaphore.rb
@@ -40,7 +40,7 @@ def initialize(job)
       end
 
       def wait
-        if semaphore = Semaphore.find_by(key: key)
+        if semaphore = Semaphore.lock.find_by(key: key)
           semaphore.value > 0 && attempt_decrement
         else
           attempt_creation
diff --git a/gemfiles/rails_8_1.gemfile b/gemfiles/rails_8_1.gemfile
new file mode 100644
index 000000000..5e111e3d1
--- /dev/null
+++ b/gemfiles/rails_8_1.gemfile
@@ -0,0 +1,7 @@
+# This file was generated by Appraisal
+
+source "https://rubygems.org"
+
+gem "railties", "~> 8.1.0"
+
+gemspec path: "../"
diff --git a/lib/puma/plugin/solid_queue.rb b/lib/puma/plugin/solid_queue.rb
index 434b8f65d..8a7aea28b 100644
--- a/lib/puma/plugin/solid_queue.rb
+++ b/lib/puma/plugin/solid_queue.rb
@@ -1,5 +1,13 @@
 require "puma/plugin"
 
+module Puma
+  class DSL
+    def solid_queue_mode(mode = :fork)
+      @options[:solid_queue_mode] = mode.to_sym
+    end
+  end
+end
+
 Puma::Plugin.create do
   attr_reader :puma_pid, :solid_queue_pid, :log_writer, :solid_queue_supervisor
 
@@ -7,38 +15,78 @@ def start(launcher)
     @log_writer = launcher.log_writer
     @puma_pid = $$
 
-    in_background do
-      monitor_solid_queue
+    if launcher.options[:solid_queue_mode] == :async
+      start_async(launcher)
+    else
+      start_forked(launcher)
     end
+  end
 
-    if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
-      launcher.events.on_booted do
-        @solid_queue_pid = fork do
-          Thread.new { monitor_puma }
-          SolidQueue::Supervisor.start
+  private
+    def start_forked(launcher)
+      in_background do
+        monitor_solid_queue
+      end
+
+      if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
+        launcher.events.on_booted do
+          @solid_queue_pid = fork do
+            Thread.new { monitor_puma }
+            SolidQueue::Supervisor.start(mode: :fork)
+          end
+        end
+
+        launcher.events.on_stopped { stop_solid_queue_fork }
+        launcher.events.on_restart { stop_solid_queue_fork }
+      else
+        launcher.events.after_booted do
+          @solid_queue_pid = fork do
+            Thread.new { monitor_puma }
+            start_solid_queue(mode: :fork)
+          end
         end
+
+        launcher.events.after_stopped { stop_solid_queue_fork }
+        launcher.events.before_restart { stop_solid_queue_fork }
       end
+    end
 
-      launcher.events.on_stopped { stop_solid_queue }
-      launcher.events.on_restart { stop_solid_queue }
-    else
-      launcher.events.after_booted do
-        @solid_queue_pid = fork do
-          Thread.new { monitor_puma }
-          SolidQueue::Supervisor.start
+    def start_async(launcher)
+      if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
+        launcher.events.on_booted do
+          start_solid_queue(mode: :async, standalone: false)
+        end
+
+        launcher.events.on_stopped { solid_queue_supervisor&.stop }
+
+        launcher.events.on_restart do
+          solid_queue_supervisor&.stop
+          start_solid_queue(mode: :async, standalone: false)
+        end
+      else
+        launcher.events.after_booted do
+          start_solid_queue(mode: :async, standalone: false)
+        end
+
+        launcher.events.after_stopped { solid_queue_supervisor&.stop }
+
+        launcher.events.before_restart do
+          solid_queue_supervisor&.stop
+          start_solid_queue(mode: :async, standalone: false)
         end
       end
+    end
 
-      launcher.events.after_stopped { stop_solid_queue }
-      launcher.events.before_restart { stop_solid_queue }
+    def start_solid_queue(**options)
+      @solid_queue_supervisor = SolidQueue::Supervisor.start(**options)
     end
-  end
 
-  private
-    def stop_solid_queue
+    def stop_solid_queue_fork
+      return unless solid_queue_pid
+
       Process.waitpid(solid_queue_pid, Process::WNOHANG)
       log "Stopping Solid Queue..."
-      Process.kill(:INT, solid_queue_pid) if solid_queue_pid
+      Process.kill(:INT, solid_queue_pid)
       Process.wait(solid_queue_pid)
     rescue Errno::ECHILD, Errno::ESRCH
     end
@@ -48,7 +96,7 @@ def monitor_puma
     end
 
     def monitor_solid_queue
-      monitor(:solid_queue_dead?, "Detected Solid Queue has gone away, stopping Puma...")
+      monitor(:solid_queue_fork_dead?, "Detected Solid Queue has gone away, stopping Puma...")
     end
 
     def monitor(process_dead, message)
@@ -62,7 +110,7 @@ def monitor(process_dead, message)
       end
     end
 
-    def solid_queue_dead?
+    def solid_queue_fork_dead?
       if solid_queue_started?
         Process.waitpid(solid_queue_pid, Process::WNOHANG)
       end
diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb
index e0d51c8cb..bd2269e5f 100644
--- a/lib/solid_queue.rb
+++ b/lib/solid_queue.rb
@@ -43,6 +43,14 @@ module SolidQueue
 
   delegate :on_start, :on_stop, :on_exit, to: Supervisor
 
+  def schedule_recurring_task(key, **options)
+    RecurringTask.create_dynamic_task(key, **options)
+  end
+
+  def unschedule_recurring_task(key)
+    RecurringTask.delete_dynamic_task(key)
+  end
+
   [ Dispatcher, Scheduler, Worker ].each do |process|
     define_singleton_method(:"on_#{process.name.demodulize.downcase}_start") do |&block|
       process.on_start(&block)
diff --git a/lib/solid_queue/app_executor.rb b/lib/solid_queue/app_executor.rb
index 0580213f7..f315f61a2 100644
--- a/lib/solid_queue/app_executor.rb
+++ b/lib/solid_queue/app_executor.rb
@@ -17,5 +17,15 @@ def handle_thread_error(error)
         SolidQueue.on_thread_error.call(error)
       end
     end
+
+    def create_thread(&block)
+      Thread.new do
+        Thread.current.name = name
+        block.call
+      rescue Exception => exception
+        handle_thread_error(exception)
+        raise
+      end
+    end
   end
 end
diff --git a/lib/solid_queue/async_supervisor.rb b/lib/solid_queue/async_supervisor.rb
new file mode 100644
index 000000000..4a7f4aeec
--- /dev/null
+++ b/lib/solid_queue/async_supervisor.rb
@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class AsyncSupervisor < Supervisor
+    after_shutdown :terminate_gracefully, unless: :standalone?
+
+    def stop
+      super
+      @thread&.join
+    end
+
+    private
+      def supervise
+        if standalone? then super
+        else
+          @thread = create_thread { super }
+        end
+      end
+
+      def check_and_replace_terminated_processes
+        terminated_threads = process_instances.select { |thread_id, instance| !instance.alive? }
+        terminated_threads.each { |thread_id, _| replace_thread(thread_id) }
+      end
+
+      def replace_thread(thread_id)
+        SolidQueue.instrument(:replace_thread, supervisor_pid: ::Process.pid) do |payload|
+          if (instance = process_instances.delete(thread_id))
+            payload[:thread] = instance
+
+            error = Processes::ThreadTerminatedError.new(instance.name)
+            release_claimed_jobs_by(instance, with_error: error)
+
+            start_process(configured_processes.delete(thread_id))
+          end
+        end
+      end
+
+      def perform_graceful_termination
+        process_instances.values.each(&:stop)
+
+        Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
+      end
+
+      def perform_immediate_termination
+        exit!
+      end
+
+      def all_processes_terminated?
+        process_instances.values.none?(&:alive?)
+      end
+  end
+end
diff --git a/lib/solid_queue/cli.rb b/lib/solid_queue/cli.rb
index 7bfe555bc..2f3f3e13b 100644
--- a/lib/solid_queue/cli.rb
+++ b/lib/solid_queue/cli.rb
@@ -8,6 +8,10 @@ class Cli < Thor
       desc: "Path to config file (default: #{Configuration::DEFAULT_CONFIG_FILE_PATH}).",
       banner: "SOLID_QUEUE_CONFIG"
 
+    class_option :mode, type: :string, enum: %w[ fork async ],
+      desc: "Whether to fork processes for workers and dispatchers (fork) or to run these in the same process as the supervisor (async) (default: fork).",
+      banner: "SOLID_QUEUE_SUPERVISOR_MODE"
+
     class_option :recurring_schedule_file, type: :string,
       desc: "Path to recurring schedule definition (default: #{Configuration::DEFAULT_RECURRING_SCHEDULE_FILE_PATH}).",
       banner: "SOLID_QUEUE_RECURRING_SCHEDULE"
diff --git a/lib/solid_queue/configuration.rb b/lib/solid_queue/configuration.rb
index a002b41dd..e63a000ca 100644
--- a/lib/solid_queue/configuration.rb
+++ b/lib/solid_queue/configuration.rb
@@ -28,6 +28,11 @@ def instantiate
       concurrency_maintenance_interval: 600
     }
 
+    SCHEDULER_DEFAULTS = {
+      polling_interval: 5,
+      dynamic_tasks_enabled: false
+    }
+
     DEFAULT_CONFIG_FILE_PATH = "config/queue.yml"
     DEFAULT_RECURRING_SCHEDULE_FILE_PATH = "config/recurring.yml"
 
@@ -56,6 +61,14 @@ def error_messages
       end
     end
 
+    def mode
+      @options[:mode].to_s.inquiry
+    end
+
+    def standalone?
+      mode.fork? || @options[:standalone]
+    end
+
     private
       attr_reader :options
 
@@ -84,6 +97,8 @@ def ensure_correctly_sized_thread_pool
 
       def default_options
         {
+          mode: ENV["SOLID_QUEUE_SUPERVISOR_MODE"] || :fork,
+          standalone: true,
           config_file: Rails.root.join(ENV["SOLID_QUEUE_CONFIG"] || DEFAULT_CONFIG_FILE_PATH),
           recurring_schedule_file: Rails.root.join(ENV["SOLID_QUEUE_RECURRING_SCHEDULE"] || DEFAULT_RECURRING_SCHEDULE_FILE_PATH),
           only_work: false,
@@ -110,7 +125,12 @@ def skip_recurring_tasks?
 
       def workers
         workers_options.flat_map do |worker_options|
-          processes = worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
+          processes = if mode.fork?
+            worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
+          else
+            1
+          end
+
           processes.times.map { Process.new(:worker, worker_options.with_defaults(WORKER_DEFAULTS)) }
         end
       end
@@ -122,8 +142,10 @@ def dispatchers
       end
 
       def schedulers
-        if !skip_recurring_tasks? && recurring_tasks.any?
-          [ Process.new(:scheduler, recurring_tasks: recurring_tasks) ]
+        return [] if skip_recurring_tasks?
+
+        if recurring_tasks.any? || dynamic_recurring_tasks_enabled?
+          [ Process.new(:scheduler, { recurring_tasks: recurring_tasks, **scheduler_options.with_defaults(SCHEDULER_DEFAULTS) }) ]
         else
           []
         end
@@ -139,17 +161,29 @@ def dispatchers_options
           .map { |options| options.dup.symbolize_keys }
       end
 
+      def scheduler_options
+        @scheduler_options ||= processes_config.fetch(:scheduler, {}).dup.symbolize_keys
+      end
+
+      def dynamic_recurring_tasks_enabled?
+        scheduler_options.fetch(:dynamic_tasks_enabled, SCHEDULER_DEFAULTS[:dynamic_tasks_enabled])
+      end
+
       def recurring_tasks
         @recurring_tasks ||= recurring_tasks_config.map do |id, options|
-          RecurringTask.from_configuration(id, **options) if options&.has_key?(:schedule)
+          RecurringTask.from_configuration(id, **options.merge(static: true)) if options&.has_key?(:schedule)
         end.compact
       end
 
       def processes_config
         @processes_config ||= config_from \
-          options.slice(:workers, :dispatchers).presence || options[:config_file],
-          keys: [ :workers, :dispatchers ],
-          fallback: { workers: [ WORKER_DEFAULTS ], dispatchers: [ DISPATCHER_DEFAULTS ] }
+          options.slice(:workers, :dispatchers, :scheduler).presence || options[:config_file],
+          keys: [ :workers, :dispatchers, :scheduler ],
+          fallback: {
+            workers: [ WORKER_DEFAULTS ],
+            dispatchers: [ DISPATCHER_DEFAULTS ],
+            scheduler: SCHEDULER_DEFAULTS
+          }
       end
 
       def recurring_tasks_config
@@ -158,7 +192,6 @@ def recurring_tasks_config
         end
       end
 
-
       def config_from(file_or_hash, keys: [], fallback: {}, env: Rails.env)
         load_config_from(file_or_hash).then do |config|
           config = config[env.to_sym] ? config[env.to_sym] : config
@@ -188,6 +221,7 @@ def load_config_from_file(file)
         if file.exist?
           ActiveSupport::ConfigurationFile.parse(file).deep_symbolize_keys
         else
+          puts "[solid_queue] WARNING: Provided configuration file '#{file}' does not exist. Falling back to default configuration."
           {}
         end
       end
diff --git a/lib/solid_queue/dispatcher.rb b/lib/solid_queue/dispatcher.rb
index 1583e1dd5..461ce8036 100644
--- a/lib/solid_queue/dispatcher.rb
+++ b/lib/solid_queue/dispatcher.rb
@@ -3,6 +3,7 @@
 module SolidQueue
   class Dispatcher < Processes::Poller
     include LifecycleHooks
+
     attr_reader :batch_size
 
     after_boot :run_start_hooks
diff --git a/lib/solid_queue/fork_supervisor.rb b/lib/solid_queue/fork_supervisor.rb
new file mode 100644
index 000000000..b8bb7f3ff
--- /dev/null
+++ b/lib/solid_queue/fork_supervisor.rb
@@ -0,0 +1,68 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class ForkSupervisor < Supervisor
+    private
+
+    def perform_graceful_termination
+      term_forks
+
+      Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? }) do
+        reap_terminated_forks
+      end
+    end
+
+    def perform_immediate_termination
+      quit_forks
+    end
+
+    def term_forks
+      signal_processes(process_instances.keys, :TERM)
+    end
+
+    def quit_forks
+      signal_processes(process_instances.keys, :QUIT)
+    end
+
+    def check_and_replace_terminated_processes
+      loop do
+        pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
+        break unless pid
+
+        replace_fork(pid, status)
+      end
+    end
+
+    def reap_terminated_forks
+      loop do
+        pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
+        break unless pid
+
+        if (terminated_fork = process_instances.delete(pid)) && (!status.exited? || status.exitstatus.to_i > 0)
+          error = Processes::ProcessExitError.new(status)
+          release_claimed_jobs_by(terminated_fork, with_error: error)
+        end
+
+        configured_processes.delete(pid)
+      end
+    rescue SystemCallError
+      # All children already reaped
+    end
+
+    def replace_fork(pid, status)
+      SolidQueue.instrument(:replace_fork, supervisor_pid: ::Process.pid, pid: pid, status: status) do |payload|
+        if terminated_fork = process_instances.delete(pid)
+          payload[:fork] = terminated_fork
+          error = Processes::ProcessExitError.new(status)
+          release_claimed_jobs_by(terminated_fork, with_error: error)
+
+          start_process(configured_processes.delete(pid))
+        end
+      end
+    end
+
+    def all_processes_terminated?
+      process_instances.empty?
+    end
+  end
+end
diff --git a/lib/solid_queue/processes/registrable.rb b/lib/solid_queue/processes/registrable.rb
index 2cc9036d6..35b4e01bd 100644
--- a/lib/solid_queue/processes/registrable.rb
+++ b/lib/solid_queue/processes/registrable.rb
@@ -18,17 +18,19 @@ def process_id
       attr_accessor :process
 
       def register
-        @process = SolidQueue::Process.register \
-          kind: kind,
-          name: name,
-          pid: pid,
-          hostname: hostname,
-          supervisor: try(:supervisor),
-          metadata: metadata.compact
+        wrap_in_app_executor do
+          @process = SolidQueue::Process.register \
+            kind: kind,
+            name: name,
+            pid: pid,
+            hostname: hostname,
+            supervisor: try(:supervisor),
+            metadata: metadata.compact
+        end
       end
 
       def deregister
-        process&.deregister
+        wrap_in_app_executor { process&.deregister }
       end
 
       def registered?
@@ -52,10 +54,14 @@ def stop_heartbeat
       end
 
       def heartbeat
-        process.heartbeat
+        process&.heartbeat
       rescue ActiveRecord::RecordNotFound
         self.process = nil
         wake_up
       end
+
+      def reload_metadata
+        wrap_in_app_executor { process&.update(metadata: metadata.compact) }
+      end
   end
 end
diff --git a/lib/solid_queue/processes/runnable.rb b/lib/solid_queue/processes/runnable.rb
index 33b441f61..c6e002e4f 100644
--- a/lib/solid_queue/processes/runnable.rb
+++ b/lib/solid_queue/processes/runnable.rb
@@ -7,20 +7,26 @@ module Runnable
     attr_writer :mode
 
     def start
-      boot
-
-      if running_async?
-        @thread = create_thread { run }
-      else
+      run_in_mode do
+        boot
         run
       end
     end
 
     def stop
       super
-
       wake_up
-      @thread&.join
+
+      # When not supervised, block until the thread terminates for backward
+      # compatibility with code that expects stop to be synchronous.
+      # When supervised, the supervisor controls the shutdown timeout.
+      unless supervised?
+        @thread&.join
+      end
+    end
+
+    def alive?
+      !running_async? || @thread&.alive?
     end
 
     private
@@ -30,6 +36,18 @@ def mode
         (@mode || DEFAULT_MODE).to_s.inquiry
       end
 
+      def run_in_mode(&block)
+        case
+        when running_as_fork?
+          fork(&block)
+        when running_async?
+          @thread = create_thread(&block)
+          @thread.object_id
+        else
+          block.call
+        end
+      end
+
       def boot
         SolidQueue.instrument(:start_process, process: self) do
           run_callbacks(:boot) do
@@ -74,16 +92,5 @@ def running_async?
       def running_as_fork?
         mode.fork?
       end
-
-
-      def create_thread(&block)
-        Thread.new do
-          Thread.current.name = name
-          block.call
-        rescue Exception => exception
-          handle_thread_error(exception)
-          raise
-        end
-      end
   end
 end
diff --git a/lib/solid_queue/processes/thread_terminated_error.rb b/lib/solid_queue/processes/thread_terminated_error.rb
new file mode 100644
index 000000000..7f1d6f7af
--- /dev/null
+++ b/lib/solid_queue/processes/thread_terminated_error.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  module Processes
+    class ThreadTerminatedError < RuntimeError
+      def initialize(name)
+        super("Thread #{name} terminated unexpectedly")
+      end
+    end
+  end
+end
diff --git a/lib/solid_queue/scheduler.rb b/lib/solid_queue/scheduler.rb
index 3cec90fa7..6022b3382 100644
--- a/lib/solid_queue/scheduler.rb
+++ b/lib/solid_queue/scheduler.rb
@@ -5,7 +5,7 @@ class Scheduler < Processes::Base
     include Processes::Runnable
     include LifecycleHooks
 
-    attr_reader :recurring_schedule
+    attr_reader :recurring_schedule, :polling_interval
 
     after_boot :run_start_hooks
     after_boot :schedule_recurring_tasks
@@ -14,7 +14,10 @@ class Scheduler < Processes::Base
     after_shutdown :run_exit_hooks
 
     def initialize(recurring_tasks:, **options)
-      @recurring_schedule = RecurringSchedule.new(recurring_tasks)
+      options = options.dup.with_defaults(SolidQueue::Configuration::SCHEDULER_DEFAULTS)
+      @dynamic_tasks_enabled = options[:dynamic_tasks_enabled]
+      @polling_interval = options[:polling_interval]
+      @recurring_schedule = RecurringSchedule.new(recurring_tasks, dynamic_tasks_enabled: @dynamic_tasks_enabled)
 
       super(**options)
     end
@@ -24,13 +27,16 @@ def metadata
     end
 
     private
-      SLEEP_INTERVAL = 60 # Right now it doesn't matter, can be set to 1 in the future for dynamic tasks
+
+      STATIC_SLEEP_INTERVAL = 60
 
       def run
         loop do
           break if shutting_down?
 
-          interruptible_sleep(SLEEP_INTERVAL)
+          reload_dynamic_schedule if dynamic_tasks_enabled?
+
+          interruptible_sleep(sleep_interval)
         end
       ensure
         SolidQueue.instrument(:shutdown_process, process: self) do
@@ -46,10 +52,23 @@ def unschedule_recurring_tasks
         recurring_schedule.unschedule_tasks
       end
 
+      def reload_dynamic_schedule
+        recurring_schedule.reschedule_dynamic_tasks
+        reload_metadata
+      end
+
+      def dynamic_tasks_enabled?
+        @dynamic_tasks_enabled
+      end
+
       def all_work_completed?
         recurring_schedule.empty?
       end
 
+      def sleep_interval
+        dynamic_tasks_enabled? ? polling_interval : STATIC_SLEEP_INTERVAL
+      end
+
       def set_procline
         procline "scheduling #{recurring_schedule.task_keys.join(",")}"
       end
diff --git a/lib/solid_queue/scheduler/recurring_schedule.rb b/lib/solid_queue/scheduler/recurring_schedule.rb
index 4070a0eca..a1e2409e0 100644
--- a/lib/solid_queue/scheduler/recurring_schedule.rb
+++ b/lib/solid_queue/scheduler/recurring_schedule.rb
@@ -4,21 +4,28 @@ module SolidQueue
   class Scheduler::RecurringSchedule
     include AppExecutor
 
-    attr_reader :configured_tasks, :scheduled_tasks
+    attr_reader :scheduled_tasks
+
+    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
 
-    def initialize(tasks)
-      @configured_tasks = Array(tasks).map { |task| SolidQueue::RecurringTask.wrap(task) }.select(&:valid?)
       @scheduled_tasks = Concurrent::Hash.new
     end
 
+    def configured_tasks
+      static_tasks + dynamic_tasks
+    end
+
     def empty?
-      configured_tasks.empty?
+      scheduled_tasks.empty? && dynamic_tasks.empty?
     end
 
     def schedule_tasks
       wrap_in_app_executor do
-        persist_tasks
-        reload_tasks
+        persist_static_tasks
+        reload_static_tasks
+        reload_dynamic_tasks
       end
 
       configured_tasks.each do |task|
@@ -39,14 +46,57 @@ def task_keys
       configured_tasks.map(&:key)
     end
 
+    def reschedule_dynamic_tasks
+      wrap_in_app_executor do
+        reload_dynamic_tasks
+        schedule_created_dynamic_tasks
+        unschedule_deleted_dynamic_tasks
+      end
+    end
+
     private
-      def persist_tasks
-        SolidQueue::RecurringTask.static.where.not(key: task_keys).delete_all
-        SolidQueue::RecurringTask.create_or_update_all configured_tasks
+      attr_reader :static_tasks
+
+      def static_task_keys
+        static_tasks.map(&:key)
+      end
+
+      def dynamic_tasks
+        @dynamic_tasks ||= load_dynamic_tasks
+      end
+
+      def dynamic_tasks_enabled?
+        @dynamic_tasks_enabled
+      end
+
+      def schedule_created_dynamic_tasks
+        RecurringTask.dynamic.where.not(key: scheduled_tasks.keys).each do |task|
+          schedule_task(task)
+        end
+      end
+
+      def unschedule_deleted_dynamic_tasks
+        (scheduled_tasks.keys - RecurringTask.pluck(:key)).each do |key|
+          scheduled_tasks[key].cancel
+          scheduled_tasks.delete(key)
+        end
+      end
+
+      def persist_static_tasks
+        RecurringTask.static.where.not(key: static_task_keys).delete_all
+        RecurringTask.create_or_update_all static_tasks
+      end
+
+      def reload_static_tasks
+        @static_tasks = RecurringTask.static.where(key: static_task_keys).to_a
+      end
+
+      def reload_dynamic_tasks
+        @dynamic_tasks = load_dynamic_tasks
       end
 
-      def reload_tasks
-        @configured_tasks = SolidQueue::RecurringTask.where(key: task_keys)
+      def load_dynamic_tasks
+        dynamic_tasks_enabled? ? RecurringTask.dynamic.to_a : []
       end
 
       def schedule(task)
diff --git a/lib/solid_queue/supervisor.rb b/lib/solid_queue/supervisor.rb
index 05deaa52d..ae17ec95c 100644
--- a/lib/solid_queue/supervisor.rb
+++ b/lib/solid_queue/supervisor.rb
@@ -13,42 +13,46 @@ def start(**options)
         configuration = Configuration.new(**options)
 
         if configuration.valid?
-          new(configuration).tap(&:start)
+          klass = configuration.mode.fork? ? ForkSupervisor : AsyncSupervisor
+          klass.new(configuration).tap(&:start)
         else
           abort configuration.errors.full_messages.join("\n") + "\nExiting..."
         end
       end
     end
 
+    delegate :mode, :standalone?, to: :configuration
+
     def initialize(configuration)
       @configuration = configuration
-      @forks = {}
+
       @configured_processes = {}
+      @process_instances = {}
 
       super
     end
 
     def start
-      wrap_in_app_executor do
-        boot
-        run_start_hooks
+      boot
+      run_start_hooks
 
-        start_processes
-        launch_maintenance_task
+      start_processes
+      launch_maintenance_task
 
-        supervise
-      end
+      supervise
     end
 
     def stop
-      wrap_in_app_executor do
-        super
-        run_stop_hooks
-      end
+      super
+      run_stop_hooks
+    end
+
+    def kind
+      "Supervisor(#{mode})"
     end
 
     private
-      attr_reader :configuration, :forks, :configured_processes
+      attr_reader :configuration, :configured_processes, :process_instances
 
       def boot
         SolidQueue.instrument(:start_process, process: self) do
@@ -66,11 +70,13 @@ def supervise
         loop do
           break if stopped?
 
-          set_procline
-          process_signal_queue
+          if standalone?
+            set_procline
+            process_signal_queue
+          end
 
           unless stopped?
-            reap_and_replace_terminated_forks
+            check_and_replace_terminated_processes
             interruptible_sleep(1.second)
           end
         end
@@ -81,30 +87,23 @@ def supervise
       def start_process(configured_process)
         process_instance = configured_process.instantiate.tap do |instance|
           instance.supervised_by process
-          instance.mode = :fork
+          instance.mode = mode
         end
 
-        pid = fork do
-          process_instance.start
-        end
+        process_id = process_instance.start
 
-        configured_processes[pid] = configured_process
-        forks[pid] = process_instance
+        configured_processes[process_id] = configured_process
+        process_instances[process_id] = process_instance
       end
 
-      def set_procline
-        procline "supervising #{supervised_processes.join(", ")}"
+      def check_and_replace_terminated_processes
       end
 
       def terminate_gracefully
-        SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do |payload|
-          term_forks
-
-          Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_forks_terminated? }) do
-            reap_terminated_forks
-          end
+        SolidQueue.instrument(:graceful_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: configured_processes.keys) do |payload|
+          perform_graceful_termination
 
-          unless all_forks_terminated?
+          unless all_processes_terminated?
             payload[:shutdown_timeout_exceeded] = true
             terminate_immediately
           end
@@ -112,82 +111,37 @@ def terminate_gracefully
       end
 
       def terminate_immediately
-        SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: supervised_processes) do
-          quit_forks
+        SolidQueue.instrument(:immediate_termination, process_id: process_id, supervisor_pid: ::Process.pid, supervised_processes: configured_processes.keys) do
+          perform_immediate_termination
         end
       end
 
-      def shutdown
-        SolidQueue.instrument(:shutdown_process, process: self) do
-          run_callbacks(:shutdown) do
-            stop_maintenance_task
-          end
-        end
+      def perform_graceful_termination
+        raise NotImplementedError
       end
 
-      def sync_std_streams
-        STDOUT.sync = STDERR.sync = true
+      def perform_immediate_termination
+        raise NotImplementedError
       end
 
-      def supervised_processes
-        forks.keys
+      def all_processes_terminated?
+        raise NotImplementedError
       end
 
-      def term_forks
-        signal_processes(forks.keys, :TERM)
-      end
-
-      def quit_forks
-        signal_processes(forks.keys, :QUIT)
-      end
-
-      def reap_and_replace_terminated_forks
-        loop do
-          pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
-          break unless pid
-
-          replace_fork(pid, status)
-        end
-      end
-
-      def reap_terminated_forks
-        loop do
-          pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
-          break unless pid
-
-          if (terminated_fork = forks.delete(pid)) && (!status.exited? || status.exitstatus > 0)
-            handle_claimed_jobs_by(terminated_fork, status)
-          end
-
-          configured_processes.delete(pid)
-        end
-      rescue SystemCallError
-        # All children already reaped
-      end
-
-      def replace_fork(pid, status)
-        SolidQueue.instrument(:replace_fork, supervisor_pid: ::Process.pid, pid: pid, status: status) do |payload|
-          if terminated_fork = forks.delete(pid)
-            payload[:fork] = terminated_fork
-            handle_claimed_jobs_by(terminated_fork, status)
-
-            start_process(configured_processes.delete(pid))
+      def shutdown
+        SolidQueue.instrument(:shutdown_process, process: self) do
+          run_callbacks(:shutdown) do
+            stop_maintenance_task
           end
         end
       end
 
-      # When a supervised fork crashes or exits we need to mark all the
-      # executions it had claimed as failed so that they can be retried
-      # by some other worker.
-      def handle_claimed_jobs_by(terminated_fork, status)
-        if registered_process = SolidQueue::Process.find_by(name: terminated_fork.name)
-          error = Processes::ProcessExitError.new(status)
-          registered_process.fail_all_claimed_executions_with(error)
-        end
+      def set_procline
+        procline "supervising #{configured_processes.keys.join(", ")}"
       end
 
-      def all_forks_terminated?
-        forks.empty?
+      def sync_std_streams
+        STDOUT.sync = STDERR.sync = true
       end
   end
 end
diff --git a/lib/solid_queue/supervisor/maintenance.rb b/lib/solid_queue/supervisor/maintenance.rb
index 1b6b52048..d92569d58 100644
--- a/lib/solid_queue/supervisor/maintenance.rb
+++ b/lib/solid_queue/supervisor/maintenance.rb
@@ -32,5 +32,16 @@ def fail_orphaned_executions
           ClaimedExecution.orphaned.fail_all_with(Processes::ProcessMissingError.new)
         end
       end
+
+      # When a supervised process crashes or exits we need to mark all the
+      # executions it had claimed as failed so that they can be retried
+      # by some other worker.
+      def release_claimed_jobs_by(terminated_process, with_error:)
+        wrap_in_app_executor do
+          if registered_process = SolidQueue::Process.find_by(name: terminated_process.name)
+            registered_process.fail_all_claimed_executions_with(with_error)
+          end
+        end
+      end
   end
 end
diff --git a/lib/solid_queue/supervisor/signals.rb b/lib/solid_queue/supervisor/signals.rb
index fe0960d5b..7bee107d5 100644
--- a/lib/solid_queue/supervisor/signals.rb
+++ b/lib/solid_queue/supervisor/signals.rb
@@ -6,8 +6,8 @@ module Signals
       extend ActiveSupport::Concern
 
       included do
-        before_boot :register_signal_handlers
-        after_shutdown :restore_default_signal_handlers
+        before_boot :register_signal_handlers, if: :standalone?
+        after_shutdown :restore_default_signal_handlers, if: :standalone?
       end
 
       private
diff --git a/lib/solid_queue/timer.rb b/lib/solid_queue/timer.rb
index ca16466d2..19e691f53 100644
--- a/lib/solid_queue/timer.rb
+++ b/lib/solid_queue/timer.rb
@@ -4,18 +4,18 @@ module SolidQueue
   module Timer
     extend self
 
-    def wait_until(timeout, condition, &block)
+    def wait_until(timeout, condition)
       if timeout > 0
         deadline = monotonic_time_now + timeout
 
         while monotonic_time_now < deadline && !condition.call
           sleep 0.1
-          block.call
+          yield if block_given?
         end
       else
         while !condition.call
           sleep 0.5
-          block.call
+          yield if block_given?
         end
       end
     end
diff --git a/lib/solid_queue/version.rb b/lib/solid_queue/version.rb
index f1718f9a8..8514496c3 100644
--- a/lib/solid_queue/version.rb
+++ b/lib/solid_queue/version.rb
@@ -1,3 +1,3 @@
 module SolidQueue
-  VERSION = "1.2.2"
+  VERSION = "1.4.0"
 end
diff --git a/solid_queue.gemspec b/solid_queue.gemspec
index 64818a2e2..3a5c9693e 100644
--- a/solid_queue.gemspec
+++ b/solid_queue.gemspec
@@ -34,6 +34,7 @@ Gem::Specification.new do |spec|
 
   spec.add_development_dependency "appraisal"
   spec.add_development_dependency "debug", "~> 1.9"
+  spec.add_development_dependency "minitest", "~> 5.0"
   spec.add_development_dependency "mocha"
   spec.add_development_dependency "puma", "~> 7.0"
   spec.add_development_dependency "mysql2"
diff --git a/test/dummy/config/initializers/sqlite3.rb b/test/dummy/config/initializers/sqlite3.rb
index bb3ee5db8..a5e45c339 100644
--- a/test/dummy/config/initializers/sqlite3.rb
+++ b/test/dummy/config/initializers/sqlite3.rb
@@ -24,7 +24,10 @@ def configure_connection
 
 ActiveSupport.on_load :active_record do
   if defined?(ActiveRecord::ConnectionAdapters::SQLite3Adapter)
-    ActiveRecord::ConnectionAdapters::SQLite3Adapter.prepend SqliteImmediateTransactions
+    # Rails 8.0+ has immediate transactions built-in
+    if Rails::VERSION::MAJOR < 8
+      ActiveRecord::ConnectionAdapters::SQLite3Adapter.prepend SqliteImmediateTransactions
+    end
     ActiveRecord::ConnectionAdapters::SQLite3Adapter.prepend SQLite3Configuration
   end
 
diff --git a/test/dummy/config/puma.rb b/test/dummy/config/puma.rb
deleted file mode 100644
index d4f1ae108..000000000
--- a/test/dummy/config/puma.rb
+++ /dev/null
@@ -1,44 +0,0 @@
-# Puma can serve each request in a thread from an internal thread pool.
-# The `threads` method setting takes two numbers: a minimum and maximum.
-# Any libraries that use thread pools should be configured to match
-# the maximum value specified for Puma. Default is set to 5 threads for minimum
-# and maximum; this matches the default thread size of Active Record.
-#
-max_threads_count = ENV.fetch("RAILS_MAX_THREADS") { 5 }
-min_threads_count = ENV.fetch("RAILS_MIN_THREADS") { max_threads_count }
-threads min_threads_count, max_threads_count
-
-# Specifies the `worker_timeout` threshold that Puma will use to wait before
-# terminating a worker in development environments.
-#
-worker_timeout 3600 if ENV.fetch("RAILS_ENV", "development") == "development"
-
-# Specifies the `port` that Puma will listen on to receive requests; default is 3000.
-#
-port ENV.fetch("PORT") { 3000 }
-
-# Specifies the `environment` that Puma will run in.
-#
-environment ENV.fetch("RAILS_ENV") { "development" }
-
-# Specifies the `pidfile` that Puma will use.
-pidfile ENV.fetch("PIDFILE") { "tmp/pids/server.pid" }
-
-# Specifies the number of `workers` to boot in clustered mode.
-# Workers are forked web server processes. If using threads and workers together
-# the concurrency of the application would be max `threads` * `workers`.
-# Workers do not work on JRuby or Windows (both of which do not support
-# processes).
-#
-# workers ENV.fetch("WEB_CONCURRENCY") { 2 }
-
-# Use the `preload_app!` method when specifying a `workers` number.
-# This directive tells Puma to first boot the application and load code
-# before forking the application. This takes advantage of Copy On Write
-# process behavior so workers use less memory.
-#
-# preload_app!
-
-# Allow puma to be restarted by `bin/rails restart` command.
-plugin :tmp_restart
-plugin :solid_queue
diff --git a/test/dummy/config/puma.rb b/test/dummy/config/puma.rb
new file mode 120000
index 000000000..f923a8263
--- /dev/null
+++ b/test/dummy/config/puma.rb
@@ -0,0 +1 @@
+puma_fork.rb
\ No newline at end of file
diff --git a/test/dummy/config/puma_async.rb b/test/dummy/config/puma_async.rb
new file mode 100644
index 000000000..beb652595
--- /dev/null
+++ b/test/dummy/config/puma_async.rb
@@ -0,0 +1,46 @@
+# Puma can serve each request in a thread from an internal thread pool.
+# The `threads` method setting takes two numbers: a minimum and maximum.
+# Any libraries that use thread pools should be configured to match
+# the maximum value specified for Puma. Default is set to 5 threads for minimum
+# and maximum; this matches the default thread size of Active Record.
+#
+max_threads_count = ENV.fetch("RAILS_MAX_THREADS") { 5 }
+min_threads_count = ENV.fetch("RAILS_MIN_THREADS") { max_threads_count }
+threads min_threads_count, max_threads_count
+
+# Specifies the `worker_timeout` threshold that Puma will use to wait before
+# terminating a worker in development environments.
+#
+worker_timeout 3600 if ENV.fetch("RAILS_ENV", "development") == "development"
+
+# Specifies the `port` that Puma will listen on to receive requests; default is 3000.
+#
+port ENV.fetch("PORT") { 3000 }
+
+# Specifies the `environment` that Puma will run in.
+#
+environment ENV.fetch("RAILS_ENV") { "development" }
+
+# Specifies the `pidfile` that Puma will use.
+pidfile ENV.fetch("PIDFILE") { "tmp/pids/server.pid" }
+
+# Specifies the number of `workers` to boot in clustered mode.
+# Workers are forked web server processes. If using threads and workers together
+# the concurrency of the application would be max `threads` * `workers`.
+# Workers do not work on JRuby or Windows (both of which do not support
+# processes).
+#
+# workers ENV.fetch("WEB_CONCURRENCY") { 2 }
+
+# Use the `preload_app!` method when specifying a `workers` number.
+# This directive tells Puma to first boot the application and load code
+# before forking the application. This takes advantage of Copy On Write
+# process behavior so workers use less memory.
+#
+# preload_app!
+
+# Allow puma to be restarted by `bin/rails restart` command.
+plugin :tmp_restart
+plugin :solid_queue
+
+solid_queue_mode :async
diff --git a/test/dummy/config/puma_fork.rb b/test/dummy/config/puma_fork.rb
new file mode 100644
index 000000000..4cdbbfd1d
--- /dev/null
+++ b/test/dummy/config/puma_fork.rb
@@ -0,0 +1,46 @@
+# Puma can serve each request in a thread from an internal thread pool.
+# The `threads` method setting takes two numbers: a minimum and maximum.
+# Any libraries that use thread pools should be configured to match
+# the maximum value specified for Puma. Default is set to 5 threads for minimum
+# and maximum; this matches the default thread size of Active Record.
+#
+max_threads_count = ENV.fetch("RAILS_MAX_THREADS") { 5 }
+min_threads_count = ENV.fetch("RAILS_MIN_THREADS") { max_threads_count }
+threads min_threads_count, max_threads_count
+
+# Specifies the `worker_timeout` threshold that Puma will use to wait before
+# terminating a worker in development environments.
+#
+worker_timeout 3600 if ENV.fetch("RAILS_ENV", "development") == "development"
+
+# Specifies the `port` that Puma will listen on to receive requests; default is 3000.
+#
+port ENV.fetch("PORT") { 3000 }
+
+# Specifies the `environment` that Puma will run in.
+#
+environment ENV.fetch("RAILS_ENV") { "development" }
+
+# Specifies the `pidfile` that Puma will use.
+pidfile ENV.fetch("PIDFILE") { "tmp/pids/server.pid" }
+
+# Specifies the number of `workers` to boot in clustered mode.
+# Workers are forked web server processes. If using threads and workers together
+# the concurrency of the application would be max `threads` * `workers`.
+# Workers do not work on JRuby or Windows (both of which do not support
+# processes).
+#
+# workers ENV.fetch("WEB_CONCURRENCY") { 2 }
+
+# Use the `preload_app!` method when specifying a `workers` number.
+# This directive tells Puma to first boot the application and load code
+# before forking the application. This takes advantage of Copy On Write
+# process behavior so workers use less memory.
+#
+# preload_app!
+
+# Allow puma to be restarted by `bin/rails restart` command.
+plugin :tmp_restart
+plugin :solid_queue
+
+solid_queue_mode :fork
diff --git a/test/integration/async_processes_lifecycle_test.rb b/test/integration/async_processes_lifecycle_test.rb
new file mode 100644
index 000000000..fd284210e
--- /dev/null
+++ b/test/integration/async_processes_lifecycle_test.rb
@@ -0,0 +1,232 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class AsyncProcessesLifecycleTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  setup do
+    @pid = run_supervisor_as_fork(mode: :async, workers: [ { queues: :background }, { queues: :default, threads: 5 } ])
+
+    wait_for_registered_processes(3, timeout: 3.second)
+    assert_registered_workers_for(:background, :default, supervisor_pid: @pid)
+  end
+
+  teardown do
+    terminate_process(@pid) if process_exists?(@pid)
+  end
+
+  test "enqueue jobs in multiple queues" do
+    6.times { |i| enqueue_store_result_job("job_#{i}") }
+    6.times { |i| enqueue_store_result_job("job_#{i}", :default) }
+
+    wait_for_jobs_to_finish_for(2.seconds)
+
+    assert_equal 12, JobResult.count
+    6.times { |i| assert_completed_job_results("job_#{i}", :background) }
+    6.times { |i| assert_completed_job_results("job_#{i}", :default) }
+
+    terminate_process(@pid)
+    assert_clean_termination
+  end
+
+  test "kill supervisor while there are jobs in-flight" do
+    no_pause = enqueue_store_result_job("no pause")
+    pause = enqueue_store_result_job("pause", pause: 3.second)
+
+    # Wait for the "no pause" job to complete before sending KILL
+    wait_for_jobs_to_finish_for(2.seconds, except: pause)
+
+    signal_process(@pid, :KILL, wait: 0.1.seconds)
+    wait_for_registered_processes(1, timeout: 2.second)
+
+    assert_not process_exists?(@pid)
+
+    assert_completed_job_results("no pause")
+    assert_job_status(no_pause, :finished)
+
+    # In async mode, killing the supervisor kills all threads too,
+    # so we can't complete in-flight jobs
+    assert_registered_supervisor
+    assert_registered_workers_for(:background, :default, supervisor_pid: @pid)
+    assert_started_job_result("pause")
+    assert_claimed_jobs
+  end
+
+  test "term supervisor multiple times" do
+    5.times do
+      signal_process(@pid, :TERM, wait: 0.1.second)
+    end
+
+    sleep(1.second)
+    assert_clean_termination
+  end
+
+  test "quit supervisor while there are jobs in-flight" do
+    no_pause = enqueue_store_result_job("no pause")
+    pause = enqueue_store_result_job("pause", pause: 1.second)
+
+    wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 0 }
+
+    signal_process(@pid, :QUIT, wait: 0.4.second)
+    wait_for_jobs_to_finish_for(2.seconds, except: pause)
+
+    wait_while_with_timeout(2.seconds) { process_exists?(@pid) }
+    assert_not process_exists?(@pid)
+
+    # In async mode, QUIT calls exit! which terminates immediately without cleanup.
+    # The in-flight job remains claimed and the process/workers remain registered.
+    # A future supervisor will need to prune and fail these orphaned executions.
+    assert_completed_job_results("no pause")
+    assert_job_status(no_pause, :finished)
+    assert_started_job_result("pause")
+    assert_job_status(pause, :claimed)
+
+    assert_registered_supervisor
+    assert_registered_workers_for(:background, :default, supervisor_pid: @pid)
+    assert_claimed_jobs
+  end
+
+  test "term supervisor while there are jobs in-flight" do
+    no_pause = enqueue_store_result_job("no pause")
+    pause = enqueue_store_result_job("pause", pause: 0.2.seconds)
+
+    signal_process(@pid, :TERM, wait: 0.3.second)
+    wait_for_jobs_to_finish_for(3.seconds)
+
+    assert_completed_job_results("no pause")
+    assert_completed_job_results("pause")
+
+    assert_job_status(no_pause, :finished)
+    assert_job_status(pause, :finished)
+
+    wait_for_process_termination_with_timeout(@pid, timeout: 1.second)
+    assert_clean_termination
+  end
+
+  test "int supervisor while there are jobs in-flight" do
+    no_pause = enqueue_store_result_job("no pause")
+    pause = enqueue_store_result_job("pause", pause: 0.2.seconds)
+
+    signal_process(@pid, :INT, wait: 0.3.second)
+    wait_for_jobs_to_finish_for(2.second)
+
+    assert_completed_job_results("no pause")
+    assert_completed_job_results("pause")
+
+    assert_job_status(no_pause, :finished)
+    assert_job_status(pause, :finished)
+
+    wait_for_process_termination_with_timeout(@pid, timeout: 1.second)
+    assert_clean_termination
+  end
+
+  test "term supervisor exceeding timeout while there are jobs in-flight" do
+    no_pause = enqueue_store_result_job("no pause")
+    pause = enqueue_store_result_job("pause", pause: SolidQueue.shutdown_timeout + 10.second)
+
+    # Wait for the "no pause" job to complete and the pause job to be claimed.
+    # This ensures the pause job is actively being processed.
+    wait_for_jobs_to_finish_for(3.seconds, except: pause)
+    wait_for(timeout: 2.seconds) { SolidQueue::ClaimedExecution.exists?(job_id: SolidQueue::Job.find_by(active_job_id: pause.job_id)&.id) }
+
+    signal_process(@pid, :TERM, wait: 0.2.second)
+    wait_for_jobs_to_finish_for(2.seconds, except: pause)
+
+    # Wait for process to terminate. In async mode, shutdown_timeout is used by both
+    # the supervisor and workers, creating a race: exit status may be 0 (graceful) or
+    # 1 (exit!) depending on which timeout check happens first.
+    wait_for_process_termination_with_timeout(@pid, timeout: SolidQueue.shutdown_timeout + 5.seconds, exitstatus: nil)
+    assert_not process_exists?(@pid)
+
+    assert_completed_job_results("no pause")
+    assert_job_status(no_pause, :finished)
+
+    # The pause job should have started but not completed
+    assert_started_job_result("pause")
+    assert_not_equal "completed", skip_active_record_query_cache { JobResult.find_by(value: "pause")&.status }
+
+    # 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
+  end
+
+  test "process some jobs that raise errors" do
+    2.times { enqueue_store_result_job("no error", :background) }
+    2.times { enqueue_store_result_job("no error", :default) }
+    error1 = enqueue_store_result_job("error", :background, exception: ExpectedTestError)
+    enqueue_store_result_job("no error", :background, pause: 0.03)
+    error2 = enqueue_store_result_job("error", :background, exception: ExpectedTestError, pause: 0.05)
+    2.times { enqueue_store_result_job("no error", :default, pause: 0.01) }
+    error3 = enqueue_store_result_job("error", :default, exception: ExpectedTestError)
+
+    wait_for_jobs_to_finish_for(2.second, except: [ error1, error2, error3 ])
+
+    assert_completed_job_results("no error", :background, 3)
+    assert_completed_job_results("no error", :default, 4)
+
+    wait_while_with_timeout(1.second) { SolidQueue::FailedExecution.count < 3 }
+    [ error1, error2, error3 ].each do |job|
+      assert_job_status(job, :failed)
+    end
+
+    terminate_process(@pid)
+    assert_clean_termination
+  end
+
+
+  private
+    def assert_clean_termination
+      wait_for_registered_processes 0, timeout: 0.2.second
+      assert_no_registered_processes
+      assert_no_claimed_jobs
+      assert_not process_exists?(@pid)
+    end
+
+    def assert_registered_workers_for(*queues, supervisor_pid: nil)
+      workers = find_processes_registered_as("Worker")
+      registered_queues = workers.map { |process| process.metadata["queues"] }.compact
+      assert_equal queues.map(&:to_s).sort, registered_queues.sort
+      if supervisor_pid
+        assert_equal [ supervisor_pid ], workers.map { |process| process.supervisor.pid }.uniq
+      end
+    end
+
+    def assert_registered_supervisor
+      processes = find_processes_registered_as("Supervisor(async)")
+      assert_equal 1, processes.count
+      assert_equal @pid, processes.first.pid
+    end
+
+    def assert_no_registered_workers
+      assert_empty find_processes_registered_as("Worker").to_a
+    end
+
+    def enqueue_store_result_job(value, queue_name = :background, **options)
+      StoreResultJob.set(queue: queue_name).perform_later(value, **options)
+    end
+
+    def assert_completed_job_results(value, queue_name = :background, count = 1)
+      skip_active_record_query_cache do
+        assert_equal count, JobResult.where(queue_name: queue_name, status: "completed", value: value).count
+      end
+    end
+
+    def assert_started_job_result(value, queue_name = :background, count = 1)
+      skip_active_record_query_cache do
+        assert_equal count, JobResult.where(queue_name: queue_name, status: "started", value: value).count
+      end
+    end
+
+    def assert_job_status(active_job, status)
+      skip_active_record_query_cache do
+        job = SolidQueue::Job.find_by(active_job_id: active_job.job_id)
+        assert job.public_send("#{status}?")
+      end
+    end
+end
diff --git a/test/integration/concurrency_controls_test.rb b/test/integration/concurrency_controls_test.rb
index 178c796d8..a12c48e42 100644
--- a/test/integration/concurrency_controls_test.rb
+++ b/test/integration/concurrency_controls_test.rb
@@ -36,14 +36,16 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   test "schedule several conflicting jobs over the same record sequentially" do
-    UpdateResultJob.set(wait: 0.23.seconds).perform_later(@result, name: "000", pause: 0.1.seconds)
+    # Writes to @result at 0.4s
+    UpdateResultJob.set(wait: 0.2.seconds).perform_later(@result, name: "000", pause: 0.2.seconds)
 
     ("A".."F").each_with_index do |name, i|
-      NonOverlappingUpdateResultJob.set(wait: (0.2 + i * 0.01).seconds).perform_later(@result, name: name, pause: 0.3.seconds)
+      # "A" is enqueued at 0.2s and writes to @result at 0.6s, the write at 0.4s gets overwritten
+      NonOverlappingUpdateResultJob.set(wait: (0.2 + i * 0.1).seconds).perform_later(@result, name: name, pause: 0.4.seconds)
     end
 
     ("G".."K").each_with_index do |name, i|
-      NonOverlappingUpdateResultJob.set(wait: (0.3 + i * 0.01).seconds).perform_later(@result, name: name)
+      NonOverlappingUpdateResultJob.set(wait: (1 + i * 0.1).seconds).perform_later(@result, name: name)
     end
 
     wait_for_jobs_to_finish_for(5.seconds)
@@ -176,8 +178,9 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   test "verify transactions remain valid after Job creation conflicts via limits_concurrency" do
-    # Doesn't work with enqueue_after_transaction_commit? true on SolidQueueAdapter, but only Rails 7.2 uses this
-    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 2
+    # Doesn't work when enqueue_after_transaction_commit is enabled
+    skip if ActiveJob::Base.respond_to?(:enqueue_after_transaction_commit) &&
+            [ true, :default ].include?(ActiveJob::Base.enqueue_after_transaction_commit)
 
     ActiveRecord::Base.transaction do
       NonOverlappingUpdateResultJob.perform_later(@result, name: "A", pause: 0.2.seconds)
@@ -194,6 +197,8 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
 
   test "discard jobs when concurrency limit is reached with on_conflict: :discard" do
     job1 = DiscardableUpdateResultJob.perform_later(@result, name: "1", pause: 3)
+    sleep(0.1)
+
     # should be discarded due to concurrency limit
     job2 = DiscardableUpdateResultJob.perform_later(@result, name: "2")
     # should also be discarded
@@ -216,8 +221,9 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
 
   test "discard on conflict across different concurrency keys" do
     another_result = JobResult.create!(queue_name: "default", status: "")
-    DiscardableUpdateResultJob.perform_later(@result, name: "1", pause: 0.2)
-    DiscardableUpdateResultJob.perform_later(another_result, name: "2", pause: 0.2)
+    DiscardableUpdateResultJob.perform_later(@result, name: "1", pause: 2)
+    DiscardableUpdateResultJob.perform_later(another_result, name: "2", pause: 2)
+    sleep(0.1)
     DiscardableUpdateResultJob.perform_later(@result, name: "3") # Should be discarded
     DiscardableUpdateResultJob.perform_later(another_result, name: "4") # Should be discarded
 
diff --git a/test/integration/processes_lifecycle_test.rb b/test/integration/forked_processes_lifecycle_test.rb
similarity index 92%
rename from test/integration/processes_lifecycle_test.rb
rename to test/integration/forked_processes_lifecycle_test.rb
index a86b2f15f..40495c20a 100644
--- a/test/integration/processes_lifecycle_test.rb
+++ b/test/integration/forked_processes_lifecycle_test.rb
@@ -2,7 +2,7 @@
 
 require "test_helper"
 
-class ProcessesLifecycleTest < ActiveSupport::TestCase
+class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
@@ -58,13 +58,14 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
       signal_process(@pid, :TERM, wait: 0.1.second)
     end
 
-    sleep(1.second)
+    wait_while_with_timeout(SolidQueue.shutdown_timeout + 1.second) { process_exists?(@pid) }
     assert_clean_termination
   end
 
   test "quit supervisor while there are jobs in-flight" do
     no_pause = enqueue_store_result_job("no pause")
-    pause = enqueue_store_result_job("pause", pause: 1.second)
+    # long enough pause to make sure it doesn't finish
+    pause = enqueue_store_result_job("pause", pause: 60.second)
 
     wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 0 }
 
@@ -87,7 +88,7 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
 
   test "term supervisor while there are jobs in-flight" do
     no_pause = enqueue_store_result_job("no pause")
-    pause = enqueue_store_result_job("pause", pause: 0.2.seconds)
+    pause = enqueue_store_result_job("pause", pause: 1.second)
 
     signal_process(@pid, :TERM, wait: 0.3.second)
     wait_for_jobs_to_finish_for(3.seconds)
@@ -104,10 +105,10 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
 
   test "int supervisor while there are jobs in-flight" do
     no_pause = enqueue_store_result_job("no pause")
-    pause = enqueue_store_result_job("pause", pause: 0.2.seconds)
+    pause = enqueue_store_result_job("pause", pause: 1.second)
 
     signal_process(@pid, :INT, wait: 0.3.second)
-    wait_for_jobs_to_finish_for(2.second)
+    wait_for_jobs_to_finish_for(3.second)
 
     assert_completed_job_results("no pause")
     assert_completed_job_results("pause")
@@ -121,14 +122,14 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
 
   test "term supervisor exceeding timeout while there are jobs in-flight" do
     no_pause = enqueue_store_result_job("no pause")
-    pause = enqueue_store_result_job("pause", pause: SolidQueue.shutdown_timeout + 10.second)
+    pause = enqueue_store_result_job("pause", pause: SolidQueue.shutdown_timeout + 10.seconds)
 
     wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 1 }
 
     signal_process(@pid, :TERM, wait: 0.5.second)
     wait_for_jobs_to_finish_for(2.seconds, except: pause)
 
-    wait_while_with_timeout!(SolidQueue.shutdown_timeout + 1.second) { process_exists?(@pid) }
+    wait_while_with_timeout!(SolidQueue.shutdown_timeout + 5.seconds) { process_exists?(@pid) }
     assert_not process_exists?(@pid)
 
     assert_completed_job_results("no pause")
@@ -170,10 +171,13 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
       enqueue_store_result_job("no exit", :background)
       enqueue_store_result_job("no exit", :default)
     end
-    enqueue_store_result_job("paused no exit", :default, pause: 0.5)
-    exit_job = enqueue_store_result_job("exit", :background, exit_value: 1, pause: 0.2)
-    pause_job = enqueue_store_result_job("exit", :background, pause: 0.3)
+    enqueue_store_result_job("paused no exit", :default, pause: 0.5.second)
+    # the worker for :background queue will exit abnormally
+    exit_job = enqueue_store_result_job("exit", :background, exit_value: 1, pause: 0.5.second)
+    # this will run *after* exit_job (pause: 1.second) - should also be marked as failed
+    pause_job = enqueue_store_result_job("exit", :background, pause: 1.second)
 
+    # this will run *before* exit_job (no pause) - should complete normally
     2.times { enqueue_store_result_job("no exit", :background) }
 
     wait_for_jobs_to_finish_for(3.seconds, except: [ exit_job, pause_job ])
@@ -182,12 +186,6 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
     assert_completed_job_results("no exit", :background, 4)
     assert_completed_job_results("paused no exit", :default, 1)
 
-    # The background worker exits because of the exit job,
-    # leaving the pause job claimed
-    [ exit_job, pause_job ].each do |job|
-      assert_job_status(job, :claimed)
-    end
-
     assert process_exists?(@pid)
     terminate_process(@pid)
 
@@ -234,9 +232,11 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
   end
 
   test "kill worker individually" do
-    killed_pause = enqueue_store_result_job("killed_pause", pause: 1.second)
+    killed_pause = enqueue_store_result_job("killed_pause", pause: 2.seconds)
     enqueue_store_result_job("pause", :default, pause: 0.5.seconds)
 
+    wait_for_jobs_to_finish_for(1.second, except: [ killed_pause ])
+
     worker = find_processes_registered_as("Worker").detect { |process| process.metadata["queues"].include? "background" }
     signal_process(worker.pid, :KILL, wait: 0.5.seconds)
 
@@ -284,7 +284,7 @@ def assert_registered_workers_for(*queues, supervisor_pid: nil)
     end
 
     def assert_registered_supervisor_with(pid)
-      processes = find_processes_registered_as("Supervisor")
+      processes = find_processes_registered_as("Supervisor(fork)")
       assert_equal 1, processes.count
       assert_equal pid, processes.first.pid
     end
diff --git a/test/integration/instrumentation_test.rb b/test/integration/instrumentation_test.rb
index 046700d0c..1822cf159 100644
--- a/test/integration/instrumentation_test.rb
+++ b/test/integration/instrumentation_test.rb
@@ -53,7 +53,7 @@ class InstrumentationTest < ActiveSupport::TestCase
   end
 
   test "stopping a worker with claimed executions emits release_claimed events" do
-    StoreResultJob.perform_later(42, pause: SolidQueue.shutdown_timeout + 100.second)
+    StoreResultJob.perform_later(42, pause: SolidQueue.shutdown_timeout + 15.seconds)
     process = nil
 
     events = subscribed(/release.*_claimed\.solid_queue/) do
@@ -88,7 +88,7 @@ class InstrumentationTest < ActiveSupport::TestCase
   end
 
   test "starting and stopping a worker emits register_process and deregister_process events" do
-    StoreResultJob.perform_later(42, pause: SolidQueue.shutdown_timeout + 100.second)
+    StoreResultJob.perform_later(42, pause: SolidQueue.shutdown_timeout + 15.seconds)
     process = nil
 
     events = subscribed(/(register|deregister)_process\.solid_queue/) do
@@ -166,7 +166,7 @@ class InstrumentationTest < ActiveSupport::TestCase
     SolidQueue::Process.any_instance.expects(:destroy!).raises(error).at_least_once
 
     events = subscribed("deregister_process.solid_queue") do
-      assert_raises RuntimeError do
+      assert_raises ExpectedTestError do
         worker = SolidQueue::Worker.new.tap(&:start)
         wait_for_registered_processes(1, timeout: 1.second)
 
@@ -353,7 +353,7 @@ class InstrumentationTest < ActiveSupport::TestCase
 
     events = subscribed("enqueue_recurring_task.solid_queue") do
       scheduler.start
-      sleep(1.01)
+      sleep(1.5)
       scheduler.stop
     end
 
@@ -375,7 +375,7 @@ class InstrumentationTest < ActiveSupport::TestCase
 
     events = subscribed("enqueue_recurring_task.solid_queue") do
       scheduler.start
-      sleep(1.01)
+      sleep(1.5)
       scheduler.stop
     end
 
diff --git a/test/integration/jobs_lifecycle_test.rb b/test/integration/jobs_lifecycle_test.rb
index decab5b09..8444c3759 100644
--- a/test/integration/jobs_lifecycle_test.rb
+++ b/test/integration/jobs_lifecycle_test.rb
@@ -3,6 +3,8 @@
 require "test_helper"
 
 class JobsLifecycleTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
   setup do
     @_on_thread_error = SolidQueue.on_thread_error
     SolidQueue.on_thread_error = silent_on_thread_error_for([ ExpectedTestError, RaisingJob::DefaultError ], @_on_thread_error)
@@ -34,7 +36,6 @@ class JobsLifecycleTest < ActiveSupport::TestCase
   test "enqueue and run jobs that fail without retries" do
     RaisingJob.perform_later(ExpectedTestError, "A")
     RaisingJob.perform_later(ExpectedTestError, "B")
-    jobs = SolidQueue::Job.last(2)
 
     @dispatcher.start
     @worker.start
diff --git a/test/integration/lifecycle_hooks_test.rb b/test/integration/lifecycle_hooks_test.rb
index da7feedce..7cd04a82d 100644
--- a/test/integration/lifecycle_hooks_test.rb
+++ b/test/integration/lifecycle_hooks_test.rb
@@ -87,7 +87,7 @@ class LifecycleHooksTest < ActiveSupport::TestCase
       end
 
       assert_equal %w[
-        supervisor_start supervisor_stop supervisor_exit
+        forksupervisor_start forksupervisor_stop forksupervisor_exit
         worker_first_queue_start worker_first_queue_stop worker_first_queue_exit
         worker_second_queue_start worker_second_queue_stop worker_second_queue_exit
         dispatcher_100_start dispatcher_100_stop dispatcher_100_exit
diff --git a/test/integration/puma/plugin_async_test.rb b/test/integration/puma/plugin_async_test.rb
new file mode 100644
index 000000000..551ebd634
--- /dev/null
+++ b/test/integration/puma/plugin_async_test.rb
@@ -0,0 +1,13 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require_relative "plugin_testing"
+
+class PluginAsyncTest < ActiveSupport::TestCase
+  include PluginTesting
+
+  private
+    def solid_queue_mode
+      :async
+    end
+end
diff --git a/test/integration/puma/plugin_fork_test.rb b/test/integration/puma/plugin_fork_test.rb
new file mode 100644
index 000000000..40f1fd18e
--- /dev/null
+++ b/test/integration/puma/plugin_fork_test.rb
@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require_relative "plugin_testing"
+
+class PluginForkTest < ActiveSupport::TestCase
+  include PluginTesting
+
+  test "stop puma when solid queue's supervisor dies" do
+    supervisor = find_processes_registered_as("Supervisor(fork)").first
+
+    signal_process(supervisor.pid, :KILL)
+    wait_for_process_termination_with_timeout(@pid)
+
+    assert_not process_exists?(@pid)
+
+    # When the supervisor is KILLed, the forked processes become orphans.
+    # Clean them up manually.
+    SolidQueue::Process.all.each do |process|
+      signal_process(process.pid, :KILL) if process_exists?(process.pid)
+    end
+
+    wait_for_registered_processes 0, timeout: 3.second
+  end
+
+  private
+    def solid_queue_mode
+      :fork
+    end
+end
diff --git a/test/integration/puma/plugin_test.rb b/test/integration/puma/plugin_test.rb
deleted file mode 100644
index bac98a2bb..000000000
--- a/test/integration/puma/plugin_test.rb
+++ /dev/null
@@ -1,63 +0,0 @@
-# frozen_string_literal: true
-
-require "test_helper"
-
-class PluginTest < ActiveSupport::TestCase
-  self.use_transactional_tests = false
-
-  setup do
-    FileUtils.mkdir_p Rails.root.join("tmp", "pids")
-    Dir.chdir("test/dummy") do
-      cmd = %W[
-        bundle exec puma
-          -b tcp://127.0.0.1:9222
-          -C config/puma.rb
-          -s
-          config.ru
-      ]
-      @pid = fork do
-        exec(*cmd)
-      end
-    end
-    wait_for_registered_processes 5, timeout: 3.second
-  end
-
-  teardown do
-    terminate_process(@pid, signal: :INT) if process_exists?(@pid)
-    wait_for_registered_processes 0, timeout: 2.seconds
-  end
-
-  test "perform jobs inside puma's process" do
-    StoreResultJob.perform_later(:puma_plugin)
-
-    wait_for_jobs_to_finish_for(2.seconds)
-    assert_equal 1, JobResult.where(queue_name: :background, status: "completed", value: :puma_plugin).count
-  end
-
-  test "stop the queue on puma's restart" do
-    signal_process(@pid, :SIGUSR2)
-    # Ensure the restart finishes before we try to continue with the test
-    wait_for_registered_processes(0, timeout: 3.second)
-    wait_for_registered_processes(5, timeout: 3.second)
-
-    StoreResultJob.perform_later(:puma_plugin)
-    wait_for_jobs_to_finish_for(2.seconds)
-    assert_equal 1, JobResult.where(queue_name: :background, status: "completed", value: :puma_plugin).count
-  end
-
-  test "stop puma when solid queue's supervisor dies" do
-    supervisor = find_processes_registered_as("Supervisor").first
-
-    signal_process(supervisor.pid, :KILL)
-    wait_for_process_termination_with_timeout(@pid)
-
-    assert_not process_exists?(@pid)
-
-    # Make sure all supervised processes are also terminated
-    SolidQueue::Process.all.each do |process|
-      signal_process(process.pid, :KILL) if process_exists?(process.pid)
-    end
-
-    wait_for_registered_processes 0, timeout: 3.second
-  end
-end
diff --git a/test/integration/puma/plugin_testing.rb b/test/integration/puma/plugin_testing.rb
new file mode 100644
index 000000000..14165c9b8
--- /dev/null
+++ b/test/integration/puma/plugin_testing.rb
@@ -0,0 +1,60 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+module PluginTesting
+  extend ActiveSupport::Concern
+  extend ActiveSupport::Testing::Declarative
+
+  included do
+    self.use_transactional_tests = false
+
+    setup do
+      FileUtils.mkdir_p Rails.root.join("tmp", "pids")
+
+      Dir.chdir("test/dummy") do
+        cmd = %W[
+          bundle exec puma
+            -b tcp://127.0.0.1:9222
+            -C config/puma_#{solid_queue_mode}.rb
+            -s
+            config.ru
+        ]
+
+        @pid = fork do
+          exec(*cmd)
+        end
+      end
+
+      wait_for_registered_processes(5, timeout: 5.second)
+    end
+
+    teardown do
+      terminate_process(@pid, signal: :INT) if process_exists?(@pid)
+      wait_for_registered_processes(0, timeout: 5.seconds)
+    end
+  end
+
+  test "perform jobs inside puma's process" do
+    StoreResultJob.perform_later(:puma_plugin)
+
+    wait_for_jobs_to_finish_for(5.seconds)
+    assert_equal 1, JobResult.where(queue_name: :background, status: "completed", value: :puma_plugin).count
+  end
+
+  test "stop the queue on puma's restart" do
+    signal_process(@pid, :SIGUSR2)
+    # Ensure the restart finishes before we try to continue with the test
+    wait_for_registered_processes(0, timeout: 5.second)
+    wait_for_registered_processes(5, timeout: 5.second)
+
+    StoreResultJob.perform_later(:puma_plugin)
+    wait_for_jobs_to_finish_for(2.seconds)
+    assert_equal 1, JobResult.where(queue_name: :background, status: "completed", value: :puma_plugin).count
+  end
+
+  private
+    def solid_queue_mode
+      raise NotImplementedError
+    end
+end
diff --git a/test/integration/recurring_tasks_test.rb b/test/integration/recurring_tasks_test.rb
index 7367bc062..f2fc7145f 100644
--- a/test/integration/recurring_tasks_test.rb
+++ b/test/integration/recurring_tasks_test.rb
@@ -6,26 +6,17 @@ class RecurringTasksTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
-    @pid = run_supervisor_as_fork(skip_recurring: false)
-    # 1 supervisor + 2 workers + 1 dispatcher + 1 scheduler
-    wait_for_registered_processes(5, timeout: 3.second)
+    @pid = run_recurring_supervisor
   end
 
   teardown do
-    terminate_process(@pid) if process_exists?(@pid)
-
-    SolidQueue::Process.destroy_all
-    SolidQueue::Job.destroy_all
-    SolidQueue::RecurringTask.delete_all
-    JobResult.delete_all
+    terminate_gracefully(@pid)
   end
 
   test "enqueue and process periodic tasks" do
     wait_for_jobs_to_be_enqueued(2, timeout: 2.5.seconds)
     wait_for_jobs_to_finish_for(2.5.seconds)
 
-    terminate_process(@pid)
-
     skip_active_record_query_cache do
       assert SolidQueue::Job.count >= 2
       SolidQueue::Job.all.each do |job|
@@ -43,36 +34,31 @@ class RecurringTasksTest < ActiveSupport::TestCase
 
   test "persist and delete configured tasks" do
     configured_task = { periodic_store_result: { class: "StoreResultJob", schedule: "every second" } }
-    # Wait for concurrency schedule loading after process registration
-    sleep(0.5)
 
     assert_recurring_tasks configured_task
-    terminate_process(@pid)
-
     task = SolidQueue::RecurringTask.find_by(key: "periodic_store_result")
     task.update!(class_name: "StoreResultJob", schedule: "every minute", arguments: [ 42 ])
 
-    @pid = run_supervisor_as_fork(skip_recurring: false)
-    wait_for_registered_processes(5, timeout: 3.second)
+    terminate_gracefully(@pid)
 
-    # Wait for concurrency schedule loading after process registration
-    sleep(0.5)
+    @pid = run_recurring_supervisor
 
     assert_recurring_tasks configured_task
 
     another_task = { example_task: { class: "AddToBufferJob", schedule: "every hour", args: [ 42 ] } }
     scheduler1 = SolidQueue::Scheduler.new(recurring_tasks: another_task).tap(&:start)
-    wait_for_registered_processes(6, timeout: 1.second)
+    wait_for_registered_processes(6, timeout: 2.seconds)
+    wait_for { SolidQueue::RecurringTask.find_by(key: "example_task").present? }
 
     assert_recurring_tasks another_task
 
     updated_task = { example_task: { class: "AddToBufferJob", schedule: "every minute" } }
     scheduler2 = SolidQueue::Scheduler.new(recurring_tasks: updated_task).tap(&:start)
-    wait_for_registered_processes(7, timeout: 1.second)
+    wait_for_registered_processes(7, timeout: 2.seconds)
+    wait_for { SolidQueue::RecurringTask.find_by(key: "example_task")&.schedule == "every minute" }
 
     assert_recurring_tasks updated_task
 
-    terminate_process(@pid)
     scheduler1.stop
     scheduler2.stop
   end
@@ -92,4 +78,18 @@ def assert_recurring_tasks(expected_tasks)
         end
       end
     end
+
+    def run_recurring_supervisor
+      pid = run_supervisor_as_fork(skip_recurring: false)
+      wait_for_registered_processes(5, timeout: 3.seconds) # 1 supervisor + 2 workers + 1 dispatcher + 1 scheduler
+      sleep 1.second # Wait for concurrency schedule loading after process registration
+      pid
+    end
+
+    def terminate_gracefully(pid)
+      return if pid.nil? || !process_exists?(pid)
+
+      terminate_process(pid)
+      wait_for_registered_processes(0, timeout: SolidQueue.shutdown_timeout)
+    end
 end
diff --git a/test/models/solid_queue/claimed_execution_test.rb b/test/models/solid_queue/claimed_execution_test.rb
index 98513c94e..dd1088c90 100644
--- a/test/models/solid_queue/claimed_execution_test.rb
+++ b/test/models/solid_queue/claimed_execution_test.rb
@@ -73,6 +73,60 @@ class SolidQueue::ClaimedExecutionTest < ActiveSupport::TestCase
     assert job.reload.failed?
   end
 
+  test "fail with error when a failed execution already exists updates the existing one" do
+    claimed_execution = prepare_and_claim_job AddToBufferJob.perform_later(42)
+    job = claimed_execution.job
+
+    # Simulate corrupted state: a failed execution already exists for this job
+    SolidQueue::FailedExecution.create!(job_id: job.id, exception: RuntimeError.new("old error"))
+
+    assert_no_difference -> { SolidQueue::FailedExecution.count } do
+      assert_difference -> { SolidQueue::ClaimedExecution.count }, -1 do
+        claimed_execution.failed_with(RuntimeError.new("new error"))
+      end
+    end
+
+    assert_equal "new error", job.failed_execution.message
+  end
+
+  test "perform job with missing class fails gracefully" do
+    job = create_job_with_missing_class
+    claimed_execution = claim_job(job)
+
+    assert_difference -> { SolidQueue::ClaimedExecution.count } => -1, -> { SolidQueue::FailedExecution.count } => 1 do
+      assert_raises NameError do
+        claimed_execution.perform
+      end
+    end
+
+    assert job.reload.failed?
+  end
+
+  test "perform concurrency-controlled job with missing class fails gracefully" do
+    job = create_job_with_missing_class(concurrency_key: "test_key")
+    claimed_execution = claim_job(job)
+
+    assert_difference -> { SolidQueue::ClaimedExecution.count } => -1, -> { SolidQueue::FailedExecution.count } => 1 do
+      assert_raises NameError do
+        claimed_execution.perform
+      end
+    end
+
+    assert job.reload.failed?
+  end
+
+  test "dispatch job with missing class and concurrency key skips concurrency controls" do
+    job = create_job_with_missing_class(concurrency_key: "test_key")
+
+    assert_not job.concurrency_limited?
+
+    job.prepare_for_execution
+
+    assert job.reload.ready?
+    assert_equal 0, SolidQueue::BlockedExecution.where(job_id: job.id).count
+    assert_equal 0, SolidQueue::Semaphore.where(key: "test_key").count
+  end
+
   test "provider_job_id is available within job execution" do
     job = ProviderJobIdJob.perform_later
     claimed_execution = prepare_and_claim_job job
@@ -84,8 +138,22 @@ class SolidQueue::ClaimedExecutionTest < ActiveSupport::TestCase
   private
     def prepare_and_claim_job(active_job, process: @process)
       job = SolidQueue::Job.find_by(active_job_id: active_job.job_id)
-
       job.prepare_for_execution
+      claim_job(job, process: process)
+    end
+
+    def create_job_with_missing_class(concurrency_key: nil)
+      SolidQueue::Job.create!(
+        queue_name: "background",
+        class_name: "RemovedJobClass",
+        active_job_id: SecureRandom.uuid,
+        arguments: { "job_class" => "RemovedJobClass", "arguments" => [] },
+        concurrency_key: concurrency_key,
+        scheduled_at: Time.current
+      )
+    end
+
+    def claim_job(job, process: @process)
       assert_difference -> { SolidQueue::ClaimedExecution.count } => +1 do
         SolidQueue::ReadyExecution.claim(job.queue_name, 1, process.id)
       end
diff --git a/test/models/solid_queue/job_test.rb b/test/models/solid_queue/job_test.rb
index 201e6d720..47702bd18 100644
--- a/test/models/solid_queue/job_test.rb
+++ b/test/models/solid_queue/job_test.rb
@@ -340,8 +340,9 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob
   end
 
   test "enqueue successfully inside a rolled-back transaction in the app DB" do
-    # Doesn't work with enqueue_after_transaction_commit? true on SolidQueueAdapter, but only Rails 7.2 uses this
-    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 2
+    # Doesn't work when enqueue_after_transaction_commit is enabled
+    skip if ActiveJob::Base.respond_to?(:enqueue_after_transaction_commit) &&
+            [ true, :default ].include?(ActiveJob::Base.enqueue_after_transaction_commit)
     assert_difference -> { SolidQueue::Job.count } do
       assert_no_difference -> { JobResult.count } do
         JobResult.transaction do
diff --git a/test/models/solid_queue/process_test.rb b/test/models/solid_queue/process_test.rb
index 489b2aca7..cd2430caf 100644
--- a/test/models/solid_queue/process_test.rb
+++ b/test/models/solid_queue/process_test.rb
@@ -1,5 +1,4 @@
 require "test_helper"
-require "minitest/mock"
 
 class SolidQueue::ProcessTest < ActiveSupport::TestCase
   test "prune processes with expired heartbeats" do
@@ -35,7 +34,7 @@ class SolidQueue::ProcessTest < ActiveSupport::TestCase
   end
 
   test "prune processes including their supervisor with expired heartbeats and fail claimed executions" do
-    supervisor = SolidQueue::Process.register(kind: "Supervisor", pid: 42, name: "supervisor-42")
+    supervisor = SolidQueue::Process.register(kind: "Supervisor(fork)", pid: 42, name: "supervisor-42")
     process = SolidQueue::Process.register(kind: "Worker", pid: 43, name: "worker-43", supervisor_id: supervisor.id)
     3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
     jobs = SolidQueue::Job.last(3)
@@ -58,16 +57,16 @@ class SolidQueue::ProcessTest < ActiveSupport::TestCase
 
   test "hostname's with special characters are properly loaded" do
     worker = SolidQueue::Worker.new(queues: "*", threads: 3, polling_interval: 0.2)
-    hostname = "Basecamp’s-Computer"
+    hostname = "Basecamp's-Computer"
 
-    Socket.stub :gethostname, hostname.dup.force_encoding("ASCII-8BIT") do
-      worker.start
-      wait_for_registered_processes(1, timeout: 1.second)
+    Socket.stubs(:gethostname).returns(hostname.dup.force_encoding("ASCII-8BIT"))
 
-      assert_equal hostname, SolidQueue::Process.last.hostname
+    worker.start
+    wait_for_registered_processes(1, timeout: 1.second)
 
-      worker.stop
-    end
+    assert_equal hostname, SolidQueue::Process.last.hostname
+
+    worker.stop
   end
 
   test "clear unregistered changes before locking for heartbeat" do
diff --git a/test/models/solid_queue/recurring_task_test.rb b/test/models/solid_queue/recurring_task_test.rb
index ec31dd69d..dba9d6b91 100644
--- a/test/models/solid_queue/recurring_task_test.rb
+++ b/test/models/solid_queue/recurring_task_test.rb
@@ -142,6 +142,25 @@ def perform
 
     # Empty schedule
     assert_not SolidQueue::RecurringTask.new(key: "task-id", class_name: "SolidQueue::RecurringTaskTest::JobWithoutArguments").valid?
+
+    task = recurring_task_with(class_name: "JobWithoutArguments", schedule: "every day at 00:40")
+    assert task.valid?
+    assert task.to_s.ends_with? "[ 40 0 * * * ]"
+
+    task = recurring_task_with(class_name: "JobWithoutArguments", schedule: "40 0 * * *")
+    assert task.valid?
+
+    # Single cron to represent both hours, as it's the same minute
+    task = recurring_task_with(class_name: "JobWithoutArguments", schedule: "every day at 00:40 and 15:40")
+    assert task.valid?
+    assert task.to_s.ends_with? "[ 40 0,15 * * * ]"
+
+    # Would need two cron tabs to reprenset the different hours and minutes
+    task = recurring_task_with(class_name: "JobWithoutArguments", schedule: "every day at 00:40 and 15:20")
+    assert_not task.valid?
+    error_message = task.errors[:schedule].first
+    assert_includes error_message, "generates multiple cron schedules"
+    assert_includes error_message, "Please use separate recurring tasks"
   end
 
   test "valid and invalid job class and command" do
@@ -154,7 +173,9 @@ def perform
     assert_not recurring_task_with(class_name: "UnknownJob").valid?
 
     # Empty class name and command
-    assert_not recurring_task_with(key: "task-id", schedule: "every minute").valid?
+    task = SolidQueue::RecurringTask.from_configuration("task-id", schedule: "every minute")
+    assert_not task.valid?
+    assert_includes task.errors[:base], "either command or class must be present"
   end
 
   test "task with custom queue and priority" do
@@ -189,6 +210,56 @@ def perform
     assert_equal "SolidQueue::RecurringJob", SolidQueue::Job.last.class_name
   end
 
+  test "schedule recurring tasks dynamically" do
+    SolidQueue::RecurringTask.create_dynamic_task("test 1", command: "puts 1", schedule: "every hour")
+    SolidQueue::RecurringTask.create_dynamic_task("test 2", command: "puts 2", schedule: "every minute", static: true)
+
+    assert SolidQueue::RecurringTask.exists?(key: "test 1", command: "puts 1", schedule: "every hour", static: false)
+    assert SolidQueue::RecurringTask.exists?(key: "test 2", command: "puts 2", schedule: "every minute", static: false)
+  end
+
+  test "schedule recurring tasks dynamically with class and args (same keys as YAML config)" do
+    SolidQueue::RecurringTask.create_dynamic_task("test 3", class: "AddToBufferJob", args: [ 42 ], schedule: "every hour")
+
+    task = SolidQueue::RecurringTask.find_by!(key: "test 3")
+    assert_equal "AddToBufferJob", task.class_name
+    assert_equal [ 42 ], task.arguments
+    assert_not task.static
+  end
+
+  test "unschedule recurring tasks dynamically" do
+    dynamic_task = SolidQueue::RecurringTask.create!(
+      key: "dynamic", command: "puts 'd'", schedule: "every day", static: false
+    )
+
+    static_task = SolidQueue::RecurringTask.create!(
+      key: "static", command: "puts 's'", schedule: "every week", static: true
+    )
+
+    SolidQueue::RecurringTask.delete_dynamic_task(dynamic_task.key)
+
+    assert_raises(ActiveRecord::RecordNotFound) do
+      SolidQueue::RecurringTask.delete_dynamic_task(static_task.key)
+    end
+
+    assert_not SolidQueue::RecurringTask.exists?(key: "dynamic", static: false)
+    assert SolidQueue::RecurringTask.exists?(key: "static", static: true)
+  end
+
+  test "scheduling dynamic recurring task with duplicate key raises error" do
+    SolidQueue::RecurringTask.create_dynamic_task("duplicate_test", command: "puts 1", schedule: "every hour")
+
+    assert_raises(ActiveRecord::RecordNotUnique) do
+      SolidQueue::RecurringTask.create_dynamic_task("duplicate_test", command: "puts 2", schedule: "every minute")
+    end
+  end
+
+  test "unscheduling dynamic recurring task with nonexistent key raises RecordNotFound" do
+    assert_raises(ActiveRecord::RecordNotFound) do
+      SolidQueue::RecurringTask.delete_dynamic_task("nonexistent_key")
+    end
+  end
+
   private
     def enqueue_and_assert_performed_with_result(task, result)
       assert_difference [ -> { SolidQueue::Job.count }, -> { SolidQueue::ReadyExecution.count } ], +1 do
diff --git a/test/models/solid_queue/semaphore_test.rb b/test/models/solid_queue/semaphore_test.rb
new file mode 100644
index 000000000..432b97afd
--- /dev/null
+++ b/test/models/solid_queue/semaphore_test.rb
@@ -0,0 +1,119 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class SolidQueue::SemaphoreTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  setup do
+    @result = JobResult.create!(queue_name: "default")
+  end
+
+  test "wait acquires a row lock that blocks concurrent signal" do
+    skip_on_sqlite
+
+    # Enqueue first job to create semaphore with value=0
+    NonOverlappingUpdateResultJob.perform_later(@result)
+    concurrency_key = SolidQueue::Job.last.concurrency_key
+    assert_equal 0, SolidQueue::Semaphore.find_by(key: concurrency_key).value
+
+    lock_held = Concurrent::Event.new
+
+    # Background thread: holds a FOR UPDATE lock on the semaphore row
+    locker = Thread.new do
+      SolidQueue::Record.connection_pool.with_connection do
+        SolidQueue::Record.transaction do
+          SolidQueue::Semaphore.where(key: concurrency_key).lock.first
+          lock_held.set
+          sleep 1
+        end
+      end
+    end
+
+    lock_held.wait(5)
+    sleep 0.1
+
+    # Main thread: this UPDATE should block until the locker's transaction commits
+    start = monotonic_now
+    SolidQueue::Semaphore.where(key: concurrency_key).update_all("value = value + 1")
+    elapsed = monotonic_now - start
+
+    locker.join(5)
+
+    assert elapsed >= 0.5, "UPDATE should have been blocked by FOR UPDATE lock (took #{elapsed.round(3)}s)"
+    assert_equal 1, SolidQueue::Semaphore.find_by(key: concurrency_key).value
+  end
+
+  test "blocked execution created during enqueue is visible to release_one after signal" do
+    skip_on_sqlite
+
+    # Enqueue first job to create semaphore with value=0
+    NonOverlappingUpdateResultJob.perform_later(@result)
+    job_a = SolidQueue::Job.last
+    concurrency_key = job_a.concurrency_key
+    assert_equal 0, SolidQueue::Semaphore.find_by(key: concurrency_key).value
+
+    lock_held = Concurrent::Event.new
+
+    # Background thread: simulates the enqueue path for a second job.
+    # Locks the semaphore row (as the code does), creates a BlockedExecution,
+    # then holds the transaction open to simulate the window where the race occurs.
+    enqueue_thread = Thread.new do
+      SolidQueue::Record.connection_pool.with_connection do
+        SolidQueue::Record.transaction do
+          # Lock the semaphore (same as Semaphore::Proxy#wait)
+          SolidQueue::Semaphore.where(key: concurrency_key).lock.first
+
+          # Create job and blocked execution bypassing after_create callbacks
+          # to avoid re-entering Semaphore.wait
+          uuid = SecureRandom.uuid
+          SolidQueue::Job.insert({
+            queue_name: "default",
+            class_name: "NonOverlappingUpdateResultJob",
+            concurrency_key: concurrency_key,
+            active_job_id: uuid,
+            arguments: "{}",
+            scheduled_at: Time.current
+          })
+          job_b_id = SolidQueue::Job.where(active_job_id: uuid).pick(:id)
+
+          SolidQueue::BlockedExecution.insert({
+            job_id: job_b_id,
+            queue_name: "default",
+            concurrency_key: concurrency_key,
+            expires_at: SolidQueue.default_concurrency_control_period.from_now,
+            priority: 0
+          })
+
+          lock_held.set
+
+          # Hold the transaction open so the signal path must wait
+          sleep 1
+        end
+      end
+    end
+
+    lock_held.wait(5)
+    sleep 0.1
+
+    # Main thread: simulates job_a finishing — signal + release_one.
+    # The signal UPDATE will block until the enqueue transaction commits,
+    # guaranteeing the BlockedExecution is visible to release_one.
+    assert SolidQueue::Semaphore.signal(job_a)
+    assert SolidQueue::BlockedExecution.release_one(concurrency_key),
+      "release_one should find the BlockedExecution created by the concurrent enqueue"
+
+    enqueue_thread.join(5)
+
+    assert_equal 0, SolidQueue::BlockedExecution.where(concurrency_key: concurrency_key).count
+  end
+
+  private
+    def skip_on_sqlite
+      skip "Row-level locking not supported on SQLite" if SolidQueue::Record.connection.adapter_name.downcase.include?("sqlite")
+    end
+
+    def monotonic_now
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+end
diff --git a/test/test_helper.rb b/test/test_helper.rb
index 7c1c87923..db5bd5c34 100644
--- a/test/test_helper.rb
+++ b/test/test_helper.rb
@@ -1,6 +1,9 @@
 # Configure Rails Environment
 ENV["RAILS_ENV"] = "test"
 
+# Ensure minitest gem is loaded before Rails tries to use minitest/mock
+require "minitest"
+
 require_relative "../test/dummy/config/environment"
 ActiveRecord::Migrator.migrations_paths = [ File.expand_path("../test/dummy/db/migrate", __dir__) ]
 require "rails/test_help"
@@ -30,10 +33,26 @@ class ExpectedTestError < RuntimeError; end
 class ActiveSupport::TestCase
   include ConfigurationTestHelper, ProcessesTestHelper, JobsTestHelper
 
+  def destroy_records
+    SolidQueue::Job.destroy_all
+    SolidQueue::Process.destroy_all
+    SolidQueue::Semaphore.delete_all
+    SolidQueue::RecurringTask.delete_all
+    SolidQueue::ScheduledExecution.delete_all
+    SolidQueue::ReadyExecution.delete_all
+    SolidQueue::ClaimedExecution.delete_all
+    SolidQueue::FailedExecution.delete_all
+    JobResult.delete_all
+  end
+
   setup do
     @_on_thread_error = SolidQueue.on_thread_error
     SolidQueue.on_thread_error = silent_on_thread_error_for(ExpectedTestError, @_on_thread_error)
     ActiveJob::QueueAdapters::SolidQueueAdapter.stopping = false
+
+    unless self.class.use_transactional_tests
+      destroy_records
+    end
   end
 
   teardown do
@@ -45,11 +64,18 @@ class ActiveSupport::TestCase
     end
 
     unless self.class.use_transactional_tests
-      SolidQueue::Job.destroy_all
-      SolidQueue::Process.destroy_all
-      SolidQueue::Semaphore.delete_all
-      SolidQueue::RecurringTask.delete_all
-      JobResult.delete_all
+      destroy_records
+    end
+  end
+
+  def run(...)
+    # Rails 8.1.dev changed default logging levels
+    if defined?(with_debug_event_reporting)
+      with_debug_event_reporting do
+        super
+      end
+    else
+      super
     end
   end
 
diff --git a/test/test_helpers/processes_test_helper.rb b/test/test_helpers/processes_test_helper.rb
index 927ddfded..01cec796c 100644
--- a/test/test_helpers/processes_test_helper.rb
+++ b/test/test_helpers/processes_test_helper.rb
@@ -70,7 +70,7 @@ def wait_for_process_termination_with_timeout(pid, timeout: 10, exitstatus: 0, s
       if process_exists?(pid)
         begin
           status = Process.waitpid2(pid).last
-          assert_equal exitstatus, status.exitstatus, "Expected pid #{pid} to exit with status #{exitstatus}" if status.exitstatus
+          assert_equal exitstatus, status.exitstatus, "Expected pid #{pid} to exit with status #{exitstatus}" if status.exitstatus && !exitstatus.nil?
           assert_equal signaled, Signal.list.key(status.termsig).to_sym, "Expected pid #{pid} to be terminated with signal #{signaled}" if status.termsig
         rescue Errno::ECHILD
           # Child pid already reaped
diff --git a/test/unit/async_supervisor_test.rb b/test/unit/async_supervisor_test.rb
new file mode 100644
index 000000000..6a0f3553b
--- /dev/null
+++ b/test/unit/async_supervisor_test.rb
@@ -0,0 +1,106 @@
+require "test_helper"
+
+class AsyncSupervisorTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  test "start as non-standalone" do
+    supervisor = run_supervisor_as_thread
+    wait_for_registered_processes(4, timeout: 3.seconds) # supervisor + dispatcher + 2 workers
+
+    assert_registered_processes(kind: "Supervisor(async)")
+    assert_registered_processes(kind: "Worker", supervisor_id: supervisor.process_id, count: 2)
+    assert_registered_processes(kind: "Dispatcher", supervisor_id: supervisor.process_id)
+  ensure
+    supervisor.stop
+    assert_no_registered_processes
+  end
+
+  test "start standalone" do
+    pid = run_supervisor_as_fork(mode: :async)
+    wait_for_registered_processes(4, timeout: 5.seconds) # supervisor + dispatcher + 2 workers
+
+    assert_registered_processes(kind: "Supervisor(async)")
+    assert_registered_processes(kind: "Worker", supervisor_pid: pid, count: 2)
+    assert_registered_processes(kind: "Dispatcher", supervisor_pid: pid)
+
+    terminate_process(pid)
+    assert_no_registered_processes
+  end
+
+  test "start as non-standalone with provided configuration" do
+    supervisor = run_supervisor_as_thread(workers: [], dispatchers: [ { batch_size: 100 } ], skip_recurring: false)
+    wait_for_registered_processes(3, timeout: 3.seconds) # supervisor + dispatcher + scheduler
+
+    assert_registered_processes(kind: "Supervisor(async)")
+    assert_registered_processes(kind: "Worker", count: 0)
+    assert_registered_processes(kind: "Dispatcher", supervisor_id: supervisor.process_id)
+    assert_registered_processes(kind: "Scheduler", supervisor_id: supervisor.process_id)
+  ensure
+    supervisor.stop
+    assert_no_registered_processes
+  end
+
+  test "failed orphaned executions as non-standalone" do
+    simulate_orphaned_executions 3
+
+    config = {
+      workers: [ { queues: "background", polling_interval: 10 } ],
+      dispatchers: []
+    }
+
+    supervisor = run_supervisor_as_thread(**config)
+    wait_for_registered_processes(2, timeout: 3.seconds) # supervisor + 1 worker
+    assert_registered_processes(kind: "Supervisor(async)")
+
+    wait_while_with_timeout(1.second) { SolidQueue::ClaimedExecution.count > 0 }
+
+    skip_active_record_query_cache do
+      assert_equal 0, SolidQueue::ClaimedExecution.count
+      assert_equal 3, SolidQueue::FailedExecution.count
+    end
+  ensure
+    supervisor.stop
+  end
+
+  test "failed orphaned executions as standalone" do
+    simulate_orphaned_executions 3
+
+    config = {
+      workers: [ { queues: "background", polling_interval: 10 } ],
+      dispatchers: []
+    }
+
+    pid = run_supervisor_as_fork(mode: :async, **config)
+    wait_for_registered_processes(2, timeout: 3.seconds) # supervisor + 1 worker
+    assert_registered_processes(kind: "Supervisor(async)")
+
+    wait_while_with_timeout(1.second) { SolidQueue::ClaimedExecution.count > 0 }
+
+    terminate_process(pid)
+
+    skip_active_record_query_cache do
+      assert_equal 0, SolidQueue::ClaimedExecution.count
+      assert_equal 3, SolidQueue::FailedExecution.count
+    end
+  end
+
+  private
+    def run_supervisor_as_thread(**options)
+      SolidQueue::Supervisor.start(mode: :async, standalone: false, **options.with_defaults(skip_recurring: true))
+    end
+
+    def simulate_orphaned_executions(count)
+      count.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
+      process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123")
+
+      SolidQueue::ReadyExecution.claim("*", count + 1, process.id)
+
+      assert_equal count, SolidQueue::ClaimedExecution.count
+      assert_equal 0, SolidQueue::ReadyExecution.count
+
+      assert_equal [ process.id ], SolidQueue::ClaimedExecution.last(3).pluck(:process_id).uniq
+
+      # Simulate orphaned executions by just wiping the claiming process
+      process.delete
+    end
+end
diff --git a/test/unit/cli_test.rb b/test/unit/cli_test.rb
new file mode 100644
index 000000000..9ee0e233a
--- /dev/null
+++ b/test/unit/cli_test.rb
@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require "solid_queue/cli"
+
+class CliTest < ActiveSupport::TestCase
+  test "mode defaults to fork when no env var or option" do
+    with_env("SOLID_QUEUE_SUPERVISOR_MODE" => nil) do
+      config = configuration_from_cli
+
+      assert config.mode.fork?
+    end
+  end
+
+  test "mode respects SOLID_QUEUE_SUPERVISOR_MODE env var" do
+    with_env("SOLID_QUEUE_SUPERVISOR_MODE" => "async") do
+      config = configuration_from_cli
+
+      assert config.mode.async?
+    end
+  end
+
+  test "mode option overrides env var" do
+    with_env("SOLID_QUEUE_SUPERVISOR_MODE" => "async") do
+      config = configuration_from_cli(mode: "fork")
+
+      assert config.mode.fork?
+    end
+  end
+
+  test "mode option works without env var" do
+    with_env("SOLID_QUEUE_SUPERVISOR_MODE" => nil) do
+      config = configuration_from_cli(mode: "async")
+
+      assert config.mode.async?
+    end
+  end
+
+  private
+    def configuration_from_cli(**cli_options)
+      cli = SolidQueue::Cli.new([], cli_options)
+      options = cli.options.symbolize_keys.compact
+
+      SolidQueue::Configuration.new(**options)
+    end
+end
diff --git a/test/unit/configuration_test.rb b/test/unit/configuration_test.rb
index 2ccaa7288..34f69658b 100644
--- a/test/unit/configuration_test.rb
+++ b/test/unit/configuration_test.rb
@@ -26,6 +26,13 @@ class ConfigurationTest < ActiveSupport::TestCase
     assert_processes configuration, :dispatcher, 1, batch_size: SolidQueue::Configuration::DISPATCHER_DEFAULTS[:batch_size]
   end
 
+  test "warns if provided configuration file does not exist" do
+    assert_output "[solid_queue] WARNING: Provided configuration file '/path/to/nowhere.yml' does not exist. Falling back to default configuration.\n" do
+      configuration = SolidQueue::Configuration.new(config_file: Pathname.new("/path/to/nowhere.yml"))
+      assert configuration.valid?
+    end
+  end
+
   test "read configuration from default file" do
     configuration = SolidQueue::Configuration.new
     assert 3, configuration.configured_processes.count
@@ -80,6 +87,24 @@ class ConfigurationTest < ActiveSupport::TestCase
     assert_has_recurring_task scheduler, key: "periodic_store_result", class_name: "StoreResultJob", schedule: "every second"
   end
 
+  test "scheduler starts with dynamic_tasks_enabled even without static tasks" do
+    configuration = SolidQueue::Configuration.new(
+      recurring_schedule_file: config_file_path(:empty_configuration),
+      scheduler: { dynamic_tasks_enabled: true }
+    )
+
+    assert_processes configuration, :scheduler, 1, dynamic_tasks_enabled: true
+  end
+
+  test "no scheduler without static tasks or dynamic_tasks_enabled" do
+    configuration = SolidQueue::Configuration.new(
+      recurring_schedule_file: config_file_path(:empty_configuration),
+      scheduler: { dynamic_tasks_enabled: false }
+    )
+
+    assert_processes configuration, :scheduler, 0
+  end
+
   test "no recurring tasks configuration when explicitly excluded" do
     configuration = SolidQueue::Configuration.new(dispatchers: [ { polling_interval: 0.1 } ], skip_recurring: true)
     assert_processes configuration, :dispatcher, 1, polling_interval: 0.1, recurring_tasks: nil
@@ -120,16 +145,20 @@ class ConfigurationTest < ActiveSupport::TestCase
     assert error.include?("periodic_invalid_class: Class name doesn't correspond to an existing class")
     assert error.include?("periodic_incorrect_schedule: Schedule is not a supported recurring schedule")
 
-    assert SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:empty_recurring)).valid?
+    assert_output(/Provided configuration file '[^']+' does not exist\./) do
+      assert SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:empty_recurring)).valid?
+    end
     assert SolidQueue::Configuration.new(skip_recurring: true).valid?
 
     configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_production_only))
     assert configuration.valid?
     assert_processes configuration, :scheduler, 0
 
-    configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_empty))
-    assert configuration.valid?
-    assert_processes configuration, :scheduler, 0
+    assert_output(/Provided configuration file '[^']+' does not exist\./) do
+      configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_empty))
+      assert configuration.valid?
+      assert_processes configuration, :scheduler, 0
+    end
 
     # No processes
     configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [])
diff --git a/test/unit/dispatcher_test.rb b/test/unit/dispatcher_test.rb
index 9aa2196ea..7df0591f5 100644
--- a/test/unit/dispatcher_test.rb
+++ b/test/unit/dispatcher_test.rb
@@ -40,8 +40,9 @@ class DispatcherTest < ActiveSupport::TestCase
     log = StringIO.new
     with_active_record_logger(ActiveSupport::Logger.new(log)) do
       with_polling(silence: false) do
+        rewind_io(log)
         @dispatcher.start
-        sleep 0.2
+        sleep 0.5.second
       end
     end
 
@@ -52,8 +53,9 @@ class DispatcherTest < ActiveSupport::TestCase
     log = StringIO.new
     with_active_record_logger(ActiveSupport::Logger.new(log)) do
       with_polling(silence: true) do
+        rewind_io(log)
         @dispatcher.start
-        sleep 0.2
+        sleep 0.5.second
       end
     end
 
@@ -64,7 +66,7 @@ class DispatcherTest < ActiveSupport::TestCase
     with_active_record_logger(nil) do
       with_polling(silence: true) do
         @dispatcher.start
-        sleep 0.2
+        sleep 0.5.second
       end
     end
 
@@ -75,8 +77,9 @@ class DispatcherTest < ActiveSupport::TestCase
 
   test "run more than one instance of the dispatcher" do
     15.times do
-      AddToBufferJob.set(wait: 0.2).perform_later("I'm scheduled")
+      AddToBufferJob.set(wait: 0.5.second).perform_later("I'm scheduled")
     end
+    sleep 0.5.second
     assert_equal 15, SolidQueue::ScheduledExecution.count
 
     another_dispatcher = SolidQueue::Dispatcher.new(polling_interval: 0.1, batch_size: 10)
@@ -86,8 +89,10 @@ class DispatcherTest < ActiveSupport::TestCase
 
     wait_while_with_timeout(1.second) { SolidQueue::ScheduledExecution.any? }
 
-    assert_equal 0, SolidQueue::ScheduledExecution.count
-    assert_equal 15, SolidQueue::ReadyExecution.count
+    skip_active_record_query_cache do
+      assert_equal 0, SolidQueue::ScheduledExecution.count
+      assert_equal 15, SolidQueue::ReadyExecution.count
+    end
   ensure
     another_dispatcher&.stop
   end
@@ -98,15 +103,19 @@ class DispatcherTest < ActiveSupport::TestCase
     dispatcher.expects(:interruptible_sleep).with(dispatcher.polling_interval).at_least_once
     dispatcher.expects(:handle_thread_error).never
 
-    3.times { AddToBufferJob.set(wait: 0.1).perform_later("I'm scheduled") }
+    3.times { AddToBufferJob.set(wait: 0.5.second).perform_later("I'm scheduled") }
+    sleep 0.5.second
     assert_equal 3, SolidQueue::ScheduledExecution.count
-    sleep 0.1
 
     dispatcher.start
     wait_while_with_timeout(1.second) { SolidQueue::ScheduledExecution.any? }
 
-    assert_equal 0, SolidQueue::ScheduledExecution.count
-    assert_equal 3, SolidQueue::ReadyExecution.count
+    skip_active_record_query_cache do
+      assert_equal 0, SolidQueue::ScheduledExecution.count
+      assert_equal 3, SolidQueue::ReadyExecution.count
+    end
+  ensure
+    dispatcher.stop
   end
 
   test "sleeps `polling_interval` between polls if there are no un-dispatched jobs" do
@@ -117,6 +126,8 @@ class DispatcherTest < ActiveSupport::TestCase
 
     dispatcher.start
     wait_while_with_timeout(1.second) { !SolidQueue::ScheduledExecution.exists? }
+  ensure
+    dispatcher.stop
   end
 
   private
@@ -133,4 +144,9 @@ def with_active_record_logger(logger)
     ensure
       ActiveRecord::Base.logger = old_logger
     end
+
+    def rewind_io(log)
+      log.truncate(0)
+      log.rewind
+    end
 end
diff --git a/test/unit/supervisor_test.rb b/test/unit/fork_supervisor_test.rb
similarity index 91%
rename from test/unit/supervisor_test.rb
rename to test/unit/fork_supervisor_test.rb
index 7a531ad2b..9ec81b510 100644
--- a/test/unit/supervisor_test.rb
+++ b/test/unit/fork_supervisor_test.rb
@@ -1,6 +1,6 @@
 require "test_helper"
 
-class SupervisorTest < ActiveSupport::TestCase
+class ForkSupervisorTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
@@ -186,8 +186,8 @@ class SupervisorTest < ActiveSupport::TestCase
   end
 
   # Regression test for supervisor failing to handle claimed jobs when its own
-  # process record has been pruned (NoMethodError in #handle_claimed_jobs_by).
-  test "handle_claimed_jobs_by fails claimed executions even if supervisor record is missing" do
+  # process record has been pruned (NoMethodError in #release_claimed_jobs_by).
+  test "release_claimed_jobs_by fails claimed executions even if supervisor record is missing" do
     worker_name = "worker-test-#{SecureRandom.hex(4)}"
 
     worker_process = SolidQueue::Process.register(kind: "Worker", pid: 999_999, name: worker_name)
@@ -196,20 +196,14 @@ class SupervisorTest < ActiveSupport::TestCase
     claimed_execution = SolidQueue::ReadyExecution.claim("*", 1, worker_process.id).first
 
     terminated_fork = Struct.new(:name).new(worker_name)
+    supervisor = SolidQueue::ForkSupervisor.allocate
+    error = RuntimeError.new
 
-    DummyStatus = Struct.new(:pid, :exitstatus) do
-      def signaled? = false
-      def termsig = nil
-    end
-    status = DummyStatus.new(worker_process.pid, 1)
-
-    supervisor = SolidQueue::Supervisor.allocate
-
-    supervisor.send(:handle_claimed_jobs_by, terminated_fork, status)
+    supervisor.send(:release_claimed_jobs_by, terminated_fork, with_error: error)
 
     failed = SolidQueue::FailedExecution.find_by(job_id: claimed_execution.job_id)
     assert failed.present?
-    assert_equal "SolidQueue::Processes::ProcessExitError", failed.exception_class
+    assert_equal "RuntimeError", failed.exception_class
   end
 
   private
@@ -223,7 +217,7 @@ def assert_registered_dispatcher(supervisor_pid: nil)
 
     def assert_registered_supervisor(pid)
       skip_active_record_query_cache do
-        processes = find_processes_registered_as("Supervisor")
+        processes = find_processes_registered_as("Supervisor(fork)")
         assert_equal 1, processes.count
         assert_nil processes.first.supervisor
         assert_equal pid, processes.first.pid
diff --git a/test/unit/process_recovery_test.rb b/test/unit/process_recovery_test.rb
index 620fbd515..296d6b958 100644
--- a/test/unit/process_recovery_test.rb
+++ b/test/unit/process_recovery_test.rb
@@ -20,10 +20,11 @@ class ProcessRecoveryTest < ActiveSupport::TestCase
     @pid = run_supervisor_as_fork(workers: [ { queues: "*", polling_interval: 0.1, processes: 1 } ])
     wait_for_registered_processes(2, timeout: 1.second) # Supervisor + 1 worker
 
-    supervisor_process = SolidQueue::Process.find_by(kind: "Supervisor", pid: @pid)
+    supervisor_process = SolidQueue::Process.find_by(kind: "Supervisor(fork)", pid: @pid)
     assert supervisor_process
 
-    worker_process = SolidQueue::Process.find_by(kind: "Worker")
+    # Find the worker supervised by this specific supervisor to avoid interference from other tests
+    worker_process = SolidQueue::Process.find_by(kind: "Worker", supervisor_id: supervisor_process.id)
     assert worker_process
 
     # Enqueue a job and wait for it to be claimed
diff --git a/test/unit/scheduler_test.rb b/test/unit/scheduler_test.rb
index 9478b9f18..e914a23ca 100644
--- a/test/unit/scheduler_test.rb
+++ b/test/unit/scheduler_test.rb
@@ -3,7 +3,7 @@
 class SchedulerTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
-  test "recurring schedule" do
+  test "recurring schedule (only static)" do
     recurring_tasks = { example_task: { class: "AddToBufferJob", schedule: "every hour", args: 42 } }
     scheduler = SolidQueue::Scheduler.new(recurring_tasks: recurring_tasks).tap(&:start)
 
@@ -17,6 +17,57 @@ class SchedulerTest < ActiveSupport::TestCase
     scheduler.stop
   end
 
+  test "recurring schedule (only dynamic)" do
+    SolidQueue::RecurringTask.create(
+      key: "dynamic_task", static: false, class_name: "AddToBufferJob", schedule: "every second", arguments: [ 42 ]
+    )
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}, dynamic_tasks_enabled: true).tap(&:start)
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    assert_equal "Scheduler", process.kind
+
+    assert_metadata process, recurring_schedule: [ "dynamic_task" ]
+  ensure
+    scheduler.stop
+  end
+
+  test "recurring schedule (static + dynamic)" do
+    SolidQueue::RecurringTask.create(
+      key: "dynamic_task", static: false, class_name: "AddToBufferJob", schedule: "every second", arguments: [ 42 ]
+    )
+
+    recurring_tasks = { static_task: { class: "AddToBufferJob", schedule: "every hour", args: 42 } }
+
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: recurring_tasks, dynamic_tasks_enabled: true).tap(&:start)
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    assert_equal "Scheduler", process.kind
+
+    assert_metadata process, recurring_schedule: [ "static_task", "dynamic_task" ]
+  ensure
+    scheduler.stop
+  end
+
+  test "dynamic tasks in DB are ignored when dynamic_tasks_enabled is false" do
+    SolidQueue::RecurringTask.create!(
+      key: "ignored_task", static: false, class_name: "AddToBufferJob", schedule: "every second", arguments: [ 42 ]
+    )
+
+    recurring_tasks = { static_task: { class: "AddToBufferJob", schedule: "every hour", args: 42 } }
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: recurring_tasks).tap(&:start)
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    process = SolidQueue::Process.first
+    assert_metadata process, recurring_schedule: [ "static_task" ]
+  ensure
+    scheduler.stop
+  end
+
   test "run more than one instance of the scheduler with recurring tasks" do
     recurring_tasks = { example_task: { class: "AddToBufferJob", schedule: "every second", args: 42 } }
     schedulers = 2.times.collect do
@@ -24,13 +75,85 @@ class SchedulerTest < ActiveSupport::TestCase
     end
 
     schedulers.each(&:start)
-    sleep 2
+    wait_while_with_timeout(3.seconds) { SolidQueue::RecurringExecution.count < 2 }
     schedulers.each(&:stop)
 
-    assert_equal SolidQueue::Job.count, SolidQueue::RecurringExecution.count
-    run_at_times = SolidQueue::RecurringExecution.all.map(&:run_at).sort
-    0.upto(run_at_times.length - 2) do |i|
-      assert_equal 1, run_at_times[i + 1] - run_at_times[i]
+    skip_active_record_query_cache do
+      assert SolidQueue::RecurringExecution.count >= 2, "Expected at least 2 recurring executions, got #{SolidQueue::RecurringExecution.count}"
+      assert_equal SolidQueue::Job.count, SolidQueue::RecurringExecution.count
+      run_at_times = SolidQueue::RecurringExecution.all.map(&:run_at).sort
+      0.upto(run_at_times.length - 2) do |i|
+        time_diff = run_at_times[i + 1] - run_at_times[i]
+        assert_in_delta 1, time_diff, 0.001, "Expected run_at times to be 1 second apart, got #{time_diff}. All run_at times: #{run_at_times.inspect}"
+      end
     end
   end
+
+  test "dynamic task actually enqueues jobs" do
+    SolidQueue::RecurringTask.create!(
+      key: "dynamic_enqueue_task", static: false, class_name: "AddToBufferJob", schedule: "every second", arguments: [ 42 ]
+    )
+
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}, dynamic_tasks_enabled: true, polling_interval: 0.1).tap(&:start)
+
+    wait_for_registered_processes(1, timeout: 1.second)
+    wait_while_with_timeout(3.seconds) { SolidQueue::Job.count < 1 }
+
+    skip_active_record_query_cache do
+      assert SolidQueue::Job.count >= 1, "Expected at least one job to be enqueued by the dynamic task"
+      assert_equal SolidQueue::Job.count, SolidQueue::RecurringExecution.count
+    end
+  ensure
+    scheduler&.stop
+  end
+
+  test "updates metadata after adding dynamic task post-start" do
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}, dynamic_tasks_enabled: true, polling_interval: 0.1).tap(&:start)
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    skip_active_record_query_cache do
+      process = SolidQueue::Process.first
+      assert_empty process.metadata
+
+      SolidQueue::RecurringTask.create!(
+        key: "new_dynamic_task",
+        static: false,
+        class_name: "AddToBufferJob",
+        schedule: "every second",
+        arguments: [ 42 ]
+      )
+
+      wait_while_with_timeout(3.seconds) { process.reload.metadata.empty? }
+      assert_metadata process, recurring_schedule: [ "new_dynamic_task" ]
+    end
+  ensure
+    scheduler&.stop
+  end
+
+  test "updates metadata after removing dynamic task post-start" do
+    old_dynamic_task = SolidQueue::RecurringTask.create!(
+      key: "old_dynamic_task",
+      static: false,
+      class_name: "AddToBufferJob",
+      schedule: "every second",
+      arguments: [ 42 ]
+    )
+
+    scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}, dynamic_tasks_enabled: true, polling_interval: 0.1).tap(&:start)
+
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    skip_active_record_query_cache do
+      process = SolidQueue::Process.first
+      assert_metadata process, recurring_schedule: [ "old_dynamic_task" ]
+
+      old_dynamic_task.destroy
+
+      wait_while_with_timeout(3.seconds) { process.reload.metadata.present? }
+      assert_empty process.metadata
+    end
+  ensure
+    scheduler&.stop
+  end
 end
diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb
index 8db67912b..3d692404b 100644
--- a/test/unit/worker_test.rb
+++ b/test/unit/worker_test.rb
@@ -35,6 +35,7 @@ class WorkerTest < ActiveSupport::TestCase
     worker = SolidQueue::Worker.new(queues: "background", threads: 3, polling_interval: 0.2).tap(&:start)
     sleep(1)
 
+    # stop calls join internally when not supervised, which re-raises the error
     assert_raises ExpectedTestError do
       worker.stop
     end