{"record":{"id":"6aae24d2f1f8cebb","repo":"nautechsystems/nautilus_trader","slug":"stream-message-missing-topic-stream-msg","errorCode":null,"errorMessage":"Stream message missing topic: {stream_msg:?}","messagePattern":"Stream message missing topic: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":879,"sourceCode":"                };\n                let value = std::str::from_utf8(bytes)\n                    .map_err(|e| anyhow::anyhow!(\"Error parsing encoding: {e}\"))?;\n                encoding = value\n                    .parse()\n                    .map_err(|e| anyhow::anyhow!(\"Error parsing encoding: {e}\"))?;\n            }\n            b\"payload\" => {\n                let redis::Value::BulkString(bytes) = &pair[1] else {\n                    anyhow::bail!(\"Invalid payload format: {stream_msg:?}\");\n                };\n                payload = Some(Bytes::copy_from_slice(bytes));\n            }\n            _ => {}\n        }\n    }\n\n    let Some(topic) = topic else {\n        anyhow::bail!(\"Stream message missing topic: {stream_msg:?}\");\n    };\n    let Some(payload) = payload else {\n        anyhow::bail!(\"Stream message missing payload: {stream_msg:?}\");\n    };\n    let payload_type = match type_name {\n        Some(type_name) if typed_payload => BusPayloadType::from_typed_name(&type_name)\n            .ok_or_else(|| anyhow::anyhow!(\"Unknown typed payload '{type_name}'\"))?,\n        Some(type_name) => BusPayloadType::from_name(&type_name),\n        None if typed_payload => {\n            anyhow::bail!(\"Typed stream message missing type: {stream_msg:?}\")\n        }\n        None => BusPayloadType::Custom(Ustr::default()),\n    };\n\n    Ok(BusMessage::with_str_topic(\n        topic,\n        payload_type,\n        payload,","sourceCodeStart":861,"sourceCodeEnd":897,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L861-L897","documentation":"Every Redis stream entry decoded by decode_bus_message must carry a 'topic' header identifying where the message was published. If the entry's fields contain no 'topic' pair, the decoder cannot construct a BusMessage and bails. This guards against foreign or corrupt entries in the stream.","triggerScenarios":"stream_messages encounters a stream entry written without the 'topic' field — e.g. by an older producer version, a hand-crafted XADD, or a different application sharing the same stream key.","commonSituations":"Redis stream keys reused by other services; manual testing with redis-cli; producer/consumer version skew where the header name changed; partially written entries from an aborted pipeline.","solutions":["Check the stream entry (XRANGE) and confirm it includes the 'topic' field; re-publish with a topic.","Fix or upgrade the producer so every published BusMessage includes the topic header.","Delete the malformed entry (XDEL) so decoding can proceed.","Confirm the consumer is subscribed to the intended stream (correct topic/stream key)."],"exampleFix":"// before (manual XADD without topic)\nXADD mystream * payload \"...\"\n// after\nXADD mystream * topic \"quotes.BTC-USD\" payload \"...\"","handlingStrategy":"try-catch","validationCode":"// Validate required headers before writing\nfn entry_is_publishable(fields: &[(&str, &str)]) -> bool {\n    fields.iter().any(|(k, _)| *k == \"topic\")\n}","typeGuard":"fn topic_of<'a>(fields: &'a [(String, redis::Value)]) -> Option<&'a str> {\n    fields.iter().find(|(k, _)| k == \"topic\")\n        .and_then(|(_, v)| match v { redis::Value::BulkString(b) => std::str::from_utf8(b).ok(), _ => None })\n}","tryCatchPattern":"match stream_messages(&mut con, stream, count).await {\n    Ok(msgs) => { /* use msgs */ }\n    Err(e) if e.to_string().contains(\"missing topic\") => {\n        tracing::warn!(\"skipping entry without topic: {e}\");\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Always include the 'topic' header in every published BusMessage.","Use the library publisher rather than hand-rolled XADD commands.","Avoid sharing bus stream keys with other applications.","Keep producer and consumer versions aligned on the wire format."],"tags":["redis","decoding","missing-field","msgbus"],"backgroundTag":"missing-required-argument","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"}