{"record":{"id":"99ea946c8faaf576","repo":"nautechsystems/nautilus_trader","slug":"flush-command-should-not-be-drained","errorCode":null,"errorMessage":"Flush command should not be drained","messagePattern":"Flush command should not be drained","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/cache.rs","lineNumber":870,"sourceCode":"                    log::error!(\"Null `payload` for `replace_list`\");\n                }\n            }\n            DatabaseOperation::Delete => {\n                log::debug!(\n                    \"Processing DELETE for collection: {}, key: {}, payload: {:?}\",\n                    collection,\n                    key,\n                    msg.payload.as_ref().map(std::vec::Vec::len)\n                );\n                // `payload` can be `None` for a delete operation\n                if let Err(e) = delete(&mut pipe, collection, &key, msg.payload) {\n                    log::error!(\"{e}\");\n                } else {\n                    has_pending_ops = true;\n                }\n            }\n            DatabaseOperation::Close => panic!(\"Close command should not be drained\"),\n            DatabaseOperation::Flush(_) => panic!(\"Flush command should not be drained\"),\n        }\n    }\n\n    flush_pending_pipeline(conn, &mut pipe, &mut has_pending_ops).await;\n}\n\nasync fn flush_pending_pipeline(\n    conn: &mut ConnectionManager,\n    pipe: &mut Pipeline,\n    has_pending_ops: &mut bool,\n) {\n    if !*has_pending_ops {\n        return;\n    }\n\n    if let Err(e) = pipe.query_async::<()>(conn).await {\n        log::error!(\"{e}\");\n    }","sourceCodeStart":852,"sourceCodeEnd":888,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/cache.rs#L852-L888","documentation":"Companion to the Close panic: in `drain_buffer` (crates/infrastructure/src/redis/cache.rs), encountering `DatabaseOperation::Flush(_)` in the drained buffer panics with \"Flush command should not be drained\". Flush is a control/lifecycle command that the command processor must intercept; finding it in the drainable operation buffer indicates the internal separation between control commands and data operations broke.","triggerScenarios":"A `DatabaseOperation::Flush` being enqueued into the operation buffer and then drained by `drain_buffer` (via `process_commands`, `handle_command`, or `flush_buffer`) instead of being handled as an explicit cache flush command.","commonSituations":"Calling cache flush concurrently with buffered writes so the Flush op lands in the queue; custom integrations enqueueing Flush into the data channel; version mismatches where Flush routing changed.","solutions":["Route Flush through the dedicated control path (handle_command/process_commands match on Flush) and never push it into the drained buffer.","Intercept Flush/Close variants before enqueueing database operations.","Reproduce with RUST_BACKTRACE=1 to find the enqueue site and fix the routing."],"exampleFix":"// before\nbuffer_tx.send(DatabaseOperation::Flush(pattern)).await?; // lands in drained buffer\n\n// after\n// invoke the cache's flush command API so handle_command consumes it directly","handlingStrategy":"validation","validationCode":"debug_assert!(!matches!(op, DatabaseOperation::Flush(_)), \"Flush must go through the control path\");","typeGuard":null,"tryCatchPattern":"// guard enqueue sites: if is_control_op(&op) { control_tx.send(op).await?; } else { buffer_tx.send(op).await?; }","preventionTips":["Expose Flush only through the cache's command API, never through the op buffer.","Avoid racing user-triggered flush with background buffer draining.","Add an assertion/test that the drained stream contains only data operations."],"tags":["rust","redis","panic","invariant","flush"],"backgroundTag":"internal-invariant-violation","analyzedSha":"18893faf8b356be3320add8de2f861b0b647cf06","analyzedAt":"2026-09-08T20:49:34.690Z","contentChangedAt":"2026-09-08T20:49:34.690Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}