nautechsystems/nautilus_trader · error

Invalid topic format: {stream_msg:?}

Error message

Invalid topic format: {stream_msg:?}

What it means

When decoding the `topic` field, its value must be a BulkString that can be decoded as UTF-8. If the value paired with key `topic` is not a BulkString, this error is thrown before the UTF-8 conversion (which would produce the separate 'Error parsing topic' error).

Source

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

    if fields.len() < 4 || fields.len() % 2 != 0 {
        anyhow::bail!("Invalid stream message format: {stream_msg:?}");
    }

    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:?}");

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Fix the producer to always XADD topic as a UTF-8 string value
  2. Inspect {stream_msg:?} to see the actual value variant stored under `topic`
  3. Remove or skip the malformed entry (XDEL) so the consumer can proceed
  4. Correct test fixture data to use BulkString for the topic value

Example fix

// before
redis.xadd(stream, "*", &[("topic", 42)]); // non-string
// after
redis.xadd(stream, "*", &[("topic", "events.trades")]);
Defensive patterns

Strategy: validation

Validate before calling

fn topic_is_valid(entry: &redis::Value) -> bool {
    matches!(entry, redis::Value::Array(f) if f.chunks(2).any(|c|
        matches!((&c[0], &c[1]), (redis::Value::BulkString(k), redis::Value::BulkString(_)) if k == b"topic")))
}

Try / catch

match decode_bus_message(&entry) {
    Err(e) if e.to_string().contains("Invalid topic format") => {
        log::warn!("bad topic field, skipping entry: {e}");
    }
    other => other?,
}

Prevention

When it happens

Trigger: A stream entry containing a `topic` field whose value is Nil, Int, or an Array instead of a byte string — e.g. a producer XADDing a non-string topic value or corrupted/mocked entry data.

Common situations: Custom producers writing non-string field values; test fixtures with wrong value variants; intermediary tooling transforming the stream entry.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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