{"record":{"id":"5ee2f80aca305acb","repo":"nautechsystems/nautilus_trader","slug":"invalid-payload-format-stream-msg","errorCode":null,"errorMessage":"Invalid payload format: {stream_msg:?}","messagePattern":"Invalid payload format: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":870,"sourceCode":"                anyhow::ensure!(\n                    value == PAYLOAD_KIND_TYPED,\n                    \"Unknown payload kind '{value}'\"\n                );\n                typed_payload = true;\n            }\n            b\"encoding\" => {\n                let redis::Value::BulkString(bytes) = &pair[1] else {\n                    anyhow::bail!(\"Invalid encoding format: {stream_msg:?}\");\n                };\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 => {","sourceCodeStart":852,"sourceCodeEnd":888,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L852-L888","documentation":"decode_bus_message parses a Redis stream entry into a BusMessage. When the entry contains a 'payload' header, its value must be a Redis BulkString; if it is any other redis::Value variant (e.g. Nil, Int, or a nested array), the decoder cannot turn it into raw bytes and bails with this error. It indicates the stream entry is malformed or was not written by the expected message-bus producer.","triggerScenarios":"Calling stream_messages (or decode_bus_message directly) on a Redis stream whose entry has a 'payload' field stored as a non-BulkString redis::Value — e.g. a NIL reply from a truncated entry, or a hand-written/foreign producer that stored an int, array, or simple string under 'payload'.","commonSituations":"Manual redis-cli writes into the bus stream with wrong value types; corrupted or truncated stream entries (XTRIM/XACK edge cases); a producer version writing a different payload representation than the consumer expects; pointing the consumer at a stream used by another application.","solutions":["Inspect the raw stream entry (XRANGE the stream ID) and confirm the 'payload' field is a binary string; re-publish the message correctly.","Fix the producer to write payload bytes as a BulkString (Bytes::copy_from_slice / raw bytes), matching the message-bus wire format.","Remove or skip the malformed stream entry (XDEL) so the consumer can continue past it.","Verify the consumer is pointed at the correct Redis stream/topic and not a foreign stream with a different schema."],"exampleFix":"// before (foreign producer, wrong type)\nredis.xadd(stream, &[(\"payload\", \"42\")]); // stores as integer-ish\n// after\nredis.xadd(stream, &[(\"payload\", bytes_arg) ]); // pass raw bytes / BulkString","handlingStrategy":"try-catch","validationCode":"// Rust: skip or reject entries whose payload field is not raw bytes before decode\nfn has_bulk_payload(fields: &[(String, redis::Value)]) -> bool {\n    fields.iter().any(|(k, v)| k == \"payload\" && matches!(v, redis::Value::BulkString(_)))\n}","typeGuard":"fn as_bulk_string(v: &redis::Value) -> Option<&Vec<u8>> {\n    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(\"Invalid payload format\") => {\n        // log stream entry, skip/XDEL malformed entry, continue consuming\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Only write bus stream entries via the library's publish API, not manual XADD.","Never store non-BulkString redis values under the 'payload' field.","Monitor for decode errors and alert on foreign writers to bus streams.","Pin producer and consumer versions of the message-bus wire format."],"tags":["redis","decoding","malformed-payload","msgbus"],"backgroundTag":"unexpected-response-shape","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"}