nautechsystems/nautilus_trader · error

Stream message missing topic: {stream_msg:?}

Error message

Stream message missing topic: {stream_msg:?}

What it means

Every Redis stream entry decoded by decode_bus_message must carry a 'topic' header identifying where the message was published. If the entry's fields contain no 'topic' pair, the decoder cannot construct a BusMessage and bails. This guards against foreign or corrupt entries in the stream.

Source

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

                };
                let value = std::str::from_utf8(bytes)
                    .map_err(|e| anyhow::anyhow!("Error parsing encoding: {e}"))?;
                encoding = value
                    .parse()
                    .map_err(|e| anyhow::anyhow!("Error parsing encoding: {e}"))?;
            }
            b"payload" => {
                let redis::Value::BulkString(bytes) = &pair[1] else {
                    anyhow::bail!("Invalid payload format: {stream_msg:?}");
                };
                payload = Some(Bytes::copy_from_slice(bytes));
            }
            _ => {}
        }
    }

    let Some(topic) = topic else {
        anyhow::bail!("Stream message missing topic: {stream_msg:?}");
    };
    let Some(payload) = payload else {
        anyhow::bail!("Stream message missing payload: {stream_msg:?}");
    };
    let payload_type = match type_name {
        Some(type_name) if typed_payload => BusPayloadType::from_typed_name(&type_name)
            .ok_or_else(|| anyhow::anyhow!("Unknown typed payload '{type_name}'"))?,
        Some(type_name) => BusPayloadType::from_name(&type_name),
        None if typed_payload => {
            anyhow::bail!("Typed stream message missing type: {stream_msg:?}")
        }
        None => BusPayloadType::Custom(Ustr::default()),
    };

    Ok(BusMessage::with_str_topic(
        topic,
        payload_type,
        payload,

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Check the stream entry (XRANGE) and confirm it includes the 'topic' field; re-publish with a topic.
  2. Fix or upgrade the producer so every published BusMessage includes the topic header.
  3. Delete the malformed entry (XDEL) so decoding can proceed.
  4. Confirm the consumer is subscribed to the intended stream (correct topic/stream key).

Example fix

// before (manual XADD without topic)
XADD mystream * payload "..."
// after
XADD mystream * topic "quotes.BTC-USD" payload "..."
Defensive patterns

Strategy: try-catch

Validate before calling

// Validate required headers before writing
fn entry_is_publishable(fields: &[(&str, &str)]) -> bool {
    fields.iter().any(|(k, _)| *k == "topic")
}

Type guard

fn topic_of<'a>(fields: &'a [(String, redis::Value)]) -> Option<&'a str> {
    fields.iter().find(|(k, _)| k == "topic")
        .and_then(|(_, v)| match v { redis::Value::BulkString(b) => std::str::from_utf8(b).ok(), _ => None })
}

Try / catch

match stream_messages(&mut con, stream, count).await {
    Ok(msgs) => { /* use msgs */ }
    Err(e) if e.to_string().contains("missing topic") => {
        tracing::warn!("skipping entry without topic: {e}");
    }
    Err(e) => return Err(e),
}

Prevention

When it happens

Trigger: stream_messages encounters a stream entry written without the 'topic' field — e.g. by an older producer version, a hand-crafted XADD, or a different application sharing the same stream key.

Common situations: Redis stream keys reused by other services; manual testing with redis-cli; producer/consumer version skew where the header name changed; partially written entries from an aborted pipeline.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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