{"record":{"id":"d81d592d2966172b","repo":"nautechsystems/nautilus_trader","slug":"stream-message-missing-payload-stream-msg","errorCode":null,"errorMessage":"Stream message missing payload: {stream_msg:?}","messagePattern":"Stream message missing payload: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":882,"sourceCode":" 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,\n encoding,\n ))\n}","sourceCodeStart":864,"sourceCodeEnd":900,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L864-L900","documentation":"A Redis stream entry decoded by decode_bus_message must include a 'payload' header with the message body. When the entry parsed successfully but no 'payload' pair was present, the decoder bails because a BusMessage without a body is meaningless. It signals an entry that does not conform to the message-bus wire format.","triggerScenarios":"stream_messages encounters an entry with headers such as topic/type/encoding but no 'payload' field — e.g. a heartbeat or control entry written by a different component, a producer bug that drops the payload, or a foreign entry in the stream.","commonSituations":"Stream key shared with control/heartbeat entries from other tooling; producer version skew; manual redis-cli XADD during testing; consumer attached to the wrong stream.","solutions":["Inspect the entry (XRANGE) and verify whether it legitimately lacks a payload; if it is a control entry, exclude that stream from the bus consumer.","Fix the producer so every published message includes a 'payload' BulkString.","Skip or XDEL the payload-less entry so the consumer can continue.","Verify the stream key/topic the consumer is reading matches what the producer publishes."],"exampleFix":"// before (producer bug omits payload)\nXADD mystream * topic \"quotes.X\" encoding \"json\"\n// after\nXADD mystream * topic \"quotes.X\" payload \"\" encoding \"json\"","handlingStrategy":"try-catch","validationCode":"// Ensure payload present before publishing\nfn entry_has_payload(fields: &[(&str, &[u8])]) -> bool {\n fields.iter().any(|(k, _)| *k == \"payload\")\n}","typeGuard":"fn payload_of(fields: &[(String, redis::Value)]) -> Option<&Vec> {\n fields.iter().find(|(k, _)| k == \"payload\")\n .and_then(|(_, v)| match v { redis::Value::BulkString(b) => Some(b), _ => 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 payload\") => {\n tracing::warn!(\"skipping payload-less entry: {e}\");\n }\n Err(e) => return Err(e),\n}","preventionTips":["Always set a 'payload' BulkString when publishing to the bus.","Route heartbeat/control entries to separate streams, not the bus stream.","Test producers end-to-end before pointing consumers at the stream.","Verify the consumer reads the same stream key the producer writes."],"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-14T00:17:10.932Z"}