{"record":{"id":"56b953efca8f2c99","repo":"nautechsystems/nautilus_trader","slug":"close-command-should-not-be-drained","errorCode":null,"errorMessage":"Close command should not be drained","messagePattern":"Close command should not be drained","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/infrastructure/src/redis/cache.rs","lineNumber":869,"sourceCode":"                } else {\n                    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}\");","sourceCodeStart":851,"sourceCodeEnd":887,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/infrastructure/src/redis/cache.rs#L851-L887","documentation":"In the Redis cache's buffer-draining loop (crates/infrastructure/src/redis/cache.rs), `drain_buffer` matches on `DatabaseOperation` variants and panics if it encounters `DatabaseOperation::Close`. Control commands like Close are lifecycle signals that must be handled by the command-processing logic, not persisted/drained as data operations; reaching this branch means an internal invariant was violated — a Close op leaked into the operation buffer.","triggerScenarios":"A `DatabaseOperation::Close` being enqueued into the database command buffer and then processed by `drain_buffer` (called from `process_commands`, `handle_command`, or `flush_buffer`) instead of being consumed as a shutdown/control signal.","commonSituations":"Racing a cache shutdown/flush with in-flight writes so a Close command lands in the op queue; custom code or an older version constructing/enqueueing Close into the command channel; bugs in internal command routing during teardown.","solutions":["Audit how DatabaseOperation::Close enters the buffer; handle it in the control path (e.g. match it in handle_command/process_commands and stop the worker) rather than enqueueing it.","Filter out Close/Flush variants before pushing operations into the buffer.","Reproduce with the exact shutdown sequence, capture a backtrace (RUST_BACKTRACE=1), and file/check for an upstream fix."],"exampleFix":"// before\nif let Err(e) = cmd_tx.send(DatabaseOperation::Close).await { /* ... */ } // Close goes into data buffer\n\n// after\n// signal shutdown through the dedicated control channel / handle_command match arm,\n// not by enqueueing DatabaseOperation::Close into the drained buffer","handlingStrategy":"validation","validationCode":"// never enqueue control ops into the data buffer\nmatch op {\n    DatabaseOperation::Close | DatabaseOperation::Flush(_) => handle_control(op),\n    _ => buffer_tx.send(op).await?,\n}","typeGuard":null,"tryCatchPattern":"// panics are unrecoverable here; serialize shutdown so Close is handled before flush_buffer runs\nshutdown_token.cancel().await; worker.join().await;","preventionTips":["Use separate channels for control commands (Close/Flush) and data operations.","Coordinate shutdown ordering: drain buffers before issuing Close.","Fuzz/soak-test concurrent write + shutdown paths."],"tags":["rust","redis","panic","invariant","shutdown"],"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-14T00:17:10.932Z"}