{"record":{"id":"41b52cb5ff7e8d32","repo":"nautechsystems/nautilus_trader","slug":"invalid-stream-field-key-stream-msg","errorCode":null,"errorMessage":"Invalid stream field key: {stream_msg:?}","messagePattern":"Invalid stream field key: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":824,"sourceCode":"// 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() {\n            b\"topic\" => {\n                let redis::Value::BulkString(bytes) = &pair[1] else {\n                    anyhow::bail!(\"Invalid topic format: {stream_msg:?}\");\n                };\n                topic = Some(\n                    String::from_utf8(bytes.clone())\n                        .map_err(|e| anyhow::anyhow!(\"Error parsing topic: {e}\"))?,\n                );\n            }\n            b\"type\" => {\n                let redis::Value::BulkString(bytes) = &pair[1] else {\n                    anyhow::bail!(\"Invalid type format: {stream_msg:?}\");\n                };\n                type_name = Some(\n                    String::from_utf8(bytes.clone())","sourceCodeStart":806,"sourceCodeEnd":842,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L806-L842","documentation":"decode_bus_message iterates key/value pairs and each key must be a redis::Value::BulkString. If a key is some other variant (Int, Nil, simple status string, etc.) the decoder bails with this error rather than guessing the field name.","triggerScenarios":"A stream entry whose odd-indexed elements are not BulkStrings — normally impossible from real Redis XADD replies, so usually caused by mock/test data, a proxy rewriting replies, or misusing the decoder with fabricated values.","commonSituations":"Unit or integration fixtures constructing redis::Value entries incorrectly; RESP3 protocol negotiation with a client/server mismatch altering variant representations; custom middleware translating replies.","solutions":["Inspect the entry ({stream_msg:?}) to see which key element has the unexpected variant","Ensure test fixtures build entries as Value::BulkString keys paired with values","Force RESP2 (or match server/client protocol versions) so replies use BulkString representations","Bypass or fix any proxy that rewrites reply types"],"exampleFix":null,"handlingStrategy":"type-guard","validationCode":null,"typeGuard":"fn has_bulkstring_keys(v: &redis::Value) -> bool {\n    matches!(v, redis::Value::Array(f) if f.len() % 2 == 0 &&\n        f.chunks(2).all(|c| matches!(c[0], redis::Value::BulkString(_))))\n}","tryCatchPattern":"match decode_bus_message(&entry) {\n    Err(e) if e.to_string().contains(\"Invalid stream field key\") => {\n        log::warn!(\"non-string key in entry, skipping: {e}\");\n    }\n    other => other?,\n}","preventionTips":["Build test fixtures with Value::BulkString keys","Pin RESP2/RESP3 negotiation consistently across client and server","Skip proxy layers that transform reply value variants"],"tags":["redis","deserialization","protocol"],"backgroundTag":"type-mismatch","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"}