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

[aradng]

parsed_consumer() always wraps the decoded body as {first_arg: body}, even when the message is empty. So solve_dependencies() (real, unmodified FastAPI internals, imported straight from fastapi.dependencies.utils) never sees a genuinely absent body - only a present dict like {"msg": None}. It has no way to take the no-body shortcut it already uses correctly for bodyless HTTP requests, so a required field on the target model fails validation before the handler body ever runs. On a tombstone-only topic that's every single message, forever.

The fix

AsyncConfluentParser.parse_message() now maps a literal null value to TOMBSTONE, a dedicated sentinel exported from faststream.message. decode_message() maps TOMBSTONE to None. parsed_consumer() passes real None through instead of wrapping it, so solve_dependencies() takes the same no-body shortcut it already has for HTTP. A genuinely empty (but not null) payload stays as raw bytes the whole way through and is still correctly rejected as invalid - only a true null value resolves to None.

Why a sentinel instead of a byte pattern

An earlier version of this fix just rewrote message.value() or b"" to ... or b"null", since json.loads(b"null") decodes cleanly to None (unlike b"", which raises). That's simpler, but it's still a real byte sequence someone could legitimately publish as actual data - some services already special-case value == b"null" for exactly this reason (conflating "the field is null" with "this is a delete"). A dedicated sentinel object can never collide with real content, however unlikely that collision would be.

Also worth flagging: or (truthiness) can't be swapped for if (an is None check) here without losing the fix entirely - b"" is also falsy, so value() or SENTINEL would silently swallow a genuinely empty payload into the sentinel too. The actual condition has to check identity, not truthiness.

Test plan

