25343054113

Committed 04 May 2026 08:55PM UTC coverage: 87.768% (+0.2%) from 87.56%

Build # 25343054113

Build Type

push

github

Committed by

web-flow

Commit Message

feat(transport): route warnings with request_id as Notice items (PR 3) (#506)

* feat(transport): route warnings with request_id as Notice items (PR 3)

Warnings (codes 2100..=2169) bound to a real request_id now flow to their
owning subscription as non-terminal SubscriptionItem::Notice items instead
of being diverted to the global error log. Hard errors with a real
request_id continue to terminate the subscription as Err.

- routing.rs: add ErrorDelivery struct (Routing × Severity), classify_error_delivery
- Notice: widen with advanced_order_reject_json, add notice_from_decoded
- sync/async dispatchers: classify, deliver_to_request_id (request first, order fallback), log_unrouted
- Subscription<T>: handle_response pattern-matches on RoutedItem; surfaces Notice
- delete log_error_payload, transport/common.rs log import; DecodedError::is_log_only

Tests: classify_error_delivery boundary codes; sync+async dispatcher tests for
owned warning, unrouted warning, hard error, order-channel fallback;
sync+async Subscription<T> end-to-end Notice surface tests.

* review cleanups: ERROR_TIME guard, doc comments, notice_from_decoded tests

- ResponseMessage::advanced_order_reject_json() gains a server_version >= ERROR_TIME guard, mirroring error_time(); old-format messages always return empty.
- routing::extract_text_error migrated to use the accessor, single source of truth.
- /// doc comments on next_routed/try_next_routed/next_timeout_routed in transport/mod.rs.
- New direct unit tests for notice_from_decoded covering rich-payload preservation and missing-optionals path.
- Plan doc: pruned superseded "Scope — classification" / "Scope — dispatcher rewrite" sections; as-shipped notes are now the source of truth.

* review cleanups: From<&DecodedError> impls, log_unrouted free function

Composability cleanups from the duplication/SRP review:

- log_unrouted body extracted to transport::common::log_unrouted_notice as a free function. Per-transport log_unrouted met... (continued)

Coverage Stats

102 of 118 new or added lines in 10 files covered. (86.44%)

3 existing lines in 2 files now uncovered.

17077 of 19457 relevant lines covered (87.77%)

26.29 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.0

/src/subscriptions/common.rs

//! Common utilities for subscription processing

use serde::{Deserialize, Serialize};
use time_tz::Tz;

use crate::errors::Error;
use crate::messages::{IncomingMessages, Notice, OutgoingMessages, ResponseMessage};

/// An item yielded by a [`Subscription`](crate::subscriptions::Subscription).
///
/// Subscriptions return `Option<Result<SubscriptionItem<T>, Error>>` from `next`,
/// `try_next`, and `next_timeout`. `Data(T)` is the decoded payload; `Notice` is a
/// non-fatal IB notice (warning codes 2100..=2169) bound to this subscription —
/// the stream stays open. Use [`Subscription::iter_data`](crate::subscriptions::Subscription::iter_data)
/// (or async [`Subscription::data_stream`](crate::subscriptions::Subscription::data_stream))
/// when you only care about data and want notices logged automatically.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum SubscriptionItem<T> {
    /// A successfully decoded payload from the subscription stream.
    Data(T),
    /// A non-fatal IB notice (warning codes 2100..=2169) bound to this subscription.
    /// Receiving a notice does not terminate the stream.
    Notice(Notice),
}

impl<T> SubscriptionItem<T> {
    /// Returns the inner data value, dropping notices. Pure conversion — no side effects.
    pub fn into_data(self) -> Option<T> {
        match self {
            SubscriptionItem::Data(t) => Some(t),
            SubscriptionItem::Notice(_) => None,
        }
    }
}

/// Maps `Ok(Notice)` to `None` (logged at `warn!`); passes `Data` and `Err`
/// through unchanged.
pub(crate) fn filter_notice<T>(item: Result<SubscriptionItem<T>, Error>) -> Option<Result<T, Error>> {
    match item {
        Ok(SubscriptionItem::Data(t)) => Some(Ok(t)),
        Ok(SubscriptionItem::Notice(n)) => {
            log::warn!("ib notice on subscription: {n}");
            None
        }
        Err(e) => Some(Err(e)),
    }
}

