nautechsystems/nautilus_trader · error

Invalid payload format: {stream_msg:?}

Error message

Invalid payload format: {stream_msg:?}

What it means

decode_bus_message parses a Redis stream entry into a BusMessage. When the entry contains a 'payload' header, its value must be a Redis BulkString; if it is any other redis::Value variant (e.g. Nil, Int, or a nested array), the decoder cannot turn it into raw bytes and bails with this error. It indicates the stream entry is malformed or was not written by the expected message-bus producer.

Source

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

                anyhow::ensure!(
                    value == PAYLOAD_KIND_TYPED,
                    "Unknown payload kind '{value}'"
                );
                typed_payload = true;
            }
            b"encoding" => {
                let redis::Value::BulkString(bytes) = &pair[1] else {
                    anyhow::bail!("Invalid encoding format: {stream_msg:?}");
                };
                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 => {

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Inspect the raw stream entry (XRANGE the stream ID) and confirm the 'payload' field is a binary string; re-publish the message correctly.
  2. Fix the producer to write payload bytes as a BulkString (Bytes::copy_from_slice / raw bytes), matching the message-bus wire format.
  3. Remove or skip the malformed stream entry (XDEL) so the consumer can continue past it.
  4. Verify the consumer is pointed at the correct Redis stream/topic and not a foreign stream with a different schema.

Example fix

// before (foreign producer, wrong type)
redis.xadd(stream, &[("payload", "42")]); // stores as integer-ish
// after
redis.xadd(stream, &[("payload", bytes_arg) ]); // pass raw bytes / BulkString
Defensive patterns

Strategy: try-catch

Validate before calling

// Rust: skip or reject entries whose payload field is not raw bytes before decode
fn has_bulk_payload(fields: &[(String, redis::Value)]) -> bool {
    fields.iter().any(|(k, v)| k == "payload" && matches!(v, redis::Value::BulkString(_)))
}

Type guard

fn as_bulk_string(v: &redis::Value) -> Option<&Vec<u8>> {
    match v { redis::Value::BulkString(b) => Some(b), _ => None }
}

Try / catch

match stream_messages(&mut con, stream, count).await {
    Ok(msgs) => { /* use msgs */ }
    Err(e) if e.to_string().contains("Invalid payload format") => {
        // log stream entry, skip/XDEL malformed entry, continue consuming
    }
    Err(e) => return Err(e),
}

Prevention

When it happens

Trigger: Calling stream_messages (or decode_bus_message directly) on a Redis stream whose entry has a 'payload' field stored as a non-BulkString redis::Value — e.g. a NIL reply from a truncated entry, or a hand-written/foreign producer that stored an int, array, or simple string under 'payload'.

Common situations: Manual redis-cli writes into the bus stream with wrong value types; corrupted or truncated stream entries (XTRIM/XACK edge cases); a producer version writing a different payload representation than the consumer expects; pointing the consumer at a stream used by another application.

Related errors

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