diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 523bc4c..3942934 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -96,9 +96,13 @@ jobs:
       - name: Installing dependencies (Python)
         run: uv sync --all-extras
 
-      - name: Running tests
+      - name: Running tests (parallel)
         run: |
-          uv run pytest -s -v --cov=chancy --cov-report=xml
+          uv run pytest -n auto -v --cov=chancy --cov-report=xml
+
+      - name: Running Django ORM tests (sequential)
+        run: |
+          uv run pytest tests/contrib/django/test_models.py -v --cov=chancy --cov-report=xml --cov-append
 
       - name: Uploading coverage
         uses: codecov/codecov-action@v4
diff --git a/chancy/contrib/django/backend.py b/chancy/contrib/django/backend.py
index 1b14a65..fd80cd8 100644
--- a/chancy/contrib/django/backend.py
+++ b/chancy/contrib/django/backend.py
@@ -31,7 +31,7 @@
 Chancy documentation for details.
 """
 
-import asyncio
+import inspect
 from datetime import datetime, timezone
 from functools import cached_property
 from typing import TYPE_CHECKING, Any
@@ -192,7 +192,7 @@ async def aenqueue(
 
         wrapper_func = (
             ASYNC_WRAPPER_FUNC
-            if asyncio.iscoroutinefunction(task.func)
+            if inspect.iscoroutinefunction(task.func)
             else WRAPPER_FUNC
         )
 
diff --git a/chancy/executors/asyncex.py b/chancy/executors/asyncex.py
index ec4c33b..7d188c8 100644
--- a/chancy/executors/asyncex.py
+++ b/chancy/executors/asyncex.py
@@ -1,4 +1,5 @@
 import asyncio
+import inspect
 from asyncio import CancelledError
 
 from chancy import Reference
@@ -47,7 +48,7 @@ def __len__(self):
     async def _job_wrapper(self, job: QueuedJob):
         try:
             func, kwargs = Executor.get_function_and_kwargs(job)
-            if not asyncio.iscoroutinefunction(func):
+            if not inspect.iscoroutinefunction(func):
                 raise ValueError(
                     f"Function {job.func!r} is not an async function, which is"
                     f" required for the AsyncExecutor. Please use the"
@@ -75,6 +76,9 @@ async def cancel(self, ref: Reference):
                 task.cancel()
                 return
 
+    def get_running_jobs(self) -> list["QueuedJob"]:
+        return list(self.jobs.values())
+
     async def stop(self):
         """
         Stop the executor, giving it a chance to clean up any resources it
diff --git a/chancy/executors/base.py b/chancy/executors/base.py
index 325ed89..2eefe80 100644
--- a/chancy/executors/base.py
+++ b/chancy/executors/base.py
@@ -188,6 +188,35 @@ async def cancel(self, ref: Reference):
         :param ref: The reference to the job to cancel.
         """
 
+    def get_running_job(self, ref: Reference) -> QueuedJob | None:
+        """
+        Get a running job by its reference.
+
+        :param ref: The reference to the job to find.
+        :return: The job if found, None otherwise.
+        """
+        for job in self.get_running_jobs():
+            if job.id == ref.identifier:
+                return job
+        return None
+
+    @abc.abstractmethod
+    def get_running_jobs(self) -> list[QueuedJob]:
+        """
+        Get all jobs currently running in this executor.
+
+        :return: A list of running jobs.
+        """
+
+    def is_job_running(self, ref: Reference) -> bool:
+        """
+        Check if a job is currently running in this executor.
+
+        :param ref: The reference to the job to check.
+        :return: True if the job is running, False otherwise.
+        """
+        return self.get_running_job(ref) is not None
+
     @abc.abstractmethod
     def get_default_concurrency(self) -> int:
         """
@@ -244,5 +273,8 @@ async def cancel(self, ref: Reference):
                 future.cancel()
                 return
 