/// Pre-classified channel item delivered from the dispatcher to subscriptions.
/// `Response` carries raw bytes the decoder must still interpret; `Notice` and
/// `Error` are pre-classified by the dispatcher so decoders never re-classify
/// warnings vs. hard errors.
#[derive(Debug, Clone)]
pub(crate) enum RoutedItem {
    Response(ResponseMessage),
    Notice(Notice),
    Error(Error),
}

impl From<ResponseMessage> for RoutedItem {
    fn from(message: ResponseMessage) -> Self {
        RoutedItem::Response(message)
    }
}

impl From<Error> for RoutedItem {
    fn from(error: Error) -> Self {
        RoutedItem::Error(error)
    }
}

impl RoutedItem {
    /// Translate to `Result<ResponseMessage, Error>`. Returns `None` for
    /// `Notice` so callers can skip and recv the next item.
    pub(crate) fn into_legacy(self) -> Option<Result<ResponseMessage, Error>> {
        match self {
            RoutedItem::Response(message) => Some(Ok(message)),
            RoutedItem::Error(error) => Some(Err(error)),
            RoutedItem::Notice(_) => None,
        }
    }
}

/// Checks if an error indicates the end of a stream
#[allow(dead_code)]
pub(crate) fn is_stream_end(error: &Error) -> bool {
    matches!(error, Error::EndOfStream)
}

/// Checks if an error should be stored for later retrieval
#[allow(dead_code)]
pub(crate) fn should_store_error(error: &Error) -> bool {
    !is_stream_end(error)
}

/// Common error types that can occur during subscription processing
#[derive(Debug)]
pub(crate) enum ProcessingResult<T> {
    /// Successfully processed a value
    Success(T),
    /// Message not intended for this subscription — skip silently.
    /// Occurs on shared broadcast channels where messages from other
    /// subscriptions can arrive on the same channel.
    Skip,
    /// Encountered an error that should be stored
    Error(Error),
    /// Stream has ended normally
    EndOfStream,
}

/// Process a decoding result into a common processing result
pub(crate) fn process_decode_result<T>(result: Result<T, Error>) -> ProcessingResult<T> {
    match result {
        Ok(val) => ProcessingResult::Success(val),
        Err(Error::EndOfStream) => ProcessingResult::EndOfStream,
        Err(Error::UnexpectedResponse(_)) => ProcessingResult::Skip,
        Err(err) => ProcessingResult::Error(err),
    }
}

/// Context for decoding responses, providing all necessary state for decoders.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct DecoderContext {
    /// Server version for protocol compatibility
    pub server_version: i32,
    /// Timezone for parsing timestamps (from TWS connection)
    pub time_zone: Option<&'static Tz>,
    /// Type of the original request that initiated this subscription
    pub request_type: Option<OutgoingMessages>,
    /// Whether this is a smart depth subscription
    pub is_smart_depth: bool,
}

impl DecoderContext {
    /// Create a new context with server version and optional timezone
    pub fn new(server_version: i32, time_zone: Option<&'static Tz>) -> Self {
        Self {
            server_version,
            time_zone,
            request_type: None,
            is_smart_depth: false,
        }
    }

    /// Set the request type
    #[allow(dead_code)]
    pub fn with_request_type(mut self, request_type: OutgoingMessages) -> Self {
        self.request_type = Some(request_type);
        self
    }

    /// Set the smart depth flag
    pub fn with_smart_depth(mut self, is_smart_depth: bool) -> Self {
        self.is_smart_depth = is_smart_depth;
        self
    }
}

/// Common trait for decoding streaming data responses
///
/// This trait is shared between sync and async implementations to avoid code duplication.
/// Decoders receive a `DecoderContext` containing server version, timezone, and other
/// context needed to properly decode messages.
pub(crate) trait StreamDecoder<T> {
    /// Message types this stream can handle
    #[allow(dead_code)]
    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[];

    /// Decode a response message into the stream's data type
    fn decode(context: &DecoderContext, message: &mut ResponseMessage) -> Result<T, Error>;

