{"record":{"id":"9e070c73b050e627","repo":"nautechsystems/nautilus_trader","slug":"invalid-payload-kind-format-stream-msg","errorCode":null,"errorMessage":"Invalid payload kind format: {stream_msg:?}","messagePattern":"Invalid payload kind format: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":848,"sourceCode":"                    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!(\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}\"))?;","sourceCodeStart":830,"sourceCodeEnd":866,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L830-L866","documentation":"The optional payload-kind header (PAYLOAD_KIND_FIELD) must be a BulkString, and after UTF-8 decoding it must equal PAYLOAD_KIND_TYPED. A non-BulkString value raises this 'Invalid payload kind format' error before the value check (which would yield 'Unknown payload kind').","triggerScenarios":"A stream entry whose payload-kind field value is not a byte string — a producer writing a numeric or malformed kind marker instead of the expected typed-payload constant.","commonSituations":"Producers using an incompatible/older message format with a different kind encoding; hand-written stream entries; test data built with wrong redis::Value variants.","solutions":["Write the payload-kind field using the library's constant (PAYLOAD_KIND_TYPED) rather than a custom value","Omit the payload-kind field for legacy (untyped) messages instead of sending an invalid marker","Align producer and consumer library versions so both use the same payload-kind field name and constant","Inspect the raw entry to confirm the stored value, then XDEL the malformed entry"],"exampleFix":"// before\nredis.xadd(stream, \"*\", &[(\"payload_kind\", 1)]);\n// after\nredis.xadd(stream, \"*\", &[(\"payload_kind\", \"typed\")]); // matches PAYLOAD_KIND_TYPED","handlingStrategy":"validation","validationCode":"const PAYLOAD_KIND_TYPED: &str = \"typed\"; // match library constant\nfn payload_kind_ok(entry: &redis::Value) -> bool {\n    matches!(entry, redis::Value::Array(f) if f.chunks(2).all(|c|\n        !matches!(&c[0], redis::Value::BulkString(k) if k == b\"payload_kind\") || matches!(c[1], redis::Value::BulkString(_))))\n}","typeGuard":null,"tryCatchPattern":"match decode_bus_message(&entry) {\n    Err(e) if e.to_string().contains(\"payload kind\") => {\n        log::warn!(\"unrecognized payload kind header, skipping: {e}\");\n    }\n    other => other?,\n}","preventionTips":["Use the library's PAYLOAD_KIND_FIELD/PAYLOAD_KIND_TYPED constants in producers","Omit the header for legacy untyped messages instead of inventing values","Keep producer and consumer crate versions in lockstep"],"tags":["redis","deserialization","validation"],"backgroundTag":"invalid-enum-value","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"}