+    def get_running_jobs(self) -> list[QueuedJob]:
+        return list(self.jobs.values())
+
     def __len__(self):
         return len(self.jobs)
diff --git a/chancy/executors/process.py b/chancy/executors/process.py
index 6fe8718..c75c3d8 100644
--- a/chancy/executors/process.py
+++ b/chancy/executors/process.py
@@ -1,5 +1,6 @@
 import asyncio
 import functools
+import inspect
 import multiprocessing
 import os
 import warnings
@@ -15,6 +16,7 @@
 from asyncio import Future, CancelledError
 from concurrent.futures import ProcessPoolExecutor
 from typing import Callable, Any
+from uuid import UUID
 
 from chancy import Reference
 from chancy.executors.base import ConcurrentExecutor
@@ -138,7 +140,7 @@ async def push(self, job: QueuedJob) -> Future:
 
         return future
 
-    async def _handle_timeout(self, job_id: str, time_limit: int):
+    async def _handle_timeout(self, job_id: UUID, time_limit: int):
         try:
             await asyncio.sleep(time_limit)
             pid = self.pids_for_job.get(job_id)
@@ -190,7 +192,7 @@ def job_wrapper(cls, job: QueuedJob, pids_for_job) -> tuple[QueuedJob, Any]:
                                 )
                             )
 
-            if asyncio.iscoroutinefunction(func):
+            if inspect.iscoroutinefunction(func):
                 loop = asyncio.new_event_loop()
                 asyncio.set_event_loop(loop)
                 try:
diff --git a/chancy/executors/sub.py b/chancy/executors/sub.py
index 6bb91f3..7f91df7 100644
--- a/chancy/executors/sub.py
+++ b/chancy/executors/sub.py
@@ -1,5 +1,6 @@
-import os
 import asyncio
+import inspect
+import os
 import threading
 import functools
 from concurrent.futures import Future
@@ -122,7 +123,7 @@ def job_wrapper(cls, job: QueuedJob) -> tuple[QueuedJob, Any]:
             timer.start()
 
         try:
-            if asyncio.iscoroutinefunction(func):
+            if inspect.iscoroutinefunction(func):
                 loop = asyncio.new_event_loop()
                 asyncio.set_event_loop(loop)
                 try:
diff --git a/chancy/executors/thread.py b/chancy/executors/thread.py
index df5ee74..e7575e2 100644
--- a/chancy/executors/thread.py
+++ b/chancy/executors/thread.py
@@ -1,3 +1,4 @@
+import inspect
 import os
 import threading
 from concurrent.futures import ThreadPoolExecutor, Future
@@ -77,7 +78,7 @@ def job_wrapper(self, job: QueuedJob) -> tuple[QueuedJob, Any]:
             timer.start()
 
         try:
-            if asyncio.iscoroutinefunction(func):
+            if inspect.iscoroutinefunction(func):
                 loop = asyncio.new_event_loop()
                 asyncio.set_event_loop(loop)
                 try:
diff --git a/chancy/job.py b/chancy/job.py
index 8d34819..ea76b65 100644
--- a/chancy/job.py
+++ b/chancy/job.py
@@ -221,7 +221,7 @@ class State(enum.Enum):
         SUCCEEDED = "succeeded"
 
     #: The unique identifier for this job instance.
-    id: str
+    id: UUID
     #: The time at which this job was created.
     created_at: datetime
     #: The time at which this job was started, if it has been started.
@@ -237,8 +237,9 @@ class State(enum.Enum):
 
     @classmethod
     def unpack(cls, data: dict) -> "QueuedJob":