    /// Generate a cancellation message for this stream
    fn cancel_message(_server_version: i32, _request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
        Err(Error::NotImplemented)
    }

    /// Returns true if this decoded value represents the end of a snapshot subscription
    #[allow(unused)]
    fn is_snapshot_end(&self) -> bool {
        false
    }
}

#[cfg(test)]
#[path = "common_tests.rs"]
mod tests;

1	//! Common utilities for subscription processing
2
3	use serde::{Deserialize, Serialize};
4	use time_tz::Tz;
5
6	use crate::errors::Error;
7	use crate::messages::{IncomingMessages, Notice, OutgoingMessages, ResponseMessage};
8
9	/// An item yielded by a [`Subscription`](crate::subscriptions::Subscription).
10	///
11	/// Subscriptions return `Option<Result<SubscriptionItem<T>, Error>>` from `next`,
12	/// `try_next`, and `next_timeout`. `Data(T)` is the decoded payload; `Notice` is a
13	/// non-fatal IB notice (warning codes 2100..=2169) bound to this subscription —
14	/// the stream stays open. Use [`Subscription::iter_data`](crate::subscriptions::Subscription::iter_data)
15	/// (or async [`Subscription::data_stream`](crate::subscriptions::Subscription::data_stream))
16	/// when you only care about data and want notices logged automatically.
17	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
18	pub enum SubscriptionItem<T> {
19	/// A successfully decoded payload from the subscription stream.
20	Data(T),
21	/// A non-fatal IB notice (warning codes 2100..=2169) bound to this subscription.
22	/// Receiving a notice does not terminate the stream.
23	Notice(Notice),
24	}
25
26	impl<T> SubscriptionItem<T> {
27	/// Returns the inner data value, dropping notices. Pure conversion — no side effects.
28	pub fn into_data(self) -> Option<T> {	×
29	match self {	×
30	SubscriptionItem::Data(t) => Some(t),	×
31	SubscriptionItem::Notice(_) => None,	×
32	}
33	}	×
34	}
35
36	/// Maps `Ok(Notice)` to `None` (logged at `warn!`); passes `Data` and `Err`
37	/// through unchanged.
38	pub(crate) fn filter_notice<T>(item: Result<SubscriptionItem<T>, Error>) -> Option<Result<T, Error>> {	164✔
39	match item {	155✔
40	Ok(SubscriptionItem::Data(t)) => Some(Ok(t)),	151✔
41	Ok(SubscriptionItem::Notice(n)) => {	4✔
42	log::warn!("ib notice on subscription: {n}");	4✔
43	None	4✔
44	}
45	Err(e) => Some(Err(e)),	9✔
46	}
47	}	164✔
48
49	/// Pre-classified channel item delivered from the dispatcher to subscriptions.
50	/// `Response` carries raw bytes the decoder must still interpret; `Notice` and
51	/// `Error` are pre-classified by the dispatcher so decoders never re-classify
52	/// warnings vs. hard errors.
53	#[derive(Debug, Clone)]
54	pub(crate) enum RoutedItem {
55	Response(ResponseMessage),
56	Notice(Notice),
57	Error(Error),
58	}
59
60	impl From<ResponseMessage> for RoutedItem {
61	fn from(message: ResponseMessage) -> Self {	443✔
62	RoutedItem::Response(message)	443✔
63	}	443✔
64	}
65
66	impl From<Error> for RoutedItem {
67	fn from(error: Error) -> Self {	551✔
68	RoutedItem::Error(error)	551✔
69	}	551✔
70	}
71
72	impl RoutedItem {
73	/// Translate to `Result<ResponseMessage, Error>`. Returns `None` for
74	/// `Notice` so callers can skip and recv the next item.
75	pub(crate) fn into_legacy(self) -> Option<Result<ResponseMessage, Error>> {	145✔
76	match self {	145✔
77	RoutedItem::Response(message) => Some(Ok(message)),	139✔
78	RoutedItem::Error(error) => Some(Err(error)),	6✔
UNCOV 79	RoutedItem::Notice(_) => None,	×
80	}
81	}	145✔
82	}
83
84	/// Checks if an error indicates the end of a stream
85	#[allow(dead_code)]
86	pub(crate) fn is_stream_end(error: &Error) -> bool {	12✔
87	matches!(error, Error::EndOfStream)	12✔
88	}	12✔
89
90	/// Checks if an error should be stored for later retrieval
91	#[allow(dead_code)]
92	pub(crate) fn should_store_error(error: &Error) -> bool {	6✔
93	!is_stream_end(error)	6✔
94	}	6✔
95
96	/// Common error types that can occur during subscription processing
97	#[derive(Debug)]
98	pub(crate) enum ProcessingResult<T> {
99	/// Successfully processed a value
100	Success(T),
101	/// Message not intended for this subscription — skip silently.
102	/// Occurs on shared broadcast channels where messages from other
103	/// subscriptions can arrive on the same channel.
104	Skip,
105	/// Encountered an error that should be stored
106	Error(Error),
107	/// Stream has ended normally
108	EndOfStream,
109	}
110
111	/// Process a decoding result into a common processing result
112	pub(crate) fn process_decode_result<T>(result: Result<T, Error>) -> ProcessingResult<T> {	212✔
113	match result {	60✔
114	Ok(val) => ProcessingResult::Success(val),	152✔
115	Err(Error::EndOfStream) => ProcessingResult::EndOfStream,	10✔
116	Err(Error::UnexpectedResponse(_)) => ProcessingResult::Skip,	43✔
117	Err(err) => ProcessingResult::Error(err),	7✔
118	}
119	}	212✔
120
121	/// Context for decoding responses, providing all necessary state for decoders.
122	#[derive(Debug, Clone, Default, PartialEq)]
123	pub struct DecoderContext {
124	/// Server version for protocol compatibility
125	pub server_version: i32,
126	/// Timezone for parsing timestamps (from TWS connection)
127	pub time_zone: Option<&'static Tz>,
128	/// Type of the original request that initiated this subscription
129	pub request_type: Option<OutgoingMessages>,
130	/// Whether this is a smart depth subscription
131	pub is_smart_depth: bool,
132	}
133
134	impl DecoderContext {
135	/// Create a new context with server version and optional timezone
136	pub fn new(server_version: i32, time_zone: Option<&'static Tz>) -> Self {	305✔
137	Self {	305✔
138	server_version,	305✔
139	time_zone,	305✔
140	request_type: None,	305✔
141	is_smart_depth: false,	305✔
142	}	305✔
143	}	305✔
144
145	/// Set the request type
146	#[allow(dead_code)]
147	pub fn with_request_type(mut self, request_type: OutgoingMessages) -> Self {	19✔
148	self.request_type = Some(request_type);	19✔
149	self	19✔
150	}	19✔
151
152	/// Set the smart depth flag
153	pub fn with_smart_depth(mut self, is_smart_depth: bool) -> Self {	12✔
154	self.is_smart_depth = is_smart_depth;	12✔
155	self	12✔
156	}	12✔
157	}
158
159	/// Common trait for decoding streaming data responses
160	///
161	/// This trait is shared between sync and async implementations to avoid code duplication.
162	/// Decoders receive a `DecoderContext` containing server version, timezone, and other
163	/// context needed to properly decode messages.
164	pub(crate) trait StreamDecoder<T> {
165	/// Message types this stream can handle
166	#[allow(dead_code)]
167	const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[];
168
169	/// Decode a response message into the stream's data type
170	fn decode(context: &DecoderContext, message: &mut ResponseMessage) -> Result<T, Error>;
171
172	/// Generate a cancellation message for this stream
173	fn cancel_message(_server_version: i32, _request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {	26✔
174	Err(Error::NotImplemented)	26✔
175	}	26✔
176
177	/// Returns true if this decoded value represents the end of a snapshot subscription
178	#[allow(unused)]
179	fn is_snapshot_end(&self) -> bool {	67✔
180	false	67✔
181	}	67✔
182	}
183
184	#[cfg(test)]
185	#[path = "common_tests.rs"]
186	mod tests;

wboayue / rust-ibapi / 25343054113

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous