refactor(message): name the envelope-managed header keys

faststream_outbox/envelope.py

@@ -5,6 +5,8 @@
 from faststream.message import gen_cor_id
 from faststream.message.utils import encode_message
 
+from faststream_outbox.message import CONTENT_TYPE_HEADER, CORRELATION_ID_HEADER
+
 
 if TYPE_CHECKING:
     from fast_depends.library.serializer import SerializerProto
@@ -28,25 +30,25 @@ def _encode_payload(
     """
     payload, content_type = encode_message(body, serializer=serializer)
     out_headers: dict[str, str] = dict(headers or {})
-    if "content-type" in out_headers and content_type and out_headers["content-type"] != content_type:
+    if CONTENT_TYPE_HEADER in out_headers and content_type and out_headers[CONTENT_TYPE_HEADER] != content_type:
         msg = (
-            f"headers['content-type']={out_headers['content-type']!r} conflicts with the "
+            f"headers['content-type']={out_headers[CONTENT_TYPE_HEADER]!r} conflicts with the "
             f"encoder's output ({content_type!r}). Drop content-type from headers, or pass "
             "body as bytes if you need to label the payload yourself."
         )
         raise ValueError(msg)
     if content_type:
-        out_headers["content-type"] = content_type
+        out_headers[CONTENT_TYPE_HEADER] = content_type
     # An explicit ``correlation_id`` kwarg used to lose silently to a
     # ``headers["correlation_id"]`` of a different value (the kwarg was dropped).
     # Treat a genuine mismatch as a conflict (like content-type above); otherwise the
     # kwarg wins when set, falling back to the header, then a fresh id (P2).
-    header_cid = out_headers.get("correlation_id")
+    header_cid = out_headers.get(CORRELATION_ID_HEADER)
     if correlation_id is not None and header_cid is not None and correlation_id != header_cid:
         msg = (
             f"correlation_id={correlation_id!r} conflicts with headers['correlation_id']="
             f"{header_cid!r}. Pass correlation_id one way, not both."
         )
         raise ValueError(msg)
-    out_headers["correlation_id"] = correlation_id or header_cid or gen_cor_id()
+    out_headers[CORRELATION_ID_HEADER] = correlation_id or header_cid or gen_cor_id()
     return payload, out_headers

faststream_outbox/message.py

@@ -36,6 +36,16 @@
 DLQFailureReason = Literal["max_deliveries", "retry_terminal", "rejected"]
 
 
+# Header keys the outbox envelope owns on a row's ``headers`` dict. ``_encode_payload``
+# (``envelope.py``) writes them, ``OutboxParser.parse_message`` (``parser.py``) reads
+# them back, and the relay header-propagation (``_maybe_propagate_inbound_headers``)
+# strips them before re-encode. Named here so those sites share one definition instead
+# of repeating the string literals.
+CONTENT_TYPE_HEADER = "content-type"
+CORRELATION_ID_HEADER = "correlation_id"
+ENVELOPE_MANAGED_HEADERS = frozenset({CONTENT_TYPE_HEADER, CORRELATION_ID_HEADER})
+
+
 @dataclass(kw_only=True)
 class OutboxInnerMessage:
     """In-memory copy of a claimed outbox row, plus ack/nack/reject intent helpers.

faststream_outbox/parser/parser.py

@@ -2,7 +2,12 @@
 
 from faststream.message import decode_message
 
-from faststream_outbox.message import OutboxInnerMessage, OutboxMessage
+from faststream_outbox.message import (
+    CONTENT_TYPE_HEADER,
+    CORRELATION_ID_HEADER,
+    OutboxInnerMessage,
+    OutboxMessage,
+)
 
 
 class OutboxParser:
@@ -12,9 +17,9 @@ async def parse_message(self, msg: OutboxInnerMessage) -> OutboxMessage:
             raw_message=msg,
             body=msg.payload,
             headers=headers,
-            content_type=headers.get("content-type"),
+            content_type=headers.get(CONTENT_TYPE_HEADER),
             message_id=str(msg.id),
-            correlation_id=headers.get("correlation_id", str(msg.id)),
+            correlation_id=headers.get(CORRELATION_ID_HEADER, str(msg.id)),
             # Set so ``SubscriberUsecase.__get_response_publisher`` (gated on a
             # truthy ``reply_to``) wires up the response publisher when a handler
             # returns ``OutboxResponse``. The inbound queue is the semantically

faststream_outbox/subscriber/usecase.py

@@ -39,7 +39,7 @@
 from faststream.specification.schema import Message, Operation, SubscriberSpec
 from typing_extensions import override
 
-from faststream_outbox.message import OutboxInnerMessage
+from faststream_outbox.message import ENVELOPE_MANAGED_HEADERS, OutboxInnerMessage
 from faststream_outbox.parser.parser import OutboxParser
 from faststream_outbox.publisher.fake import OutboxFakePublisher
 from faststream_outbox.response import OutboxResponse
@@ -1089,7 +1089,7 @@ def _maybe_propagate_inbound_headers(
             return
         propagated = dict(message.headers)
         if isinstance(result_msg, OutboxResponse):
-            for managed in ("content-type", "correlation_id"):
+            for managed in ENVELOPE_MANAGED_HEADERS:
                 propagated.pop(managed, None)
         result_msg.headers = propagated