From 54463ca664c4617d5b67cd3d8a57fa6686ef962e Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Thu, 6 Jun 2024 17:37:35 +0200
Subject: [PATCH] Add test

---
 cpp/src/arrow/filesystem/s3fs_test.cc | 31 +++++++++++++++++++++++++--
 1 file changed, 29 insertions(+), 2 deletions(-)

diff --git a/cpp/src/arrow/filesystem/s3fs_test.cc b/cpp/src/arrow/filesystem/s3fs_test.cc
index 7bfa120eda678..5a160a78ceea0 100644
--- a/cpp/src/arrow/filesystem/s3fs_test.cc
+++ b/cpp/src/arrow/filesystem/s3fs_test.cc
@@ -614,9 +614,26 @@ class TestS3FS : public S3TestMixin {
     // after CloseAsync or synchronously after stream.reset() since we're just
     // checking that `closeAsyncFut` keeps the stream alive until completion
     // rather than segfaulting on a dangling stream
-    auto closeAsyncFut = stream->CloseAsync();
+    auto close_fut = stream->CloseAsync();
     stream.reset();
-    ASSERT_OK(closeAsyncFut.MoveResult());
+    ASSERT_OK(close_fut.MoveResult());
+    AssertObjectContents(client_.get(), "bucket", "somefile", "new data");
+  }
+
+  void TestOpenOutputStreamCloseAsyncFutureDeadlock() {
+    // This is inspired by GH-41862, though it fails to reproduce the actual
+    // issue reported there (actual preconditions might be required).
+    // Here we lose our reference to an output stream from its CloseAsync callback.
+    // This should not deadlock.
+    std::shared_ptr<io::OutputStream> stream;
+    ASSERT_OK_AND_ASSIGN(stream, fs_->OpenOutputStream("bucket/somefile"));
+    ASSERT_OK(stream->Write("new data"));
+    auto close_fut = stream->CloseAsync();
+    close_fut.AddCallback([stream = std::move(stream)](Status st) mutable {
+      // Trigger stream destruction from callback
+      stream.reset();
+    });
+    ASSERT_OK(close_fut.MoveResult());
     AssertObjectContents(client_.get(), "bucket", "somefile", "new data");
   }
 
@@ -1254,6 +1271,16 @@ TEST_F(TestS3FS, OpenOutputStreamAsyncDestructorSyncWrite) {
   TestOpenOutputStreamCloseAsyncDestructor();
 }
 
+TEST_F(TestS3FS, OpenOutputStreamCloseAsyncFutureDeadlockBackgroundWrites) {
+  TestOpenOutputStreamCloseAsyncFutureDeadlock();
+}
+
+TEST_F(TestS3FS, OpenOutputStreamCloseAsyncFutureDeadlockSyncWrite) {
+  options_.background_writes = false;
+  MakeFileSystem();
+  TestOpenOutputStreamCloseAsyncFutureDeadlock();
+}
+
 TEST_F(TestS3FS, OpenOutputStreamMetadata) {
   std::shared_ptr<io::OutputStream> stream;