fix: include positional args in the deduplication fingerprint (#81)

Bumps [pymdown-extensions](https://github.com/facelessuser/pymdown-extensions) from 10.21.3 to 11.0.
- [Release notes](https://github.com/facelessuser/pymdown-extensions/releases)
- [Commits](https://github.com/facelessuser/pymdown-extensions/compare/10.21.3...11.0)

---
updated-dependencies:
- dependency-name: pymdown-extensions
  dependency-version: '11.0'
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

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.
+- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments (positional arguments are excluded).
 - **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.
+- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments (positional arguments are excluded).
 - **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,11 +147,19 @@ 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. Ignored if `deduplication_key` is set. |
+| `deduplication_key_fields` | `list[str]` | Subset of kwargs to include in the fingerprint. Positional arguments are excluded. Ignored if `deduplication_key` is set. |
 
 ## Fingerprint and key customisation
 
-By default the lock key is a SHA-256 fingerprint of the task name and all kwargs.
+By default the lock key is a SHA-256 fingerprint of the task name, its positional
+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
 
@@ -182,6 +190,10 @@ 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,12 +1,10 @@
 """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,34 +137,38 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
             await self._redis.aclose()
 
     def _build_deduplication_key(self, message: TaskiqMessage) -> str | None:
-        return self._build_key(message.task_name, message.labels, message.kwargs)
+        explicit_key: str | None = message.labels.get(DEDUP_EXPLICIT_KEY_LABEL)
-
-    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(
-            labels.get(DEDUP_KEY_FIELDS_LABEL), DEDUP_KEY_FIELDS_LABEL
+            message.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 kwargs]
+            missing = [field for field in key_fields if field not in message.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.",
-                    task_name,
+                    message.task_name,
                     missing,
                 )
-            filtered_kwargs = {k: v for k, v in kwargs.items() if k in key_fields}
+            if message.args:
+                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:
-            filtered_kwargs = kwargs
+            kwargs = message.kwargs
+            args = message.args
         try:
             payload = json.dumps(
-                {"task": task_name, "kwargs": filtered_kwargs},
+                {"task": message.task_name, "args": args, "kwargs": kwargs},
                 sort_keys=True,
             )
         except TypeError:
@@ -177,31 +181,6 @@ 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),
@@ -210,9 +189,12 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         )
 
     async def _release_if_owned(self, key: str, task_id: str) -> None:
-        redis = self._require_redis()
+        if self._redis is None:
+            raise RuntimeError(
+                "RedisDeduplicationMiddleware.startup() was never called."
+            )
         if self._release_script is None:
-            self._release_script = redis.register_script(RELEASE_LUA_SCRIPT)
+            self._release_script = self._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)
@@ -220,9 +202,12 @@ 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:
-        redis = self._require_redis()
+        if self._redis is None:
+            raise RuntimeError(
+                "RedisDeduplicationMiddleware.startup() was never called."
+            )
         if self._refresh_script is None:
-            self._refresh_script = redis.register_script(REFRESH_LUA_SCRIPT)
+            self._refresh_script = self._redis.register_script(REFRESH_LUA_SCRIPT)
         return await check_and_refresh(self._refresh_script, key, task_id, ttl)
 
     @staticmethod
@@ -237,7 +222,10 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         if not self._is_enabled(message.labels):
             return message
 
-        redis = self._require_redis()
+        if self._redis is None:
+            raise RuntimeError(
+                "RedisDeduplicationMiddleware.startup() was never called."
+            )
         key = self._build_deduplication_key(message)
         self._cache_key(message, key)
         if key is None:
@@ -250,11 +238,11 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
         ttl = self._get_ttl(message.labels)
 
         logger.debug("Acquiring lock %s for task %s", key, message.task_name)
-        acquired = await redis.set(key, message.task_id, ex=ttl, nx=True)
+        acquired = await self._redis.set(key, message.task_id, ex=ttl, nx=True)
         if not acquired:
-            holder_task_id = await redis.get(key)
+            holder_task_id = await self._redis.get(key)
-            if holder_task_id is not None:
+            if isinstance(holder_task_id, bytes):
-                holder_task_id = self._decode_task_id(holder_task_id)
+                holder_task_id = holder_task_id.decode()
             logger.warning(
                 "Duplicate task %s dropped (key=%s, holder_task_id=%s).",
                 message.task_name,

src/taskiq_deduplication/schedule.py

@@ -1,77 +0,0 @@
-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,9 +1,7 @@
 import pytest
 import fakeredis.aioredis
 from redis.asyncio import Redis
-from taskiq import ScheduledTask, TaskiqMessage, TaskiqResult
+from taskiq import TaskiqMessage, TaskiqResult
-
-from taskiq_deduplication import RedisDeduplicationMiddleware
 
 
 @pytest.fixture
@@ -18,13 +16,6 @@ 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")
@@ -42,13 +33,15 @@ async def real_redis():
 
 @pytest.fixture
 def make_message():
-    def _make(task_name="my_task", task_id="task-1", labels=None, kwargs=None):
+    def _make(
+        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=args or [],
             kwargs=kwargs or {},
         )
 
@@ -66,24 +59,3 @@ 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,6 +12,13 @@ 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})
@@ -79,6 +86,27 @@ 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

@@ -1,235 +0,0 @@
-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 = "10.21.3"
+version = "11.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "markdown" },
     { name = "pyyaml" },
 ]
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]
 
 [[package]]