feat: add skip_result label to skip result backend storage

docs/available-components/result-backends.md

@@ -10,6 +10,32 @@ This includes:
 - return value;
 - Execution time in seconds.
 
+## Skipping result storage
+
+Sometimes you don't need task results at all (fire-and-forget jobs, notifications, side effects only).
+You can skip writing to the result backend for selected tasks with the `skip_result` label:
+
+```python
+@broker.task(skip_result=True)
+async def push_notification(user_id: int) -> None:
+    ...
+```
+
+Or only for a single call:
+
+```python
+await push_notification.kicker().with_labels(skip_result=True).kiq(user_id=1)
+```
+
+When `skip_result` is enabled:
+
+- worker does **not** call `result_backend.set_result`;
+- `post_save` middleware hooks are **not** executed;
+- callers that use `wait_result()` will time out, because nothing is stored.
+
+This is complementary to raising `NoResultError` from inside a task.
+Use the label when the decision is static; raise `NoResultError` when you decide at runtime.
+
 ## Built-in result backends
 
 ### DummyResultBackend

docs/guide/architecture-overview.md

@@ -118,6 +118,15 @@ async def main():
     ).kiq()
 ```
 
+Built-in labels include things like `timeout`, `ack_type`, and `skip_result`.
+For example, fire-and-forget tasks can avoid writing to the result backend:
+
+```python
+@broker.task(skip_result=True)
+async def send_email(user_id: int) -> None:
+    ...
+```
+
 Also you can assign custom task names using decorator.
 This is useful to be sure that task names are unique and resolved correctly.
 Also it may be useful to balance message routing in some brokers.

docs/guide/getting-started.md

@@ -219,6 +219,31 @@ Returned value: 2
 Continue reading to get more information about taskiq internals.
 
 
+## Skipping results
+
+If a task is fire-and-forget and you do not need `wait_result()`,
+set the `skip_result` label so the worker will not store anything in the result backend.
+
+::: tabs
+
+@tab decorator
+
+```python
+@broker.task(skip_result=True)
+async def push_event(payload: dict) -> None:
+    ...
+```
+
+@tab when calling
+
+```python
+await push_event.kicker().with_labels(skip_result=True).kiq(payload={...})
+```
+
+:::
+
+You can also raise `NoResultError` inside a task to skip storage dynamically.
+
 ## Timeouts
 
 If you want to restrict amount of time you want to run task,

taskiq/kicker.py

@@ -55,11 +55,14 @@ def __init__(
 
     def with_labels(
         self,
-        **labels: str | float,
+        **labels: str | float | bool | bytes,
     ) -> "AsyncKicker[_FuncParams, _ReturnType]":
         """
         Update function's labels before sending.
 
+        Supported value types match label serialization:
+        ``str``, ``float`` (also accepts ``int``), ``bool`` and ``bytes``.
+
         :param labels: new labels.
         :return: kicker with new labels.
         """

taskiq/receiver/receiver.py

@@ -174,8 +174,9 @@ async def callback(  # noqa: C901, PLR0912
             if middleware.__class__.post_execute != TaskiqMiddleware.post_execute:
                 await maybe_awaitable(middleware.post_execute(taskiq_msg, result))
 
+        should_save = self._should_save_result(taskiq_msg, result)
         try:
-            if not isinstance(result.error, NoResultError):
+            if should_save:
                 await self.broker.result_backend.set_result(taskiq_msg.task_id, result)
 
                 for middleware in reversed(self.broker.middlewares):
@@ -210,6 +211,46 @@ def _get_ack_time(self, message: TaskiqMessage) -> AcknowledgeType:
                 f"Invalid ack_type label {ack_type!r} for task {message.task_name}.",
             ) from exc
 
+    def _should_save_result(
+        self,
+        message: TaskiqMessage,
+        result: TaskiqResult[Any],
+    ) -> bool:
+        """
+        Decide whether execution result should be stored.
+
+        Results are skipped when:
+        * task raised ``NoResultError``;
+        * task has ``skip_result`` label set to ``True``.
+        """
+        if isinstance(result.error, NoResultError):
+            return False
+        if self._is_skip_result(message):
+            logger.debug(
+                "Task %s with id %s has skip_result label. Skipping result backend.",
+                message.task_name,
+                message.task_id,
+            )
+            return False
+        return True
+
+    @staticmethod
+    def _is_skip_result(message: TaskiqMessage) -> bool:
+        """
+        Check whether ``skip_result`` label is enabled.
+
+        Only ``True`` and ``False`` are allowed. Missing label means ``False``.
+        """
+        skip_result = message.labels.get("skip_result")
+        if skip_result is None:
+            return False
+        if isinstance(skip_result, bool):
+            return skip_result
+        raise ValueError(
+            f"Invalid skip_result label {skip_result!r} for task "
+            f"{message.task_name}. Expected True or False.",
+        )
+
     async def run_task(  # noqa: C901, PLR0912, PLR0915
         self,
         target: Callable[..., Any],

tests/receiver/test_receiver.py

@@ -772,6 +772,142 @@ async def task_no_result() -> int:
     assert not broker._running_tasks
 
 
+async def test_skip_result_label_on_decorator() -> None:
+    """Task skip_result label skips result backend and post_save."""
+    events: list[str] = []
+    broker = (
+        InMemoryBroker()
+        .with_result_backend(
+            _EventResultBackend(events),
+        )
+        .with_middlewares(_EventMiddleware(events))
+    )
+
+    @broker.task(skip_result=True)
+    async def my_task() -> int:
+        events.append("task")
+        return 1
+
+    receiver = get_receiver(broker)
+    broker_message = broker.formatter.dumps(my_task.kicker()._prepare_message())
+
+    await receiver.callback(broker_message.message)
+
+    assert events == ["task", "post_execute"]
+    assert not await broker.result_backend.is_result_ready(
+        broker_message.task_id,
+    )
+
+
+async def test_skip_result_label_via_kicker() -> None:
+    """skip_result can be set per call with kicker labels."""
+    events: list[str] = []
+    broker = (
+        InMemoryBroker()
+        .with_result_backend(
+            _EventResultBackend(events),
+        )
+        .with_middlewares(_EventMiddleware(events))
+    )
+
+    @broker.task
+    async def my_task() -> int:
+        events.append("task")
+        return 1
+
+    receiver = get_receiver(broker)
+    broker_message = broker.formatter.dumps(
+        my_task.kicker().with_labels(skip_result=True)._prepare_message(),
+    )
+
+    await receiver.callback(broker_message.message)
+
+    assert events == ["task", "post_execute"]
+    assert not await broker.result_backend.is_result_ready(
+        broker_message.task_id,
+    )
+
+
+async def test_skip_result_false_still_saves_result() -> None:
+    """Explicit skip_result=False keeps default result storage behavior."""
+    events: list[str] = []
+    broker = (
+        InMemoryBroker()
+        .with_result_backend(
+            _EventResultBackend(events),
+        )
+        .with_middlewares(_EventMiddleware(events))
+    )
+
+    @broker.task(skip_result=False)
+    async def my_task() -> int:
+        events.append("task")
+        return 1
+
+    receiver = get_receiver(broker)
+    broker_message = broker.formatter.dumps(my_task.kicker()._prepare_message())
+
+    await receiver.callback(broker_message.message)
+
+    assert events == ["task", "post_execute", "save", "post_save"]
+    assert await broker.result_backend.is_result_ready(broker_message.task_id)
+
+
+async def test_skip_result_invalid_value_raises() -> None:
+    """Non-bool skip_result values are rejected."""
+    broker = InMemoryBroker()
+
+    @broker.task
+    async def my_task() -> int:
+        return 1
+
+    receiver = get_receiver(broker)
+    broker_message = broker.formatter.dumps(
+        TaskiqMessage(
+            task_id="skip-result-invalid",
+            task_name=my_task.task_name,
+            labels={"skip_result": "true"},
+            args=[],
+            kwargs={},
+        ),
+    )
+
+    with pytest.raises(ValueError, match="Invalid skip_result label"):
+        await receiver.callback(broker_message.message)
+
+
+async def test_skip_result_still_acks_when_saved() -> None:
+    """WHEN_SAVED ack still happens if result storage is skipped."""
+    events: list[str] = []
+    broker = (
+        InMemoryBroker()
+        .with_result_backend(
+            _EventResultBackend(events),
+        )
+        .with_middlewares(_EventMiddleware(events))
+    )
+
+    @broker.task(skip_result=True)
+    async def my_task() -> int:
+        events.append("task")
+        return 1
+
+    def ack_callback() -> None:
+        events.append("ack")
+
+    receiver = get_receiver(broker, ack_type=AcknowledgeType.WHEN_SAVED)
+    broker_message = broker.formatter.dumps(my_task.kicker()._prepare_message())
+
+    await receiver.callback(
+        AckableMessage(
+            data=broker_message.message,
+            ack=ack_callback,
+        ),
+    )
+
+    assert events == ["task", "post_execute", "ack"]
+
+
 async def test_result() -> None:
     broker = InMemoryBroker()