nautechsystems/nautilus_trader · error

Error parsing topic: {e}

Error message

Error parsing topic: {e}

What it means

decode_bus_message parses Redis stream entries into bus messages. The `topic` field must be a BulkString containing valid UTF-8; when the raw bytes cannot be decoded as UTF-8, this anyhow error is returned with the underlying FromUtf8Error message. It indicates the stream entry's topic bytes are corrupt or were written with a non-UTF-8 encoding.

Source

Thrown at crates/infrastructure/src/redis/msgbus.rs:834

    let mut topic: Option<String> = None;
    let mut type_name: Option<String> = None;
    let mut typed_payload = false;
    let mut encoding = SerializationEncoding::default();
    let mut payload: Option<Bytes> = None;

    for pair in fields.as_chunks::<2>().0 {
        let redis::Value::BulkString(key) = &pair[0] else {
            anyhow::bail!("Invalid stream field key: {stream_msg:?}");
        };

        match key.as_slice() {
            b"topic" => {
                let redis::Value::BulkString(bytes) = &pair[1] else {
                    anyhow::bail!("Invalid topic format: {stream_msg:?}");
                };
                topic = Some(
                    String::from_utf8(bytes.clone())
                        .map_err(|e| anyhow::anyhow!("Error parsing topic: {e}"))?,
                );
            }
            b"type" => {
                let redis::Value::BulkString(bytes) = &pair[1] else {
                    anyhow::bail!("Invalid type format: {stream_msg:?}");
                };
                type_name = Some(
                    String::from_utf8(bytes.clone())
                        .map_err(|e| anyhow::anyhow!("Error parsing type: {e}"))?,
                );
            }
            key if key == PAYLOAD_KIND_FIELD.as_bytes() => {
                let redis::Value::BulkString(bytes) = &pair[1] else {
                    anyhow::bail!("Invalid payload kind format: {stream_msg:?}");
                };
                let value = std::str::from_utf8(bytes)
                    .map_err(|e| anyhow::anyhow!("Error parsing payload kind: {e}"))?;
                anyhow::ensure!(

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Inspect the offending Redis stream entry (XRANGE/XRSTREAM dump) and find which producer wrote non-UTF-8 topic bytes.
  2. Ensure all publishers write topics as UTF-8 strings (standard Redis clients encode strings as UTF-8; avoid custom binary encoders for headers).
  3. Delete or skip the corrupt stream entries, then re-publish with correct encoding.
  4. Verify no intermediate proxy/serializer (e.g. msgpack round-trip misconfiguration) mangles the topic field.

Example fix

// before: publishing with raw bytes
redis.xadd(stream, [(b"topic", non_utf8_bytes), ...]);
// after: ensure UTF-8 encoding
let topic = String::from_utf8(bytes).expect("topic must be UTF-8");
redis.xadd(stream, &[("topic", topic.as_str()), ...]);
Defensive patterns

Strategy: try-catch

Validate before calling

// Rust: pre-validate header bytes before publish
if let Err(e) = std::str::from_utf8(&topic_bytes) {
    return Err(format!("topic not UTF-8: {e}"));
}

Try / catch

match decode_bus_message(entry) {
    Ok(msg) => process(msg),
    Err(e) if e.to_string().contains("Error parsing topic") => {
        log::warn!("skipping entry with corrupt topic: {e}");
        // skip / dead-letter the entry
    }
    Err(e) => return Err(e),
}

Prevention

When it happens

Trigger: Calling stream_messages (which calls decode_bus_message) on a Redis stream entry whose `topic` value is a BulkString containing invalid UTF-8 bytes (String::from_utf8 fails).

Common situations: A producer wrote binary/corrupted data into the topic field; manual edits or tooling inserted raw bytes into the Redis stream; a non-Nautilus client publishes entries with different byte encodings (e.g. UTF-16 or Latin-1 strings).

Related errors

AI-assisted analysis of nautechsystems/nautilus_trader@18893faf8b (2026-09-08). Data as JSON: /api/errors/1de86328bee38f1f. Report an issue: GitHub.