+        id_ = data["id"]
         return cls(
-            id=str(data["id"]),
+            id=id_ if isinstance(id_, UUID) else UUID(id_),
             func=data["func"],
             kwargs=data["kwargs"],
             priority=data["priority"],
diff --git a/chancy/migrations/v1.py b/chancy/migrations/v1.py
index ab0c3f9..76048e1 100644
--- a/chancy/migrations/v1.py
+++ b/chancy/migrations/v1.py
@@ -60,10 +60,15 @@ async def up(self, migrator: Migrator, cursor: AsyncCursor[DictRow]):
                     expires_at TIMESTAMPTZ NOT NULL
                 );
                 ALTER TABLE {leader}
-                    ADD CONSTRAINT leader_worker_id_unique UNIQUE
+                    ADD CONSTRAINT {leader_worker_id_unique} UNIQUE
                         (worker_id);
                 """
-            ).format(leader=sql.Identifier(f"{migrator.prefix}leader"))
+            ).format(
+                leader=sql.Identifier(f"{migrator.prefix}leader"),
+                leader_worker_id_unique=sql.Identifier(
+                    f"{migrator.prefix}leader_worker_id_unique"
+                ),
+            )
         )
 
         await cursor.execute(
diff --git a/chancy/migrations/v7.py b/chancy/migrations/v7.py
new file mode 100644
index 0000000..e0fc6f2
--- /dev/null
+++ b/chancy/migrations/v7.py
@@ -0,0 +1,62 @@
+from psycopg import AsyncCursor, sql
+from psycopg.rows import DictRow
+
+from chancy.migrate import Migration, Migrator
+
+
+class V7Migration(Migration):
+    async def up(self, migrator: Migrator, cursor: AsyncCursor[DictRow]):
+        """
+        Rename the leader_worker_id_unique constraint to use the prefix.
+
+        This fixes an oversight where the constraint name was hardcoded
+        instead of using the configurable prefix, which prevented running
+        multiple Chancy instances with different prefixes in the same database.
+        """
+        old_name = "leader_worker_id_unique"
+        new_name = f"{migrator.prefix}leader_worker_id_unique"
+
+        # Check if the old unprefixed constraint exists
+        await cursor.execute(
+            """
+            SELECT 1 FROM pg_constraint
+            WHERE conname = %s
+            """,
+            (old_name,),
+        )
+        if await cursor.fetchone():
+            await cursor.execute(
+                sql.SQL(
+                    "ALTER TABLE {table} RENAME CONSTRAINT {old} TO {new}"
+                ).format(
+                    table=sql.Identifier(f"{migrator.prefix}leader"),
+                    old=sql.Identifier(old_name),
+                    new=sql.Identifier(new_name),
+                )
+            )
+
+    async def down(self, migrator: Migrator, cursor: AsyncCursor[DictRow]):
+        """
+        Rename the constraint back to the unprefixed name.
+        """
+        old_name = f"{migrator.prefix}leader_worker_id_unique"
+        new_name = "leader_worker_id_unique"
+
+        # Check if the prefixed constraint exists
+        await cursor.execute(
+            """
+            SELECT 1 FROM pg_constraint
+            WHERE conname = %s
+            """,
+            (old_name,),
+        )
+        if await cursor.fetchone():
+            await cursor.execute(
+                sql.SQL(
+                    "ALTER TABLE {table} RENAME CONSTRAINT {old} TO {new}"
+                ).format(
+                    table=sql.Identifier(f"{migrator.prefix}leader"),
+                    old=sql.Identifier(old_name),
+                    new=sql.Identifier(new_name),
+                )
+            )
diff --git a/chancy/utils.py b/chancy/utils.py
index 7d236f3..65d63c0 100644
--- a/chancy/utils.py
+++ b/chancy/utils.py
@@ -121,7 +121,7 @@ def import_string(name):
     return getattr(module, func_name)
 
 
-def chancy_uuid() -> str:
+def chancy_uuid() -> uuid.UUID:
     """
     Generate a UUID suitable for use as a job ID.
 
@@ -129,12 +129,12 @@ def chancy_uuid() -> str:
 
         It's UUID7, kinda, since the draft keeps changing.
 
