Crash during shutdown

[ianton-ru]

Node failed during shutdown

E           Exception: Sanitizer assert found for instance =================================================================
E           ==9==ERROR: AddressSanitizer: heap-use-after-free on address 0x5040020a6df0 at pc 0x55a785ef9736 bp 0x7f74b18c1eb0 sp 0x7f74b18c1ea8
E           READ of size 8 at 0x5040020a6df0 thread T773 (QueryPipelineEx)
E               #0 0x55a785ef9735 in DB::StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica(unsigned long) ci/tmp/build/./src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp:312:28
E               #1 0x55a78776d15c in DB::RemoteQueryExecutor::processPacket(DB::Packet) ci/tmp/build/./src/QueryPipeline/RemoteQueryExecutor.cpp:739:47
E               #2 0x55a78776f914 in DB::RemoteQueryExecutor::readAsync() ci/tmp/build/./src/QueryPipeline/RemoteQueryExecutor.cpp:595:28
E               #3 0x55a79631eeab in DB::RemoteSource::tryGenerate() ci/tmp/build/./src/Processors/Sources/RemoteSource.cpp:189:36
E               #4 0x55a79599b85b in DB::ISource::work() ci/tmp/build/./src/Processors/ISource.cpp:110:26
E               #5 0x55a7959e21b3 in DB::executeJob(DB::ExecutingGraph::Node*, DB::ReadProgressCallback*) ci/tmp/build/./src/Processors/Executors/ExecutionThreadContext.cpp:53:26
E               #6 0x55a7959e21b3 in DB::ExecutionThreadContext::executeTask() ci/tmp/build/./src/Processors/Executors/ExecutionThreadContext.cpp:102:9
E               #7 0x55a7959bd19b in DB::PipelineExecutor::executeStepImpl(unsigned long, DB::IAcquiredSlot*, std::__1::atomic<bool>*) ci/tmp/build/./src/Processors/Executors/PipelineExecutor.cpp:351:26
E               #8 0x55a7959c0c43 in DB::PipelineExecutor::executeSingleThread(unsigned long, DB::IAcquiredSlot*) ci/tmp/build/./src/Processors/Executors/PipelineExecutor.cpp:279:5
E               #9 0x55a7959c0c43 in DB::PipelineExecutor::spawnThreads(std::__1::shared_ptr<DB::IAcquiredSlot>)::$_0::operator()() const ci/tmp/build/./src/Processors/Executors/PipelineExecutor.cpp:565:17
E               #10 0x55a7959c0c43 in decltype(std::declval<DB::PipelineExecutor::spawnThreads(std::__1::shared_ptr<DB::IAcquiredSlot>)::$_0&>()()) std::__1::__invoke[abi:ne190107]<DB::PipelineExecutor::spawnThreads(std::__1::shared_ptr<DB::IAcquiredSlot>)::$_0&>(DB::PipelineExecutor::spawnThreads(std::__1::shared_ptr<DB::IAcquiredSlot>)::$_0&) ci/tmp/build/./contrib/llvm-project/libcxx/include/__type_traits/invoke.h:149:25
E               #11 0x55a7959c0c43 in void std::__1::__invoke_void_return_wrapper<void, true>::__call[abi:ne190107]<DB::PipelineExecutor::spawnThreads(std::__1::shared_ptr<DB::IAcquiredSlot>)::$_0&>(DB::PipelineExecutor::spawnThreads(std::__1::shared_ptr<DB::IAcquiredSlot>)::$_0&) ci/tmp/build/./contrib/llvm-project/libcxx/include/__type_traits/invoke.h:224:5
E               #12 0x55a7959c0c43 in std::__1::__function::__default_alloc_func<DB::PipelineExecutor::spawnThreads(std::__1::shared_ptr<DB::IAcquiredSlot>)::$_0, void ()>::operator()[abi:ne190107]() ci/tmp/build/./contrib/llvm-project/libcxx/include/__functional/function.h:210:12
E               #13 0x55a7959c0c43 in void std::__1::__function::__policy_invoker<void ()>::__call_impl[abi:ne190107]<std::__1::__function::__default_alloc_func<DB::PipelineExecutor::spawnThreads(std::__1::shared_ptr<DB::IAcquiredSlot>)::$_0, void ()>>(std::__1::__function::__policy_storage const*) ci/tmp/build/./contrib/llvm-project/libcxx/include/__functional/function.h:610:12
E               #14 0x55a77a62f8d2 in std::__1::__function::__policy_func<void ()>::operator()[abi:ne190107]() const ci/tmp/build/./contrib/llvm-project/libcxx/include/__functional/function.h:716:12
E               #15 0x55a77a62f8d2 in std::__1::function<void ()>::operator()() const ci/tmp/build/./contrib/llvm-project/libcxx/include/__functional/function.h:989:10
E               #16 0x55a77a62f8d2 in ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::worker() ci/tmp/build/./src/Common/ThreadPool.cpp:802:17
E               #17 0x55a77a63df62 in decltype(*std::declval<ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*&>().*std::declval<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&)()>()()) std::__1::__invoke[abi:ne190107]<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*&, void>(void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*&) ci/tmp/build/./contrib/llvm-project/libcxx/include/__type_traits/invoke.h:117:25
E               #18 0x55a77a63df62 in decltype(auto) std::__1::__apply_tuple_impl[abi:ne190107]<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&)(), std::__1::tuple<ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>&, 0ul>(void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&)(), std::__1::tuple<ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>&, std::__1::__tuple_indices<0ul>) ci/tmp/build/./contrib/llvm-project/libcxx/include/tuple:1354:5
E               #19 0x55a77a63df62 in decltype(auto) std::__1::apply[abi:ne190107]<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&)(), std::__1::tuple<ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>&>(void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&)(), std::__1::tuple<ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>&) ci/tmp/build/./contrib/llvm-project/libcxx/include/tuple:1358:5
E               #20 0x55a77a63df62 in ThreadFromGlobalPoolImpl<false, true>::ThreadFromGlobalPoolImpl<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>(void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&&)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*&&)::'lambda'()::operator()() ci/tmp/build/./src/Common/ThreadPool.h:312:13
E               #21 0x55a77a629e2d in std::__1::__function::__policy_func<void ()>::operator()[abi:ne190107]() const ci/tmp/build/./contrib/llvm-project/libcxx/include/__functional/function.h:716:12
E               #22 0x55a77a629e2d in std::__1::function<void ()>::operator()() const ci/tmp/build/./contrib/llvm-project/libcxx/include/__functional/function.h:989:10
E               #23 0x55a77a629e2d in ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool::worker() ci/tmp/build/./src/Common/ThreadPool.cpp:812:17
E               #24 0x55a77a6389cf in decltype(*std::declval<ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool*>().*std::declval<void (ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool::*)()>()()) std::__1::__invoke[abi:ne190107]<void (ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool::*)(), ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool*, void>(void (ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool::*&&)(), ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool*&&) ci/tmp/build/./contrib/llvm-project/libcxx/include/__type_traits/invoke.h:117:25
E               #25 0x55a77a6389cf in void std::__1::__thread_execute[abi:ne190107]<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::__1::__thread_struct>>, void (ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool::*)(), ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool*, 2ul>(std::__1::tuple<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::__1::__thread_struct>>, void (ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool::*)(), ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool*>&, std::__1::__tuple_indices<2ul>) ci/tmp/build/./contrib/llvm-project/libcxx/include/__thread/thread.h:192:3
E               #26 0x55a77a6389cf in void* std::__1::__thread_proxy[abi:ne190107]<std::__1::tuple<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::__1::__thread_struct>>, void (ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool::*)(), ThreadPoolImpl<std::__1::thread>::ThreadFromThreadPool*>>(void*) ci/tmp/build/./contrib/llvm-project/libcxx/include/__thread/thread.h:201:3
E               #27 0x55a767a96e66 in asan_thread_start(void*) crtstuff.c
E               #28 0x7f7a14307ac2  (/lib/x86_64-linux-gnu/libc.so.6+0x94ac2) (BuildId: d5197096f709801829b118af1b7cf6631efa2dcd)
E               #29 0x7f7a1439984f  (/lib/x86_64-linux-gnu/libc.so.6+0x12684f) (BuildId: d5197096f709801829b118af1b7cf6631efa2dcd)

