{"record":{"id":"e1cbcfef98c98b2f","repo":"nautechsystems/nautilus_trader","slug":"invalid-stream-message-format-stream-msg","errorCode":null,"errorMessage":"Invalid stream message format: {stream_msg:?}","messagePattern":"Invalid stream message format: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":809,"sourceCode":"        () = &mut retry_timer => true,\n        () = &mut terminate => false,\n    }\n}\n\nasync fn wait_for_stream_signal(stream_signal: &Arc<AtomicBool>) {\n    let check_timer = tokio::time::interval(Duration::from_millis(100));\n\n    tokio::pin!(check_timer);\n\n    while !stream_signal.load(Ordering::Relaxed) {\n        check_timer.tick().await;\n    }\n}\n\n// Redis fields are unordered, and older streams may omit type or encoding headers\nfn decode_bus_message(stream_msg: &redis::Value) -> anyhow::Result<BusMessage> {\n    let redis::Value::Array(fields) = stream_msg else {\n        anyhow::bail!(\"Invalid stream message format: {stream_msg:?}\");\n    };\n\n    if fields.len() < 4 || fields.len() % 2 != 0 {\n        anyhow::bail!(\"Invalid stream message format: {stream_msg:?}\");\n    }\n\n    let mut topic: Option<String> = None;\n    let mut type_name: Option<String> = None;\n    let mut typed_payload = false;\n    let mut encoding = SerializationEncoding::default();\n    let mut payload: Option<Bytes> = None;\n\n    for pair in fields.as_chunks::<2>().0 {\n        let redis::Value::BulkString(key) = &pair[0] else {\n            anyhow::bail!(\"Invalid stream field key: {stream_msg:?}\");\n        };\n\n        match key.as_slice() {","sourceCodeStart":791,"sourceCodeEnd":827,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L791-L827","documentation":"decode_bus_message expects each stream entry to be a redis::Value::Array of alternating key/value BulkStrings. If the entry is not an Array at all, this error is thrown. It indicates the raw XREAD reply shape differs from what the decoder requires.","triggerScenarios":"Feeding decode_bus_message with a redis::Value that is Nil, BulkString, Int, or a Map variant instead of an Array — e.g. a malformed response from a proxy, or passing the wrong nesting level of the XREAD reply.","commonSituations":"Redis proxies/mocking layers returning non-standard reply shapes; custom code piping entries into the decoder from a differently-shaped response; decoder version mismatch with the redis crate's Value representation.","solutions":["Confirm you are passing the per-entry value from the XREAD reply, not the outer nested response","Log the offending stream_msg to see the actual redis::Value variant returned","Check for middleware/proxy rewriting RESP replies; connect directly to Redis to compare","Align the redis crate version used by the library with the reply format your server emits"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"fn is_array_entry(v: &redis::Value) -> bool { matches!(v, redis::Value::Array(_)) }","typeGuard":"fn as_entry_array(v: &redis::Value) -> Option<&Vec<redis::Value>> {\n    if let redis::Value::Array(fields) = v { Some(fields) } else { None }\n}","tryCatchPattern":"match decode_bus_message(&entry) {\n    Err(e) if e.to_string().contains(\"Invalid stream message format\") => {\n        log::warn!(\"skipping malformed entry: {e}\");\n    }\n    other => other?,\n}","preventionTips":["Pass only per-entry XREAD reply values into the decoder","Keep the redis crate and server RESP protocol versions consistent","Avoid proxies/middleware that reshape replies"],"tags":["redis","deserialization","protocol"],"backgroundTag":"unexpected-response-shape","analyzedSha":"18893faf8b356be3320add8de2f861b0b647cf06","analyzedAt":"2026-09-08T20:49:34.690Z","contentChangedAt":"2026-09-08T20:49:34.690Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}