fix: memory stomping in CompletionQueue::RunAsync

google/cloud/completion_queue.h

@@ -177,14 +177,17 @@ class CompletionQueue {
             typename std::enable_if<
                 internal::CheckRunAsyncCallback<Functor>::value, int>::type = 0>
   void RunAsync(Functor&& functor) {
+    auto impl = impl_;
     MakeRelativeTimer(std::chrono::seconds(0))
-        .then([this, functor](
-                  future<StatusOr<std::chrono::system_clock::time_point>>) {
-          // We intentionally ignore the status here; the functor is always
-          // called, even after a call to `CancelAll`.
-          CompletionQueue cq(impl_);
-          functor(cq);
-        });
+        .then(
+            [impl, functor](
+                future<
+                    StatusOr<std::chrono::system_clock::time_point>>) mutable {
+              // We intentionally ignore the status here; the functor is always
+              // called, even after a call to `CancelAll`.
+              CompletionQueue cq(impl);
+              functor(cq);
+            });
   }
 
  private:

google/cloud/completion_queue_test.cc

@@ -328,6 +328,22 @@ TEST(CompletionQueueTest, RunAsync) {
   runner.join();
 }
 
+TEST(CompletionQueueTest, RunAsyncCompletionQueueDestroyed) {
+  auto cq_impl = std::make_shared<MockCompletionQueue>();
+
+  std::promise<void> done_promise;
+  {
+    CompletionQueue cq(cq_impl);
+    cq.RunAsync([&done_promise](CompletionQueue& cq) {
+      done_promise.set_value();
+      cq.Shutdown();
+    });
+  }
+  cq_impl->SimulateCompletion(true);
+
+  done_promise.get_future().get();
+}
+
 // Sets up a timer that reschedules itself and verifies we can shut down
 // cleanly whether we call `CancelAll()` on the queue first or not.
 namespace {