feat: add RedisDeduplicationScheduleSource for scheduled task deduplication

README.md

@@ -55,7 +55,7 @@ except DuplicateTaskError:
 - **Configurable TTL** — set a global default or override per task with the `deduplication_ttl` label.
 - **Lock heartbeat** — a background task re-extends the lock TTL while the task runs, so long-running tasks keep their lock instead of expiring mid-execution and admitting a duplicate.
 - **Explicit lock key** — pin any task to a fixed Redis key with `deduplication_key`, bypassing fingerprint computation entirely.
-- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments (positional arguments are excluded).
+- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments.
 - **Per-task opt-out** — disable deduplication for individual tasks with the `deduplication` label.
 - **Startup resilience** — automatic reconnection with exponential backoff if Redis is unavailable at broker startup.
 

docs/index.md

@@ -55,7 +55,7 @@ except DuplicateTaskError:
 - **Configurable TTL** — set a global default or override per task with the `deduplication_ttl` label.
 - **Lock heartbeat** — a background task re-extends the lock TTL while the task runs, so long-running tasks keep their lock instead of expiring mid-execution and admitting a duplicate.
 - **Explicit lock key** — pin any task to a fixed Redis key with `deduplication_key`, bypassing fingerprint computation entirely.
-- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments (positional arguments are excluded).
+- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments.
 - **Per-task opt-out** — disable deduplication for individual tasks with the `deduplication` label.
 - **Startup resilience** — automatic reconnection with exponential backoff if Redis is unavailable at broker startup.
 

docs/usage.md

@@ -147,19 +147,11 @@ await my_task.kicker().with_labels(deduplication_ttl=60).kiq(user_id=42)
 | `deduplication` | `bool` | Set `False` to opt out of deduplication entirely for this task. |
 | `deduplication_ttl` | `int` | Lock TTL in seconds. Overrides the middleware `default_ttl`. |
 | `deduplication_key` | `str` | Explicit lock key. Skips fingerprint computation entirely. |
-| `deduplication_key_fields` | `list[str]` | Subset of kwargs to include in the fingerprint. Positional arguments are excluded. Ignored if `deduplication_key` is set. |
+| `deduplication_key_fields` | `list[str]` | Subset of kwargs to include in the fingerprint. Ignored if `deduplication_key` is set. |
 
 ## Fingerprint and key customisation
 
-By default the lock key is a SHA-256 fingerprint of the task name, its positional
+By default the lock key is a SHA-256 fingerprint of the task name and all kwargs.
-arguments and all kwargs.
-
-!!! warning "Positional and keyword calls fingerprint differently"
-
-    taskiq serialises arguments as they were passed, without binding them to the
-    task signature. `my_task.kiq(42)` and `my_task.kiq(user_id=42)` are therefore
-    *not* recognised as duplicates of each other. Call a deduplicated task
-    consistently, preferably always with keyword arguments.
 
 ### Explicit key
 
@@ -190,10 +182,6 @@ If a listed field is absent from a task's kwargs, it is dropped from the
 fingerprint and a warning is logged, since this can make genuinely different
 calls collide on the same lock.
 
