{"record":{"id":"1c18d7b159abd538","repo":"nautechsystems/nautilus_trader","slug":"error-reading-from-stream-e","errorCode":null,"errorMessage":"Error reading from stream: {e:?}","messagePattern":"Error reading from stream: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"crates/infrastructure/src/redis/msgbus.rs","lineNumber":702,"sourceCode":"                                match decode_bus_message(array) {\n                                    Ok(msg) => {\n                                        if let Err(e) = tx.send(msg).await {\n                                            log::debug!(\"Channel closed: {e:?}\");\n                                            break 'outer; // End streaming\n                                        }\n                                    }\n                                    Err(e) => {\n                                        log::error!(\"{e:?}\");\n                                    }\n                                }\n                            }\n                        }\n                    }\n                }\n            }\n            Err(e) => {\n                if !is_retryable_stream_error(&e) {\n                    anyhow::bail!(\"Error reading from stream: {e:?}\");\n                }\n\n                log::error!(\"Error reading from stream: {e:?}\");\n\n                let Some(reconnected) =\n                    reconnect_stream_connection(&config, &stream_signal, &mut read_error_count)\n                        .await?\n                else {\n                    break;\n                };\n                con = reconnected;\n            }\n        }\n    }\n\n    log_task_stopped(MSGBUS_STREAM);\n    Ok(())\n}","sourceCodeStart":684,"sourceCodeEnd":720,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/msgbus.rs#L684-L720","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Inspect the wrapped error ({e:?}) to identify the exact Redis error (NOGROUP, WRONGTYPE, auth, etc.)","If NOGROUP: recreate the consumer group (XGROUP CREATE ... MKSTREAM) and restart the consumer","If WRONGTYPE: check what is writing to the stream key — delete/rename the conflicting key and fix the writer","If auth/ACL errors: update the connection credentials or grant the user stream read permissions","Restart the message bus after fixing the server-side condition; the loop does not self-heal non-retryable errors"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"match bus.stream_messages().await {\n    Err(e) if e.to_string().contains(\"Error reading from stream\") => {\n        // inspect debug payload; recreate consumer group / fix key then restart with backoff\n        tokio::time::sleep(Duration::from_secs(5)).await;\n    }\n    other => other?,\n}","preventionTips":["Ensure the consumer group exists (XGROUP CREATE ... MKSTREAM) before starting the bus","Never write a different type to the stream key; reserve the key for the bus","Use stable credentials/ACLs that grant stream read for the bus user","Monitor for stream key deletion (FLUSHALL/DEL) in shared Redis instances"],"tags":["redis","streams","network","consumer"],"backgroundTag":"database-query-failed","analyzedSha":"18893faf8b356be3320add8de2f861b0b647cf06","analyzedAt":"2026-09-08T20:49:34.690Z","contentChangedAt":"2026-09-08T20:49:34.690Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}