• tests/message/test_utils.py: decode_message() maps TOMBSTONE to None; a genuine b"" body stays b"".
• tests/brokers/confluent/test_parser.py: parse_message() maps a null value to TOMBSTONE; a genuine empty value stays bytes.
• tests/brokers/confluent/test_fastapi.py: connected test against a real broker - Optional[Model] = None resolves to None for an actual tombstone (constructed via the raw producer, same technique used in Bug: Broker subscribe handler falls in Error when try to handle message without value (but key exists!!) #1967's own repro, so this doesn't depend on any producer-side fix), while a valid payload still parses correctly.
• Full tests/message/ + tests/brokers/confluent/ suite passes (not connected/slow).
• ruff + mypy clean.

[Lancetnik]

Thanks for the follow-up — the problem is real (consuming a compacted topic through the FastAPI integration crash-loops on every tombstone today), and the FastAPI-side symptom fix works. But the chosen mechanism — a sentinel object traveling through the public message.body — moves the crash instead of removing it, and changes behavior outside the stated scope. Requesting changes on the points below (1–3 are blocking).

1. The sentinel crashes Prometheus / OpenTelemetry middlewares on every tombstone

ConfluentMetricsSettingsProvider.get_consume_attrs_from_message does len(msg.body) (faststream/confluent/prometheus/provider.py:36), and the OTel provider does the same (faststream/confluent/opentelemetry/provider.py:56). Both run in the consume path before the user handler. _Tombstone has no __len__, so any instrumented app gets TypeError: object of type '_Tombstone' has no len() on every tombstone — the exact bug class this PR fixes for FastAPI, reintroduced one layer up. The batch variants are worse: bytearray().join(msg.body) (prometheus :51, otel :87-88) fails the whole batch if it contains a single tombstone. Verified by running the PR code end-to-end.

Whatever representation is chosen, both providers (single + batch) need to be updated in this PR.

2. elif body is None in route.py silently changes behavior for all brokers, not just Kafka tombstones

parsed_consumer is shared cross-broker code. Any message whose decoded body is None — e.g. a plain b"null" JSON payload on RabbitMQ/NATS — now takes the new branch. Reproduced on in-memory Rabbit: a handler async def h(msg: dict) was previously invoked (receiving the leaked {"msg": None} wrapper), and now raises RequestValidationError: Field required before the handler runs. The old behavior was arguably broken too, but this is a user-visible semantic change on every broker and should be deliberate: documented, and covered by a base testcase in tests/brokers/base/fastapi.py so all brokers inherit it — not only by a confluent-local test.

3. The sibling aiokafka broker keeps the bug — and the sentinel is likely unnecessary

faststream/kafka/parser.py:45 / :77 still do body = message.value or b"", so KafkaBroker users get none of this fix while the special cases live in shared layers (decode_message, fastapi route). Both Kafka parsers should implement the same contract.

Related design point: StreamMessage.body is already annotated bytes | Any, so plain None expresses "tombstone" with zero new vocabulary — the parser sets body = message.value(), and decode_message needs the same one-line guard either way. That would (a) avoid a new forever-public TOMBSTONE symbol, (b) make finding #1 visible to mypy as len(Optional[bytes]), and (c) drop the __bool__ = False footgun, which makes if not msg.body: conflate tombstones with b"" again — the very distinction the sentinel exists to preserve. If the sentinel stays, note there's already an established sentinel idiom in the codebase (EMPTY in faststream/_internal/constants.py) it should follow, and the mapping expression is duplicated in parse_message/parse_batch (a third copy will appear with the aiokafka fix) — worth one helper.

4. In-memory TestKafkaBroker now diverges from the real broker

faststream/confluent/testing.py:349 encodes publish(None) through encode_message (→ b"") and MockConfluentMessage.value() can't represent a null value at all. So the same user code sees b"" under TestKafkaBroker and None against a real broker — user test suites will pass while production behaves differently. The testing path needs the same null-value story.

5. Test nits

• await asyncio.sleep(self.timeout) burns a fixed 10s per connected run and flakes if delivery is slower; the neighboring tests (test_batch_real, base FastAPITestcase) use an asyncio.Event + asyncio.wait(..., timeout=...) — same assertion, ~1s, no flake window.
• br._producer._producer.producer reaches through three levels of private internals; fine as a deliberate encoder bypass, but the exact-order list assert additionally relies on the topic having a single partition — an Event set at len(received) == 2 plus order-insensitive comparison (like the batch-key tests do) would be sturdier.
• TOMBSTONE is a new public export with no docs; if it survives the redesign discussion it needs a documentation entry.

Also: this PR and #2932 are two halves of one story (produce vs consume) — worth coordinating the merge so tombstones work end-to-end.

🤖 Generated with Claude Code

[aradng]

Related: #2932 (merged, producer side) and #2939 (test broker side, same gap). This PR covers the consumer/fastapi side. All three together make Model | Tombstone actually usable end to end.

[aradng]

Thanks, fixed all three blocking points, keeping the sentinel.

1. Prometheus and otel now check for the sentinel instead of assuming bytes, single and batch, confluent and kafka. Added body_size/batch_body_size helpers so this isn't duplicated per provider.

2. Scoped the route.py check to message.body is TOMBSTONE instead of body is None. Reproduced your rabbit repro before and after: async def h(msg: dict) receiving a null body still gets {'msg': None} exactly like before, unaffected now. Only a real tombstone takes the new path.

3. aiokafka parser now does the same TOMBSTONE mapping as confluent (pulled into a shared value_or_tombstone helper so it's one place, not three). Two existing kafka tests asserted the old b""-on-tombstone behavior, updated them to expect None.

Also fixed the sleep-based test, uses an event now.

On dropping the sentinel for plain None: I'd rather not. Your own point 2 is the reason why, None already means other things on other brokers, and your rabbit repro showed that collision is real. Folding tombstone into plain None would make that worse, not better, since a handler treating None as "optional field" would now silently also fire on "this was deleted" with no error at all. The sentinel is what lets us keep the two apart.

Separate question: does #2939 (TestKafkaBroker mocking) need to fold into this PR, or does it stay independent? Happy either way.

[aradng]

Added the TOMBSTONE note to kafka/message.md and confluent/message.md, next to the value() field list.

[aradng]

Pushed one more change. publish(None, ...) used to always send a real tombstone, but now None goes through the codec like any other value, so it's symmetric with what consume already does. TOMBSTONE is the explicit way to send a real tombstone now.

Two things I wasn't sure about here, and #2939 merged in the meantime assuming the old None = tombstone behavior, so you might not have caught this redesign yet. Happy to go either way:

1. Is this publish side change even in scope for this PR? It's breaking since it flips the existing tombstone on None behavior, I can split it out if you'd rather keep this PR scoped to consume/FastAPI.
2. Small thing, should plain None keep encoding to b"" like it does now, or match how everything else gets encoded and produce b"null"?

[aradng]

Ran a hard self review on the publish side change before you look again, found and fixed a real bug plus a few smaller things.

Real bug: publish_batch never checked for TOMBSTONE, only single message publish() did. A tombstone inside a batch fell through to the codec and crashed with a serialization error instead of producing a real tombstone. Fixed in both confluent and kafka, real producer and the test client.

Also cleaned up:

• extracted encode_or_tombstone() and ensure_tombstone_key() into message/utils.py so the tombstone check and the key requirement aren't copied across four places
• confluent now requires a key for a tombstone too, same as aiokafka. A keyless tombstone deletes nothing on either broker so there was no good reason for the difference
• switched the key check from SetupError to ValueError. Every other SetupError in this repo is a router or broker construction time error, not a per call publish argument problem
• renamed _Tombstone to Tombstone and exported it, since the test client needs it for an isinstance check at runtime, not just for typing
• moved the FastAPI tombstone test into a shared testcase so confluent and kafka both run it, added a negative case for a genuinely empty non-null body
• added tests for a tombstone inside a batch, and gave confluent its own consume tests for a plain tombstone, it had none before

All green, mypy and ruff clean.

[aradng]

hey @Lancetnik, would appreciate a re-review! :)