{"record":{"id":"0692746f270586be","repo":"nautechsystems/nautilus_trader","slug":"typed-stream-message-missing-type-stream-msg","errorCode":null,"errorMessage":"Typed stream message missing type: {stream_msg:?}","messagePattern":"Typed stream message missing type: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":889,"sourceCode":"                };\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,\n        encoding,\n    ))\n}\n\nasync fn run_heartbeat(\n    heartbeat_interval_secs: u16,\n    signal: Arc<AtomicBool>,\n    pub_tx: tokio::sync::mpsc::UnboundedSender<BusMessage>,\n) {\n    log_task_started(\"heartbeat\");","sourceCodeStart":871,"sourceCodeEnd":907,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L871-L907","documentation":"When a stream entry is marked with the typed payload kind header, it must also carry a 'type' header naming the payload type. decode_bus_message bails if kind is typed but type_name is None, since it cannot resolve a BusPayloadType. This catches producers that set kind=typed without the accompanying type name.","triggerScenarios":"decode_bus_message (via stream_messages) reads an entry whose payload-kind header is set to the typed kind constant but which has no 'type' field — typically a producer writing inconsistent headers or a hand-crafted entry.","commonSituations":"Partial adoption of the typed-payload protocol in a producer; manual testing with redis-cli where the type header was forgotten; version skew between producer and consumer message formats.","solutions":["Add the 'type' header to the published entry (e.g. the instrument/data type name) so BusPayloadType::from_typed_name can resolve it.","Fix the producer to always set both the typed kind flag and the type name together.","Republish the message with correct headers and XDEL the bad entry.","If the payload is genuinely untyped, publish without the typed-kind header so the decoder falls back to BusPayloadType::Custom."],"exampleFix":"// before (typed kind without type)\nXADD bus * topic \"q\" payload \"...\" kind \"typed\"\n// after\nXADD bus * topic \"q\" payload \"...\" kind \"typed\" type \"data.QuoteTick\"","handlingStrategy":"try-catch","validationCode":"// Typed publishes must include both kind and type headers\nfn typed_entry_ok(fields: &[(&str, &str)]) -> bool {\n    let typed = fields.iter().any(|(k, v)| *k == \"kind\" && *v == \"typed\");\n    let has_type = fields.iter().any(|(k, _)| *k == \"type\");\n    !typed || has_type\n}","typeGuard":"fn typed_name_of(fields: &[(String, redis::Value)]) -> Option<String> {\n    fields.iter().find(|(k, _)| k == \"type\")\n        .and_then(|(_, v)| match v { redis::Value::BulkString(b) => String::from_utf8(b.clone()).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(\"Typed stream message missing type\") => {\n        tracing::error!(\"producer bug: typed entry lacks type header: {e}\");\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Set kind and type headers together when publishing typed payloads.","Cover the producer's typed-publish path with integration tests.","Omit the typed-kind header if the payload has no resolvable type name.","Keep type names consistent with BusPayloadType::from_typed_name lookups."],"tags":["redis","decoding","missing-field","typed-payload"],"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"}