{"record":{"id":"fb4df3a098cd8610","repo":"nautechsystems/nautilus_trader","slug":"invalid-topic-format-stream-msg","errorCode":null,"errorMessage":"Invalid topic format: {stream_msg:?}","messagePattern":"Invalid topic format: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":830,"sourceCode":"    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())\n                        .map_err(|e| anyhow::anyhow!(\"Error parsing type: {e}\"))?,\n                );\n            }\n            key if key == PAYLOAD_KIND_FIELD.as_bytes() => {\n                let redis::Value::BulkString(bytes) = &pair[1] else {\n                    anyhow::bail!(\"Invalid payload kind format: {stream_msg:?}\");","sourceCodeStart":812,"sourceCodeEnd":848,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L812-L848","documentation":"When decoding the `topic` field, its value must be a BulkString that can be decoded as UTF-8. If the value paired with key `topic` is not a BulkString, this error is thrown before the UTF-8 conversion (which would produce the separate 'Error parsing topic' error).","triggerScenarios":"A stream entry containing a `topic` field whose value is Nil, Int, or an Array instead of a byte string — e.g. a producer XADDing a non-string topic value or corrupted/mocked entry data.","commonSituations":"Custom producers writing non-string field values; test fixtures with wrong value variants; intermediary tooling transforming the stream entry.","solutions":["Fix the producer to always XADD topic as a UTF-8 string value","Inspect {stream_msg:?} to see the actual value variant stored under `topic`","Remove or skip the malformed entry (XDEL) so the consumer can proceed","Correct test fixture data to use BulkString for the topic value"],"exampleFix":"// before\nredis.xadd(stream, \"*\", &[(\"topic\", 42)]); // non-string\n// after\nredis.xadd(stream, \"*\", &[(\"topic\", \"events.trades\")]);","handlingStrategy":"validation","validationCode":"fn topic_is_valid(entry: &redis::Value) -> bool {\n    matches!(entry, redis::Value::Array(f) if f.chunks(2).any(|c|\n        matches!((&c[0], &c[1]), (redis::Value::BulkString(k), redis::Value::BulkString(_)) if k == b\"topic\")))\n}","typeGuard":null,"tryCatchPattern":"match decode_bus_message(&entry) {\n    Err(e) if e.to_string().contains(\"Invalid topic format\") => {\n        log::warn!(\"bad topic field, skipping entry: {e}\");\n    }\n    other => other?,\n}","preventionTips":["Always XADD topic as a UTF-8 string value","Reject non-string field values at the producer boundary","Scan streams for malformed entries after third-party writers touch them"],"tags":["redis","deserialization","validation"],"backgroundTag":"type-mismatch","analyzedSha":"18893faf8b356be3320add8de2f861b0b647cf06","analyzedAt":"2026-09-08T20:49:34.690Z","contentChangedAt":"2026-09-08T20:49:34.690Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}