Make Abort() interrupt the libcurl and NSURLSession transports

Source/UrlRequest_Apple.mm

@@ -228,6 +228,23 @@ void Open(UrlMethod method, const std::string& url)
             NSURLSessionDataTask* task{[session dataTaskWithRequest:request completionHandler:completionHandler]};
             [task resume];
 
+            // Observe Abort(): NSURLSession runs the request asynchronously and does not watch
+            // m_cancellationSource on its own. Cancelling the task makes its completion handler fire
+            // with NSURLErrorCancelled (recorded as the transport error). The task is captured
+            // *weakly* so the listener does not keep a finished task alive -- NSURLSession releases
+            // the task once its completion handler has run, after which a late Abort() loads a nil
+            // strong reference and the -cancel is a safe no-op. The listener fires synchronously if
+            // the request was already aborted; the ticket is reset on each send and released before
+            // m_cancellationSource (a base member) is destroyed. emplace() (not assignment) is used
+            // because arcana::cancellation::ticket is a move-only final_action whose assignment
+            // operators are deleted, so std::optional::operator= would not compile; emplace destroys
+            // any prior ticket (releasing the previous send's listener) and move-constructs the new one.
+            __weak NSURLSessionDataTask* weakTask = task;
+            m_cancellationTicket.emplace(m_cancellationSource.add_listener([weakTask]() {
+                NSURLSessionDataTask* strongTask = weakTask;
+                [strongTask cancel];
+            }));
+
             return taskCompletionSource.as_task();
         }
 
@@ -244,6 +261,7 @@ void Open(UrlMethod method, const std::string& url)
     private:
         NSURL* m_url{};
         NSData* m_responseBuffer{};
+        std::optional<arcana::cancellation::ticket> m_cancellationTicket{};
     };
 }
 

Source/UrlRequest_Unix.cpp

@@ -53,6 +53,7 @@ namespace
             case CURLE_BAD_CONTENT_ENCODING: return "CURLE_BAD_CONTENT_ENCODING";
             case CURLE_SSL_CACERT_BADFILE: return "CURLE_SSL_CACERT_BADFILE";
             case CURLE_REMOTE_FILE_NOT_FOUND: return "CURLE_REMOTE_FILE_NOT_FOUND";
+            case CURLE_ABORTED_BY_CALLBACK: return "CURLE_ABORTED_BY_CALLBACK";
             default: return "CURLE_" + std::to_string(static_cast<int>(code));
         }
     }
@@ -146,7 +147,7 @@ namespace UrlLib
                 curl_check(curl_easy_setopt(m_curl, CURLOPT_HEADERFUNCTION, HeaderCallback));
                 curl_check(curl_easy_setopt(m_curl, CURLOPT_FOLLOWLOCATION, 1L));
                 // Request-specific failure detail (host/port/path specifics) lands here during
-                // curl_easy_perform; see the error handling in PerformAsync.
+                // the transfer; see the error handling in PerformAsync.
                 curl_check(curl_easy_setopt(m_curl, CURLOPT_ERRORBUFFER, m_curlErrorBuffer.data()));
             }
         }
@@ -195,7 +196,6 @@ namespace UrlLib
             }
         }
 
-
         static void Append(std::string& string, char* buffer, size_t nitems)
         {
             string.insert(string.end(), buffer, buffer + nitems);
@@ -207,6 +207,65 @@ namespace UrlLib
             byteVector.insert(byteVector.end(), bytes, bytes + nitems);   
         }
 