-    :return: str
+    :return: UUID
     """
     t = (time.time_ns() // 100) & 0xFFFFFFFFFFFFFF
     rand = secrets.randbits(62)
     uuid7 = (t << 68) | (7 << 64) | (2 << 62) | rand
-    return f"{uuid7:032x}"
+    return uuid.UUID(f"{uuid7:032x}")
 
 
 def json_dumps(obj, **kwargs):
diff --git a/pyproject.toml b/pyproject.toml
index 739d9ad..0c29bf5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -57,6 +57,7 @@ dev = [
     "pytest-django>=4.8.0",
     "pre-commit>=4.3.0",
     "ruff>=0.14.2",
+    "pytest-xdist>=3.8.0",
 ]
 
 [project.scripts]
diff --git a/tests/conftest.py b/tests/conftest.py
index 52b1828..c2cc62b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,4 +1,5 @@
 import asyncio
+import os
 from typing import AsyncIterator
 
 import pytest
@@ -8,6 +9,36 @@
 from chancy import Chancy, Worker
 
 
+@pytest.fixture(scope="session")
+def xdist_worker() -> str | None:
+    """Returns the xdist worker ID (e.g., 'gw0') or None if not running parallel."""
+    return os.environ.get("PYTEST_XDIST_WORKER")
+
+
+@pytest.fixture(scope="session")
+def test_prefix(xdist_worker) -> str:
+    """
+    Unique table prefix for parallel test isolation.
+
+    Returns 'chancy_' normally, or 'chancy_gw0_' etc. when running with xdist.
+    """
+    if xdist_worker is None:
+        return "chancy_"
+    return f"chancy_{xdist_worker}_"
+
+
+@pytest.fixture(scope="session")
+def test_suffix(xdist_worker) -> str:
+    """
+    Unique suffix for test resources (tables, schemas) that need isolation.
+
+    Returns '' normally, or '_gw0' etc. when running with xdist.
+    """
+    if xdist_worker is None:
+        return ""
+    return f"_{xdist_worker}"
+
+
 @pytest.fixture(scope="session")
 def event_loop_policy():
     # Since psycopg's asyncio implementation cannot use the default
@@ -18,14 +49,22 @@ def event_loop_policy():
 
 
 @pytest_asyncio.fixture()
-async def chancy(request):
+async def chancy(request, test_prefix):
     """
     Provides a Chancy application instance with an open connection pool
     to the test database.
+
+    When running with pytest-xdist, each worker gets a unique table prefix
+    to enable parallel test execution without conflicts.
     """
+    params = getattr(request, "param", {})
+    # Allow tests to override prefix, but default to worker-specific prefix
+    if "prefix" not in params:
+        params = {**params, "prefix": test_prefix}
+
     async with Chancy(
         "postgresql://postgres:localtest@localhost:8190/postgres",
-        **getattr(request, "param", {}),
+        **params,
     ) as chancy:
         await chancy.migrate()
         yield chancy
@@ -33,13 +72,16 @@ async def chancy(request):
 
 
 @pytest.fixture
-def chancy_just_app():
+def chancy_just_app(test_prefix):
     """
     Provides just a configured chancy instance with no open connection pool
     or migrations.
