What causes "Invalid stream message format: {stream_msg:?}" in nautechsystems/nautilus_trader?

Confirm you are passing the per-entry value from the XREAD reply, not the outer nested response

Invalid stream message format: {stream_msg:?}

Error message

Invalid stream message format: {stream_msg:?}

What it means

decode_bus_message expects each stream entry to be a redis::Value::Array of alternating key/value BulkStrings. If the entry is not an Array at all, this error is thrown. It indicates the raw XREAD reply shape differs from what the decoder requires.

Source

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

        () = &mut retry_timer => true,
        () = &mut terminate => false,
    }
}

async fn wait_for_stream_signal(stream_signal: &Arc<AtomicBool>) {
    let check_timer = tokio::time::interval(Duration::from_millis(100));

    tokio::pin!(check_timer);

    while !stream_signal.load(Ordering::Relaxed) {
        check_timer.tick().await;
    }
}

// Redis fields are unordered, and older streams may omit type or encoding headers
fn decode_bus_message(stream_msg: &redis::Value) -> anyhow::Result<BusMessage> {
    let redis::Value::Array(fields) = stream_msg else {
        anyhow::bail!("Invalid stream message format: {stream_msg:?}");
    };

    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() {

View on GitHub (pinned to 18893faf8b)

Solutions

Confirm you are passing the per-entry value from the XREAD reply, not the outer nested response
Log the offending stream_msg to see the actual redis::Value variant returned
Check for middleware/proxy rewriting RESP replies; connect directly to Redis to compare
Align the redis crate version used by the library with the reply format your server emits

Defensive patterns

Strategy: validation

Validate before calling

fn is_array_entry(v: &redis::Value) -> bool { matches!(v, redis::Value::Array(_)) }

Type guard

fn as_entry_array(v: &redis::Value) -> Option<&Vec<redis::Value>> {
    if let redis::Value::Array(fields) = v { Some(fields) } else { None }
}

Try / catch

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

Prevention

Pass only per-entry XREAD reply values into the decoder
Keep the redis crate and server RESP protocol versions consistent
Avoid proxies/middleware that reshape replies

When it happens

Trigger: Feeding decode_bus_message with a redis::Value that is Nil, BulkString, Int, or a Map variant instead of an Array — e.g. a malformed response from a proxy, or passing the wrong nesting level of the XREAD reply.

Common situations: Redis proxies/mocking layers returning non-standard reply shapes; custom code piping entries into the decoder from a differently-shaped response; decoder version mismatch with the redis crate's Value representation.

Related errors

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