fix(fastapi): resolve Optional body params to None for a real tombstone

docs/docs/en/confluent/message.md

@@ -26,6 +26,11 @@ This object serves as a unified **FastStream** wrapper around the native broker
 * `#!python topic(): str`
 * `#!python value(): Optional[Union[str, bytes]]`
 
+!!! note
+    A `None` value() is a Kafka tombstone, the delete marker on a compacted topic. FastStream represents it as `faststream.message.TOMBSTONE`, kept distinct from an empty payload (`b""`).
+
+    To publish a tombstone, pass `faststream.message.TOMBSTONE` as the message body, along with a key. `broker.publish(TOMBSTONE, key=b"...")`
+
 For example, if you would like to access the headers of an incoming message, you would do so like this:
 
 ```python hl_lines="1 6"

docs/docs/en/kafka/message.md

@@ -31,6 +31,11 @@ This object serves as a unified **FastStream** wrapper around the native broker
 * `#!python topic: str`
 * `#!python value: Optional[aiokafka.structs.VT]`
 
+!!! note
+    A `None` value is a Kafka tombstone, the delete marker on a compacted topic. FastStream represents it as `faststream.message.TOMBSTONE`, kept distinct from an empty payload (`b""`).
+
+    To publish a tombstone, pass `faststream.message.TOMBSTONE` as the message body and a key. `broker.publish(TOMBSTONE, key=b"...")`
+
 For example, if you would like to access the headers of an incoming message, you would do so like this:
 
 ```python hl_lines="1 6"

faststream/_internal/fastapi/route.py

@@ -18,6 +18,7 @@
 from faststream._internal.context import Context, ContextRepo
 from faststream._internal.types import P_HandlerParams, T_HandlerReturn
 from faststream.exceptions import SetupError
+from faststream.message import TOMBSTONE
 from faststream.response import Response, ensure_response
 
 from ._compat import (
@@ -51,13 +52,13 @@ class StreamMessage(Request):
     scope: "dict[str, Any]"
     _cookies: "dict[str, Any]"
     _headers: "dict[str, Any]"  # type: ignore[assignment]
-    _body: Union["dict[str, Any]", list[Any]]  # type: ignore[assignment]
+    _body: Union["dict[str, Any]", list[Any], None]  # type: ignore[assignment]
     _query_params: "dict[str, Any]"  # type: ignore[assignment]
 
     def __init__(
         self,
         *,
-        body: Union["dict[str, Any]", list[Any]],
+        body: Union["dict[str, Any]", list[Any], None],
         headers: "dict[str, Any]",
         path: "dict[str, Any]",
     ) -> None:
@@ -172,9 +173,15 @@ async def parsed_consumer(message: "NativeMessage[Any]") -> Any:
         """Wrapper, that parser FastStream message to FastAPI compatible one."""
         body = await message.decode()
 
-        fastapi_body: dict[str, Any] | list[Any]
+        fastapi_body: dict[str, Any] | list[Any] | None
         if first_arg is not None:
-            if isinstance(body, dict):
+            # NOTE: checks the raw pre-decode body, not the already-decoded
+            # `body` local (also None here) - a real b"null" payload decodes
+            # to None too but must still fall through to {first_arg: body}
+            # below, unchanged from before this sentinel existed.
+            if message.body is TOMBSTONE:
+                fastapi_body, path = None, {}
+            elif isinstance(body, dict):
                 path = fastapi_body = body or {}
             elif isinstance(body, list):
                 fastapi_body, path = body, {}

faststream/confluent/opentelemetry/provider.py

@@ -5,6 +5,7 @@
 
 from faststream._internal.types import MsgType
 from faststream.confluent.response import KafkaPublishCommand
+from faststream.message import batch_body_size, body_size
 from faststream.opentelemetry import TelemetrySettingsProvider
 from faststream.opentelemetry.consts import MESSAGING_DESTINATION_PUBLISH_NAME
 
@@ -53,7 +54,7 @@ def get_consume_attrs_from_message(
             SpanAttributes.MESSAGING_SYSTEM: self.messaging_system,
             SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id,
             SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id,
-            SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len(msg.body),
+            SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: body_size(msg.body),
             SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: msg.raw_message.partition(),
             SpanAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET: msg.raw_message.offset(),
             MESSAGING_DESTINATION_PUBLISH_NAME: msg.raw_message.topic(),
@@ -84,8 +85,8 @@ def get_consume_attrs_from_message(
             SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id,
             SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id,
             SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT: len(msg.raw_message),
-            SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len(
-                bytearray().join(cast("Sequence[bytes]", msg.body)),
+            SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: batch_body_size(
+                msg.body
             ),
             SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: raw_message.partition(),
             MESSAGING_DESTINATION_PUBLISH_NAME: raw_message.topic(),

faststream/confluent/parser.py

@@ -1,6 +1,6 @@
 from typing import TYPE_CHECKING, Any, cast
 
-from faststream.message import StreamMessage, decode_message
+from faststream.message import StreamMessage, decode_message, value_or_tombstone
 
 from .message import FAKE_CONSUMER, KafkaMessage
 
@@ -38,7 +38,7 @@ async def parse_message(
         """Parses a Kafka message."""
         headers = _parse_msg_headers(cast("_HeadersInput", message.headers() or ()))
 
-        body = message.value() or b""
+        body = value_or_tombstone(message.value())
         offset = message.offset()
         _, timestamp = message.timestamp()
 
@@ -66,7 +66,7 @@ async def parse_batch(
         last = message[-1]
 
         for m in message:
-            body.append(m.value() or b"")
+            body.append(value_or_tombstone(m.value()))
             batch_headers.append(
                 _parse_msg_headers(cast("_HeadersInput", m.headers() or ()))
             )

faststream/confluent/prometheus/provider.py

@@ -1,6 +1,7 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Union, cast
 
+from faststream.message import batch_body_size, body_size
 from faststream.message.message import MsgType, StreamMessage
 from faststream.prometheus import (
     ConsumeAttrs,
@@ -33,7 +34,7 @@ def get_consume_attrs_from_message(
     ) -> ConsumeAttrs:
         return {
             "destination_name": cast("str", msg.raw_message.topic()),
-            "message_size": len(msg.body),
+            "message_size": body_size(msg.body),
             "messages_count": 1,
         }
 
@@ -48,7 +49,7 @@ def get_consume_attrs_from_message(
         raw_message = msg.raw_message[0]
         return {
             "destination_name": cast("str", raw_message.topic()),
-            "message_size": len(bytearray().join(cast("Sequence[bytes]", msg.body))),
+            "message_size": batch_body_size(msg.body),
             "messages_count": len(msg.raw_message),
         }
 

faststream/confluent/publisher/producer.py

@@ -9,11 +9,14 @@
 from faststream.confluent.parser import AsyncConfluentParser
 from faststream.confluent.response import KafkaPublishCommand
 from faststream.exceptions import FeatureNotSupportedException
+from faststream.message import TOMBSTONE, Tombstone
+from faststream.message.utils import encode_or_tombstone, ensure_tombstone_key
 
 from .state import EmptyProducerState, ProducerState, RealProducer
 
 if TYPE_CHECKING:
     import asyncio
+    from collections.abc import Sequence
 
     from confluent_kafka import Message
     from fast_depends.library.serializer import SerializerProto
@@ -139,10 +142,11 @@ async def publish(
         cmd: "KafkaPublishCommand",
     ) -> "asyncio.Future[Message | None] | Message | None":
         """Publish a message to a topic."""
-        if cmd.body is None:
-            message, content_type = None, None
-        else:
-            message, content_type = await self.codec.encode(cmd.body, self.serializer)
+        if cmd.body is TOMBSTONE:
+            ensure_tombstone_key(cmd.key)
+        message, content_type = await encode_or_tombstone(
+            cmd.body, self.codec, self.serializer
+        )
 
         headers_to_send = {
             "content-type": content_type or "",
@@ -166,13 +170,23 @@ async def publish_batch(self, cmd: "KafkaPublishCommand") -> None:
 
         headers_to_send = cmd.headers_to_publish()
 
+        encoded_batch: Sequence[tuple[bytes | None, str | None]]
         if isinstance(self.codec, BatchCodecProto):
+            if any(isinstance(body, Tombstone) for body in cmd.batch_bodies):
+                msg = (
+                    "a tombstone in a batch isn't supported with a custom BatchCodecProto"
+                )
+                raise ValueError(msg)
             encoded_batch = await self.codec.encode_batch(
                 cmd.batch_bodies, self.serializer
             )
         else:
+            for message_position, body in enumerate(cmd.batch_bodies):
+                if body is TOMBSTONE:
+                    ensure_tombstone_key(cmd.key_for(message_position))
             encoded_batch = [
-                await self.codec.encode(msg, self.serializer) for msg in cmd.batch_bodies
+                await encode_or_tombstone(msg, self.codec, self.serializer)
+                for msg in cmd.batch_bodies
             ]
 
         for message_position, (message, content_type) in enumerate(encoded_batch):

faststream/confluent/response.py

@@ -10,6 +10,7 @@
     Response,
     extract_per_message_keys_and_bodies,
     key_for_index,
+    realign_keys,
 )
 
 if TYPE_CHECKING:
@@ -100,9 +101,9 @@ def __init__(
 
         # per-message keys support
         keys, normalized = extract_per_message_keys_and_bodies(self.batch_bodies)
+        self._per_message_keys = keys
         if normalized is not None:
             self.batch_bodies = normalized
-        self._per_message_keys = keys
 
     @classmethod
     def from_cmd(
@@ -153,29 +154,12 @@ def batch_bodies(self, value: Sequence["Any"]) -> None:
 
     def _align_keys(self, value: Sequence["Any"]) -> None:
         """Align the per-message keys with the batch_bodies."""
-        if len(self.batch_bodies) == 0:
-            return
-        if isinstance(self.batch_bodies[0], KafkaResponse):
+        if not self._per_message_keys:
             return
-        new_indexes = self._form_indexes(value)
-        self._per_message_keys = tuple(self._per_message_keys[i] for i in new_indexes)
-
-    def _form_indexes(self, value: Sequence["Any"]) -> tuple[int, ...]:
-        """Form a list of indexes for the given value sequence based on batch_bodies."""
-        bodies_seen: dict[int, Any] = {}
-        for body in value:
-            index = self.batch_bodies.index(body)
-            self._update_bodies(bodies_seen, index, body)
-
-        return tuple(i for i in bodies_seen)
-
-    def _update_bodies(self, bodies_seen: dict[int, Any], index: int, body: Any) -> None:
-        """Update the bodies_seen dictionary with the given index and body."""
-        if self.batch_bodies[index] == body and bodies_seen.get(index) is None:
-            bodies_seen.update({index: body})
-        else:
-            index += 1
-            self._update_bodies(bodies_seen, index, body)
+
+        self._per_message_keys = realign_keys(
+            self._per_message_keys, self.batch_bodies, value
+        )
 
 
 # Semantic alias for publish operations

faststream/confluent/testing.py

@@ -21,7 +21,8 @@
 from faststream.confluent.schemas import TopicPartition
 from faststream.confluent.subscriber.usecase import BatchSubscriber
 from faststream.exceptions import SubscriberNotFound
-from faststream.message import gen_cor_id
+from faststream.message import TOMBSTONE, Tombstone, gen_cor_id
+from faststream.message.utils import encode_or_tombstone, ensure_tombstone_key
 
 if TYPE_CHECKING:
     from fast_depends.library.serializer import SerializerProto
@@ -194,11 +195,21 @@ async def publish_batch(self, cmd: "KafkaPublishCommand") -> None:
         """Publish a batch of messages to the Kafka broker."""
         serializer = self.broker.config.fd_config._serializer
 
+        encoded: Sequence[tuple[bytes | None, str | None]]
         if isinstance(self.codec, BatchCodecProto):
+            if any(isinstance(body, Tombstone) for body in cmd.batch_bodies):
+                msg = (
+                    "a tombstone in a batch isn't supported with a custom BatchCodecProto"
+                )
+                raise ValueError(msg)
             encoded = await self.codec.encode_batch(cmd.batch_bodies, serializer)
         else:
+            for message_position, body in enumerate(cmd.batch_bodies):
+                if body is TOMBSTONE:
+                    ensure_tombstone_key(cmd.key_for(message_position))
             encoded = [
-                await self.codec.encode(body, serializer) for body in cmd.batch_bodies
+                await encode_or_tombstone(body, self.codec, serializer)
+                for body in cmd.batch_bodies
             ]
 
         for handler in _find_handler(
@@ -332,7 +343,7 @@ def value(self) -> bytes | None:
 
 
 async def build_message(
-    message: "SendableMessage",
+    message: "SendableMessage | Tombstone",
     topic: str,
     *,
     correlation_id: str | None = None,
@@ -345,12 +356,11 @@ async def build_message(
     codec: Optional["CodecProto"] = None,
 ) -> MockConfluentMessage:
     """Build a mock confluent_kafka.Message for a sendable message."""
-    if message is None:
-        # keep a real tombstone (message.value() is None) distinct from b""
-        msg, content_type = None, None
-    else:
-        codec_instance = codec or DefaultCodec()
-        msg, content_type = await codec_instance.encode(message, serializer)
+    if message is TOMBSTONE:
+        ensure_tombstone_key(key)
+    msg, content_type = await encode_or_tombstone(
+        message, codec or DefaultCodec(), serializer
+    )
     k = key or b""
     headers = {
         "content-type": content_type or "",
@@ -373,7 +383,7 @@ async def build_message(
 
 
 def _build_mock_message(
-    body: bytes,
+    body: bytes | None,
     content_type: str | None,
     topic: str,
     partition: int | None = None,

faststream/kafka/opentelemetry/provider.py

@@ -5,6 +5,7 @@
 
 from faststream._internal.types import MsgType
 from faststream.kafka.response import KafkaPublishCommand
+from faststream.message import batch_body_size, body_size
 from faststream.opentelemetry import TelemetrySettingsProvider
 from faststream.opentelemetry.consts import MESSAGING_DESTINATION_PUBLISH_NAME
 
@@ -59,7 +60,7 @@ def get_consume_attrs_from_message(
             SpanAttributes.MESSAGING_SYSTEM: self.messaging_system,
             SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id,
             SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id,
-            SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len(msg.body),
+            SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: body_size(msg.body),
             SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: msg.raw_message.partition,
             SpanAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET: msg.raw_message.offset,
             MESSAGING_DESTINATION_PUBLISH_NAME: msg.raw_message.topic,
@@ -90,8 +91,8 @@ def get_consume_attrs_from_message(
             SpanAttributes.MESSAGING_SYSTEM: self.messaging_system,
             SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id,
             SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id,
-            SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len(
-                bytearray().join(cast("Sequence[bytes]", msg.body)),
+            SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: batch_body_size(
+                msg.body
             ),
             SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT: len(msg.raw_message),
             SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION: raw_message.partition,

faststream/kafka/parser.py

@@ -7,7 +7,7 @@
     KafkaMessage,
     KafkaRawMessage,
 )
-from faststream.message import decode_message
+from faststream.message import decode_message, value_or_tombstone
 
 if TYPE_CHECKING:
     from re import Pattern
@@ -42,7 +42,7 @@ async def parse_message(
         headers = {i: j.decode() for i, j in message.headers}
 
         return self.msg_class(
-            body=message.value or b"",
+            body=value_or_tombstone(message.value),
             headers=headers,
             reply_to=headers.get("reply_to", ""),
             content_type=headers.get("content-type"),
@@ -74,7 +74,7 @@ async def parse_batch(
         last = message[-1]
 
         for m in message:
-            body.append(m.value or b"")
+            body.append(value_or_tombstone(m.value))
             batch_headers.append({i: j.decode() for i, j in m.headers})
 
         headers = next(iter(batch_headers), {})