nautechsystems/nautilus_trader · critical

Error reading from stream: {e:?}

Error message

Error reading from stream: {e:?}

What it means

stream_messages reads entries from a Redis stream in a loop. If XREAD fails with a non-retryable error (is_retryable_stream_error returns false), the loop aborts by propagating this error with the debug representation of the underlying redis error. Retryable errors instead trigger reconnection logic and logging.

Source

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

                                match decode_bus_message(array) {
                                    Ok(msg) => {
                                        if let Err(e) = tx.send(msg).await {
                                            log::debug!("Channel closed: {e:?}");
                                            break 'outer; // End streaming
                                        }
                                    }
                                    Err(e) => {
                                        log::error!("{e:?}");
                                    }
                                }
                            }
                        }
                    }
                }
            }
            Err(e) => {
                if !is_retryable_stream_error(&e) {
                    anyhow::bail!("Error reading from stream: {e:?}");
                }

                log::error!("Error reading from stream: {e:?}");

                let Some(reconnected) =
                    reconnect_stream_connection(&config, &stream_signal, &mut read_error_count)
                        .await?
                else {
                    break;
                };
                con = reconnected;
            }
        }
    }

    log_task_stopped(MSGBUS_STREAM);
    Ok(())
}

View on GitHub (pinned to 18893faf8b)

Solutions

1. Inspect the wrapped error ({e:?}) to identify the exact Redis error (NOGROUP, WRONGTYPE, auth, etc.)
2. If NOGROUP: recreate the consumer group (XGROUP CREATE ... MKSTREAM) and restart the consumer
3. If WRONGTYPE: check what is writing to the stream key — delete/rename the conflicting key and fix the writer
4. If auth/ACL errors: update the connection credentials or grant the user stream read permissions
5. Restart the message bus after fixing the server-side condition; the loop does not self-heal non-retryable errors

match bus.stream_messages().await {
    Err(e) if e.to_string().contains("Error reading from stream") => {
        // inspect debug payload; recreate consumer group / fix key then restart with backoff
        tokio::time::sleep(Duration::from_secs(5)).await;
    }
    other => other?,
}

When it happens

Trigger: A persistent failure while calling XREAD on the stream — e.g. NOGROUP (stream deleted/consumer group missing), wrong-type errors (key holds a non-stream value), auth failures after connect, or protocol corruption that the retry classifier deems unrecoverable.

Common situations: The stream key was deleted or FLUSHALL'd while the bus was running; another client overwrote the key with a different type; Redis restarted with the consumer group gone; ACL/permission changes mid-session.

Understand the failure class

Background: Database query failed: Internal Server Error 500s wrapping SQL, Prisma, and connection failures — what to check first — this error's family across 16 libraries.

Related errors

• rate limiter decision lock poisoned
• Close command should not be drained
• Flush command should not be drained
• Redis config error: username supplied without password. Eith
• Invalid `ConnectionMode` value: {value}

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