nautechsystems/nautilus_trader · error

Invalid stream field key: {stream_msg:?}

Error message

Invalid stream field key: {stream_msg:?}

What it means

decode_bus_message iterates key/value pairs and each key must be a redis::Value::BulkString. If a key is some other variant (Int, Nil, simple status string, etc.) the decoder bails with this error rather than guessing the field name.

Source

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

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

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Inspect the entry ({stream_msg:?}) to see which key element has the unexpected variant
  2. Ensure test fixtures build entries as Value::BulkString keys paired with values
  3. Force RESP2 (or match server/client protocol versions) so replies use BulkString representations
  4. Bypass or fix any proxy that rewrites reply types
Defensive patterns

Strategy: type-guard

Type guard

fn has_bulkstring_keys(v: &redis::Value) -> bool {
    matches!(v, redis::Value::Array(f) if f.len() % 2 == 0 &&
        f.chunks(2).all(|c| matches!(c[0], redis::Value::BulkString(_))))
}

Try / catch

match decode_bus_message(&entry) {
    Err(e) if e.to_string().contains("Invalid stream field key") => {
        log::warn!("non-string key in entry, skipping: {e}");
    }
    other => other?,
}

Prevention

When it happens

Trigger: A stream entry whose odd-indexed elements are not BulkStrings — normally impossible from real Redis XADD replies, so usually caused by mock/test data, a proxy rewriting replies, or misusing the decoder with fabricated values.

Common situations: Unit or integration fixtures constructing redis::Value entries incorrectly; RESP3 protocol negotiation with a client/server mismatch altering variant representations; custom middleware translating replies.

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/41b52cb5ff7e8d32. Report an issue: GitHub.