nautechsystems/nautilus_trader · error

Typed stream message missing type: {stream_msg:?}

Error message

Typed stream message missing type: {stream_msg:?}

What it means

When a stream entry is marked with the typed payload kind header, it must also carry a 'type' header naming the payload type. decode_bus_message bails if kind is typed but type_name is None, since it cannot resolve a BusPayloadType. This catches producers that set kind=typed without the accompanying type name.

Source

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

                };
                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,
        encoding,
    ))
}

async fn run_heartbeat(
    heartbeat_interval_secs: u16,
    signal: Arc<AtomicBool>,
    pub_tx: tokio::sync::mpsc::UnboundedSender<BusMessage>,
) {
    log_task_started("heartbeat");

View on GitHub (pinned to 18893faf8b)

Solutions

1. Add the 'type' header to the published entry (e.g. the instrument/data type name) so BusPayloadType::from_typed_name can resolve it.
2. Fix the producer to always set both the typed kind flag and the type name together.
3. Republish the message with correct headers and XDEL the bad entry.
4. If the payload is genuinely untyped, publish without the typed-kind header so the decoder falls back to BusPayloadType::Custom.

Example fix

// before (typed kind without type)
XADD bus * topic "q" payload "..." kind "typed"
// after
XADD bus * topic "q" payload "..." kind "typed" type "data.QuoteTick"

// Typed publishes must include both kind and type headers
fn typed_entry_ok(fields: &[(&str, &str)]) -> bool {
    let typed = fields.iter().any(|(k, v)| *k == "kind" && *v == "typed");
    let has_type = fields.iter().any(|(k, _)| *k == "type");
    !typed || has_type
}

fn typed_name_of(fields: &[(String, redis::Value)]) -> Option<String> {
    fields.iter().find(|(k, _)| k == "type")
        .and_then(|(_, v)| match v { redis::Value::BulkString(b) => String::from_utf8(b.clone()).ok(), _ => None })
}

match stream_messages(&mut con, stream, count).await {
    Ok(msgs) => { /* use msgs */ }
    Err(e) if e.to_string().contains("Typed stream message missing type") => {
        tracing::error!("producer bug: typed entry lacks type header: {e}");
    }
    Err(e) => return Err(e),
}

When it happens

Trigger: decode_bus_message (via stream_messages) reads an entry whose payload-kind header is set to the typed kind constant but which has no 'type' field — typically a producer writing inconsistent headers or a hand-crafted entry.

Common situations: Partial adoption of the typed-payload protocol in a producer; manual testing with redis-cli where the type header was forgotten; version skew between producer and consumer message formats.

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

• Stream message missing topic: {stream_msg:?}
• Stream message missing payload: {stream_msg:?}
• Unknown payload kind '{value}'
• Invalid payload format: {stream_msg:?}
• Error parsing topic: {e}

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