-Positional arguments are excluded from the fingerprint entirely when this label is
-set: you asked to deduplicate on named fields, so pass them as keyword arguments.
-A warning is logged if the task is called with positional arguments anyway.
-
 ## Opting out per task
 
 ```python

src/taskiq_deduplication/__init__.py

@@ -1,10 +1,12 @@
 """Redis-backed deduplication middleware for Taskiq."""
 
 from .middleware import DuplicateTaskError, RedisDeduplicationMiddleware
+from .schedule import RedisDeduplicationScheduleSource
 
 __version__ = "1.1.0"
 
 __all__ = [
     "DuplicateTaskError",
     "RedisDeduplicationMiddleware",
+    "RedisDeduplicationScheduleSource",
 ]

src/taskiq_deduplication/middleware.py

@@ -137,38 +137,34 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
             await self._redis.aclose()
 
     def _build_deduplication_key(self, message: TaskiqMessage) -> str | None:
-        explicit_key: str | None = message.labels.get(DEDUP_EXPLICIT_KEY_LABEL)
+        return self._build_key(message.task_name, message.labels, message.kwargs)
+
+    def _build_key(
+        self, task_name: str, labels: dict[str, Any], kwargs: dict[str, Any]
+    ) -> str | None:
+        explicit_key: str | None = labels.get(DEDUP_EXPLICIT_KEY_LABEL)
         if explicit_key is not None:
             return f"{self.key_prefix}:{explicit_key}"
 
         key_fields = parse_list_label(
-            message.labels.get(DEDUP_KEY_FIELDS_LABEL), DEDUP_KEY_FIELDS_LABEL
+            labels.get(DEDUP_KEY_FIELDS_LABEL), DEDUP_KEY_FIELDS_LABEL
         )
         if key_fields is not None:
-            missing = [field for field in key_fields if field not in message.kwargs]
+            missing = [field for field in key_fields if field not in kwargs]
             if missing:
                 logger.warning(
                     "Task %s requested deduplication_key_fields %r but they are "
                     "absent from kwargs; they are dropped from the fingerprint, which "
                     "may cause distinct calls to collide.",
-                    message.task_name,
+                    task_name,
                     missing,
                 )
-            if message.args:
+            filtered_kwargs = {k: v for k, v in kwargs.items() if k in key_fields}
-                logger.warning(
-                    "Task %s was called with positional arguments but uses "
-                    "deduplication_key_fields; positional arguments are excluded "
-                    "from the fingerprint.",
-                    message.task_name,
-                )
-            kwargs = {k: v for k, v in message.kwargs.items() if k in key_fields}
-            args: list[Any] = []
         else:
-            kwargs = message.kwargs
+            filtered_kwargs = kwargs
-            args = message.args
         try:
             payload = json.dumps(
-                {"task": message.task_name, "args": args, "kwargs": kwargs},
+                {"task": task_name, "kwargs": filtered_kwargs},
                 sort_keys=True,
             )
         except TypeError:
@@ -181,6 +177,31 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
             labels.get(DEDUP_LABEL), self.default_deduplication, DEDUP_LABEL
         )
 
+    def _require_redis(self) -> Redis:
+        if self._redis is None:
+            raise RuntimeError(
+                "RedisDeduplicationMiddleware.startup() was never called."
+            )
+        return self._redis
+
+    @staticmethod
+    def _decode_task_id(value: bytes | str) -> str:
+        return value.decode() if isinstance(value, bytes) else value
+
+    async def _peek(
+        self, task_name: str, labels: dict[str, Any], kwargs: dict[str, Any]
+    ) -> tuple[str, str] | None:
+        if not self._is_enabled(labels):
+            return None
+        redis = self._require_redis()
+        key = self._build_key(task_name, labels, kwargs)
+        if key is None:
+            return None
+        holder_task_id = await redis.get(key)
+        if holder_task_id is None:
+            return None
+        return key, self._decode_task_id(holder_task_id)
+
     def _get_ttl(self, labels: dict[str, Any]) -> int:
         return parse_int_label(
             labels.get(DEDUP_TTL_LABEL, self.default_ttl),
@@ -189,12 +210,9 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         )
 
     async def _release_if_owned(self, key: str, task_id: str) -> None:
-        if self._redis is None:
+        redis = self._require_redis()
-            raise RuntimeError(
-                "RedisDeduplicationMiddleware.startup() was never called."
-            )
         if self._release_script is None:
-            self._release_script = self._redis.register_script(RELEASE_LUA_SCRIPT)
+            self._release_script = redis.register_script(RELEASE_LUA_SCRIPT)
         released = await check_and_delete(self._release_script, key, task_id)
         if released:
             logger.debug("Released lock %s", key)
@@ -202,12 +220,9 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
             logger.debug("Skipped release of lock %s: not owned by this task", key)
 
     async def _refresh_if_owned(self, key: str, task_id: str, ttl: int) -> bool:
-        if self._redis is None:
+        redis = self._require_redis()
-            raise RuntimeError(
-                "RedisDeduplicationMiddleware.startup() was never called."
-            )
         if self._refresh_script is None:
-            self._refresh_script = self._redis.register_script(REFRESH_LUA_SCRIPT)
+            self._refresh_script = redis.register_script(REFRESH_LUA_SCRIPT)
         return await check_and_refresh(self._refresh_script, key, task_id, ttl)
 
     @staticmethod
@@ -222,10 +237,7 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         if not self._is_enabled(message.labels):
             return message
 
-        if self._redis is None:
+        redis = self._require_redis()
-            raise RuntimeError(
-                "RedisDeduplicationMiddleware.startup() was never called."
-            )
         key = self._build_deduplication_key(message)
         self._cache_key(message, key)
         if key is None:
@@ -238,11 +250,11 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         ttl = self._get_ttl(message.labels)
 
         logger.debug("Acquiring lock %s for task %s", key, message.task_name)
-        acquired = await self._redis.set(key, message.task_id, ex=ttl, nx=True)
+        acquired = await redis.set(key, message.task_id, ex=ttl, nx=True)
         if not acquired:
-            holder_task_id = await self._redis.get(key)
+            holder_task_id = await redis.get(key)
-            if isinstance(holder_task_id, bytes):
+            if holder_task_id is not None:
-                holder_task_id = holder_task_id.decode()
+                holder_task_id = self._decode_task_id(holder_task_id)
             logger.warning(
                 "Duplicate task %s dropped (key=%s, holder_task_id=%s).",
                 message.task_name,

src/taskiq_deduplication/schedule.py

@@ -0,0 +1,77 @@
+import logging
+
+from taskiq import ScheduledTask, ScheduleSource
+from taskiq.exceptions import ScheduledTaskCancelledError
+from taskiq.utils import maybe_awaitable
+
+from .middleware import RedisDeduplicationMiddleware
+
+logger = logging.getLogger(__name__)
+
+
+class RedisDeduplicationScheduleSource(ScheduleSource):
+    """Skips scheduled firings whose fingerprint is already locked.
+
+    Wraps a ``ScheduleSource`` and peeks the lock in ``pre_send``, raising
+    ``ScheduledTaskCancelledError`` on a hit so the scheduler skips the firing
+    cleanly instead of raising ``DuplicateTaskError`` out of ``kiq()``. The
+    atomic acquire/release lifecycle stays owned by ``middleware``.
+
+    Attributes:
+        source: The wrapped ``ScheduleSource``.
+        middleware: The ``RedisDeduplicationMiddleware`` instance registered
+            on the broker. Must be the same instance, so both share one Redis
+            connection and configuration. Its ``startup()`` must have run
+            before ``pre_send()`` is invoked.
+    """
+
+    def __init__(
+        self,
+        source: ScheduleSource,
+        middleware: RedisDeduplicationMiddleware,
+    ) -> None:
+        self.source = source
+        self.middleware = middleware
+
+    async def startup(self) -> None:
+        await self.source.startup()
+
+    async def shutdown(self) -> None:
+        await self.source.shutdown()
+
+    async def get_schedules(self) -> list[ScheduledTask]:
+        return await self.source.get_schedules()
+
+    async def add_schedule(self, schedule: ScheduledTask) -> None:
+        await self.source.add_schedule(schedule)
+
+    async def delete_schedule(self, schedule_id: str) -> None:
+        await self.source.delete_schedule(schedule_id)
+
+    async def post_send(self, task: ScheduledTask) -> None:
+        await maybe_awaitable(self.source.post_send(task))
+
+    async def pre_send(self, task: ScheduledTask) -> None:
+        await maybe_awaitable(self.source.pre_send(task))
+
+        try:
+            held = await self.middleware._peek(task.task_name, task.labels, task.kwargs)
+        except RuntimeError:
+            logger.error(
+                "RedisDeduplicationMiddleware.startup() was never called; "
+                "cannot deduplicate scheduled task %s.",
+                task.task_name,
+            )
+            raise
+
+        if held is None:
+            return
+        key, holder_task_id = held
+        logger.warning(
+            "Duplicate scheduled task %s skipped before dispatch "
+            "(key=%s, holder_task_id=%s).",
+            task.task_name,
+            key,
+            holder_task_id,
+        )
+        raise ScheduledTaskCancelledError()

tests/conftest.py

@@ -1,7 +1,9 @@
 import pytest
 import fakeredis.aioredis
 from redis.asyncio import Redis
-from taskiq import TaskiqMessage, TaskiqResult
+from taskiq import ScheduledTask, TaskiqMessage, TaskiqResult
+
+from taskiq_deduplication import RedisDeduplicationMiddleware
 
 
 @pytest.fixture
@@ -16,6 +18,13 @@ async def fake_redis():
     await client.aclose()
 
 
+@pytest.fixture
+def middleware(fake_redis):
+    mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+    mw._redis = fake_redis
+    return mw
+
+
 @pytest.fixture
 async def real_redis():
     client = Redis.from_url("redis://localhost:6379/15")
@@ -33,15 +42,13 @@ async def real_redis():
 
 @pytest.fixture
 def make_message():
-    def _make(
+    def _make(task_name="my_task", task_id="task-1", labels=None, kwargs=None):
-        task_name="my_task", task_id="task-1", labels=None, kwargs=None, args=None
-    ):
         return TaskiqMessage(
             task_id=task_id,
             task_name=task_name,
             labels=labels or {},
             labels_types={},
-            args=args or [],
+            args=[],
             kwargs=kwargs or {},
         )
 
@@ -59,3 +66,24 @@ def make_result():
         )
 
     return _make
+
+
+@pytest.fixture
+def make_scheduled_task():
+    def _make(
+        task_name="my_task",
+        schedule_id="schedule-1",
+        labels=None,
+        kwargs=None,
+        cron="* * * * *",
+    ):
+        return ScheduledTask(
+            task_name=task_name,
+            schedule_id=schedule_id,
+            labels=labels or {},
+            args=[],
+            kwargs=kwargs or {},
+            cron=cron,
+        )
+
+    return _make

tests/test_middleware.py

@@ -12,13 +12,6 @@ from taskiq_deduplication.middleware import (
 )
 
 
-@pytest.fixture
-def middleware(fake_redis):
-    mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
-    mw._redis = fake_redis
-    return mw
-
-
 class TestDefaultBuildDeduplicationKey:
     def test_same_kwargs_same_key(self, middleware, make_message):
         m1 = make_message(kwargs={"a": 1, "b": 2})
@@ -86,27 +79,6 @@ class TestDefaultBuildDeduplicationKey:
             m1
         ) == middleware._build_deduplication_key(m2)
 
-    def test_different_args_different_key(self, middleware, make_message):
-        m1 = make_message(args=["a"])
-        m2 = make_message(args=["b"])
-        assert middleware._build_deduplication_key(
-            m1
-        ) != middleware._build_deduplication_key(m2)
-
-    def test_arg_order_matters(self, middleware, make_message):
-        m1 = make_message(args=["a", "b"])
-        m2 = make_message(args=["b", "a"])
-        assert middleware._build_deduplication_key(
-            m1
-        ) != middleware._build_deduplication_key(m2)
-
-    def test_key_fields_ignores_args(self, middleware, make_message):
-        m1 = make_message(args=["a"], labels={DEDUP_KEY_FIELDS_LABEL: ["x"]})
-        m2 = make_message(args=["b"], labels={DEDUP_KEY_FIELDS_LABEL: ["x"]})
-        assert middleware._build_deduplication_key(
-            m1
-        ) == middleware._build_deduplication_key(m2)
-
     def test_key_prefix_in_output(self, make_message):
         mw = RedisDeduplicationMiddleware(
             redis_url="redis://localhost", key_prefix="myapp:locks"

tests/test_schedule.py

@@ -0,0 +1,235 @@
+import logging
+
+import pytest
+from taskiq import InMemoryBroker, ScheduleSource, TaskiqScheduler
+from taskiq.exceptions import ScheduledTaskCancelledError
+
+from taskiq_deduplication import RedisDeduplicationMiddleware
+from taskiq_deduplication.middleware import (
+    DEDUP_EXPLICIT_KEY_LABEL,
+    DEDUP_KEY_FIELDS_LABEL,
+    DEDUP_LABEL,
+)
+from taskiq_deduplication.schedule import RedisDeduplicationScheduleSource
+
+
+class FakeScheduleSource(ScheduleSource):
+    def __init__(self):
+        self.startup_called = False
+        self.shutdown_called = False
+        self.schedules_to_return = []
+        self.added = []
+        self.deleted = []
+        self.pre_send_calls = []
+        self.post_send_calls = []
+        self.pre_send_raises = None
+
+    async def startup(self):
+        self.startup_called = True
+
+    async def shutdown(self):
+        self.shutdown_called = True
+
+    async def get_schedules(self):
+        return self.schedules_to_return
+
+    async def add_schedule(self, schedule):
+        self.added.append(schedule)
+
+    async def delete_schedule(self, schedule_id):
+        self.deleted.append(schedule_id)
+
+    async def pre_send(self, task):
+        self.pre_send_calls.append(task)
+        if self.pre_send_raises is not None:
+            raise self.pre_send_raises
+
+    async def post_send(self, task):
+        self.post_send_calls.append(task)
+
+
+@pytest.fixture
+def fake_source():
+    return FakeScheduleSource()
+
+
+@pytest.fixture
+def wrapper(fake_source, middleware):
+    return RedisDeduplicationScheduleSource(fake_source, middleware)
+
+
+class TestDelegation:
+    async def test_startup_delegates(self, wrapper, fake_source):
+        await wrapper.startup()
+        assert fake_source.startup_called
+
+    async def test_shutdown_delegates(self, wrapper, fake_source):
+        await wrapper.shutdown()
+        assert fake_source.shutdown_called
+
+    async def test_get_schedules_delegates(
+        self, wrapper, fake_source, make_scheduled_task
+    ):
+        task = make_scheduled_task()
+        fake_source.schedules_to_return = [task]
+        assert await wrapper.get_schedules() == [task]
+
+    async def test_add_schedule_delegates(
+        self, wrapper, fake_source, make_scheduled_task
+    ):
+        task = make_scheduled_task()
+        await wrapper.add_schedule(task)
+        assert fake_source.added == [task]
+
+    async def test_delete_schedule_delegates(self, wrapper, fake_source):
+        await wrapper.delete_schedule("some-id")
+        assert fake_source.deleted == ["some-id"]
+
+    async def test_post_send_delegates(self, wrapper, fake_source, make_scheduled_task):
+        task = make_scheduled_task()
+        await wrapper.post_send(task)
+        assert fake_source.post_send_calls == [task]
+
+    async def test_pre_send_delegates(self, wrapper, fake_source, make_scheduled_task):
+        task = make_scheduled_task()
+        await wrapper.pre_send(task)
+        assert fake_source.pre_send_calls == [task]
+
+
+class TestPreSend:
+    async def test_no_lock_held_passes(self, wrapper, make_scheduled_task):
+        task = make_scheduled_task()
+        result = await wrapper.pre_send(task)
+        assert result is None
+
+    async def test_wrapped_source_cancellation_propagates(
+        self, wrapper, fake_source, make_scheduled_task
+    ):
+        fake_source.pre_send_raises = ScheduledTaskCancelledError()
+        with pytest.raises(ScheduledTaskCancelledError):
+            await wrapper.pre_send(make_scheduled_task())
+
+    async def test_lock_held_raises_scheduled_task_cancelled_error(
+        self, wrapper, middleware, make_message, make_scheduled_task
+    ):
+        await middleware.pre_send(make_message(task_name="my_task", kwargs={"a": 1}))
+        task = make_scheduled_task(task_name="my_task", kwargs={"a": 1})
+        with pytest.raises(ScheduledTaskCancelledError):
+            await wrapper.pre_send(task)
+
+    async def test_peek_does_not_acquire_or_mutate(
+        self, wrapper, middleware, fake_redis, make_scheduled_task
+    ):
+        task = make_scheduled_task(task_name="my_task", kwargs={"a": 1})
+        await wrapper.pre_send(task)
+        key = middleware._build_key(task.task_name, task.labels, task.kwargs)
+        assert not await fake_redis.exists(key)
+        assert task.labels == {}
+        assert task.kwargs == {"a": 1}
+
+    async def test_peek_is_read_only_when_lock_held(
+        self, wrapper, middleware, fake_redis, make_message, make_scheduled_task
+    ):
+        held_msg = make_message(task_name="my_task", task_id="holder", kwargs={"a": 1})
+        await middleware.pre_send(held_msg)
+        key = middleware._build_deduplication_key(held_msg)
+        ttl_before = await fake_redis.ttl(key)
+        holder_before = await fake_redis.get(key)
+
+        task = make_scheduled_task(task_name="my_task", kwargs={"a": 1})
+        with pytest.raises(ScheduledTaskCancelledError):
+            await wrapper.pre_send(task)
+
+        # The peek must not have re-set the key (TTL untouched) or changed
+        # its owner.
+        assert await fake_redis.get(key) == holder_before
+        assert await fake_redis.ttl(key) <= ttl_before
+
+    async def test_deduplication_disabled_label_bypasses_peek(
+        self, wrapper, middleware, make_message, make_scheduled_task
+    ):
+        await middleware.pre_send(make_message(task_name="my_task", kwargs={"a": 1}))
+        task = make_scheduled_task(
+            task_name="my_task", kwargs={"a": 1}, labels={DEDUP_LABEL: False}
+        )
+        await wrapper.pre_send(task)  # should not raise
+
+    async def test_deduplication_key_label_respected(
+        self, wrapper, middleware, make_message, make_scheduled_task
+    ):
+        await middleware.pre_send(
+            make_message(kwargs={"a": 1}, labels={DEDUP_EXPLICIT_KEY_LABEL: "fixed"})
+        )
+        task = make_scheduled_task(
+            kwargs={"a": 999}, labels={DEDUP_EXPLICIT_KEY_LABEL: "fixed"}
+        )
+        with pytest.raises(ScheduledTaskCancelledError):
+            await wrapper.pre_send(task)
+
+    async def test_deduplication_key_fields_label_respected(
+        self, wrapper, middleware, make_message, make_scheduled_task
+    ):
+        await middleware.pre_send(
+            make_message(
+                kwargs={"a": 1, "b": 2},
+                labels={DEDUP_KEY_FIELDS_LABEL: ["a"]},
+            )
+        )
+        task = make_scheduled_task(
+            kwargs={"a": 1, "b": 999},
+            labels={DEDUP_KEY_FIELDS_LABEL: ["a"]},
+        )
+        with pytest.raises(ScheduledTaskCancelledError):
+            await wrapper.pre_send(task)
+
+    async def test_non_serializable_kwargs_skips_peek_silently(
+        self, wrapper, make_scheduled_task, caplog
+    ):
+        task = make_scheduled_task(kwargs={"dt": object()})
+        with caplog.at_level(logging.WARNING, logger="taskiq_deduplication.schedule"):
+            await wrapper.pre_send(task)  # should not raise
+        assert not any("non-JSON-serializable" in r.message for r in caplog.records)
+
+    async def test_pre_send_without_middleware_startup_raises_runtime_error(
+        self, fake_source, make_scheduled_task
+    ):
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+        w = RedisDeduplicationScheduleSource(fake_source, mw)
+        with pytest.raises(RuntimeError, match="startup"):
+            await w.pre_send(make_scheduled_task())
+
+    async def test_pre_send_without_middleware_startup_logs_before_raising(
+        self, fake_source, make_scheduled_task, caplog
+    ):
+        mw = RedisDeduplicationMiddleware(redis_url="redis://localhost")
+        w = RedisDeduplicationScheduleSource(fake_source, mw)
+        with caplog.at_level(logging.ERROR, logger="taskiq_deduplication.schedule"):
+            with pytest.raises(RuntimeError):
+                await w.pre_send(make_scheduled_task())
+        assert any("startup" in r.message for r in caplog.records)
+
+    async def test_different_kwargs_both_pass(self, wrapper, make_scheduled_task):
+        await wrapper.pre_send(make_scheduled_task(kwargs={"x": 1}))
+        await wrapper.pre_send(make_scheduled_task(kwargs={"x": 2}))
+
+
+class TestSchedulerIntegration:
+    async def test_second_firing_cancelled_without_uncaught_exception(
+        self, middleware, fake_source, make_message, make_scheduled_task
+    ):
+        broker = InMemoryBroker().with_middlewares(middleware)
+        wrapper = RedisDeduplicationScheduleSource(fake_source, middleware)
+        scheduler = TaskiqScheduler(broker=broker, sources=[wrapper])
+
+        task_name = "my_task"
+
+        # Simulate a still-running first firing by holding the lock directly.
+        await middleware.pre_send(
+            make_message(task_name=task_name, task_id="holder", kwargs={})
+        )
+
+        scheduled = make_scheduled_task(task_name=task_name, kwargs={})
+
+        # Must be cancelled gracefully by on_ready()'s own except clause, not
+        # raise DuplicateTaskError out of kiq() into an uncaught exception.
+        await scheduler.on_ready(wrapper, scheduled)

uv.lock

@@ -1266,15 +1266,15 @@ wheels = [
 
 [[package]]
 name = "pymdown-extensions"
-version = "11.0"
+version = "10.21.3"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "markdown" },
     { name = "pyyaml" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/47/67/f1e79672a5f91985577c7984c9709ca110e4fd37fe7fd167b60422e6ccc2/pymdown_extensions-11.0.tar.gz", hash = "sha256:8269cef0247f9e2d0a62fcea10860aba05c1cbab5470fd4b63230b96434dc589", size = 857049, upload-time = "2026-06-23T02:27:45.146Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/9e/26/d1015444da4d952a1ca487a236b522eb979766f0295a0bd0c5fc089989a9/pymdown_extensions-10.21.3.tar.gz", hash = "sha256:72cfcf55f07aea0d4af2c4f11dd4e52466ddfb1bb819673146398e0bd3a77354", size = 854140, upload-time = "2026-05-13T12:57:32.267Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/af/b6/1ae53367e28b9cffa3be7574e13fbe4589694272fd47710fbdbafd3d63c6/pymdown_extensions-11.0-py3-none-any.whl", hash = "sha256:fbc4acb641814fa9d17521bbd21a5240ef739a662f11c06330c4b78c93e954d6", size = 269415, upload-time = "2026-06-23T02:27:43.826Z" },
+    { url = "https://files.pythonhosted.org/packages/7e/85/545a951eecc270fcd688288c600017e2050a1aacb56c711d208586d3e470/pymdown_extensions-10.21.3-py3-none-any.whl", hash = "sha256:d7a5d08014fc571e80ca21dd6f854e31f94c489800350564d55d15b3c41e76b6", size = 269002, upload-time = "2026-05-13T12:57:30.296Z" },
 ]
 
 [[package]]