{"record":{"id":"ded0ad8c4880a918","repo":"nautechsystems/nautilus_trader","slug":"invalid-type-format-stream-msg","errorCode":null,"errorMessage":"Invalid type format: {stream_msg:?}","messagePattern":"Invalid type format: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":839,"sourceCode":"\n    for pair in fields.as_chunks::<2>().0 {\n        let redis::Value::BulkString(key) = &pair[0] else {\n            anyhow::bail!(\"Invalid stream field key: {stream_msg:?}\");\n        };\n\n        match key.as_slice() {\n            b\"topic\" => {\n                let redis::Value::BulkString(bytes) = &pair[1] else {\n                    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            }","sourceCodeStart":821,"sourceCodeEnd":857,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L821-L857","documentation":"When decoding the optional `type` header field, its value must be a BulkString. Any other value variant paired with the `type` key aborts decoding with this error, since the message type name cannot be read.","triggerScenarios":"A stream entry with a `type` field whose value is not a byte string (Nil, Int, Array, etc.) — from a producer writing a non-string type header or corrupted/mock entry data.","commonSituations":"Custom or buggy producers serializing the type header incorrectly; test fixtures with wrong value variants; stream data written by a different tool with binary/odd encodings.","solutions":["Fix the producer to write the type header as a UTF-8 string (e.g. 'data.TradeTick')","Inspect the raw entry ({stream_msg:?}) to confirm the value variant under `type`","Omit the `type` field entirely if the message is a legacy/custom payload — it is optional","Delete the malformed entry (XDEL) to unblock the consumer"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"fn type_header_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\"type\") || matches!(c[1], redis::Value::BulkString(_))))\n}","typeGuard":null,"tryCatchPattern":"match decode_bus_message(&entry) {\n    Err(e) if e.to_string().contains(\"Invalid type format\") => {\n        log::warn!(\"bad type header, skipping entry: {e}\");\n    }\n    other => other?,\n}","preventionTips":["Write the type header as a plain UTF-8 string or omit it","Keep producer serialization of headers in one shared module","Version the stream protocol and gate producers on it"],"tags":["redis","deserialization","validation"],"backgroundTag":"type-mismatch","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"}