From 18efc848fe7b79c84a2e4311ac9ce3492b7aaa82 Mon Sep 17 00:00:00 2001
From: eldritch horrors <pennae@lix.systems>
Date: Fri, 24 Oct 2025 00:16:15 +0200
Subject: [PATCH] libstore: move curl-multi wrapper into own class

the wrapper is needed by transfer streams to restart a failed transfer
if desired. curlFileTransfer itself is more of a fancy handler for the
thread we're dedicating to curl io handling. the thread will stay with
the multi handle for now because quit handling needs to stay there. we
could have CurlMulti keep only a flag, but that does not help us much.

Change-Id: I99550f0bbb635b75898ca7260f08275df86050e3
---
 lix/libstore/filetransfer.cc        | 41 ++++++++++++++++++++---------
 tests/unit/libstore/filetransfer.cc | 23 ++++++++++++++++
 2 files changed, 52 insertions(+), 12 deletions(-)

diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc
index d1268d98a..1b3e036a5 100644
--- a/lix/libstore/filetransfer.cc
+++ b/lix/libstore/filetransfer.cc
@@ -22,6 +22,7 @@
 #include <kj/async.h>
 #include <kj/encoding.h>
 #include <kj/time.h>
+#include <memory>
 
 #if ENABLE_DTRACE
 #include "trace-probes.gen.hh"
@@ -553,7 +554,7 @@ struct TransferItem
     }
 };
 
-struct curlFileTransfer : public FileTransfer
+struct CurlMulti
 {
     std::unique_ptr<CURLM, decltype([](auto * m) { curl_multi_cleanup(m); })> curlm;
 
@@ -593,12 +594,12 @@ struct curlFileTransfer : public FileTransfer
 
     std::thread workerThread;
 
-    curlFileTransfer(unsigned int baseRetryTimeMs)
+    CurlMulti(unsigned int baseRetryTimeMs)
         : curlm(curl_multi_init())
         , baseRetryTimeMs(baseRetryTimeMs)
     {
         if (curlm == nullptr) {
-            throw FileTransferError(Misc, {}, "could not allocate curl handle");
+            throw FileTransferError(FileTransfer::Misc, {}, "could not allocate curl handle");
         }
 
         static std::once_flag globalInit;
@@ -614,7 +615,7 @@ struct curlFileTransfer : public FileTransfer
         });
     }
 
-    ~curlFileTransfer()
+    ~CurlMulti()
     {
         try {
             stopWorkerThread();
@@ -779,6 +780,21 @@ struct curlFileTransfer : public FileTransfer
         }
         wakeup();
     }
+};
+
+struct curlFileTransfer : public FileTransfer
+{
+    std::shared_ptr<CurlMulti> multi;
+
+    curlFileTransfer(unsigned int baseRetryTimeMs)
+        : multi(std::make_shared<CurlMulti>(baseRetryTimeMs))
+    {
+    }
+
+    ~curlFileTransfer()
+    {
+        multi->stopWorkerThread();
+    }
 
 #if ENABLE_S3
     static std::tuple<std::string, std::string, StoreConfig::Params> parseS3Uri(std::string uri)
@@ -952,7 +968,7 @@ struct curlFileTransfer : public FileTransfer
 
     struct TransferStream : AsyncInputStream
     {
-        curlFileTransfer & parent;
+        std::shared_ptr<CurlMulti> parent;
         std::string uri;
         FileTransferOptions options;
         std::optional<std::string> data;
@@ -977,7 +993,7 @@ struct curlFileTransfer : public FileTransfer
             bool noBody,
             const Activity * context
         )
-            : parent(parent)
+            : parent(parent.multi)
             , uri(uri)
             , options(options)
             , data(std::move(data))
@@ -987,7 +1003,7 @@ struct curlFileTransfer : public FileTransfer
                   fileTransferSettings.tries,
                   std::chrono::seconds(fileTransferSettings.maxConnectTimeout.get()),
                   std::chrono::seconds(fileTransferSettings.initialConnectTimeout.get()),
-                  std::chrono::milliseconds(parent.baseRetryTimeMs)
+                  std::chrono::milliseconds(this->parent->baseRetryTimeMs)
               ))
         {
         }
@@ -997,7 +1013,7 @@ struct curlFileTransfer : public FileTransfer
             // wake up the download thread if it's still going and have it abort
             try {
                 if (transfer) {
-                    parent.cancel(transfer);
+                    parent->cancel(transfer);
                 }
             } catch (...) {
                 ignoreExceptionInDestructor();
@@ -1069,7 +1085,7 @@ struct curlFileTransfer : public FileTransfer
                 std::move(pfp.fulfiller),
                 timeout
             );
-            parent.enqueueItem(transfer);
+            parent->enqueueItem(transfer);
             co_return TRY_AWAIT(pfp.promise);
         } catch (...) {
             co_return result::current_exception();
@@ -1152,13 +1168,13 @@ struct curlFileTransfer : public FileTransfer
                     chunk = std::move(state->data);
                     buffered = chunk;
                     totalReceived += chunk.size();
-                    parent.unpause(transfer);
+                    parent->unpause(transfer);
                 } else if (state->exc) {
                     std::rethrow_exception(state->exc);
                 } else if (state->done) {
                     co_return false;
                 } else {
-                    parent.unpause(transfer);
+                    parent->unpause(transfer);
                     signal = state->wait();
                 }
             }
@@ -1246,8 +1262,9 @@ ref<FileTransfer> getFileTransfer()
 {
     static ref<curlFileTransfer> fileTransfer = makeCurlFileTransfer({});
 
-    if (fileTransfer->state_.lock()->quit)
+    if (fileTransfer->multi->state_.lock()->quit) {
         fileTransfer = makeCurlFileTransfer({});
+    }
 
     return fileTransfer;
 }
diff --git a/tests/unit/libstore/filetransfer.cc b/tests/unit/libstore/filetransfer.cc
index 7d4f227a8..92f47a1f1 100644
--- a/tests/unit/libstore/filetransfer.cc
+++ b/tests/unit/libstore/filetransfer.cc
@@ -470,4 +470,27 @@ TEST(FileTransfer, setupErrorsAreMetadata)
     ASSERT_THROW(aio.blockOn(ft->upload(fmt("http://[::1]:%d", port), "")), FileTransferError);
 }
 
+TEST(FileTransfer, shutdownKillsTransfers)
+{
+    auto [port, srv] = serveHTTP({
+        {"200 ok", "content-length: 999999999\r\n", [&](int) { return std::string(1024, 'X'); }},
+    });
+    AsyncIoRoot aio;
+    char buf;
+    std::optional<box_ptr<AsyncInputStream>> s;
+    {
+        auto ft = makeFileTransfer(0);
+        auto [_r, stream] = aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port)));
+        ASSERT_EQ(aio.blockOn(stream->read(&buf, 1)), 1);
+        s = std::move(stream);
+    }
+    ASSERT_THROW(
+        {
+            while (true) {
+                aio.blockOn((*s)->drain());
+            }
+        },
+        FileTransferError
+    );
+}
 }