[alsugiliazova]

Fix: Use-after-free in StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica

Summary

Fixed a heap-use-after-free in StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica that caused AddressSanitizer failures during node shutdown when replicas disconnect.

Root Cause

Iterator invalidation. The code erased an element from replica_to_files_to_be_processed and then iterated over the erased element's data using a now-invalidated iterator.

// BEFORE (buggy)
replica_to_files_to_be_processed.erase(number_of_current_replica);  // invalidates processed_file_list_ptr
for (const auto & file : processed_file_list_ptr->second)          // use-after-free

processed_file_list_ptr pointed to the element being erased. After erase(), the iterator is invalid and dereferencing it is undefined behavior.

When It Triggers

Triggered when a replica is marked as lost (e.g. ConnectionLost packet):

1. RemoteQueryExecutor::processPacket receives Protocol::Server::ConnectionLost
2. Calls extension->task_iterator->rescheduleTasksFromReplica(extension->replica_info->number_of_current_replica)
3. rescheduleTasksFromReplica runs the buggy path

During shutdown, replicas disconnect and this path runs more often, so the bug is more likely to appear.

Fix

Copy or move the list before erasing so the iterator is not used after invalidation:

// AFTER (fixed)
auto files_to_reschedule = std::move(processed_file_list_ptr->second);
replica_to_files_to_be_processed.erase(number_of_current_replica);
for (const auto & file : files_to_reschedule)
{
    // ...
}

Files Changed

• src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp

ASan Stack Trace (Reference)

AddressSanitizer: heap-use-after-free on address 0x5040020a6df0
READ of size 8 at 0x5040020a6df0 thread T773 (QueryPipelineEx)
    #0 StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica(unsigned long) ... StorageObjectStorageStableTaskDistributor.cpp:312
    #1 RemoteQueryExecutor::processPacket(DB::Packet) ... RemoteQueryExecutor.cpp:739
    #2 RemoteQueryExecutor::readAsync() ... RemoteQueryExecutor.cpp:595
    ...