{"record":{"id":"2b68d838f0063afa","repo":"nautechsystems/nautilus_trader","slug":"python-stream-processor-failed-e","errorCode":null,"errorMessage":"Python stream processor failed: {e}","messagePattern":"Python stream processor failed: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/live/src/python/node.rs","lineNumber":927,"sourceCode":"    /// MessagePack typed payloads. Each callback receives an encoding-independent Python mapping\n    /// that follows the concrete payload's serialized object shape and adds a `payload_type` field.\n    /// Other encodings are skipped with a warning. External egress is suppressed while the callback\n    /// runs, so synchronous publications remain local. Callback exceptions are logged, stop the\n    /// remaining processors, and skip internal republishing.\n    ///\n    /// # Errors\n    ///\n    /// Returns an error if the node is no longer available for mutation.\n    #[pyo3(name = \"add_stream_processor\")]\n    fn py_add_stream_processor(&self, callback: Py<PyAny>) -> PyResult<()> {\n        self.node_mut()?\n            .add_stream_processor_with_mapping(move |_, mapping| {\n                Python::attach(|py| {\n                    let payload = json_value_to_py(py, mapping)?;\n                    callback.call1(py, (payload,))?;\n                    Ok(())\n                })\n                .map_err(|e: PyErr| anyhow::anyhow!(\"Python stream processor failed: {e}\"))\n            });\n        Ok(())\n    }\n\n    /// Runs the live node on the caller's asyncio event loop.\n    ///\n    /// Takes the node and returns an awaitable that resolves once the node has stopped. The host\n    /// owns the loop and its signal handling, so this installs no signal handlers. Stop the node\n    /// through the handle from `handle()`; cancelling the awaiting task requests the same graceful\n    /// shutdown, waits for it to finish, then re-raises the cancellation.\n    ///\n    /// Capture `cache`, `portfolio`, and `handle()` before calling this. They stay usable while the\n    /// node runs, whereas the node itself is owned by the returned awaitable.\n    ///\n    /// # Limitations\n    ///\n    /// A node configured with a cache database backing is rejected. Those backings wait for their\n    /// worker task by blocking the calling thread, which stalls the host loop rather than merely","sourceCodeStart":909,"sourceCodeEnd":945,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/live/src/python/node.rs#L909-L945","documentation":"This error is raised when the Python callback registered for the message bus stream processor raises an exception. The Rust closure `add_stream_processor_with_mapping` invokes the callback via PyO3; any `PyErr` it returns is converted with `anyhow!` into this message containing the Python traceback text. The stream processor mapping itself failed, so the mapped payload is not delivered.","triggerScenarios":"Registering a stream-processing callback via the live node's Python API where the callback raises — e.g. `callback(payload)` throws a TypeError (wrong signature/arity), the handler references missing state, or JSON conversion of the mapping produces an object the callback mishandles.","commonSituations":"User callbacks with the wrong signature (expecting raw bytes instead of a dict); handlers raising on the first messages after config changes; exceptions in long-running signal-processing callbacks attached to the message bus.","solutions":["Fix the exception reported in `{e}` — the message contains the full Python traceback.","Ensure the callback accepts exactly one argument (the mapped payload dict) and is callable.","Test the callback standalone with a representative payload before attaching it to the stream.","Wrap the callback body in try/except if failures should be logged rather than aborting the stream registration."],"exampleFix":"# before\ndef handler(mapping):\n    return process(mapping['data'])  # KeyError if 'data' missing\n# after\ndef handler(mapping):\n    try:\n        return process(mapping.get('data'))\n    except Exception as e:\n        logging.exception('stream handler failed: %s', e)","handlingStrategy":"try-catch","validationCode":"# Validate the callback before registering\nassert callable(cb) and cb.__code__.co_argcount == 1, \"callback must take one payload arg\"\ntry:\n    cb({\"test\": 1})\nexcept Exception as e:\n    raise RuntimeError(f\"stream callback not safe: {e}\")","typeGuard":null,"tryCatchPattern":"try:\n    node.add_stream_processor(callback)\nexcept Exception as e:\n    if \"Python stream processor failed\" in str(e):\n        logging.exception(\"stream callback raised: %s\", e)\n        # fix or replace the callback\n    else:\n        raise","preventionTips":["Give the callback exactly one parameter (the mapped payload dict)","Unit-test the callback with representative payloads before attaching it","Wrap callback bodies in try/except with logging for non-fatal failures","Don't mutate shared state from the callback without locks"],"tags":["python","pyo3","callback","streaming"],"backgroundTag":"python-callback-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"}