+
+    When running with pytest-xdist, uses a unique table prefix for isolation.
     """
     return Chancy(
         "postgresql://postgres:localtest@localhost:8190/postgres",
+        prefix=test_prefix,
     )
 
 
diff --git a/tests/contrib/django/test_connection.py b/tests/contrib/django/test_connection.py
index 7f6f484..6110e2b 100644
--- a/tests/contrib/django/test_connection.py
+++ b/tests/contrib/django/test_connection.py
@@ -5,7 +5,7 @@
 
 @pytest.mark.django_db
 @pytest.mark.asyncio
-async def test_django_style_connection(settings):
-    async with Chancy(settings.DATABASES["default"]) as app:
+async def test_django_style_connection(settings, test_prefix):
+    async with Chancy(settings.DATABASES["default"], prefix=test_prefix) as app:
         await app.migrate()
         await app.migrate(to_version=0)
diff --git a/tests/contrib/django/test_django_tasks.py b/tests/contrib/django/test_django_tasks.py
index 812353f..4d179b7 100644
--- a/tests/contrib/django/test_django_tasks.py
+++ b/tests/contrib/django/test_django_tasks.py
@@ -55,7 +55,7 @@ async def async_add(a: int, b: int) -> int:
 
 
 @pytest.fixture
-def django_tasks_settings(settings):
+def django_tasks_settings(settings, chancy):
     """Configure Django Tasks to use the Chancy backend."""
     settings.TASKS = {
         "default": {
@@ -63,7 +63,7 @@ def django_tasks_settings(settings):
             "QUEUES": ["default", "async_queue"],
             "OPTIONS": {
                 "dsn": "postgresql://postgres:localtest@localhost:8190/postgres",
-                "prefix": "chancy_",
+                "prefix": chancy.prefix,
             },
         },
     }
diff --git a/tests/contrib/django/test_models.py b/tests/contrib/django/test_models.py
index c721f5b..2a8cb6d 100644
--- a/tests/contrib/django/test_models.py
+++ b/tests/contrib/django/test_models.py
@@ -1,12 +1,24 @@
 """
 Tests for the Django models integration.
+
+Note: These tests cannot run in parallel (pytest-xdist) because Django ORM
+models have table names set at import time, not runtime.
 """
 
+import os
 import pytest
 
 from chancy import job
 
 
+# Skip these tests when running with pytest-xdist since Django model
+# table names are set at import time and can't be made worker-specific
+pytestmark = pytest.mark.skipif(
+    os.environ.get("PYTEST_XDIST_WORKER") is not None,
+    reason="Django ORM models have static table names incompatible with parallel tests",
+)
+
+
 @job()
 def test_job():
     pass
diff --git a/tests/plugins/test_trigger.py b/tests/plugins/test_trigger.py
index b767727..f1b4c64 100644
--- a/tests/plugins/test_trigger.py
+++ b/tests/plugins/test_trigger.py
@@ -12,7 +12,7 @@
 def handle_insert(*, j: QueuedJob):
     """Test job for INSERT operations"""
     assert j.meta["trigger"]["operation"] == "INSERT"
-    assert j.meta["trigger"]["table_name"] == "test_users"
+    assert j.meta["trigger"]["table_name"].startswith("test_users")
     assert j.meta["trigger"]["schema_name"] == "public"
 
 
@@ -37,7 +37,7 @@ def handle_any_change(*, j: QueuedJob):
 @job(queue="trigger_events", priority=10)
 def handle_with_priority(*, j: QueuedJob):
     """Test job with priority"""
-    assert j.meta["trigger"]["table_name"] == "test_users"
+    assert j.meta["trigger"]["table_name"].startswith("test_users")
 
 
 # Keep old name for backward compatibility with existing test
@@ -45,8 +45,9 @@ def handle_with_priority(*, j: QueuedJob):
 
 
 @pytest_asyncio.fixture
-async def test_table(chancy):
-    table_name = "test_users"
+async def test_table(chancy, test_suffix):
+    # Use unique table name per xdist worker to avoid conflicts
+    table_name = f"test_users{test_suffix}"
 
     async with chancy.pool.connection() as conn:
         async with conn.cursor() as cursor:
@@ -65,7 +66,7 @@ async def test_table(chancy):
     async with chancy.pool.connection() as conn:
         async with conn.cursor() as cursor:
             await cursor.execute(
-                sql.SQL("DROP TABLE IF EXISTS {table}").format(
+                sql.SQL("DROP TABLE IF EXISTS {table} CASCADE").format(
                     table=sql.Identifier(table_name)
                 )
             )
@@ -710,9 +711,11 @@ async def test_trigger_disabled_at_registration(
 
 
 @pytest_asyncio.fixture
-async def test_schema(chancy):
+async def test_schema(chancy, test_suffix):
     """Create a test schema for schema-specific tests"""
-    schema_name = "test_schema"
+    # Use unique schema/table names per xdist worker
+    schema_name = f"test_schema{test_suffix}"
+    table_name = f"test_users{test_suffix}"
 
     async with chancy.pool.connection() as conn:
         async with conn.cursor() as cursor:
@@ -730,11 +733,11 @@ async def test_schema(chancy):
                     )
                 """).format(
                     schema=sql.Identifier(schema_name),
-                    table=sql.Identifier("test_users"),
+                    table=sql.Identifier(table_name),
                 )
             )
 
