From 5f3a6fcf0364ca212f72eb1f2ee38451b772b75e Mon Sep 17 00:00:00 2001 From: Mila Klimkina <256179787+klimkina-eris@users.noreply.github.com> Date: Fri, 10 Jul 2026 16:39:04 -0500 Subject: [PATCH 1/5] fix(pubsub): run lease refresh-timer continuation via RunAsync StartRefreshTimer attached its .then() continuation while mu_ was held. When the timer future was already satisfied at attach time the continuation ran inline on the calling thread, re-entered OnRefreshTimer -> RefreshMessageLeases, and self-deadlocked re-acquiring the non-recursive mu_, freezing the subscription until process restart. Dispatch it through the CompletionQueue (RunAsync) so it always runs on a CQ thread with no locks held, mirroring the Bigtable MutationBatcher fix (#4083 / #4327). --- .../internal/subscription_lease_management.cc | 22 ++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/google/cloud/pubsub/internal/subscription_lease_management.cc b/google/cloud/pubsub/internal/subscription_lease_management.cc index 1d91ab9e722c5..be2647ecca5ad 100644 --- a/google/cloud/pubsub/internal/subscription_lease_management.cc +++ b/google/cloud/pubsub/internal/subscription_lease_management.cc @@ -144,9 +144,25 @@ void SubscriptionLeaseManagement::StartRefreshTimer( shutdown_manager_->StartOperation(__func__, "OnRefreshTimer", [&] { using TimerFuture = future>; if (refresh_timer_.valid()) refresh_timer_.cancel(); - refresh_timer_ = - cq_.MakeDeadlineTimer(deadline).then([weak](TimerFuture tp) { - if (auto self = weak.lock()) self->OnRefreshTimer(!tp.get()); + // The caller holds mu_ (the unnamed unique_lock parameter) for the full + // duration of this function. If the timer future is already satisfied + // when .then() attaches its continuation -- a deadline in the past + // (RefreshMessageLeases can compute extensions as small as 1s, and + // kAckDeadlineSlack then puts the deadline before now), or a CQ thread + // winning the satisfaction race -- the continuation runs INLINE on this + // thread, re-enters OnRefreshTimer -> RefreshMessageLeases, and + // self-deadlocks re-acquiring the non-recursive mu_. Every other path + // (AckMessage, OnRead, Shutdown) then blocks behind mu_ and the + // subscription freezes until the process restarts. Dispatch the + // continuation through the CompletionQueue so OnRefreshTimer always runs + // on a CQ thread with no locks held. + auto cq = cq_; + refresh_timer_ = cq_.MakeDeadlineTimer(deadline).then( + [weak, cq](TimerFuture tp) mutable { + auto const cancelled = !tp.get(); + cq.RunAsync([weak, cancelled] { + if (auto self = weak.lock()) self->OnRefreshTimer(cancelled); + }); }); }); } From f1fe8dec0a063df1bb00f02fe77c6c8543591be3 Mon Sep 17 00:00:00 2001 From: Mila Klimkina <256179787+klimkina-eris@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:35:03 -0500 Subject: [PATCH 2/5] test(pubsub): cover lease refresh-timer RunAsync dispatch Regression test for the SubscriptionLeaseManagement self-deadlock: verifies the refresh-timer continuation is dispatched via CompletionQueue::RunAsync and not run inline in the timer callback (which re-entered the held mutex). --- .../subscription_lease_management_test.cc | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/google/cloud/pubsub/internal/subscription_lease_management_test.cc b/google/cloud/pubsub/internal/subscription_lease_management_test.cc index c1e28b105dc23..75d1f82215a73 100644 --- a/google/cloud/pubsub/internal/subscription_lease_management_test.cc +++ b/google/cloud/pubsub/internal/subscription_lease_management_test.cc @@ -280,6 +280,73 @@ TEST(SubscriptionLeaseManagementTest, ExpiredMessage) { EXPECT_THAT(done.get(), IsOk()); } +/// @test Regression test for the self-deadlock in StartRefreshTimer: the lease +/// refresh-timer continuation must be dispatched via `CompletionQueue::RunAsync` +/// rather than run inline in the timer callback. Running it inline re-entered +/// the already-held `mu_` and self-deadlocked the subscription. Here we verify +/// that firing the timer only *schedules* the refresh (a RunAsync task); the +/// lease extension runs on a subsequent completion, never synchronously inside +/// the timer callback. +TEST(SubscriptionLeaseManagementTest, + RefreshTimerContinuationDispatchedViaRunAsync) { + auto mock = std::make_shared(); + std::shared_ptr batch_callback; + EXPECT_CALL(*mock, Start).WillOnce([&](std::shared_ptr cb) { + batch_callback = std::move(cb); + }); + + auto mock_batch_callback = + std::make_shared(); + EXPECT_CALL(*mock_batch_callback, callback).Times(1); + + int extend_calls = 0; + EXPECT_CALL(*mock, ExtendLeases) + .WillRepeatedly([&](std::vector const&, + std::chrono::seconds) { + ++extend_calls; + return make_ready_future(Status{}); + }); + EXPECT_CALL(*mock, BulkNack).WillRepeatedly([](std::vector + const&) { + return make_ready_future(Status{}); + }); + EXPECT_CALL(*mock, Shutdown).Times(1); + + auto fake_cq = std::make_shared(); + CompletionQueue cq(fake_cq); + + auto shutdown_manager = std::make_shared(); + auto uut = SubscriptionLeaseManagement::Create( + cq, shutdown_manager, mock, std::chrono::seconds(345), + std::chrono::seconds(600)); + + auto done = shutdown_manager->Start({}); + uut->Start(mock_batch_callback); + batch_callback->callback( + BatchCallback::StreamingPullResponse{GenerateMessages("0-", 1)}); + ASSERT_EQ(1U, fake_cq->size()); // the refresh timer is pending + + // Fire the timer. With the fix, the timer callback only *schedules* + // OnRefreshTimer via RunAsync; it must NOT extend leases inline (doing so + // re-enters the held mutex and deadlocks). + fake_cq->SimulateCompletion(true); + EXPECT_EQ(0, extend_calls) + << "lease refresh ran inline in the timer callback (deadlock path)"; + EXPECT_EQ(1U, fake_cq->size()); // a RunAsync task is now pending + + // Draining the RunAsync task runs OnRefreshTimer -> ExtendLeases. + fake_cq->SimulateCompletion(true); + EXPECT_EQ(1, extend_calls); + + // Drain any remaining scheduled work so shutdown can complete. + shutdown_manager->MarkAsShutdown(__func__, Status{}); + uut->Shutdown(); + for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + fake_cq->SimulateCompletion(false); + } + EXPECT_THAT(done.get(), IsOk()); +} + } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace pubsub_internal From 87947083127dc7b0aeb27f1946f5b62db2b23b3c Mon Sep 17 00:00:00 2001 From: Mila Klimkina <256179787+klimkina-eris@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:35:34 -0500 Subject: [PATCH 3/5] test(pubsub): wrap doc comment to 80 columns --- .../internal/subscription_lease_management_test.cc | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/google/cloud/pubsub/internal/subscription_lease_management_test.cc b/google/cloud/pubsub/internal/subscription_lease_management_test.cc index 75d1f82215a73..74fd0fc143cc7 100644 --- a/google/cloud/pubsub/internal/subscription_lease_management_test.cc +++ b/google/cloud/pubsub/internal/subscription_lease_management_test.cc @@ -281,12 +281,11 @@ TEST(SubscriptionLeaseManagementTest, ExpiredMessage) { } /// @test Regression test for the self-deadlock in StartRefreshTimer: the lease -/// refresh-timer continuation must be dispatched via `CompletionQueue::RunAsync` -/// rather than run inline in the timer callback. Running it inline re-entered -/// the already-held `mu_` and self-deadlocked the subscription. Here we verify -/// that firing the timer only *schedules* the refresh (a RunAsync task); the -/// lease extension runs on a subsequent completion, never synchronously inside -/// the timer callback. +/// refresh-timer continuation must be dispatched via `RunAsync` rather than run +/// inline in the timer callback. Running it inline re-entered the already-held +/// `mu_` and self-deadlocked the subscription. Here we verify that firing the +/// timer only *schedules* the refresh (a RunAsync task); the lease extension +/// runs on a subsequent completion, never synchronously in the timer callback. TEST(SubscriptionLeaseManagementTest, RefreshTimerContinuationDispatchedViaRunAsync) { auto mock = std::make_shared(); From 82827a0ca26df37b0db56aea26e183cc75ab9a49 Mon Sep 17 00:00:00 2001 From: Mila Klimkina <256179787+klimkina-eris@users.noreply.github.com> Date: Fri, 10 Jul 2026 18:05:17 -0500 Subject: [PATCH 4/5] test(pubsub): drain deferred OnRefreshTimer (RunAsync) in lease tests The fix dispatches the refresh-timer continuation via CompletionQueue::RunAsync, which FakeCompletionQueueImpl only executes on SimulateCompletion(true) (false drops it). Add a drain after each timer fire and switch teardown to a true-drain so the deferred OnRefreshTimer runs, --outstanding fires, and shutdown completes. Verified green locally via bazel test. --- .../subscription_lease_management_test.cc | 30 ++++++++++++------- 1 file changed, 20 insertions(+), 10 deletions(-) diff --git a/google/cloud/pubsub/internal/subscription_lease_management_test.cc b/google/cloud/pubsub/internal/subscription_lease_management_test.cc index 74fd0fc143cc7..53a744c359a7a 100644 --- a/google/cloud/pubsub/internal/subscription_lease_management_test.cc +++ b/google/cloud/pubsub/internal/subscription_lease_management_test.cc @@ -109,18 +109,21 @@ TEST(SubscriptionLeaseManagementTest, NormalLifecycle) { // will verify that only the remaining messages have their lease extended. uut->AckMessage("ack-0-1"); fake_cq->SimulateCompletion(true); + fake_cq->SimulateCompletion(true); // drain the deferred OnRefreshTimer ASSERT_EQ(1U, fake_cq->size()); // Ack one more message and trigger the new timer. uut->NackMessage("ack-0-2"); fake_cq->SimulateCompletion(true); + fake_cq->SimulateCompletion(true); // drain the deferred OnRefreshTimer ASSERT_EQ(1U, fake_cq->size()); shutdown_manager->MarkAsShutdown(__func__, Status{}); uut->Shutdown(); - fake_cq->SimulateCompletion(false); - ASSERT_EQ(0U, fake_cq->size()); + for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + fake_cq->SimulateCompletion(true); + } EXPECT_THAT(done.get(), IsOk()); } @@ -154,8 +157,9 @@ TEST(SubscriptionLeaseManagementTest, ShutdownOnError) { Status(StatusCode::kPermissionDenied, "uh-oh"))}); ASSERT_EQ(1U, fake_cq->size()); - fake_cq->SimulateCompletion(false); - ASSERT_EQ(0U, fake_cq->size()); + for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + fake_cq->SimulateCompletion(true); + } EXPECT_THAT(done.get(), StatusIs(StatusCode::kPermissionDenied)); } @@ -219,12 +223,14 @@ TEST(SubscriptionLeaseManagementTest, UsesDeadlineExtension) { // Ignore message and then fire the timer. This will extend the deadline. fake_cq->SimulateCompletion(true); + fake_cq->SimulateCompletion(true); // drain the deferred OnRefreshTimer ASSERT_EQ(1U, fake_cq->size()); shutdown_manager->MarkAsShutdown(__func__, Status{}); uut->Shutdown(); - fake_cq->SimulateCompletion(false); - ASSERT_EQ(0U, fake_cq->size()); + for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + fake_cq->SimulateCompletion(true); + } EXPECT_THAT(done.get(), IsOk()); } @@ -270,13 +276,15 @@ TEST(SubscriptionLeaseManagementTest, ExpiredMessage) { // will verify that only the remaining messages have their lease extended. uut->AckMessage("ack-0-1"); fake_cq->SimulateCompletion(true); + fake_cq->SimulateCompletion(true); // drain the deferred OnRefreshTimer ASSERT_EQ(1U, fake_cq->size()); shutdown_manager->MarkAsShutdown(__func__, Status{}); uut->Shutdown(); - fake_cq->SimulateCompletion(false); - ASSERT_EQ(0U, fake_cq->size()); + for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + fake_cq->SimulateCompletion(true); + } EXPECT_THAT(done.get(), IsOk()); } @@ -337,11 +345,13 @@ TEST(SubscriptionLeaseManagementTest, fake_cq->SimulateCompletion(true); EXPECT_EQ(1, extend_calls); - // Drain any remaining scheduled work so shutdown can complete. + // The fix defers OnRefreshTimer to RunAsync, which the fake CQ only executes + // on SimulateCompletion(true) (false drops the task). Drain with true so the + // deferred refresh operation finishes and the session shutdown completes. shutdown_manager->MarkAsShutdown(__func__, Status{}); uut->Shutdown(); for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { - fake_cq->SimulateCompletion(false); + fake_cq->SimulateCompletion(true); } EXPECT_THAT(done.get(), IsOk()); } From ccce4f19209b75345741b0770efeb2736efeb3a9 Mon Sep 17 00:00:00 2001 From: Mila Klimkina <256179787+klimkina-eris@users.noreply.github.com> Date: Fri, 10 Jul 2026 18:35:02 -0500 Subject: [PATCH 5/5] test(pubsub): annotate deferred-RunAsync drains (mirror #4327) Use the same own-line `// RunAsync` marker convention as the bigtable MutationBatcher deadlock fix (#4327) so the deferred-continuation drains are immediately recognizable as the same idiom. --- .../subscription_lease_management_test.cc | 36 ++++++++++--------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/google/cloud/pubsub/internal/subscription_lease_management_test.cc b/google/cloud/pubsub/internal/subscription_lease_management_test.cc index 53a744c359a7a..92015f91bf24d 100644 --- a/google/cloud/pubsub/internal/subscription_lease_management_test.cc +++ b/google/cloud/pubsub/internal/subscription_lease_management_test.cc @@ -109,19 +109,21 @@ TEST(SubscriptionLeaseManagementTest, NormalLifecycle) { // will verify that only the remaining messages have their lease extended. uut->AckMessage("ack-0-1"); fake_cq->SimulateCompletion(true); - fake_cq->SimulateCompletion(true); // drain the deferred OnRefreshTimer + // RunAsync, drain the deferred OnRefreshTimer + fake_cq->SimulateCompletion(true); ASSERT_EQ(1U, fake_cq->size()); // Ack one more message and trigger the new timer. uut->NackMessage("ack-0-2"); fake_cq->SimulateCompletion(true); - fake_cq->SimulateCompletion(true); // drain the deferred OnRefreshTimer + // RunAsync, drain the deferred OnRefreshTimer + fake_cq->SimulateCompletion(true); ASSERT_EQ(1U, fake_cq->size()); shutdown_manager->MarkAsShutdown(__func__, Status{}); uut->Shutdown(); - for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + for (int i = 0; i != 16 && !fake_cq->empty(); ++i) { fake_cq->SimulateCompletion(true); } EXPECT_THAT(done.get(), IsOk()); @@ -157,7 +159,7 @@ TEST(SubscriptionLeaseManagementTest, ShutdownOnError) { Status(StatusCode::kPermissionDenied, "uh-oh"))}); ASSERT_EQ(1U, fake_cq->size()); - for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + for (int i = 0; i != 16 && !fake_cq->empty(); ++i) { fake_cq->SimulateCompletion(true); } EXPECT_THAT(done.get(), StatusIs(StatusCode::kPermissionDenied)); @@ -223,12 +225,13 @@ TEST(SubscriptionLeaseManagementTest, UsesDeadlineExtension) { // Ignore message and then fire the timer. This will extend the deadline. fake_cq->SimulateCompletion(true); - fake_cq->SimulateCompletion(true); // drain the deferred OnRefreshTimer + // RunAsync, drain the deferred OnRefreshTimer + fake_cq->SimulateCompletion(true); ASSERT_EQ(1U, fake_cq->size()); shutdown_manager->MarkAsShutdown(__func__, Status{}); uut->Shutdown(); - for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + for (int i = 0; i != 16 && !fake_cq->empty(); ++i) { fake_cq->SimulateCompletion(true); } EXPECT_THAT(done.get(), IsOk()); @@ -276,13 +279,14 @@ TEST(SubscriptionLeaseManagementTest, ExpiredMessage) { // will verify that only the remaining messages have their lease extended. uut->AckMessage("ack-0-1"); fake_cq->SimulateCompletion(true); - fake_cq->SimulateCompletion(true); // drain the deferred OnRefreshTimer + // RunAsync, drain the deferred OnRefreshTimer + fake_cq->SimulateCompletion(true); ASSERT_EQ(1U, fake_cq->size()); shutdown_manager->MarkAsShutdown(__func__, Status{}); uut->Shutdown(); - for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + for (int i = 0; i != 16 && !fake_cq->empty(); ++i) { fake_cq->SimulateCompletion(true); } EXPECT_THAT(done.get(), IsOk()); @@ -308,15 +312,15 @@ TEST(SubscriptionLeaseManagementTest, int extend_calls = 0; EXPECT_CALL(*mock, ExtendLeases) - .WillRepeatedly([&](std::vector const&, - std::chrono::seconds) { - ++extend_calls; + .WillRepeatedly( + [&](std::vector const&, std::chrono::seconds) { + ++extend_calls; + return make_ready_future(Status{}); + }); + EXPECT_CALL(*mock, BulkNack) + .WillRepeatedly([](std::vector const&) { return make_ready_future(Status{}); }); - EXPECT_CALL(*mock, BulkNack).WillRepeatedly([](std::vector - const&) { - return make_ready_future(Status{}); - }); EXPECT_CALL(*mock, Shutdown).Times(1); auto fake_cq = std::make_shared(); @@ -350,7 +354,7 @@ TEST(SubscriptionLeaseManagementTest, // deferred refresh operation finishes and the session shutdown completes. shutdown_manager->MarkAsShutdown(__func__, Status{}); uut->Shutdown(); - for (int i = 0; i != 16 && fake_cq->size() != 0; ++i) { + for (int i = 0; i != 16 && !fake_cq->empty(); ++i) { fake_cq->SimulateCompletion(true); } EXPECT_THAT(done.get(), IsOk());