From 60e984f0cfdbacaf9918750f3832bfc456ba98ea Mon Sep 17 00:00:00 2001
From: eldritch horrors <pennae@lix.systems>
Date: Sat, 9 Nov 2024 01:17:28 +0100
Subject: [PATCH] libstore: allow explicit cancellation of transfers

we only wait for the worker thread to remove the handle from its multi
and then signal us back. after that's done the transfer can be cleaned
up independently of the multi handle and no transfer progress is made.

we also no longer remove transfer items from multi handles in the item
destructor; only the worker thread can add a transfer and as such only
the worker thread should be responsible for removing them again later.

Change-Id: I6ff57bb5e7c1e91faf8657b257e73d6a72aa928e
---
 src/libstore/filetransfer.cc | 36 ++++++++++++++++++++++++++----------
 1 file changed, 26 insertions(+), 10 deletions(-)

diff --git a/src/libstore/filetransfer.cc b/src/libstore/filetransfer.cc
index 0abf6ccb3..352feb720 100644
--- a/src/libstore/filetransfer.cc
+++ b/src/libstore/filetransfer.cc
@@ -130,7 +130,6 @@ struct curlFileTransfer : public FileTransfer
 
         ~TransferItem()
         {
-            curl_multi_remove_handle(fileTransfer.curlm, req);
             curl_easy_cleanup(req);
             if (requestHeaders) curl_slist_free_all(requestHeaders);
             try {
@@ -517,6 +516,21 @@ struct curlFileTransfer : public FileTransfer
             lock->unpause.push_back(shared_from_this());
             fileTransfer.wakeup();
         }
+
+        void cancel()
+        {
+            std::promise<void> promise;
+            auto wait = promise.get_future();
+            {
+                auto lock = fileTransfer.state_.lock();
+                if (lock->quit) {
+                    return;
+                }
+                lock->cancel[shared_from_this()] = std::move(promise);
+            }
+            fileTransfer.wakeup();
+            wait.get();
+        }
     };
 
     struct State
@@ -529,6 +543,7 @@ struct curlFileTransfer : public FileTransfer
         bool quit = false;
         std::priority_queue<std::shared_ptr<TransferItem>, std::vector<std::shared_ptr<TransferItem>>, EmbargoComparator> incoming;
         std::vector<std::shared_ptr<TransferItem>> unpause;
+        std::map<std::shared_ptr<TransferItem>, std::promise<void>> cancel;
     };
 
     Sync<State> state_;
@@ -605,6 +620,14 @@ struct curlFileTransfer : public FileTransfer
         while (!quit) {
             checkInterrupt();
 
+            {
+                auto cancel = [&] { return std::move(state_.lock()->cancel); }();
+                for (auto & [item, promise] : cancel) {
+                    curl_multi_remove_handle(curlm, item->req);
+                    promise.set_value();
+                }
+            }
+
             /* Let curl do its thing. */
             int running;
             CURLMcode mc = curl_multi_perform(curlm, &running);
@@ -826,7 +849,7 @@ struct curlFileTransfer : public FileTransfer
         }
 
         struct State {
-            bool done = false, failed = false;
+            bool done = false;
             std::exception_ptr exc;
             std::string data;
             std::condition_variable avail;
@@ -848,11 +871,6 @@ struct curlFileTransfer : public FileTransfer
             [_state](std::string_view data) {
                 auto state(_state->lock());
 
-                if (state->failed) {
-                    // actual exception doesn't matter, the other end is already dead
-                    throw std::exception{};
-                }
-
                 /* If the buffer is full, then go to sleep until the calling
                 thread wakes us up (i.e. when it has removed data from the
                 buffer). We don't wait forever to prevent stalling the
@@ -890,10 +908,8 @@ struct curlFileTransfer : public FileTransfer
             ~TransferSource()
             {
                 // wake up the download thread if it's still going and have it abort
-                auto state(_state->lock());
-                state->failed |= !state->done;
                 try {
-                    transfer->unpause();
+                    transfer->cancel();
                 } catch (...) {
                     ignoreExceptionInDestructor();
                 }