-    yield schema_name
+    yield (schema_name, table_name)
 
     async with chancy.pool.connection() as conn:
         async with conn.cursor() as cursor:
@@ -753,14 +756,15 @@ async def test_schema(chancy):
 @pytest.mark.asyncio
 async def test_trigger_different_schema(chancy: Chancy, worker, test_schema):
     """Test trigger on a table in a non-public schema"""
+    schema_name, table_name = test_schema
     await chancy.declare(Queue("trigger_events"))
 
     await Trigger.register_trigger(
         chancy,
-        table_name="test_users",
+        table_name=table_name,
         operations=["INSERT"],
         job_template=handle_insert,
-        schema_name=test_schema,
+        schema_name=schema_name,
     )
 
     async with chancy.pool.connection() as conn:
@@ -771,8 +775,8 @@ async def test_trigger_different_schema(chancy: Chancy, worker, test_schema):
                         INSERT INTO {schema}.{table} (name, email)
                         VALUES ('Schema User', 'schema@example.com')
                     """).format(
-                        schema=sql.Identifier(test_schema),
-                        table=sql.Identifier("test_users"),
+                        schema=sql.Identifier(schema_name),
+                        table=sql.Identifier(table_name),
                     )
                 )
 
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index fa53760..a1869b3 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -1,9 +1,10 @@
+import sys
 import time
 import asyncio
 
 import pytest
 
-from chancy import Chancy, Worker, Queue, QueuedJob, Reference, job, Job
+from chancy import Chancy, Worker, Queue, QueuedJob, job, Job
 from test_worker import job_that_fails
 
 
@@ -171,30 +172,6 @@ async def test_async_job_instance_kwarg_on_sync_executor(
     assert j.meta.get("received_instance") is True
 
 
-@pytest.mark.asyncio
-async def test_job_cancellation(chancy: Chancy, worker: Worker):
-    """
-    Test that jobs can be cancelled on supporting executors.
-    """
-
-    async def cancel_in_a_bit(to_cancel: Reference):
-        await asyncio.sleep(10)
-        await chancy.cancel_job(to_cancel)
-
-    await chancy.declare(Queue("async", executor=Chancy.Executor.Async))
-    await chancy.declare(Queue("sync", executor=Chancy.Executor.Process))
-
-    ref = await chancy.push(very_long_job.job.with_queue("async"))
-    asyncio.create_task(cancel_in_a_bit(ref))
-    j = await chancy.wait_for_job(ref, timeout=30)
-    assert j.state == j.State.FAILED
-
-    ref = await chancy.push(sync_very_long_job.job.with_queue("sync"))
-    asyncio.create_task(cancel_in_a_bit(ref))
-    j = await chancy.wait_for_job(ref, timeout=30)
-    assert j.state == j.State.FAILED
-
-
 @pytest.mark.asyncio
 async def test_job_signature_with_kwarg_marker(chancy, worker):
     """
