nautechsystems/nautilus_trader · error

Invalid encoding format: {stream_msg:?}

Error message

Invalid encoding format: {stream_msg:?}

What it means

The optional `encoding` header must be a BulkString containing a value parseable into SerializationEncoding. A non-BulkString value raises this error before the parse step (a bad string value would instead produce 'Error parsing encoding').

Source

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

                    String::from_utf8(bytes.clone())
                        .map_err(|e| anyhow::anyhow!("Error parsing type: {e}"))?,
                );
            }
            key if key == PAYLOAD_KIND_FIELD.as_bytes() => {
                let redis::Value::BulkString(bytes) = &pair[1] else {
                    anyhow::bail!("Invalid payload kind format: {stream_msg:?}");
                };
                let value = std::str::from_utf8(bytes)
                    .map_err(|e| anyhow::anyhow!("Error parsing payload kind: {e}"))?;
                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 {

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Write the encoding header as a string the SerializationEncoding parser accepts (e.g. 'json' or 'msgpack')
  2. Omit the encoding field to use the default encoding (SerializationEncoding::default())
  3. Align producer and consumer versions so the encoding header format matches
  4. Inspect the raw entry and XDEL the malformed one to unblock the stream consumer

Example fix

// before
redis.xadd(stream, "*", &[("encoding", 2)]);
// after
redis.xadd(stream, "*", &[("encoding", "json")]);
Defensive patterns

Strategy: validation

Validate before calling

fn encoding_header_ok(entry: &redis::Value) -> bool {
    matches!(entry, redis::Value::Array(f) if f.chunks(2).all(|c|
        !matches!(&c[0], redis::Value::BulkString(k) if k == b"encoding") || matches!(c[1], redis::Value::BulkString(_))))
}

Try / catch

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

Prevention

When it happens

Trigger: A stream entry with an `encoding` field whose value is not a byte string (Nil, Int, etc.) — from a producer writing a non-string encoding header or corrupted/mock entry data.

Common situations: Producers from a different library version writing the header differently; custom pipelines stamping binary or numeric encoding markers; hand-crafted stream entries in tests.

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