From 7b5dd0c8a937d451e1b3d76ffc6665307027f352 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sat, 6 Jun 2026 08:42:36 -0500 Subject: [PATCH 1/2] feat(fix): surface inbound decode failures (closes #432) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit decode_app_message returns None for two distinct cases — no registered model (admin/unimplemented) and a registered model whose payload failed validation — so a genuine ExecutionReport lost to one off-spec field was silently dropped. - errors.py: FixDecodeError (carries raw + msg_type; chains the underlying validation error) for the registered-but-malformed case. - dispatch.py: decode_app_message_strict raises FixDecodeError on a malformed known message (None only for unregistered types); decode_app_message now delegates to it and swallows (its lenient None contract is unchanged). - FixSession: optional on_decode_error hook (the issue's recommended option 3). When set, the session decodes each inbound app message to detect a malformed known one and routes it (raw + FixDecodeError) so nothing is silently lost; the raw is still delivered to on_message. Zero behaviour change when unset; threaded through all client session factories. 7 new tests: strict vs lenient decode (valid / malformed-raises / unregistered- None / swallowed), and FixSession routing — malformed ExecutionReport surfaces via on_decode_error (raw still delivered), a valid report does not trigger it, and without the hook the raw is still delivered and the session survives. ruff + mypy --strict clean; 259 FIX tests pass. Closes #432. Part of #402. Co-Authored-By: Claude Opus 4.8 (1M context) --- kalshi/fix/__init__.py | 4 + kalshi/fix/client.py | 39 +++++-- kalshi/fix/errors.py | 26 +++++ kalshi/fix/messages/__init__.py | 7 +- kalshi/fix/messages/dispatch.py | 37 +++++-- kalshi/fix/session.py | 26 ++++- tests/fix/test_decode_routing.py | 178 +++++++++++++++++++++++++++++++ 7 files changed, 301 insertions(+), 16 deletions(-) create mode 100644 tests/fix/test_decode_routing.py diff --git a/kalshi/fix/__init__.py b/kalshi/fix/__init__.py index 88bd004..f4be120 100644 --- a/kalshi/fix/__init__.py +++ b/kalshi/fix/__init__.py @@ -73,6 +73,7 @@ from kalshi.fix.errors import ( FixCodecError, FixConnectionError, + FixDecodeError, FixLogonError, FixRejectError, FixSequenceError, @@ -133,6 +134,7 @@ SequenceReset, TestRequest, decode_app_message, + decode_app_message_strict, groupfield, ) from kalshi.fix.orderbook import BookLevel, FixOrderBook, MarketDataBook @@ -166,6 +168,7 @@ "FixConfig", "FixConnection", "FixConnectionError", + "FixDecodeError", "FixEnvironment", "FixGroup", "FixGroupMeta", @@ -248,6 +251,7 @@ "TimeInForce", "decode", "decode_app_message", + "decode_app_message_strict", "encode", "groupfield", ] diff --git a/kalshi/fix/client.py b/kalshi/fix/client.py index e9be68a..4b734f6 100644 --- a/kalshi/fix/client.py +++ b/kalshi/fix/client.py @@ -15,7 +15,12 @@ from kalshi.auth import KalshiAuth from kalshi.fix.auth import FixSigner from kalshi.fix.config import FixConfig, FixEnvironment, FixProduct, FixSessionType -from kalshi.fix.session import FixSession, MessageHandler, StateChangeHandler +from kalshi.fix.session import ( + DecodeErrorHandler, + FixSession, + MessageHandler, + StateChangeHandler, +) class _BaseFixClient: @@ -80,6 +85,7 @@ def session( *, on_message: MessageHandler | None = None, on_state_change: StateChangeHandler | None = None, + on_decode_error: DecodeErrorHandler | None = None, ) -> FixSession: """Construct (but do not start) a :class:`FixSession` for ``session_type``.""" return FixSession( @@ -88,6 +94,7 @@ def session( session_type, on_message=on_message, on_state_change=on_state_change, + on_decode_error=on_decode_error, ssl_context=self._ssl_context, ) @@ -97,13 +104,17 @@ def order_entry( retransmission: bool = False, on_message: MessageHandler | None = None, on_state_change: StateChangeHandler | None = None, + on_decode_error: DecodeErrorHandler | None = None, ) -> FixSession: """An order-entry session (KalshiNR, or KalshiRT when ``retransmission``).""" session_type = ( FixSessionType.ORDER_ENTRY_RT if retransmission else FixSessionType.ORDER_ENTRY_NR ) return self.session( - session_type, on_message=on_message, on_state_change=on_state_change + session_type, + on_message=on_message, + on_state_change=on_state_change, + on_decode_error=on_decode_error, ) def drop_copy( @@ -111,10 +122,14 @@ def drop_copy( *, on_message: MessageHandler | None = None, on_state_change: StateChangeHandler | None = None, + on_decode_error: DecodeErrorHandler | None = None, ) -> FixSession: """A drop-copy (KalshiDC) session for historical execution-report queries.""" return self.session( - FixSessionType.DROP_COPY, on_message=on_message, on_state_change=on_state_change + FixSessionType.DROP_COPY, + on_message=on_message, + on_state_change=on_state_change, + on_decode_error=on_decode_error, ) def market_data( @@ -122,10 +137,14 @@ def market_data( *, on_message: MessageHandler | None = None, on_state_change: StateChangeHandler | None = None, + on_decode_error: DecodeErrorHandler | None = None, ) -> FixSession: """A market-data (KalshiMD) session for order-book snapshots and updates.""" return self.session( - FixSessionType.MARKET_DATA, on_message=on_message, on_state_change=on_state_change + FixSessionType.MARKET_DATA, + on_message=on_message, + on_state_change=on_state_change, + on_decode_error=on_decode_error, ) @@ -166,10 +185,14 @@ def post_trade( *, on_message: MessageHandler | None = None, on_state_change: StateChangeHandler | None = None, + on_decode_error: DecodeErrorHandler | None = None, ) -> FixSession: """A post-trade (KalshiPT) session for market-settlement reports.""" return self.session( - FixSessionType.POST_TRADE, on_message=on_message, on_state_change=on_state_change + FixSessionType.POST_TRADE, + on_message=on_message, + on_state_change=on_state_change, + on_decode_error=on_decode_error, ) def rfq( @@ -177,8 +200,12 @@ def rfq( *, on_message: MessageHandler | None = None, on_state_change: StateChangeHandler | None = None, + on_decode_error: DecodeErrorHandler | None = None, ) -> FixSession: """An RFQ (KalshiRFQ) market-maker session.""" return self.session( - FixSessionType.RFQ, on_message=on_message, on_state_change=on_state_change + FixSessionType.RFQ, + on_message=on_message, + on_state_change=on_state_change, + on_decode_error=on_decode_error, ) diff --git a/kalshi/fix/errors.py b/kalshi/fix/errors.py index b7f40eb..fdcaf5f 100644 --- a/kalshi/fix/errors.py +++ b/kalshi/fix/errors.py @@ -9,8 +9,13 @@ from __future__ import annotations +from typing import TYPE_CHECKING + from kalshi.errors import KalshiError +if TYPE_CHECKING: + from kalshi.fix.codec import RawMessage + class KalshiFixError(KalshiError): """Base exception for all Kalshi FIX errors. @@ -95,6 +100,27 @@ def __init__( super().__init__(message) +class FixDecodeError(KalshiFixError): + """A registered inbound application message failed schema validation. + + Distinguishes a *malformed* known message (a real message with an off-spec + field — e.g. a bad ``DollarDecimal`` / ``FixedPointCount``) from an + *unregistered* message type. :func:`~kalshi.fix.messages.decode_app_message` + collapses both to ``None``; + :func:`~kalshi.fix.messages.decode_app_message_strict` raises this for the + former so the failure is observable (see ``FixSession``'s ``on_decode_error`` + hook). ``raw`` carries the offending message; the underlying validation error + is chained via ``__cause__``. + """ + + def __init__( + self, message: str, *, raw: RawMessage | None = None, msg_type: str | None = None + ) -> None: + self.raw = raw + self.msg_type = msg_type + super().__init__(message) + + class FixRejectError(KalshiFixError): """The acceptor rejected a message we sent. diff --git a/kalshi/fix/messages/__init__.py b/kalshi/fix/messages/__init__.py index bafbd53..7579b29 100644 --- a/kalshi/fix/messages/__init__.py +++ b/kalshi/fix/messages/__init__.py @@ -23,7 +23,11 @@ MultivariateSelectedLeg, Party, ) -from kalshi.fix.messages.dispatch import APP_MESSAGE_MODELS, decode_app_message +from kalshi.fix.messages.dispatch import ( + APP_MESSAGE_MODELS, + decode_app_message, + decode_app_message_strict, +) from kalshi.fix.messages.drop_copy import ( EventResendComplete, EventResendReject, @@ -132,6 +136,7 @@ "SequenceReset", "TestRequest", "decode_app_message", + "decode_app_message_strict", "fixfield", "groupfield", ] diff --git a/kalshi/fix/messages/dispatch.py b/kalshi/fix/messages/dispatch.py index 50c4dcf..9d33248 100644 --- a/kalshi/fix/messages/dispatch.py +++ b/kalshi/fix/messages/dispatch.py @@ -16,6 +16,7 @@ from kalshi.fix.codec import RawMessage from kalshi.fix.enums import MsgType +from kalshi.fix.errors import FixDecodeError from kalshi.fix.messages.base import FixMessage from kalshi.fix.messages.drop_copy import EventResendComplete, EventResendReject from kalshi.fix.messages.market_data import ( @@ -77,21 +78,41 @@ ) -def decode_app_message(raw: RawMessage) -> FixMessage | None: - """Decode an inbound application :class:`RawMessage` to its typed model. +def decode_app_message_strict(raw: RawMessage) -> FixMessage | None: + """Decode an inbound application :class:`RawMessage`, raising on a malformed one. - Returns ``None`` for message types without a registered model (an admin - message or a not-yet-implemented application flow), and also ``None`` if the - payload fails schema validation — a malformed inbound message is logged and - swallowed rather than raised into the consumer's ``on_message`` handler. - See GH #432 for surfacing decode failures to consumers. + Returns ``None`` only for an *unregistered* message type (an admin message or + a not-yet-implemented flow). A registered model whose payload fails schema + validation raises :class:`~kalshi.fix.errors.FixDecodeError` (chaining the + underlying error) rather than returning ``None``, so a genuine message lost to + a single off-spec field is observable. Used by ``FixSession``'s + ``on_decode_error`` hook; direct callers wanting the distinction can call this. """ model = APP_MESSAGE_MODELS.get(raw.msg_type or "") if model is None: return None try: return model.from_raw(raw) - except (ValidationError, ValueError, ArithmeticError): + except (ValidationError, ValueError, ArithmeticError) as exc: # ValueError: bad bool / int; ArithmeticError: bad Decimal (InvalidOperation). + raise FixDecodeError( + f"failed to decode inbound {raw.msg_type}", raw=raw, msg_type=raw.msg_type + ) from exc + + +def decode_app_message(raw: RawMessage) -> FixMessage | None: + """Decode an inbound application :class:`RawMessage` to its typed model. + + Returns ``None`` for message types without a registered model (an admin + message or a not-yet-implemented application flow), and also ``None`` if the + payload fails schema validation — a malformed inbound message is logged and + swallowed rather than raised into the consumer's ``on_message`` handler. To + distinguish those two cases (e.g. to route a malformed known message to a + dead-letter), use :func:`decode_app_message_strict` or ``FixSession``'s + ``on_decode_error`` hook. + """ + try: + return decode_app_message_strict(raw) + except FixDecodeError: logger.warning("failed to decode inbound %s; returning None", raw.msg_type, exc_info=True) return None diff --git a/kalshi/fix/session.py b/kalshi/fix/session.py index 2e097f1..71b05f1 100644 --- a/kalshi/fix/session.py +++ b/kalshi/fix/session.py @@ -42,8 +42,15 @@ from kalshi.fix.config import FixConfig, FixSessionType from kalshi.fix.connection import FixConnection from kalshi.fix.enums import MsgType -from kalshi.fix.errors import FixConnectionError, FixLogonError, FixSequenceError, FixSessionError +from kalshi.fix.errors import ( + FixConnectionError, + FixDecodeError, + FixLogonError, + FixSequenceError, + FixSessionError, +) from kalshi.fix.messages.base import FixMessage +from kalshi.fix.messages.dispatch import decode_app_message_strict from kalshi.fix.messages.session import ( Heartbeat, Logon, @@ -59,6 +66,7 @@ MessageHandler = Callable[[RawMessage], Awaitable[None]] StateChangeHandler = Callable[["FixSessionState", "FixSessionState"], Awaitable[None]] +DecodeErrorHandler = Callable[[RawMessage, FixDecodeError], Awaitable[None]] def _heartbeat_due(now: float, last_send: float, interval: float) -> bool: @@ -112,6 +120,7 @@ def __init__( *, on_message: MessageHandler | None = None, on_state_change: StateChangeHandler | None = None, + on_decode_error: DecodeErrorHandler | None = None, ssl_context: ssl.SSLContext | None = None, ) -> None: config._check_session(session_type) @@ -122,6 +131,11 @@ def __init__( self._supports_retransmission = config.supports_retransmission(session_type) self._on_message = on_message self._on_state_change = on_state_change + # When set, the session decodes each inbound application message to detect + # a registered-but-malformed one and routes it here (a genuine message lost + # to one off-spec field is otherwise silently dropped). Costs one decode + # per app message; pair with on_message for the happy path. See GH #432. + self._on_decode_error = on_decode_error self._ssl_context = ssl_context self._hb = config.heartbeat_interval @@ -619,6 +633,16 @@ async def _process(self, raw: RawMessage, mt: str | None) -> None: # Application message. Isolate consumer callback failures so a buggy # handler cannot tear down the session (the message's seq is already # accounted for by the caller). + if self._on_decode_error is not None: + # Detect a registered-but-malformed message and route it, so a fill + # lost to one off-spec field is observable rather than silently None. + try: + decode_app_message_strict(raw) + except FixDecodeError as exc: + try: + await self._on_decode_error(raw, exc) + except Exception: + logger.exception("FIX on_decode_error callback raised; continuing session") if self._on_message is not None: try: await self._on_message(raw) diff --git a/tests/fix/test_decode_routing.py b/tests/fix/test_decode_routing.py new file mode 100644 index 0000000..1672e4c --- /dev/null +++ b/tests/fix/test_decode_routing.py @@ -0,0 +1,178 @@ +"""Tests for inbound decode-error routing (GH #432). + +``decode_app_message`` collapses "no registered model" and "registered but +malformed" to ``None``; ``decode_app_message_strict`` and ``FixSession``'s +``on_decode_error`` hook make the malformed case observable. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from decimal import Decimal + +import pytest + +from kalshi.fix.auth import FixSigner +from kalshi.fix.codec import RawMessage, decode, encode +from kalshi.fix.config import FixConfig, FixSessionType +from kalshi.fix.errors import FixDecodeError +from kalshi.fix.messages import ( + ExecutionReport, + decode_app_message, + decode_app_message_strict, +) +from kalshi.fix.session import FixSession +from kalshi.fix.tags import Tag + +from .conftest import MockAcceptor + + +def _exec_report_raw(*body: tuple[int, str]) -> RawMessage: + return decode( + encode( + [ + (int(Tag.MSG_TYPE), "8"), + (int(Tag.MSG_SEQ_NUM), "2"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + *body, + ] + ) + ) + + +# --------------------------------------------------------------------------- +# decode_app_message_strict vs decode_app_message +# --------------------------------------------------------------------------- + + +def test_strict_decode_valid_message() -> None: + msg = decode_app_message_strict(_exec_report_raw((int(Tag.AVG_PX), "0.66"))) + assert isinstance(msg, ExecutionReport) + assert msg.avg_px == Decimal("0.66") + + +def test_strict_decode_malformed_raises() -> None: + raw = _exec_report_raw((int(Tag.AVG_PX), "notanumber")) # bad Decimal + with pytest.raises(FixDecodeError) as ei: + decode_app_message_strict(raw) + assert ei.value.msg_type == "8" + assert ei.value.raw is raw + assert ei.value.__cause__ is not None # underlying validation error chained + + +def test_strict_decode_unregistered_returns_none() -> None: + # An admin / unregistered MsgType has no model — None, not raised. + raw = decode( + encode( + [ + (int(Tag.MSG_TYPE), "0"), # Heartbeat + (int(Tag.MSG_SEQ_NUM), "2"), + (int(Tag.SENDING_TIME), "20250101-00:00:00.000"), + ] + ) + ) + assert decode_app_message_strict(raw) is None + + +def test_decode_app_message_still_swallows_malformed() -> None: + # The lenient contract is unchanged: malformed -> None (logged, not raised). + assert decode_app_message(_exec_report_raw((int(Tag.AVG_PX), "notanumber"))) is None + + +# --------------------------------------------------------------------------- +# FixSession on_decode_error routing +# --------------------------------------------------------------------------- + + +async def test_session_routes_decode_error( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + received: list[RawMessage] = [] + errors: list[tuple[RawMessage, FixDecodeError]] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + async def on_decode_error(raw: RawMessage, exc: FixDecodeError) -> None: + errors.append((raw, exc)) + + session = FixSession( + fix_signer, + fix_config, + FixSessionType.ORDER_ENTRY_NR, + on_message=on_message, + on_decode_error=on_decode_error, + ) + await session.start() + try: + await acceptor.push("8", [(int(Tag.AVG_PX), "notanumber")], seq=2) + await until(lambda: bool(errors)) + err_raw, exc = errors[0] + assert isinstance(exc, FixDecodeError) + assert exc.msg_type == "8" + assert err_raw.msg_type == "8" + # The raw message is still delivered to on_message — nothing silently lost. + await until(lambda: bool(received)) + assert received[0].msg_type == "8" + finally: + await session.close() + + +async def test_session_no_decode_error_on_valid_message( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + received: list[RawMessage] = [] + errors: list[tuple[RawMessage, FixDecodeError]] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + async def on_decode_error(raw: RawMessage, exc: FixDecodeError) -> None: + errors.append((raw, exc)) + + session = FixSession( + fix_signer, + fix_config, + FixSessionType.ORDER_ENTRY_NR, + on_message=on_message, + on_decode_error=on_decode_error, + ) + await session.start() + try: + await acceptor.push("8", [(int(Tag.ORDER_ID), "OID-1")], seq=2) + await until(lambda: bool(received)) + assert errors == [] # a valid ExecutionReport does not trigger the hook + finally: + await session.close() + + +async def test_session_without_hook_delivers_raw_on_malformed( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + # No on_decode_error: behaviour is unchanged — the raw still reaches + # on_message and the session survives the malformed payload. + received: list[RawMessage] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + session = FixSession( + fix_signer, fix_config, FixSessionType.ORDER_ENTRY_NR, on_message=on_message + ) + await session.start() + try: + await acceptor.push("8", [(int(Tag.AVG_PX), "notanumber")], seq=2) + await until(lambda: bool(received)) + assert received[0].msg_type == "8" + assert session.state.value == "active" + finally: + await session.close() From a5c54bd5a11e4b0d89c4d5f6c2e9e8de0a2c1079 Mon Sep 17 00:00:00 2001 From: Jeff West Date: Sat, 6 Jun 2026 08:57:40 -0500 Subject: [PATCH 2/2] =?UTF-8?q?fix(fix):=20address=20#439=20review=20?= =?UTF-8?q?=E2=80=94=20decode-error=20docs=20+=20FixDecodeError=20typing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Document the on_decode_error decode cost prominently in the FixSession class docstring (it runs a full Pydantic validation on every inbound app message to detect a malformed one; pairs with on_message as a second pass on a busy session) and accept it for v1; trim the verbose __init__ comment to a pointer. - FixDecodeError.raw / .msg_type are now required (non-optional) — they are always present when raised, so consumers no longer null-check. dispatch binds the registered MsgType to a local so it is a real str (mypy-clean). - tests: valid-message no-trigger test now uses a typed AVG_PX value (exercises decode, not just field presence); add a test that a raising on_decode_error callback is isolated (raw still delivered, session stays active). Double-decode design: kept option 3 (on_decode_error hook) per the issue's recommendation; the happy-path re-decode is opt-in (zero cost when unset) and now explicitly documented. A decode-once typed-delivery path is a future enhancement (the issue noted this lands best with typed inbound delivery). ruff + mypy --strict clean; 260 FIX tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- kalshi/fix/errors.py | 8 +++----- kalshi/fix/messages/dispatch.py | 8 ++++---- kalshi/fix/session.py | 16 +++++++++++---- tests/fix/test_decode_routing.py | 35 +++++++++++++++++++++++++++++++- 4 files changed, 53 insertions(+), 14 deletions(-) diff --git a/kalshi/fix/errors.py b/kalshi/fix/errors.py index fdcaf5f..463ce47 100644 --- a/kalshi/fix/errors.py +++ b/kalshi/fix/errors.py @@ -109,13 +109,11 @@ class FixDecodeError(KalshiFixError): collapses both to ``None``; :func:`~kalshi.fix.messages.decode_app_message_strict` raises this for the former so the failure is observable (see ``FixSession``'s ``on_decode_error`` - hook). ``raw`` carries the offending message; the underlying validation error - is chained via ``__cause__``. + hook). ``raw`` carries the offending message and ``msg_type`` its ``MsgType`` + (both always present); the underlying validation error chains via ``__cause__``. """ - def __init__( - self, message: str, *, raw: RawMessage | None = None, msg_type: str | None = None - ) -> None: + def __init__(self, message: str, *, raw: RawMessage, msg_type: str) -> None: self.raw = raw self.msg_type = msg_type super().__init__(message) diff --git a/kalshi/fix/messages/dispatch.py b/kalshi/fix/messages/dispatch.py index 9d33248..735045f 100644 --- a/kalshi/fix/messages/dispatch.py +++ b/kalshi/fix/messages/dispatch.py @@ -88,16 +88,16 @@ def decode_app_message_strict(raw: RawMessage) -> FixMessage | None: a single off-spec field is observable. Used by ``FixSession``'s ``on_decode_error`` hook; direct callers wanting the distinction can call this. """ - model = APP_MESSAGE_MODELS.get(raw.msg_type or "") + mt = raw.msg_type or "" + model = APP_MESSAGE_MODELS.get(mt) if model is None: return None try: return model.from_raw(raw) except (ValidationError, ValueError, ArithmeticError) as exc: # ValueError: bad bool / int; ArithmeticError: bad Decimal (InvalidOperation). - raise FixDecodeError( - f"failed to decode inbound {raw.msg_type}", raw=raw, msg_type=raw.msg_type - ) from exc + # mt is the registered key here, so it is a real (non-empty) MsgType. + raise FixDecodeError(f"failed to decode inbound {mt}", raw=raw, msg_type=mt) from exc def decode_app_message(raw: RawMessage) -> FixMessage | None: diff --git a/kalshi/fix/session.py b/kalshi/fix/session.py index 71b05f1..5f5edb0 100644 --- a/kalshi/fix/session.py +++ b/kalshi/fix/session.py @@ -110,6 +110,16 @@ class FixSession: on_message=handle) async with session: ... # send order-entry messages, receive execution reports + + ``on_message`` receives every inbound application message as a raw + :class:`~kalshi.fix.codec.RawMessage` (decode it with + :func:`~kalshi.fix.messages.decode_app_message`). Setting ``on_decode_error`` + additionally routes a *registered-but-malformed* message (a real message lost + to one off-spec field) — but it makes the session run a full Pydantic + validation on **every** inbound application message to detect failures, so on a + high-rate session that is real per-message overhead (and the consumer's own + decode in ``on_message`` is then a second pass). Leave it unset unless you + need malformed messages surfaced; there is no cost when it is ``None``. """ def __init__( @@ -131,10 +141,8 @@ def __init__( self._supports_retransmission = config.supports_retransmission(session_type) self._on_message = on_message self._on_state_change = on_state_change - # When set, the session decodes each inbound application message to detect - # a registered-but-malformed one and routes it here (a genuine message lost - # to one off-spec field is otherwise silently dropped). Costs one decode - # per app message; pair with on_message for the happy path. See GH #432. + # Routes registered-but-malformed inbound messages; see the class docstring + # for the per-message decode cost this opts into. GH #432. self._on_decode_error = on_decode_error self._ssl_context = ssl_context diff --git a/tests/fix/test_decode_routing.py b/tests/fix/test_decode_routing.py index 1672e4c..196eed9 100644 --- a/tests/fix/test_decode_routing.py +++ b/tests/fix/test_decode_routing.py @@ -145,13 +145,46 @@ async def on_decode_error(raw: RawMessage, exc: FixDecodeError) -> None: ) await session.start() try: - await acceptor.push("8", [(int(Tag.ORDER_ID), "OID-1")], seq=2) + await acceptor.push("8", [(int(Tag.AVG_PX), "0.66")], seq=2) await until(lambda: bool(received)) assert errors == [] # a valid ExecutionReport does not trigger the hook finally: await session.close() +async def test_session_survives_raising_decode_error_callback( + fix_signer: FixSigner, + fix_config: FixConfig, + acceptor: MockAcceptor, + until: Callable[..., Awaitable[None]], +) -> None: + # A buggy on_decode_error must not tear down the session — the raw is still + # delivered to on_message and the session stays active. + received: list[RawMessage] = [] + + async def on_message(raw: RawMessage) -> None: + received.append(raw) + + async def on_decode_error(raw: RawMessage, exc: FixDecodeError) -> None: + raise RuntimeError("buggy consumer hook") + + session = FixSession( + fix_signer, + fix_config, + FixSessionType.ORDER_ENTRY_NR, + on_message=on_message, + on_decode_error=on_decode_error, + ) + await session.start() + try: + await acceptor.push("8", [(int(Tag.AVG_PX), "notanumber")], seq=2) + await until(lambda: bool(received)) + assert received[0].msg_type == "8" + assert session.state.value == "active" + finally: + await session.close() + + async def test_session_without_hook_delivers_raw_on_malformed( fix_signer: FixSigner, fix_config: FixConfig,