[BugFix] Unschedule global RF timer in PipelineDriver destructor (#76252)

Signed-off-by: Jenwit Amonpongitsara <jenwit.amonpongitsara@agoda.com>

Author: jenwitteng

be/src/exec/runtime/pipeline_driver.cpp

@@ -102,6 +102,10 @@ PipelineDriver::PipelineDriver()
           _driver_id(0) {}
 
 PipelineDriver::~PipelineDriver() noexcept {
+    // Queued or blocked drivers abandoned when the driver executor is closed during shutdown are
+    // destroyed without going through finalize(), so unschedule the global runtime-filter timer here
+    // as well; otherwise the timer thread may run a task whose owning shared_ptr is already gone.
+    _unschedule_global_rf_timer();
     if (_workgroup != nullptr) {
         _workgroup->decr_num_running_drivers();
     }
@@ -884,11 +888,7 @@ void PipelineDriver::finalize(RuntimeState* runtime_state, DriverState state) {
 
     _update_driver_level_timer();
 
-    if (_global_rf_timer != nullptr) {
-        if (_pipeline_timer_context != nullptr) {
-            _pipeline_timer_context->unschedule_and_join(_global_rf_timer.get());
-        }
-    }
+    _unschedule_global_rf_timer();
 
     if (_driver_observer != nullptr) {
         _driver_observer->on_driver_finished(runtime_state);
@@ -939,6 +939,16 @@ void PipelineDriver::_update_global_rf_timer() {
     WARN_IF_ERROR(_pipeline_timer_context->schedule(_global_rf_timer.get(), abstime), "schedule:");
 }
 
+void PipelineDriver::_unschedule_global_rf_timer() noexcept {
+    // Move the task out first so it stays alive while we synchronize with a callback that may already
+    // be running on the timer thread but has not yet taken its shared_from_this() reference. Otherwise
+    // doRun() could observe a zero use_count and throw std::bad_weak_ptr.
+    auto task = std::move(_global_rf_timer);
+    if (task != nullptr && _pipeline_timer_context != nullptr) {
+        _pipeline_timer_context->unschedule_and_join(task.get());
+    }
+}
+
 std::string PipelineDriver::_build_readable_string(bool use_raw_name) const {
     std::stringstream ss;
     std::string block_reasons = "";

be/src/exec/runtime/pipeline_driver.h

@@ -446,6 +446,7 @@ class PipelineDriver {
 
     // used in event scheduler
     void _update_global_rf_timer();
+    void _unschedule_global_rf_timer() noexcept;
 
     // Helper function to build readable string with option to use raw operator names
     std::string _build_readable_string(bool use_raw_name) const;

be/test/exec/pipeline/schedule/pipeline_timer_test.cpp

@@ -28,6 +28,7 @@
 #include "butil/time.h"
 #include "common/brpc/brpc_stub_cache.h"
 #include "compute_env/pipeline/pipeline_timer_context.h"
+#include "exec/runtime/pipeline_driver.h"
 #include "exec_primitive/pipeline/primitives/pipeline_observer.h"
 #include "gtest/gtest.h"
 #include "runtime/runtime_state.h"
@@ -73,6 +74,18 @@ class LightProbe final : public LightTimerTask {
     std::atomic<bool> ran{false};
 };
 
+// Reaches PipelineDriver's protected default constructor and its protected global-RF-timer members
+// so a test can drive the destructor cleanup path without standing up a full fragment.
+class TimerTestPipelineDriver final : public PipelineDriver {
+public:
+    TimerTestPipelineDriver() = default;
+
+    void register_global_rf_timer(PipelineTimerContextPtr context, std::shared_ptr<PipelineTimerTask> task) {
+        _pipeline_timer_context = std::move(context);
+        _global_rf_timer = std::move(task);
+    }
+};
+
 class CountingObserver final : public PipelineObserver {
 public:
     void source_trigger() override {
@@ -303,6 +316,52 @@ TEST_F(PipelineTimerTaskTest, pipeline_timer_context_submits_batched_rf_timeout_
     context.clear_rf_timeout_tasks();
 }
 
+// A queued or blocked driver abandoned when the driver executor is closed during shutdown is
+// destroyed without going through finalize(). The destructor must still unschedule the global
+// runtime-filter timer, otherwise the timer thread runs a task whose owning shared_ptr is gone and
+// doRun()'s shared_from_this() throws std::bad_weak_ptr.
+TEST_F(PipelineTimerTaskTest, pipeline_driver_destructor_unschedules_global_rf_timer) {
+    auto context = std::make_shared<PipelineTimerContext>(&timer);
+    auto task = std::make_shared<ProbeTask>();
+    ASSERT_OK(context->schedule(task.get(), future_abstime(3600)));
+    ASSERT_NE(0, task->tid());
+
+    auto driver = std::make_unique<TimerTestPipelineDriver>();
+    driver->register_global_rf_timer(context, task);
+
+    // Intentionally skip finalize().
+    driver.reset();
+
+    // The destructor must already have removed the task; a second unschedule finds nothing.
+    EXPECT_EQ(-1, timer.unschedule(task.get()));
+}
+
+// The destructor must block until an already-running global RF timer task returns, mirroring
+// finalize()'s unschedule_and_join, so the task is never freed while doRun() is still executing.
+TEST_F(PipelineTimerTaskTest, pipeline_driver_destructor_waits_for_running_global_rf_timer) {
+    auto context = std::make_shared<PipelineTimerContext>(&timer);
+    auto task = std::make_shared<ProbeTask>();
+    task->block_until_released = true;
+    ASSERT_OK(context->schedule(task.get(), past_abstime()));
+    task->run_enter.acquire();
+
+    auto driver = std::make_unique<TimerTestPipelineDriver>();
+    driver->register_global_rf_timer(context, task);
+
+    std::atomic<bool> destructor_returned{false};
+    std::thread destructor([driver = std::move(driver), &destructor_returned]() mutable {
+        driver.reset();
+        destructor_returned.store(true, std::memory_order_release);
+    });
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    EXPECT_FALSE(destructor_returned.load(std::memory_order_acquire));
+
+    task->run_gate.release();
+    destructor.join();
+    EXPECT_TRUE(destructor_returned.load(std::memory_order_acquire));
+}
+
 // LightTimerTask goes through a separate schedule/unschedule pair and lacks the
 // _finished / _has_consumer protocol. Cover it to prevent regressions in the
 // other overload of PipelineTimer.