{"record":{"id":"1de86328bee38f1f","repo":"nautechsystems/nautilus_trader","slug":"error-parsing-topic-e","errorCode":null,"errorMessage":"Error parsing topic: {e}","messagePattern":"Error parsing topic: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":834,"sourceCode":"    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:?}\");\n                };\n                let value = std::str::from_utf8(bytes)\n                    .map_err(|e| anyhow::anyhow!(\"Error parsing payload kind: {e}\"))?;\n                anyhow::ensure!(","sourceCodeStart":816,"sourceCodeEnd":852,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L816-L852","documentation":"decode_bus_message parses Redis stream entries into bus messages. The `topic` field must be a BulkString containing valid UTF-8; when the raw bytes cannot be decoded as UTF-8, this anyhow error is returned with the underlying FromUtf8Error message. It indicates the stream entry's topic bytes are corrupt or were written with a non-UTF-8 encoding.","triggerScenarios":"Calling stream_messages (which calls decode_bus_message) on a Redis stream entry whose `topic` value is a BulkString containing invalid UTF-8 bytes (String::from_utf8 fails).","commonSituations":"A producer wrote binary/corrupted data into the topic field; manual edits or tooling inserted raw bytes into the Redis stream; a non-Nautilus client publishes entries with different byte encodings (e.g. UTF-16 or Latin-1 strings).","solutions":["Inspect the offending Redis stream entry (XRANGE/XRSTREAM dump) and find which producer wrote non-UTF-8 topic bytes.","Ensure all publishers write topics as UTF-8 strings (standard Redis clients encode strings as UTF-8; avoid custom binary encoders for headers).","Delete or skip the corrupt stream entries, then re-publish with correct encoding.","Verify no intermediate proxy/serializer (e.g. msgpack round-trip misconfiguration) mangles the topic field."],"exampleFix":"// before: publishing with raw bytes\nredis.xadd(stream, [(b\"topic\", non_utf8_bytes), ...]);\n// after: ensure UTF-8 encoding\nlet topic = String::from_utf8(bytes).expect(\"topic must be UTF-8\");\nredis.xadd(stream, &[(\"topic\", topic.as_str()), ...]);","handlingStrategy":"try-catch","validationCode":"// Rust: pre-validate header bytes before publish\nif let Err(e) = std::str::from_utf8(&topic_bytes) {\n    return Err(format!(\"topic not UTF-8: {e}\"));\n}","typeGuard":null,"tryCatchPattern":"match decode_bus_message(entry) {\n    Ok(msg) => process(msg),\n    Err(e) if e.to_string().contains(\"Error parsing topic\") => {\n        log::warn!(\"skipping entry with corrupt topic: {e}\");\n        // skip / dead-letter the entry\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Always publish headers through the library's encoding helpers, never raw bytes.","Add a producer-side assertion that all header fields are valid UTF-8.","Monitor stream entries for non-UTF-8 bytes in integration tests."],"tags":["redis","utf8","decoding","streaming"],"backgroundTag":"json-decode-failed","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"}