+        // Drives the transfer through the libcurl *multi* interface rather than curl_easy_perform so
+        // that Abort() is observed promptly. curl_easy_perform blocks in an internal poll that, when
+        // a peer accepts the connection but sends nothing, can wait indefinitely without invoking any
+        // callback -- so a cancelled request would hang until the peer or OS gave up. Polling with a
+        // bounded timeout lets the loop re-check m_cancellationSource between waits, bounding abort
+        // latency to ~kPollTimeoutMs regardless of peer activity. Runs on the worker thread.
+        CURLcode PerformWithCancellation()
+        {
+            CURLM* multi = curl_multi_init();
+            if (multi == nullptr)
+            {
+                return CURLE_OUT_OF_MEMORY;
+            }
+            auto multiScope = gsl::finally([this, multi] {
+                curl_multi_remove_handle(multi, m_curl);
+                curl_multi_cleanup(multi);
+            });
+
+            if (curl_multi_add_handle(multi, m_curl) != CURLM_OK)
+            {
+                return CURLE_FAILED_INIT;
+            }
+
+            constexpr int kPollTimeoutMs = 100;
+            int runningHandles = 0;
+            do
+            {
+                if (m_cancellationSource.cancelled())
+                {
+                    return CURLE_ABORTED_BY_CALLBACK;
+                }
+
+                if (curl_multi_perform(multi, &runningHandles) != CURLM_OK)
+                {
+                    return CURLE_RECV_ERROR;
+                }
+
+                // Wait for socket activity but wake at least every kPollTimeoutMs so the cancellation
+                // check above runs even when the peer is idle (curl_multi_poll waits the full timeout
+                // even when there are no fds, so this never busy-loops during DNS resolution).
+                if (runningHandles != 0 && curl_multi_poll(multi, nullptr, 0, kPollTimeoutMs, nullptr) != CURLM_OK)
+                {
+                    return CURLE_RECV_ERROR;
+                }
+            } while (runningHandles != 0);
+
+            // The transfer finished; surface the per-easy-handle result code.
+            CURLcode result = CURLE_OK;
+            int messagesInQueue = 0;
+            while (CURLMsg* message = curl_multi_info_read(multi, &messagesInQueue))
+            {
+                if (message->msg == CURLMSG_DONE && message->easy_handle == m_curl)
+                {
+                    result = message->data.result;
+                }
+            }
+            return result;
+        }
+
         template<typename DataT>
         arcana::task<void, std::exception_ptr> PerformAsync(DataT& data)
         {
@@ -226,7 +285,7 @@ namespace UrlLib
             m_thread.emplace([this, taskCompletionSource]() mutable
             {
                 m_curlErrorBuffer[0] = '\0';
-                const CURLcode performResult = curl_easy_perform(m_curl);
+                const CURLcode performResult = PerformWithCancellation();
                 if (performResult != CURLE_OK)
                 {
                     // Retain the default status code of 0 to indicate a client side error,

Tests/UrlRequestErrorReporting.cpp

@@ -17,6 +17,8 @@
 #include <memory>
 #include <regex>
 #include <string>
+#include <thread>
+#include <vector>
 
 #if defined(_WIN32)
 #include <winsock2.h>
@@ -50,6 +52,13 @@ namespace
         ::closesocket(socket);
     }
 
+    // Wakes a thread blocked in accept()/recv() on this socket. Closing the socket alone does not
+    // reliably interrupt a blocking call in another thread.
+    void ShutdownSocket(NativeSocket socket)
+    {
+        ::shutdown(socket, SD_BOTH);
+    }
+
     bool EnsureSocketsInitialized()
     {
         static const bool initialized = [] {
@@ -73,6 +82,13 @@ namespace
         ::close(socket);
     }
 
+    // Wakes a thread blocked in accept()/recv() on this socket. On Linux, close() in another thread
+    // does NOT interrupt a blocking accept(); shutdown() does.
+    void ShutdownSocket(NativeSocket socket)
+    {
+        ::shutdown(socket, SHUT_RDWR);
+    }
+
     bool EnsureSocketsInitialized()
     {
         return true;
@@ -186,6 +202,96 @@ namespace
         uint16_t m_port;
     };
 
+    // A loopback TCP server that accepts connections but never responds, so an HTTP request to it
+    // hangs until it is aborted. A background thread accept()s connections and holds them open
+    // until teardown. Used to verify that UrlRequest::Abort() interrupts an in-flight request
+    // rather than waiting for the transport's own timeout. Non-movable: the accept thread captures
+    // `this`.
+    class HangingServer
+    {
+    public:
+        HangingServer()
+        {
+            if (!EnsureSocketsInitialized())
+            {
+                return;
+            }
+
+            NativeSocket listener = ::socket(AF_INET, SOCK_STREAM, 0);
+            if (listener == InvalidSocket)
+            {
+                return;
+            }
+
+            sockaddr_in address{};
+            address.sin_family = AF_INET;
+            address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+            address.sin_port = 0;
+            SocketLength addressLength = static_cast<SocketLength>(sizeof(address));
+            if (::bind(listener, reinterpret_cast<const sockaddr*>(&address), sizeof(address)) != 0 ||
+                ::getsockname(listener, reinterpret_cast<sockaddr*>(&address), &addressLength) != 0 ||
+                ::listen(listener, 8) != 0)
+            {
+                CloseSocket(listener);
+                return;
+            }
+
+            m_listener = listener;
+            m_port = ntohs(address.sin_port);
+            m_acceptThread = std::thread{[this]() {
+                for (;;)
+                {
+                    NativeSocket connection = ::accept(m_listener, nullptr, nullptr);
+                    if (connection == InvalidSocket)
+                    {
+                        break; // listener closed during teardown
+                    }
+                    m_accepted.push_back(connection); // hold open, never respond
+                }
+            }};
+        }
+
+        HangingServer(const HangingServer&) = delete;
+        HangingServer& operator=(const HangingServer&) = delete;
+        HangingServer(HangingServer&&) = delete;
+        HangingServer& operator=(HangingServer&&) = delete;
+
+        ~HangingServer()
+        {
+            if (m_listener != InvalidSocket)
+            {
+                // shutdown() (not just close()) so a thread blocked in accept() is woken: on Linux
+                // close() in another thread does not interrupt the blocking accept().
+                ShutdownSocket(m_listener);
+                CloseSocket(m_listener);
+            }
+            if (m_acceptThread.joinable())
+            {
+                m_acceptThread.join();
+            }
+            for (NativeSocket connection : m_accepted)
+            {
+                CloseSocket(connection);
+            }
+        }
+
+        bool Valid() const
+        {
+            return m_listener != InvalidSocket;
+        }
+
+        std::string Url() const
+        {
+            return "http://127.0.0.1:" + std::to_string(m_port) + "/";
+        }
+
+    private:
+        NativeSocket m_listener{InvalidSocket};
+        uint16_t m_port{0};
+        std::vector<NativeSocket> m_accepted{};
+        std::thread m_acceptThread{};
+    };
+
     // RAII temp file with a per-process-unique name, so parallel test runs sharing a temp
     // directory don't collide and no artifacts outlive the test. Pass nullptr contents
     // for a path that is guaranteed not to exist.
@@ -352,6 +458,50 @@ TEST(UrlRequestErrorReporting, ReopenClearsPriorError)
     EXPECT_TRUE(request.ErrorString().empty()) << request.ErrorString();
 }
 
+TEST(UrlRequestErrorReporting, AbortInterruptsInFlightRequest)
+{
+    // The Windows backend already observes Abort() (its WinRT continuations are guarded by
+    // m_cancellationSource), but it does not populate the transport-error accessors, so the
+    // symbol assertions below are gated to the backends that do.
+    SKIP_WITHOUT_TRANSPORT_ERROR_DETAIL();
+
+    HangingServer server{};
+    ASSERT_TRUE(server.Valid());
+
+    UrlLib::UrlRequest request{};
+    request.Open(UrlLib::UrlMethod::Get, server.Url());
+
+    auto done = std::make_shared<std::promise<void>>();
+    auto future = done->get_future();
+    request.SendAsync().then(arcana::inline_scheduler, arcana::cancellation::none(),
+        [done](const arcana::expected<void, std::exception_ptr>&) {
+            done->set_value();
+        });
+
+    // Let the request connect to the hanging server and start waiting for a response that never
+    // comes, then abort. Without Abort() being observed on this backend the request would block
+    // until the transport's own timeout.
+    std::this_thread::sleep_for(std::chrono::milliseconds{250});
+    request.Abort();
+
+    ASSERT_EQ(future.wait_for(std::chrono::seconds{15}), std::future_status::ready)
+        << "Abort did not interrupt the in-flight request";
+
+    EXPECT_EQ(request.StatusCode(), UrlLib::UrlStatusCode::None);
+    EXPECT_FALSE(request.ErrorString().empty());
+#if defined(__APPLE__)
+    EXPECT_EQ(request.ErrorSymbol(), "NSURLErrorCancelled") << request.ErrorString();
+    EXPECT_EQ(request.ErrorCode(), -999) << request.ErrorString();
+#else
+    // The guarantee under test is that Abort() interrupts the request promptly (the bounded wait
+    // above) and records a transport error. The exact CURLcode depends on libcurl internals/timing:
+    // the progress callback returning the abort yields CURLE_ABORTED_BY_CALLBACK, but once the
+    // socket is shut down libcurl may instead surface a recv/gotnothing error -- so assert a curl
+    // transport error rather than pinning one symbol.
+    EXPECT_EQ(request.ErrorString().substr(0, 5), "curl:") << request.ErrorString();
+#endif
+}
+
 #if defined(__APPLE__)
 TEST(UrlRequestErrorReporting, MissingAppResourceReportsError)
 {