{"record":{"id":"3b521e67a339682a","repo":"databendlabs/databend","slug":"buffer-pool-working-queue-need-unbounded","errorCode":null,"errorMessage":"Buffer pool working queue need unbounded.","messagePattern":"Buffer pool working queue need unbounded\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/query/service/src/spillers/async_buffer.rs","lineNumber":200,"sourceCode":"                    let mut background = Background::create();\n                    while let Ok(op) = working_queue.recv().await {\n                        let span = Span::enter_with_parent(\"Background::recv\", op.span());\n                        background.recv(op).in_span(span).await;\n                    }\n                }),\n            );\n        }\n\n        Ok(Arc::new(SpillsBufferPool {\n            _runtime: runtime,\n            working_queue: working_tx,\n        }))\n    }\n\n    pub(crate) fn operator(&self, op: BufferOperator) {\n        self.working_queue\n            .try_send(op)\n            .expect(\"Buffer pool working queue need unbounded.\");\n    }\n\n    pub fn buffer_write(self: &Arc<Self>, writer: Writer, pool_bytes: usize) -> BufferWriter {\n        let (buffer_tx, buffer_rx) = async_channel::unbounded::<Bytes>();\n        let memory_pool = MemoryPool::create(pool_bytes);\n\n        let response = BufferOperatorResp::pending();\n\n        self.operator(BufferOperator::WriterTask(BufferWriterTaskOperator {\n            writer,\n            buffer_rx,\n            response: response.clone(),\n            memory_pool: memory_pool.clone(),\n            span: Span::enter_with_local_parent(\"BufferWriterTask\"),\n        }));\n\n        BufferWriter {\n            buffer_tx,","sourceCodeStart":182,"sourceCodeEnd":218,"githubUrl":"https://github.com/databendlabs/databend/blob/288d84d76e20a2f8f7173bda9691eb6ece301aa9/src/query/service/src/spillers/async_buffer.rs#L182-L218","documentation":"AsyncBuffer's memory-pool `operator()` submits work to a crossbeam working_queue via try_send and expects the queue to be unbounded, so try_send can never fail with Full/Disconnected. The panic fires when the channel is bounded and full, or has been disconnected (receiver dropped) — a queue configuration or lifecycle violation.","triggerScenarios":"Working queue constructed bounded (capacity set) and filled faster than the worker drains it; worker thread panicked/exited so the channel is disconnected while fetch_ranges still calls operator().","commonSituations":"Spill-heavy workloads flooding the buffer pool queue; memory pressure killing the background worker; misconfigured pool size where the worker can't keep up.","solutions":["Check logs for a prior panic in the buffer-pool worker thread — restart the query/node to recreate the worker.","Verify the working_queue is created with unbounded capacity; fix the constructor if bounded.","Reduce concurrent spill/fetch load or increase pool workers so the queue drains.","Patch operator() to use send().await or map try_send errors instead of expect."],"exampleFix":"// before\nself.working_queue.try_send(op).expect(\"Buffer pool working queue need unbounded.\");\n// after\nif let Err(e) = self.working_queue.try_send(op) {\n    log::error!(\"buffer pool queue rejected op: {e}\");\n}","handlingStrategy":"retry","validationCode":"// verify worker is alive and queue unbounded\nassert!(buffer_worker_is_running(), \"buffer pool worker died; recreate pool\");","typeGuard":"fn queue_accepts(q: &crossbeam::channel::Sender<BufferOperator>) -> bool { !q.is_full() && !q.is_disconnected() }","tryCatchPattern":"match self.working_queue.try_send(op) {\n    Err(crossbeam::channel::TrySendError::Disconnected(_)) => recreate_worker_and_retry(op),\n    Err(e) => log::error!(\"queue send failed: {e}\"),\n    Ok(()) => {}\n}","preventionTips":["Create the working queue unbounded as the name promises","Watch for worker-thread panics in logs","Cap concurrent spill/fetch pressure","Replace try_send().expect() with explicit error propagation"],"tags":["rust","panic","spill","async","queue"],"backgroundTag":"internal-invariant-violation","analyzedSha":"288d84d76e20a2f8f7173bda9691eb6ece301aa9","analyzedAt":"2026-09-11T11:29:36.208Z","contentChangedAt":"2026-09-11T11:29:36.208Z","schemaVersion":2},"datasetVersion":"2026-09-16T04:17:20.429Z"}