@@ -315,3 +292,58 @@ async def test_purge_jobs(chancy: Chancy, worker: Worker, sync_executor: str):
     await chancy.purge_jobs([ref_ok, ref_fail])
     assert await chancy.get_job(ref_ok) is None
     assert await chancy.get_job(ref_fail) is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="SIGUSR1 not available on Windows"
+)
+async def test_process_executor_job_cancellation(
+    chancy: Chancy, worker: Worker
+):
+    """
+    Test that jobs can be cancelled on the ProcessExecutor.
+    """
+    await chancy.declare(Queue("cancel_test", executor=Chancy.Executor.Process))
+
+    ref = await chancy.push(sync_very_long_job.job.with_queue("cancel_test"))
+    j = await chancy.wait_for_job(
+        ref, timeout=10, states={QueuedJob.State.RUNNING}
+    )
+    assert j.state == j.State.RUNNING
+
+    await chancy.cancel_job(ref)
+
+    executor = worker._executors.get("cancel_test")
+    async with asyncio.timeout(10):
+        while executor.is_job_running(ref):
+            await asyncio.sleep(0.1)
+
+    j = await chancy.wait_for_job(ref, timeout=10)
+    assert j.state == j.State.FAILED
+
+
+@pytest.mark.asyncio
+async def test_async_executor_job_cancellation(chancy: Chancy, worker: Worker):
+    """
+    Test that jobs can be cancelled on the AsyncExecutor.
+    """
+    await chancy.declare(
+        Queue("async_cancel_test", executor=Chancy.Executor.Async)
+    )
+
+    ref = await chancy.push(very_long_job.job.with_queue("async_cancel_test"))
+    j = await chancy.wait_for_job(
+        ref, timeout=10, states={QueuedJob.State.RUNNING}
+    )
+    assert j.state == j.State.RUNNING
+
+    await chancy.cancel_job(ref)
+
+    executor = worker._executors.get("async_cancel_test")
+    async with asyncio.timeout(10):
+        while executor.is_job_running(ref):
+            await asyncio.sleep(0.1)
+
+    j = await chancy.wait_for_job(ref, timeout=10)
+    assert j.state == j.State.FAILED
diff --git a/tests/test_scale.py b/tests/test_scale.py
index 378002f..5d28d69 100644
--- a/tests/test_scale.py
+++ b/tests/test_scale.py
@@ -83,10 +83,14 @@ async def test_busy_chancy(chancy: Chancy, worker_no_start: Worker):
 
             await conn.commit()
 
+        # In reality this is typically sub-0.1. However, we get occasional
+        # spikes in CI, especially on OS X, that make this very noisy for test
+        # failures, so we set a more generous limit to just catch egregious
+        # performance regressions.
         with timed_block() as timer:
             await worker_no_start.fetch_jobs(queue, conn, up_to=1)
-        assert timer.elapsed < 0.1
+        assert timer.elapsed < 0.5
 
         with timed_block() as timer:
             await worker_no_start.fetch_jobs(queue, conn, up_to=100)
-        assert timer.elapsed < 0.1
+        assert timer.elapsed < 0.5
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 484c562..b6f093e 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -93,8 +93,10 @@ async def test_queue_removal(chancy: Chancy, worker: Worker):
     await chancy.delete_queue("test_removal", purge_jobs=True)
     await worker.hub.wait_for("worker.queue.removed", timeout=30)
 
-    # Give the executor time to clean up
-    await asyncio.sleep(5)
+    # Wait for the executor to clean up
+    async with asyncio.timeout(10):
+        while "test_removal" in worker.executors:
+            await asyncio.sleep(0.1)
 
     assert "test_removal" not in worker.executors
 
@@ -119,7 +121,6 @@ async def test_immediate_processing(chancy: Chancy, worker: Worker):
     """
     await chancy.declare(Queue("test_immediate", polling_interval=60))
     await worker.hub.wait_for("worker.queue.started")
-    await asyncio.sleep(5)
 
     j = await chancy.push(job_to_run.job.with_queue("test_immediate"))